Sending messages from FastAPI with the Sent Python SDK

This guide shows you how to wire Sent messaging into an existing FastAPI app: install the Python SDK, manage an async client with the app lifespan, send a template message from a route, receive delivery webhooks, and verify the whole loop in sandbox mode.

Prerequisites

This guide assumes a working FastAPI app and familiarity with dependencies and Pydantic models. You also need:

A Sent API key from the API Keys page in your Sent Dashboard
A public HTTPS URL for webhook delivery. For local work, open a tunnel as described in the webhook local development guide

Install the SDK

Add the sentdm package to your existing environment:

pip install sentdm

Configure the client

Set your credentials as environment variables so they stay out of code; the webhook secret arrives in step 4:

export SENT_DM_API_KEY="your-api-key"
export SENT_DM_WEBHOOK_SECRET="whsec_your_signing_secret"

Open one AsyncSent client at startup and close it at shutdown, so every request shares the same connection pool; AsyncSent() reads SENT_DM_API_KEY by default:

# app/sent_client.py
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI
from sent_dm import AsyncSent

class SentClientManager:
    def __init__(self) -> None:
        self._client: AsyncSent | None = None

    async def startup(self) -> None:
        self._client = AsyncSent()
        await self._client.__aenter__()

    async def shutdown(self) -> None:
        if self._client:
            await self._client.__aexit__(None, None, None)
            self._client = None

    @property
    def client(self) -> AsyncSent:
        if self._client is None:
            raise RuntimeError("Sent client not initialized")
        return self._client

sent_manager = SentClientManager()

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    await sent_manager.startup()
    yield
    await sent_manager.shutdown()

Attach the lifespan and expose the client as a dependency:

# app/main.py (excerpt)
from fastapi import FastAPI
from sent_dm import AsyncSent
from app.sent_client import lifespan, sent_manager

app = FastAPI(lifespan=lifespan)

def get_sent_client() -> AsyncSent:
    return sent_manager.client

Send a template message from a route

Add a validated route that calls messages.send; the pass-through sandbox flag lets callers exercise the route without delivering anything:

# app/routers/messages.py
from fastapi import APIRouter, Depends, status
from pydantic import BaseModel, Field
from sent_dm import AsyncSent
from app.main import get_sent_client

router = APIRouter(prefix="/api/messages", tags=["Messages"])

class SendMessageRequest(BaseModel):
    phone_number: str = Field(pattern=r"^\+[1-9]\d{1,14}$")   # E.164 format
    template_name: str
    parameters: dict[str, str] = Field(default_factory=dict)
    channels: list[str] | None = None      # omit to let Sent pick per recipient
    sandbox: bool = False                  # True = validate and simulate only

@router.post("/send", status_code=status.HTTP_202_ACCEPTED)
async def send_message(
    request: SendMessageRequest,
    client: AsyncSent = Depends(get_sent_client),
) -> dict[str, str]:
    response = await client.messages.send(
        to=[request.phone_number],
        template={
            "name": request.template_name,  # reference by "name" or "id", never both
            "parameters": request.parameters,
        },
        channel=request.channels,
        sandbox=request.sandbox,
    )
    recipient = response.data.recipients[0]
    return {"message_id": recipient.message_id, "status": response.data.status}

Sent accepts sends asynchronously: the API responds with status QUEUED and one message_id per recipient-and-channel pair. Store the message_id. Delivery outcomes arrive on your webhook endpoint instead of in this response.

Receive delivery webhooks

Add a dependency that verifies the X-Webhook-Signature header against the raw request body before your handler trusts any event. The scheme is HMAC-SHA256 over {X-Webhook-ID}.{X-Webhook-Timestamp}.{rawBody}, keyed with the base64-decoded secret after stripping its whsec_ prefix. For the full scheme, refer to webhook signature verification:

# app/dependencies.py
import base64
import hashlib
import hmac
import os
import time
from fastapi import Request, Header, HTTPException, status

