Como integrar a API do WhatsApp em Python

Os dois lados da integração: sair uma mensagem e entrar um webhook. O envio é uma chamada HTTP comum e dá pouco trabalho. O recebimento tem uma armadilha específica de Python — o corpo cru — que derruba a verificação de assinatura inteira. Este guia cobre os dois com requests, httpx, FastAPI e Flask.

O que você precisa

• Uma chave de API (hub_pk_...) com o escopo messages:send. Crie em /painel/chaves. Para criar o endpoint de webhook, também webhooks:write.
• O id do canal — liste em GET /api/v1/channels.
• O segredo do webhook (whsec_...), devolvido uma única vez na criação do endpoint. Veja configurar webhook.
• Python 3.9 ou mais novo. Os trechos usam str.removeprefix (3.9) e a sintaxe str | None (3.10). Em 3.9, troque por Optional[str].

Enviar com requests

Só existe um endpoint de envio: POST /api/v1/channels/{id}/messages. O corpo é sempre { to, type, ...objeto do tipo }. O to vai com código do país e só dígitos — sem +, sem espaço, sem parênteses.

import os
import uuid

import requests

BASE = "https://crprohub.com/api/v1"
CANAL = os.environ["CRPRO_CHANNEL_ID"]

SESSAO = requests.Session()
SESSAO.headers["Authorization"] = f"Bearer {os.environ['CRPRO_API_KEY']}"


def enviar_texto(para: str, texto: str, chave_idem: str) -> dict:
    resposta = SESSAO.post(
        f"{BASE}/channels/{CANAL}/messages",
        headers={"Idempotency-Key": chave_idem},
        json={"to": para, "type": "text", "text": {"body": texto}},
        timeout=15,
    )
    corpo = resposta.json()

    if resposta.status_code != 202:
        erro = corpo["error"]
        raise RuntimeError(
            f"{erro['code']}: {erro['message']} "
            f"(x-request-id={resposta.headers.get('x-request-id')})"
        )

    return corpo["data"]


aceita = enviar_texto("5521999999999", "Olá", str(uuid.uuid4()))
print(aceita["message_id"], aceita["status"])  # wamid.EXEMPLO123 accepted

Três detalhes que não são estilo. O timeout é obrigatório na prática: requests não tem timeout padrão, e uma chamada sem ele pode pendurar o processo indefinidamente. A comparação é com 202, não com resposta.ok: a API responde 202 aqui, nunca 200. E o corpo é lido antes de levantar o erro — trocar isso por raise_for_status() joga fora o envelope {"error": {...}} e você fica com um traceback sem o code que diz o que aconteceu.

O 202 significa aceita e enfileirada para a Meta, não entregue. O message_id serve para correlacionar depois; o status vem como accepted. Marcar isso como “entregue” no seu banco é o erro clássico — o estado final chega por webhook.

Para os outros tipos, só muda o par type + objeto: image, audio, video, document, sticker, location, contacts, reaction, interactive e template. Veja enviar mídia e enviar template.

Idempotência e retentativa

O header Idempotency-Key é obrigatório neste endpoint. A razão aparece justamente em Python, onde retentativa é fácil de escrever e fácil de escrever errado: um timeout não diz se a mensagem saiu, e reenviar às cegas manda a mesma mensagem duas vezes para o cliente.

import time
import uuid

import requests

from enviar import enviar_texto


def enviar_com_retentativa(para: str, texto: str) -> dict:
    # A chave nasce FORA do laço, de propósito. Gerar um uuid4 novo a cada
    # volta transformaria a retentativa em um segundo envio de verdade.
    chave_idem = str(uuid.uuid4())

    for tentativa in range(3):
        try:
            return enviar_texto(para, texto, chave_idem)
        except (requests.Timeout, requests.ConnectionError):
            if tentativa == 2:
                raise
            time.sleep(2**tentativa)

A chave é gerada uma vez e reutilizada em todas as tentativas daquele envio lógico. Com a mesma chave e o mesmo corpo, a segunda chamada devolve a resposta original em vez de enviar de novo. Se você usar urllib3.Retry montado em um HTTPAdapter, isso já acontece de graça — o adapter repete a requisição com os mesmos headers.

Enviar com httpx

Se o seu serviço é assíncrono — FastAPI, por exemplo — usar requests dentro de uma corrotina bloqueia o event loop inteiro. O httpx resolve com a mesma forma de chamada.

import os
import uuid

import httpx

CANAL = os.environ["CRPRO_CHANNEL_ID"]

CLIENTE = httpx.AsyncClient(
    base_url="https://crprohub.com/api/v1",
    headers={"Authorization": f"Bearer {os.environ['CRPRO_API_KEY']}"},
    timeout=15.0,
)


