API de WhatsApp em Python: enviar com httpx e receber com FastAPI

O que você precisa ter em mãos

Não existe pacote Python oficial, e ele não faz falta. A API de WhatsApp da D-API responde em https://api.d-api.cloud, recebe JSON e autentica pelo header Authorization com a sua API Key pura, sem Bearer. Qualquer biblioteca HTTP funciona. Antes de escrever código, separe:

• a API Key do painel, em uma variável de ambiente como DAPI_API_KEY;
• o sessionId de um número já conectado (a conexão é feita lendo um QR Code, como no WhatsApp Web);
• uma URL pública para o webhook. Em desenvolvimento, um túnel como ngrok ou cloudflared expõe o seu localhost.

Os exemplos usam Python 3.10 ou superior, por causa da sintaxe de tipos com barra vertical nos modelos Pydantic. Em versões anteriores, troque por Optional do módulo typing e o resto continua igual.

Enviar mensagem com requests

Para um script, uma tarefa agendada ou um Django síncrono, requests resolve. O detalhe que costuma passar batido: requests não define timeout por conta própria. Sem ele, um problema de rede deixa o worker parado esperando para sempre.

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": "financeiro", "to": "5511999999999", "text": "Seu boleto vence amanhã."},
    headers=HEADERS,
    timeout=(3, 10),  # 3s para conectar, 10s para ler a resposta
)
if not resp.ok:
    erro = resp.json() if "application/json" in resp.headers.get("content-type", "") else {}
    raise RuntimeError(f"D-API {resp.status_code}: {erro.get('error', resp.text)}")

Um client reutilizável com httpx

Em código assíncrono, e em qualquer aplicação que envia com frequência, vale ter um client único com pool de conexões. O httpx oferece a mesma API em versão síncrona e assíncrona, e aceita base_url, headers padrão e timeout configurados uma vez só. A função abaixo converte o corpo de erro da API, que tem o formato {"success": false, "error": "...", "statusCode": 400}, em uma exceção própria.

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 enviar(caminho: str, payload: dict) -> dict:
    try:
        r = await dapi.post(caminho, json=payload)
    except httpx.TimeoutException as exc:
        raise DApiError(408, "sem resposta dentro do timeout") from exc
    if r.is_error:
        corpo = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        raise DApiError(r.status_code, corpo.get("error", r.text))
    return r.json()

# texto
await enviar("/api/v1/messages/send/text",
             {"sessionId": "financeiro", "to": "5511999999999", "text": "Pagamento confirmado."})

# PDF por URL pública
await enviar("/api/v1/messages/send/document",
             {"sessionId": "financeiro", "to": "5511999999999",
              "document": "https://seu-app.com/boletos/4812.pdf", "fileName": "boleto-4812.pdf"})

Imagem, áudio e vídeo seguem o mesmo padrão, trocando o caminho e o nome do campo. A lista completa de tipos e formatos aceitos está em como enviar mídia pela API.

Receber o webhook com FastAPI sem travar

Cada entrega do webhook espera uma resposta HTTP. Se o seu endpoint demorar, a D-API trata a tentativa como falha e reenvia mais tarde, e você passa a receber o mesmo evento repetido. Por isso a rota não deve chamar banco lento, IA ou outra API antes de responder. Ela valida, entrega o trabalho para outro lugar e devolve 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 Evento(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, evento: Evento, tarefas: BackgroundTasks):
    if not hmac.compare_digest(token, WEBHOOK_TOKEN):
        raise HTTPException(status_code=401)
    if evento.event == "messages.received" and not evento.data.get("fromMe"):
        tarefas.add_task(processar, evento)
    return {"ok": True}

async def processar(evento: Evento) -> None:
    dados = evento.data
    numero = dados["from"]["jid"].split("@")[0]
    if dados["type"] == "text" and "segunda via" in dados["message"].lower():
        await enviar("/api/v1/messages/send/text",
                     {"sessionId": evento.sessionId, "to": numero,
                      "text": "Vou gerar sua segunda via, só um instante."})

O token secreto no caminho substitui a assinatura, que o webhook de WhatsApp não tem. A comparação com hmac.compare_digest só evita que o tempo da comparação revele o token; não há HMAC no corpo da requisição. Para os campos de cada evento, consulte a página de webhook de WhatsApp.

Quando BackgroundTasks deixa de ser suficiente

O BackgroundTasks roda depois da resposta, no mesmo processo do Uvicorn. É ótimo para começar, mas tem limites claros:

Na migração para fila, a rota continua quase igual: em vez de add_task, ela publica o evento com processar.delay(evento.model_dump()) no Celery ou equivalente. Se o seu caso é um bot que conversa com um modelo de linguagem, o guia de chatbot de WhatsApp com IA mostra como organizar contexto e histórico.

Idempotência e boas práticas no processamento

1. Deduplique pelo id da mensagem. Um reenvio pode chegar depois que você já processou o evento. Gravar data.id com SET NX no Redis resolve em uma linha.
2. Ignore o que você mesmo mandou. Mensagens com fromMe verdadeiro vêm do próprio número em outro aparelho e não pedem resposta automática.
3. Registre o traceId. Coloque o campo nos logs estruturados do worker; com ele fica fácil cruzar o que o seu sistema fez com o que a D-API entregou.
4. Não bloqueie o loop. Dentro de função assíncrona, use o client httpx assíncrono. Chamar requests ali trava todas as outras rotas enquanto a resposta não chega.

Com o fluxo funcionando em um número, a mesma estrutura atende dezenas: o sessionId no evento diz de qual conexão veio a mensagem. A cobrança da D-API é por conexão, não por mensagem; os planos estão em preços.