--- title: "API de WhatsApp em Python: enviar e receber | D-API" description: "Use a API de WhatsApp em Python com requests ou httpx para enviar mensagens e FastAPI para receber o webhook, processando cada evento em background sem travar." url: https://d-api.cloud/api-whatsapp-python --- Por linguagem # API de WhatsApp em Python: enviar com httpx e receber com FastAPI Em Python você integra a API de WhatsApp da D-API com o cliente HTTP que já usa, requests ou httpx, porque tudo é REST com JSON. O ponto que mais dá problema não é o envio, é o webhook: ele precisa responder rápido e deixar o processamento para depois, e é nisso que este guia foca. Por Time de engenharia D-APIAtualizado em 24 de set de 20266 min de leitura ## O que você precisa ter em mãos Não existe pacote Python oficial, e ele não faz falta. A [API de WhatsApp](https://d-api.cloud/) 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](https://d-api.cloud/api-whatsapp-enviar-midia). ## 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](https://d-api.cloud/webhook-api-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: | Situação | BackgroundTasks | Fila externa (Celery, RQ, arq) | | --- | --- | --- | | Poucas mensagens por minuto | Atende bem | Exagero | | Resposta gerada por IA, vários segundos por evento | Consome o event loop e a memória do web | Workers separados, escaláveis | | Deploy ou reinício do servidor | Tarefas em andamento se perdem | Eventos ficam na fila | | Várias réplicas da API | Cada réplica processa o que recebeu | Qualquer worker pega qualquer evento | 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](https://d-api.cloud/api-whatsapp-chatbot-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](https://d-api.cloud/pricing). ## Perguntas frequentes ### Existe biblioteca Python oficial da D-API? Não. O SDK oficial é só para Node.js. Em Python a integração é REST pura: requests, httpx ou aiohttp resolvem, porque cada operação é uma requisição HTTP com JSON e o header Authorization. ### Por que minha chamada com requests ficou travada para sempre? Porque requests não tem timeout padrão. Sem o parâmetro timeout, uma conexão que não responde segura a thread indefinidamente. Passe sempre uma tupla com tempo de conexão e de leitura, por exemplo timeout=(3, 10). ### BackgroundTasks do FastAPI serve para produção? Serve para volume baixo e tarefas curtas. A tarefa roda no mesmo processo e se perde se o servidor reiniciar. Com volume maior ou quando cada evento chama uma IA, grave o evento em uma fila como Celery, RQ ou arq. ### Consigo usar Django ou Flask em vez de FastAPI? Sim. O webhook é só um POST com JSON. Em Django é uma view com csrf\_exempt, em Flask uma rota que lê request.get\_json(). A regra continua a mesma: validar, enfileirar e responder rápido. ### Como evito processar a mesma mensagem duas vezes? Use o id que vem em data.id como chave. Antes de processar, tente gravar esse id no Redis com SET NX e um prazo de expiração, ou numa tabela com índice único. Se já existir, descarte o evento. Teste a API de WhatsApp da D-API3 dias grátis, sem cartão.[Começar trial grátis](https://app.d-api.cloud/register) ## Continue lendo - [Por linguagemAPI de WhatsApp em PHP](https://d-api.cloud/api-whatsapp-php) - [Por linguagemAPI de WhatsApp em Laravel](https://d-api.cloud/api-whatsapp-laravel) - [Por linguagemAPI de WhatsApp em Java](https://d-api.cloud/api-whatsapp-java) - [Por linguagemAPI de WhatsApp em C# e .NET](https://d-api.cloud/api-whatsapp-csharp) ## Teste a API de WhatsApp da D-API Trial de 3 dias com acesso completo. Sem cartão, sem fidelidade. [Começar trial grátis](https://app.d-api.cloud/register)[Ver documentação](https://docs.d-api.cloud/)