async def verify_webhook_signature(
    request: Request,
    x_webhook_signature: str | None = Header(None),
    x_webhook_id: str = Header(""),
    x_webhook_timestamp: str = Header(""),
) -> bytes:
    if not x_webhook_signature:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing X-Webhook-Signature header")

    webhook_secret = os.environ.get("SENT_DM_WEBHOOK_SECRET")
    if not webhook_secret:
        raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, "Webhook secret not configured")

    payload = await request.body()
    # Strip the "whsec_" prefix and base64-decode to get the raw HMAC key
    key_bytes = base64.b64decode(webhook_secret.removeprefix("whsec_"))
    # Signed content = "{webhookId}.{timestamp}.{rawBody}"; signature format = "v1,{base64(hmac)}"
    signed = f"{x_webhook_id}.{x_webhook_timestamp}.{payload.decode('utf-8')}"
    digest = hmac.new(key_bytes, signed.encode("utf-8"), hashlib.sha256).digest()
    expected = "v1," + base64.b64encode(digest).decode()

    if not hmac.compare_digest(x_webhook_signature, expected):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid webhook signature")

    # Reject replayed events older than 5 minutes
    try:
        if abs(time.time() - int(x_webhook_timestamp)) > 300:
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Timestamp outside tolerance")
    except ValueError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid timestamp") from None

    return payload

Add the endpoint itself. Every event arrives in the same envelope (field, event, timestamp, payload), so one handler routes all of them; BackgroundTasks returns 200 immediately and processes the event after the response:

# app/routers/webhooks.py
import json
import logging
from typing import Any
from fastapi import APIRouter, BackgroundTasks, Depends, status
from app.dependencies import verify_webhook_signature

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/webhooks", tags=["Webhooks"])

async def process_webhook_event(event: dict[str, Any]) -> None:
    payload = event.get("payload", {})
    if event.get("field") != "message":
        logger.info("Unhandled webhook field: %s", event.get("field"))
        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"))

@router.post("/sent", status_code=status.HTTP_200_OK)
async def handle_webhook(
    background_tasks: BackgroundTasks,
    payload: bytes = Depends(verify_webhook_signature),
) -> dict[str, bool]:
    background_tasks.add_task(process_webhook_event, json.loads(payload))
    return {"received": True}

Include both routers in your app, then tell Sent where to deliver events. If you prefer a UI, use the webhooks getting started guide; otherwise register over the API:

curl -X POST https://api.sent.dm/v3/webhooks \
  -H "x-api-key: $SENT_DM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "display_name": "FastAPI integration",
    "endpoint_url": "https://your-domain.example/webhooks/sent",
    "event_types": ["message"]
  }'

Copy two values from the response: the webhook id (used to test delivery in the next step) and signing_secret. Set SENT_DM_WEBHOOK_SECRET to that value. The message event type covers every message event; the webhook event types reference lists all payload fields.

Verify the integration

Start the app with your credentials loaded:

uvicorn app.main:app --reload

Send a sandbox message through your new route, which runs full validation but delivers nothing and consumes no credits:

curl -X POST http://localhost:8000/api/messages/send \
  -H "Content-Type: application/json" \
  -d '{"phone_number": "+14155551234", "template_name": "welcome",
       "parameters": {"name": "Ada"}, "sandbox": true}'

The response should contain a message_id and "status": "QUEUED". A 422 or 400 here means the request shape is wrong: sandbox requests return real validation errors.

Now confirm webhook delivery end to end. Ask Sent to deliver a signed test event, replacing the ID with the webhook id you copied:

curl -X POST https://api.sent.dm/v3/webhooks/YOUR_WEBHOOK_ID/test \
  -H "x-api-key: $SENT_DM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"event_type": "message.delivered"}'

Your application log should show a Message ... delivered line, and the endpoint should have answered 200 {"received": true}. Test events travel the same signed delivery pipeline as real events, so a 401 in your log means the signing secret or verification code is wrong. Sent attempts a test event exactly once, so re-run the command after each fix.

Adapt this to your app

If you already manage settings with pydantic-settings, read SENT_DM_API_KEY and SENT_DM_WEBHOOK_SECRET there and pass them explicitly. The appendix below shows the pattern.
If a webhook event triggers slow work beyond a BackgroundTasks call (fan-out, long database writes), hand it to a task queue instead; see handling webhook retries.
To send to many recipients, pass them all in to; Sent creates one message per recipient-and-channel pair in a single call.
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 FastAPI stack. Adapt them to your own conventions rather than adopting them wholesale.

Next steps

Review the webhook event types reference for every payload field
Work through the webhook production checklist before going live
Explore the Python SDK reference for retries, timeouts, and error types
Read the SDK best practices guide for production deployments

Typed settings with pydantic-settings

Rate limiting with SlowAPI

Batch sends with per-recipient error capture

On this page