Sending Sent messages from Celery background tasks

This guide shows you how to move Sent message sending into Celery: install the Python SDK, share one client per worker process, add a send task with retry handling, process webhook events off the request path, and verify the task in sandbox mode.

Prerequisites

This guide assumes a working Celery setup (broker running, worker starting) inside a Python web project. You also need:

A Sent API key from the API Keys page in your Sent Dashboard
A web framework endpoint for webhooks. The Flask, Django, and FastAPI guides each build one; Celery workers do not receive HTTP requests themselves

Install the SDK

Add the sentdm package to the environment your workers run in:

pip install sentdm

Configure a worker-safe client

Export the API key where the worker can read it:

export SENT_DM_API_KEY="your-api-key"

Create the client lazily inside a base task class, so each worker process builds exactly one client after forking instead of sharing one across processes:

# celery_app/base_task.py
import os
from celery import Task

class SentTask(Task):
    abstract = True
    max_retries = 3
    default_retry_delay = 60
    retry_backoff = True          # 60s, 120s, 240s, ... between retries
    retry_jitter = True

    _sent_client = None

    @property
    def sent_client(self):
        if self._sent_client is None:
            from sent_dm import Sent
            api_key = os.getenv("SENT_DM_API_KEY")
            if not api_key:
                raise RuntimeError("SENT_DM_API_KEY not configured")
            self._sent_client = Sent(api_key=api_key)
        return self._sent_client

Add a send task

Write the task that calls messages.send. Retry on transient failures, but never on validation errors (a request that was invalid once is invalid every time):

# tasks/messages.py
import logging
from celery import shared_task
from sent_dm import APIStatusError, APIConnectionError, RateLimitError
from celery_app.base_task import SentTask

logger = logging.getLogger(__name__)

@shared_task(base=SentTask, bind=True, name="tasks.messages.send_message", queue="messages")
def send_message(self, phone_number: str, template_name: str,
                 parameters: dict | None = None, channels: list | None = None,
                 sandbox: bool = False) -> dict:
    try:
        response = self.sent_client.messages.send(
            to=[phone_number],                  # E.164 format, for example +14155551234
            template={
                "name": template_name,          # reference by "name" or "id", never both
                "parameters": parameters or {},
            },
            channel=channels,                   # omit to let Sent pick per recipient
            sandbox=sandbox,                    # True = validate and simulate only
        )
    except RateLimitError as exc:
        # Back off and retry when the API rate limit is hit
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))
    except APIConnectionError as exc:
        # Network problems are transient: retry with backoff
        raise self.retry(exc=exc)
    except APIStatusError as exc:
        if exc.status_code and exc.status_code < 500:
            logger.error("Non-retryable send failure (%s): %s", exc.status_code, exc)
            raise  # 4xx: the request itself is wrong; retrying cannot fix it
        raise self.retry(exc=exc)

    recipient = response.data.recipients[0]
    logger.info("Message queued: %s", recipient.message_id)
    return {"message_id": recipient.message_id, "status": response.data.status}

Call it from your web code with .delay(...). The request returns immediately and the worker does the sending:

send_message.delay("+14155551234", "welcome", {"name": "Ada"})

Sent accepts sends asynchronously: the API responds with status QUEUED and one message_id per recipient-and-channel pair. Store the message_id from the task result, since delivery outcomes arrive as webhooks rather than in this response.

Process webhook events off the request path

Keep your webhook endpoint fast by acknowledging immediately and handing the verified event envelope (field, event, timestamp, payload) to a task. Signature verification must stay in the web endpoint, where the raw request bytes are available. See webhook signature verification:

# tasks/webhooks.py
import logging
from celery import shared_task

logger = logging.getLogger(__name__)

@shared_task(name="tasks.webhooks.process_event", queue="webhooks")
def process_event(event: dict) -> None:
    payload = event.get("payload", {})
    if event.get("field") != "message":
        return

    match event.get("event"):
        case "message.delivered":
            logger.info("Message %s delivered", payload.get("message_id"))
        case "message.failed":
            logger.error("Message %s failed (status %s)",
                         payload.get("message_id"), payload.get("message_status"))
        case "message.received":
            logger.info("Inbound %s from %s: %s", payload.get("channel"),
                        payload.get("inbound_number"), payload.get("text"))
        case _:
            logger.info("Message %s status: %s",
                        payload.get("message_id"), payload.get("message_status"))

In the webhook endpoint from your framework guide, replace the inline processing with one line after verification succeeds:

process_event.delay(event)

Verify the task

Start a worker on the messages queue:

celery -A celery_app.app worker --queues=messages --loglevel=info

Trigger a sandbox send from a shell. Full validation runs, but nothing is delivered and no credits are consumed:

# python manage.py shell, flask shell, or plain python
from tasks.messages import send_message

result = send_message.delay("+14155551234", "welcome", {"name": "Ada"}, sandbox=True)
print(result.get(timeout=30))

The result should contain a message_id and "status": "QUEUED", and the worker log should show a Message queued: ... line. A BadRequestError in the worker log means the request shape is wrong: sandbox requests return real validation errors. If the task never runs, confirm the worker is consuming the messages queue named in the task decorator.

Adapt this to your app

If you send campaigns to large lists, fan out one task per recipient rather than looping inside a single task. The appendix below shows the batch pattern.
If you schedule recurring sends, drive them with Celery beat and call the same send_message task from the schedule.
If your workers hit API rate limits under load, lower worker concurrency for the messages queue or set a Celery rate_limit on the task before adding custom throttling.
To send free-form text instead of a template, pass text instead of template, since each send carries exactly one of the two.

Appendix: production scaffolding

The numbered steps stay on the core messaging tasks. The blocks below are optional scaffolding for a production Celery deployment. Adapt them to your own conventions rather than adopting them wholesale.

Next steps

Build the webhook endpoint in your web framework: Flask, Django, or FastAPI
Review the webhook event types reference for every payload field
Explore the Python SDK reference for error types and client options
Read the SDK best practices guide for production deployments

Queue routing and worker tuning

Batch fan-out task

Docker Compose for worker plus broker

On this page