async def enviar_texto(para: str, texto: str, chave_idem: str) -> dict:
    resposta = await CLIENTE.post(
        f"/channels/{CANAL}/messages",
        headers={"Idempotency-Key": chave_idem},
        json={"to": para, "type": "text", "text": {"body": texto}},
    )
    corpo = resposta.json()

    if resposta.status_code != 202:
        raise RuntimeError(corpo["error"]["code"])

    return corpo["data"]

Ao contrário de requests, o httpx já traz um timeout padrão de 5 segundos. Ele é curto para este endpoint, que espera a Meta aceitar a mensagem antes de responder — daí o timeout=15.0 explícito. O cliente é criado uma vez no módulo, e não por chamada: criar um AsyncClient a cada envio joga fora o pool de conexões e o handshake TLS.

Receber com FastAPI

Aqui está a parte que quebra integrações. A assinatura cobre os bytes exatos que chegaram na requisição. Em FastAPI, await request.body() devolve esses bytes; await request.json() devolve um dict, e um dict reserializado com json.dumps quase nunca produz os mesmos bytes — espaço depois dos dois pontos, ordem de chaves, escape de acento. Leia o corpo cru primeiro, sempre.

import json

from fastapi import FastAPI, Header, Request, Response

from verificacao import assinatura_valida

app = FastAPI()


@app.post("/webhooks/crprohub")
async def receber(
    request: Request,
    x_hub_delivery_id: str | None = Header(default=None),
    x_hub_timestamp: str | None = Header(default=None),
    x_hub_signature_256: str | None = Header(default=None),
) -> Response:
    # body() devolve os bytes exatos que foram assinados. Chamar
    # await request.json() primeiro consome o stream e você fica sem eles --
    # e reserializar o dict depois produz outros bytes.
    corpo = await request.body()

    if not (x_hub_delivery_id and x_hub_timestamp and x_hub_signature_256):
        return Response(status_code=401)

    if not assinatura_valida(corpo, x_hub_timestamp, x_hub_delivery_id, x_hub_signature_256):
        return Response(status_code=401)

    # registrar_entrega devolve False se este delivery id já foi visto:
    # uma retentativa reenvia o mesmo id.
    if registrar_entrega(x_hub_delivery_id):
        enfileirar(json.loads(corpo))

    return Response(status_code=200)

Os nomes dos parâmetros viram headers automaticamente: x_hub_delivery_id lê X-Hub-Delivery-Id. Eles são declarados como opcionais de propósito — sem isso, uma requisição sem os headers recebe 400 INVALID_MESSAGE do FastAPI antes de chegar ao seu código, e o log fica dizendo “erro de validação” quando a resposta certa é 401.

A deduplicação por X-Hub-Delivery-Id não é opcional: uma retentativa reenvia o mesmo id, e o seu handler vai ver o mesmo evento mais de uma vez. Um SET NX no Redis ou uma coluna única no banco resolvem.

{
  "id": "e4f5a6b7-8c9d-4e0f-1a2b-3c4d5e6f7a8b",
  "event": "message.received",
  "channel_id": "b1c2d3e4-5f6a-4b7c-8d9e-0f1a2b3c4d5e",
  "occurred_at": "2026-08-21T12:00:00.000Z",
  "data": {}
}

O envelope normalized é igual para qualquer evento. Os que interessam a quase toda integração são message.received e message.status — e um endpoint criado sem event_types recebe todos eles, porque lista vazia é curinga. Veja webhooks.

Verificar a assinatura

No formato nativo (X-Hub-Signature-Version: v2, o padrão) a assinatura não cobre só o corpo: ela cobre timestamp.delivery_id.corpo, com os valores exatos de X-Hub-Timestamp e X-Hub-Delivery-Id daquela entrega. Quem calcula o HMAC só do corpo vê 100% das entregas falharem — não algumas, todas. Amarrar a assinatura aos dois é o que impede um corpo capturado de ser transplantado para outra entrega.

import hashlib
import hmac
import os
import re

SEGREDO = os.environ["CRPRO_WEBHOOK_SECRET"].encode()
HEX_64 = re.compile("^[0-9a-f]{64}$")


def assinatura_valida(corpo: bytes, timestamp: str, delivery_id: str, header: str) -> bool:
    recebida = header.removeprefix("sha256=")

    # Valide o hexadecimal antes de comparar: compare_digest exige os dois
    # lados no mesmo tipo e tamanho, e um header malformado viraria exceção.
    if not HEX_64.match(recebida):
        return False

    # O formato nativo assina timestamp.delivery_id.corpo, não só o corpo.
    # Concatene em bytes: interpolar corpo numa f-string escreveria a repr
    # b'...' no meio da mensagem assinada e nenhuma entrega passaria.
    assinado = f"{timestamp}.{delivery_id}.".encode() + corpo
    esperada = hmac.new(SEGREDO, assinado, hashlib.sha256).hexdigest()

    return hmac.compare_digest(recebida, esperada)

