WhatsApp API in Python: send with httpx, receive with FastAPI

What you need before you start

There is no official Python package, and you won't miss it. The D-API WhatsApp API lives at https://api.d-api.cloud, accepts JSON and authenticates through the Authorization header with your raw API key, no Bearer. Any HTTP library works. Before writing code, have ready:

• the API key from the dashboard, in an environment variable such as DAPI_API_KEY;
• the sessionId of a number that is already connected (you connect by scanning a QR code, as in WhatsApp Web);
• a public URL for the webhook. In development, a tunnel such as ngrok or cloudflared exposes your localhost.

The examples use Python 3.10 or later, because of the pipe type syntax in the Pydantic models. On earlier versions, swap it for Optional from the typing module and everything else stays the same.

Send a message with requests

For a script, a scheduled job or synchronous Django, requests does the job. The detail people usually miss: requests does not set a timeout on its own. Without one, a network problem leaves the worker waiting forever.

import os
import requests

API = "https://api.d-api.cloud"
HEADERS = {"Authorization": os.environ["DAPI_API_KEY"]}

resp = requests.post(
    f"{API}/api/v1/messages/send/text",
    json={"sessionId": "billing", "to": "14155550123", "text": "Your invoice is due tomorrow."},
    headers=HEADERS,
    timeout=(3, 10),  # 3s to connect, 10s to read the response
)
if not resp.ok:
    error = resp.json() if "application/json" in resp.headers.get("content-type", "") else {}
    raise RuntimeError(f"D-API {resp.status_code}: {error.get('error', resp.text)}")

A reusable client with httpx

In async code, and in any application that sends often, it pays to have a single client with a connection pool. httpx offers the same API in sync and async flavors, and takes base_url, default headers and timeout configured once. The function below turns the API error body, shaped like {"success": false, "error": "...", "statusCode": 400}, into your own exception.

import os
import httpx

class DApiError(Exception):
    def __init__(self, status: int, message: str):
        super().__init__(f"{status}: {message}")
        self.status = status

dapi = httpx.AsyncClient(
    base_url="https://api.d-api.cloud",
    headers={"Authorization": os.environ["DAPI_API_KEY"]},
    timeout=httpx.Timeout(10.0, connect=3.0),
)

async def send(path: str, payload: dict) -> dict:
    try:
        r = await dapi.post(path, json=payload)
    except httpx.TimeoutException as exc:
        raise DApiError(408, "no response within the timeout") from exc
    if r.is_error:
        body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        raise DApiError(r.status_code, body.get("error", r.text))
    return r.json()

# text
await send("/api/v1/messages/send/text",
           {"sessionId": "billing", "to": "14155550123", "text": "Payment confirmed."})

# PDF from a public URL
await send("/api/v1/messages/send/document",
           {"sessionId": "billing", "to": "14155550123",
            "document": "https://your-app.com/invoices/4812.pdf", "fileName": "invoice-4812.pdf"})

Images, audio and video follow the same pattern, changing the path and the field name. The full list of accepted types and formats is in how to send media through the API.

Receive the webhook with FastAPI without blocking

Every webhook delivery waits for an HTTP response. If your endpoint is slow, D-API treats the attempt as a failure and retries later, and you start receiving the same event again. That is why the route should not call a slow database, an AI model or another API before responding. It validates, hands the work off and returns 200.

import hmac
import os
from typing import Any

from fastapi import BackgroundTasks, FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
WEBHOOK_TOKEN = os.environ["WEBHOOK_TOKEN"]

class Event(BaseModel):
    event: str
    sessionId: str
    timestamp: str
    traceId: str | None = None
    data: dict[str, Any]

@app.post("/webhooks/whatsapp/{token}")
async def webhook(token: str, event: Event, tasks: BackgroundTasks):
    if not hmac.compare_digest(token, WEBHOOK_TOKEN):
        raise HTTPException(status_code=401)
    if event.event == "messages.received" and not event.data.get("fromMe"):
        tasks.add_task(process, event)
    return {"ok": True}

async def process(event: Event) -> None:
    data = event.data
    number = data["from"]["jid"].split("@")[0]
    if data["type"] == "text" and "invoice copy" in data["message"].lower():
        await send("/api/v1/messages/send/text",
                   {"sessionId": event.sessionId, "to": number,
                    "text": "Generating a copy of your invoice, one moment."})

The secret token in the path replaces the signature, which the WhatsApp webhook does not have. Comparing with hmac.compare_digest only keeps the comparison time from leaking the token; there is no HMAC on the request body. For the fields of each event, see the WhatsApp webhooks page.

When BackgroundTasks is no longer enough

BackgroundTasks runs after the response, in the same Uvicorn process. It is great to start with, but it has clear limits:

When you move to a queue, the route stays almost the same: instead of add_task, it publishes the event with process.delay(event.model_dump()) in Celery or an equivalent. If your use case is a bot that talks to a language model, the WhatsApp AI chatbot guide shows how to organize context and history.

Idempotency and processing best practices

1. Deduplicate by message id. A retry can arrive after you already processed the event. Storing data.id with SET NX in Redis solves it in one line.
2. Ignore what you sent yourself. Messages with fromMe set to true come from the same number on another device and don't call for an automatic reply.
3. Log the traceId. Put the field in the worker's structured logs; it makes it easy to match what your system did with what D-API delivered.
4. Don't block the loop. Inside an async function, use the async httpx client. Calling requests there freezes every other route until the response comes back.

Once the flow works for one number, the same structure handles dozens: the sessionId in the event tells you which connection the message came from. D-API bills per connection, not per message; plans are on the pricing page.