textbee Logotextbee.dev
Plans from $9.99/mo.View Plans
Send SMS in Django and Flask: textbee Integration Guide
python
django
flask
tutorial
sms
api
celery

Send SMS in Django and Flask: textbee Integration Guide

Send SMS from Django and Flask applications using textbee and your Android phone as the gateway. Service class, Celery async tasks, Django signals, Flask blueprints: zero per-message fees.

TT

textbee team

8 min read
Share

TL;DR

  • A reusable SmsService class with requests is the foundation for both Django and Flask.
  • Never block the request thread with SMS sends. Always dispatch to a background task (Celery, Django-Q, or a thread).
  • Django: service class + Celery task + optional signal hook. Flask: blueprint + background thread or Celery.
  • Store credentials in environment variables, load them via Django's settings or Flask's app.config.
  • No per-message fee: textbee sends SMS through your Android SIM, not a CPaaS billing meter.

The standalone Python guide covers sending SMS from a script. This guide covers what changes when you're inside a web framework: service injection patterns, async task dispatch, framework-idiomatic configuration, and signal-driven sends.

Prerequisites

  • textbee installed on your Android device: download and quickstart
  • Device ID and API key from textbee.dev
  • Python 3.10+
  • requests library: pip install requests
  • For async: celery + a broker (Redis recommended): pip install celery redis

The core service class (shared by Django and Flask)

Python
# sms/service.py  (or anywhere in your project)
import os
import requests

class SmsService:
    BASE_URL = "https://api.textbee.dev/api/v1/gateway/devices/{device_id}/send-sms"

    def __init__(self, device_id: str, api_key: str):
        self.url     = self.BASE_URL.format(device_id=device_id)
        self.api_key = api_key

    def send(self, recipients: str | list[str], message: str) -> dict:
        response = requests.post(
            self.url,
            headers={"x-api-key": self.api_key, "Content-Type": "application/json"},
            json={"recipients": [recipients] if isinstance(recipients, str) else recipients,
                  "message": message},
            timeout=10,
        )
        response.raise_for_status()
        return response.json()["data"]

This class is framework-agnostic. The sections below show how to wire it into Django and Flask respectively.


Django integration

Configuration

Add credentials to your .env (and your secrets manager in production):

TEXTBEE_DEVICE_ID=your-device-id
TEXTBEE_API_KEY=your-api-key

Reference them in settings.py:

Python
import os

TEXTBEE_DEVICE_ID = os.environ["TEXTBEE_DEVICE_ID"]
TEXTBEE_API_KEY   = os.environ["TEXTBEE_API_KEY"]

Service factory (singleton pattern)

Create a factory function so the service is instantiated once and reused:

Python
# sms/service.py
from django.conf import settings

_sms_service = None

def get_sms_service() -> SmsService:
    global _sms_service
    if _sms_service is None:
        _sms_service = SmsService(
            device_id=settings.TEXTBEE_DEVICE_ID,
            api_key=settings.TEXTBEE_API_KEY,
        )
    return _sms_service

Use it in views:

Python
from sms.service import get_sms_service

class AppointmentConfirmView(View):
    def post(self, request, pk):
        appointment = get_object_or_404(Appointment, pk=pk)
        get_sms_service().send(
            appointment.client.phone,
            f"Hi {appointment.client.first_name}, your appointment on {appointment.date} is confirmed.",
        )
        return JsonResponse({"status": "confirmed"})

Don't do this in production: the requests.post call blocks the view thread and adds 1-3 seconds to response time. Use a Celery task instead (see below).

Celery async task

Python
# sms/tasks.py
from celery import shared_task
from sms.service import get_sms_service
import logging

logger = logging.getLogger(__name__)

@shared_task(bind=True, max_retries=3, default_retry_delay=30)
def send_sms_task(self, recipients, message):
    try:
        result = get_sms_service().send(recipients, message)
        logger.info("SMS sent: batch=%s recipients=%d", result["smsBatchId"], result["recipientCount"])
        return result
    except Exception as exc:
        logger.warning("SMS send failed (attempt %d): %s", self.request.retries + 1, exc)
        raise self.retry(exc=exc)

Dispatch from a view. The response returns immediately:

Python
from sms.tasks import send_sms_task