A montagem do valor assinado tem uma pegadinha só de Python. O corpo é bytes e o prefixo é str. Escrever f"{timestamp}.{delivery_id}.{corpo}" compila, roda e produz b'...' literalmente no meio da mensagem — nenhuma assinatura bate, e nada no traceback aponta para isso. Codifique só o prefixo e concatene em bytes.

A comparação usa hmac.compare_digest e nunca ==: comparar strings vaza o tempo de execução por posição de byte, o que dá a um atacante um caminho para descobrir a assinatura correta por tentativa e erro. É o equivalente ao timingSafeEqual do Node e ao hash_equals do PHP.

E o ^[0-9a-f]{64}$ vem antes da comparação porque compare_digest exige os dois lados do mesmo tipo — um header ausente, truncado ou com maiúsculas vira TypeError em vez de False, e derruba o handler no lugar de rejeitar a entrega.

Receber com Flask

Mesma lógica, outro jeito de pegar os bytes. Em Flask o corpo cru é request.get_data() — sem as_text=True, que decodificaria para str e daria o mesmo problema de tipo da seção anterior.

from flask import Flask, request

from verificacao import assinatura_valida

app = Flask(__name__)


@app.post("/webhooks/crprohub")
def receber():
    # get_data() sem as_text=True devolve bytes crus, e o Flask os mantém em
    # cache -- por isso get_json() abaixo ainda funciona.
    corpo = request.get_data()

    valida = assinatura_valida(
        corpo,
        request.headers.get("X-Hub-Timestamp", ""),
        request.headers.get("X-Hub-Delivery-Id", ""),
        request.headers.get("X-Hub-Signature-256", ""),
    )
    if not valida:
        return "", 401

    delivery_id = request.headers["X-Hub-Delivery-Id"]
    if registrar_entrega(delivery_id):
        enfileirar(request.get_json())

    return "", 200

A ordem importa: request.get_data() antes de request.get_json(). O Flask mantém o corpo em cache depois da primeira leitura, então as duas chamadas convivem — mas ler o JSON primeiro em um handler que depois tenta reconstruir os bytes é como a verificação passa a falhar sem ninguém entender por quê.

Responder rápido

O endpoint precisa responder 2xx em até 10 segundos. Não é uma recomendação: passou disso, a entrega conta como falha, entra na fila de retentativa, e depois de 20 falhas consecutivas o endpoint pausa sozinho — e aí nada chega até alguém reativar.

• Valide a assinatura, registre o delivery_id, devolva 200. Todo o resto — gravar no banco, chamar outra API, responder o cliente — vai para fora do handler.
• Em Flask, cada requisição ocupa um worker do Gunicorn até retornar. Processar a mensagem dentro do handler não atrasa só aquela entrega: esgota o pool e passa a atrasar todas.
• Em FastAPI, BackgroundTasks devolve o 200 antes de rodar a tarefa, o que resolve o prazo. Mas a tarefa morre junto com o processo — para evento que não pode se perder, use uma fila de verdade.
• Um handler def (síncrono) no FastAPI roda no threadpool; um async def roda no event loop, e qualquer I/O bloqueante ali dentro trava o servidor inteiro. Se for usar requests ou um driver síncrono de banco, declare o handler como def.

Perdeu eventos enquanto o serviço estava fora do ar? Não tem como reprocessar do seu lado — use o replay em configurar webhook.

Quando falha

• 401 — chave errada, revogada, ou você montou o header sem o prefixo Bearer. Um f"Bearer {chave}" com a variável de ambiente vazia cai aqui.
• 403 — a chave existe mas não tem o escopo messages:send.
• 404 — o canal não existe ou pertence a outra organização. São 404 nos dois casos de propósito: 403 confirmaria que o recurso existe.
• 409 — Idempotency-Key reaproveitada com corpo diferente.
• 400 INVALID_MESSAGE — JSON válido, mensagem inválida: número malformado, type sem o objeto correspondente, texto vazio. Em Python, quase sempre um None que virou null no json=.
• 429 — limite de taxa. Veja limites.

Toda resposta traz o header x-request-id — o trecho de envio acima já o inclui na mensagem de erro. Guarde-o no seu log: é o que liga a sua chamada ao registro em /painel/logs. A tabela completa de códigos está em erros.