class AppointmentConfirmView(View):
    def post(self, request, pk):
        appointment = get_object_or_404(Appointment, pk=pk)
        send_sms_task.delay(
            appointment.client.phone,
            f"Hi {appointment.client.first_name}, your appointment on {appointment.date} is confirmed.",
        )
        return JsonResponse({"status": "confirmed"})

Start a Celery worker:

Shell
celery -A your_project worker --loglevel=info -Q sms

Scheduled SMS with Celery Beat

For time-based sends (e.g., day-before appointment reminders):

Python
# sms/tasks.py
from celery import shared_task
from django.utils import timezone
from appointments.models import Appointment

@shared_task
def send_appointment_reminders():
    tomorrow = timezone.now().date() + timezone.timedelta(days=1)
    appointments = Appointment.objects.filter(
        date=tomorrow,
        reminder_sent=False,
    ).select_related("client")

    for appt in appointments:
        send_sms_task.delay(
            appt.client.phone,
            f"Reminder: your appointment is tomorrow at {appt.time}. Reply YES to confirm.",
        )
        appt.reminder_sent = True

    Appointment.objects.bulk_update(appointments, ["reminder_sent"])

Register in settings.py:

Python
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    "appointment-reminders": {
        "task": "sms.tasks.send_appointment_reminders",
        "schedule": crontab(hour=9, minute=0),  # run at 9am daily
    },
}

Signal-triggered SMS

Send SMS automatically when a model changes (no explicit call needed in the view):

Python
# orders/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order
from sms.tasks import send_sms_task

@receiver(post_save, sender=Order)
def notify_on_ship(sender, instance, created, **kwargs):
    if not created and instance.status == "shipped" and instance.client.phone:
        send_sms_task.delay(
            instance.client.phone,
            f"Your order #{instance.number} has shipped! Track it at: {instance.tracking_url}",
        )

Register the signal in your app's AppConfig.ready():

Python
# orders/apps.py
class OrdersConfig(AppConfig):
    name = "orders"

    def ready(self):
        import orders.signals  # noqa: F401

Django management command for bulk sends

Python
# management/commands/send_bulk_sms.py
from django.core.management.base import BaseCommand
from sms.tasks import send_sms_task
from customers.models import Customer
import time

class Command(BaseCommand):
    help = "Send a bulk SMS to all active customers"

    def add_arguments(self, parser):
        parser.add_argument("message", type=str)

    def handle(self, *args, **options):
        customers = Customer.objects.filter(
            is_active=True,
            phone__isnull=False,
            sms_opted_in=True,
        )
        self.stdout.write(f"Sending to {customers.count()} customers...")

        for customer in customers:
            send_sms_task.delay(customer.phone, options["message"])
            time.sleep(0.1)  # avoid flooding the queue

        self.stdout.write(self.style.SUCCESS("All tasks dispatched."))

Run: python manage.py send_bulk_sms "Our summer sale starts tomorrow. Use code SUMMER20."


Flask integration

Configuration

Python
# config.py
import os

class Config:
    TEXTBEE_DEVICE_ID = os.environ["TEXTBEE_DEVICE_ID"]
    TEXTBEE_API_KEY   = os.environ["TEXTBEE_API_KEY"]
Python
# app/__init__.py
from flask import Flask
from config import Config
from sms.service import SmsService

sms_service: SmsService = None  # set in create_app

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)

    global sms_service
    sms_service = SmsService(
        device_id=app.config["TEXTBEE_DEVICE_ID"],
        api_key=app.config["TEXTBEE_API_KEY"],
    )

    from .routes.appointments import appointments_bp
    app.register_blueprint(appointments_bp)

    return app

Blueprint with background thread send

For simple Flask apps without Celery, use a daemon thread so the response returns immediately:

Python
# routes/appointments.py
import threading
from flask import Blueprint, jsonify, request, current_app
from app import sms_service

appointments_bp = Blueprint("appointments", __name__, url_prefix="/appointments")

def _send_async(app, phone, message):
    with app.app_context():
        try:
            sms_service.send(phone, message)
        except Exception as e:
            app.logger.error("SMS send failed: %s", e)

@appointments_bp.post("/<int:pk>/confirm")
def confirm(pk):
    appointment = get_appointment_or_404(pk)
    thread = threading.Thread(
        target=_send_async,
        args=(
            current_app._get_current_object(),
            appointment.client_phone,
            f"Your appointment on {appointment.date} is confirmed.",
        ),
        daemon=True,
    )
    thread.start()
    return jsonify({"status": "confirmed"})

The app_context() push is required for Flask's application context to be available in the thread.

Flask + Celery async task

For production Flask, Celery is more reliable than raw threads:

Python
# celery_app.py
from celery import Celery

def make_celery(app):
    celery = Celery(app.import_name)
    celery.conf.broker_url  = app.config["CELERY_BROKER_URL"]
    celery.conf.result_backend = app.config["CELERY_RESULT_BACKEND"]

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery
Python
# tasks.py
from celery_app import make_celery
from app import create_app, sms_service

flask_app = create_app()
celery = make_celery(flask_app)

@celery.task(bind=True, max_retries=3)
def send_sms_task(self, recipients, message):
    try:
        return sms_service.send(recipients, message)
    except Exception as exc:
        raise self.retry(exc=exc, countdown=30)

Dispatch from a route:

Python
from tasks import send_sms_task

@appointments_bp.post("/<int:pk>/confirm")
def confirm(pk):
    appointment = get_appointment_or_404(pk)
    send_sms_task.delay(appointment.client_phone, "Your appointment is confirmed.")
    return jsonify({"status": "confirmed"})

Don't block the request: why it matters

Every requests.post() to the textbee API takes 200ms-3s depending on network conditions. In a synchronous Django or Flask view, that time is added directly to your response time.

What happens if you block:

  • A 1.5s SMS send inside a 50ms view = 30× slower response
  • Under any load, workers pile up waiting for SMS responses
  • If the textbee API is slow or times out, your view times out

What happens with async dispatch (Celery/thread):

  • View returns in <5ms. The SMS task runs independently
  • SMS failures don't cause 500 errors in your API
  • You can retry failed sends without re-running the full request

The Celery @shared_task(max_retries=3) setup above handles transient failures automatically.

Frequently asked questions

Can I use textbee with Django Channels?

Yes. Django Channels runs async consumers. In an async context, use asyncio.to_thread to run the blocking requests.post call without blocking the event loop:

Python
import asyncio
from sms.service import get_sms_service

async def notify_user(phone, message):
    await asyncio.to_thread(get_sms_service().send, phone, message)

Or use httpx with AsyncClient instead of requests for a native async HTTP call.

How do I send SMS in a Celery task in Django?

The send_sms_task example above is the recommended pattern: @shared_task with max_retries and default_retry_delay. Call it with .delay(phone, message) anywhere in your Django code. The task runs in the Celery worker process, completely outside the request thread.

Should I use Django signals or explicit task dispatch?

Explicit dispatch (calling send_sms_task.delay() directly in the view) is more readable and easier to test. Signals are useful when you want to decouple SMS sending from the view code: for example, sending an SMS whenever an Order model's status changes, regardless of which part of the codebase made the change. Use signals when the trigger is model-level and decoupling adds clarity; use explicit dispatch when the trigger is request-level.

How do I test SMS sends without actually sending?

In Django tests, mock the requests.post call:

Python
from unittest.mock import patch

@patch("requests.post")
def test_confirms_appointment(self, mock_post):
    mock_post.return_value.ok = True
    mock_post.return_value.json.return_value = {
        "data": {"success": True, "smsBatchId": "test", "recipientCount": 1}
    }
    response = self.client.post(f"/appointments/{self.appt.pk}/confirm/")
    self.assertEqual(response.status_code, 200)
    mock_post.assert_called_once()
    call_kwargs = mock_post.call_args.kwargs
    self.assertIn(self.appt.client.phone, call_kwargs["json"]["recipients"])

For Celery, use CELERY_TASK_ALWAYS_EAGER = True in test settings to run tasks synchronously in tests.

Can I send SMS from a Flask CLI command?

Yes. Flask's @app.cli.command() decorator works well for batch sends:

Python
@app.cli.command("send-alert")
@click.argument("message")
def send_alert(message):
    phones = db.session.query(User.phone).filter(User.sms_opted_in == True).all()
    for (phone,) in phones:
        sms_service.send(phone, message)
        time.sleep(1.5)
    click.echo(f"Sent to {len(phones)} users.")

Run: flask send-alert "Scheduled maintenance tonight at 11pm UTC."

Keep going

Download textbee and send your first Django or Flask SMS in under 10 minutes.