Como enviar mensagem de WhatsApp com Python
Do token ao fluxo de duas vias com requests e Flask: cliente com sessão e timeout, envio de texto, imagem e botões, verificação de número e um receptor de webhook que responde 200 e processa numa fila.
Do token ao fluxo de duas vias com requests e Flask: cliente com sessão e timeout, envio de texto, imagem e botões, verificação de número e um receptor de webhook que responde 200 e processa numa fila.
Muita automação de WhatsApp em Python começa como um script de dez linhas com requests.post e termina como uma rotina que o financeiro roda todo dia. A distância entre as duas coisas cabe nesta página: sessão reaproveitada, timeout explícito, erro que não passa despercebido e um receptor de webhook que não segura a resposta enquanto grava no banco.
O exemplo que atravessa o texto é um aviso de vencimento: a rotina lê as cobranças que vencem amanhã, avisa cada cliente, oferece segunda via em botões e recebe a resposta por webhook. Tudo com requests e Flask, que é o que a maior parte dos projetos em Python já tem instalado.
Python 3.9 ou mais novo, a biblioteca requests e um número conectado no zapon. Da conexão sai o token, que vai no header de toda chamada — um token por número, isolado dos demais. Não existe SDK para instalar: a API é REST com JSON, e qualquer cliente HTTP fala com ela.
pip install requests para enviar e pip install flask para receber. Nada além disso é necessário no caminho mínimo.token: ele autentica as chamadas daquele número.https://api.zapon.dev. A autenticação é o header token — não é Authorization, não é Bearer, não há OAuth.Em variável de ambiente, lida com os.environ["ZAPON_TOKEN"]. Use a forma com colchetes, não os.environ.get: assim o processo quebra no início, com KeyError, se a variável não estiver definida — bem melhor do que descobrir mais tarde que o header foi enviado com None e cada chamada volta 401. Em desenvolvimento, um .env fora do controle de versão com python-dotenv resolve; em produção, use o mecanismo de segredos do seu ambiente.
Um POST para https://api.zapon.dev/chat/send/text, com o header token e um corpo JSON de dois campos: Phone, o número só com dígitos, incluindo país e DDD, e Body, o texto. Use o parâmetro json= do requests, que serializa e já define o Content-Type, e sempre passe timeout.
Em vez de repetir requests.post pelo projeto, concentre tudo num módulo. Esta versão usa uma sessão — que reaproveita a conexão TCP entre chamadas, diferença sensível num laço de centenas de envios:
# zapon.py — pip install requests import os import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BASE = "https://api.zapon.dev" TIMEOUT = (5, 20) # (conectar, ler) — chamada sem timeout pode travar para sempre class ErroZapon(Exception): def __init__(self, status, detalhe): super().__init__(f"zapon {status}: {detalhe}") self.status = status # 0 = nem chegou na API (rede ou timeout) self.detalhe = detalhe def _nova_sessao(): s = requests.Session() s.headers["token"] = os.environ["ZAPON_TOKEN"] # falta a variável? quebra agora, não depois politica = Retry( total=2, backoff_factor=1, status_forcelist=[500, 502, 503, 504], allowed_methods=["POST"], # requests não repete POST por padrão ) s.mount("https://", HTTPAdapter(max_retries=politica)) return s SESSAO = _nova_sessao() def chamar(rota: str, payload: dict) -> dict: try: r = SESSAO.post(BASE + rota, json=payload, timeout=TIMEOUT) except requests.exceptions.RequestException as e: raise ErroZapon(0, repr(e)) from e try: corpo = r.json() except ValueError: # 500 pode voltar em texto puro corpo = None if not r.ok or not (corpo or {}).get("success"): raise ErroZapon(r.status_code, (corpo or {}).get("error") or r.text[:200]) return corpo["data"] def enviar_texto(phone: str, body: str) -> dict: return chamar("/chat/send/text", {"Phone": phone, "Body": body})
O uso, numa rotina que o agendador dispara toda manhã:
from zapon import enviar_texto, ErroZapon
try:
dados = enviar_texto(
"5511999999999",
"Olá, Ana. A parcela 3 do contrato 4821 vence amanhã, dia 04/09, no valor de R$ 340,00.",
)
print("enviada:", dados["Id"])
except ErroZapon as e:
print("falhou:", e.status, e.detalhe)A resposta de sucesso vem sempre neste formato:
{
"code": 200,
"success": true,
"data": {
"Details": "Sent",
"Id": "90B2F8B13FAC8A9CF6B06E99C7834DC5",
"Timestamp": "2026-09-03T09:12:08-03:00"
}
}A política de Retry acima cobre 500 e falhas de gateway, e isso ajuda quando a conexão oscila. Só que uma requisição pode falhar depois de a mensagem ter saído — o cliente recebe duas vezes e ninguém entende por quê. Por isso o total é baixo (duas tentativas), e por isso a rotina grava o Id devolvido antes de considerar a cobrança avisada. Se o seu volume é alto, prefira desligar a repetição automática e decidir caso a caso, com o registro do envio na mão.
O requests é o padrão de fato e cobre tudo que esta página faz. O httpx é a escolha natural se o seu serviço é assíncrono, porque tem a mesma interface em async — troque SESSAO.post por await cliente.post e o resto do código continua igual. Já o urllib.request da biblioteca padrão funciona sem instalar nada, mas você reescreve à mão o tratamento de corpo, de erro e de timeout. Não vale a economia.
POST /chat/send/image com Phone, Image e o opcional Caption. O campo Image é o arquivo em base64 no formato data URI. Em Python, é base64.b64encode seguido de .decode() — sem o decode, o valor continua sendo bytes e o serializador JSON levanta TypeError.
import base64 from pathlib import Path from zapon import chamar conteudo = Path("boleto-4821.png").read_bytes() imagem = "data:image/png;base64," + base64.b64encode(conteudo).decode() chamar("/chat/send/image", { "Phone": "5511999999999", "Image": imagem, "Caption": "Boleto da parcela 3 — vence em 04/09.", })
Para PDF, o caminho é /chat/send/document, que preserva o nome do arquivo — melhor escolha para boleto, nota fiscal e contrato. A imagem serve quando o cliente precisa ver sem abrir nada, como um código de barras ou um QR Code de pagamento.
O POST /chat/send/buttons aceita quantos botões você precisar — três ou quatro é o que se lê num toque. Cada um tem um id definido por você e um text curto — mantenha o rótulo em até 20 caracteres, porque o WhatsApp corta o excedente. O id volta inteiro no webhook, então é ali que você carimba a chave do registro: Além da resposta rápida, o mesmo endpoint monta botão que abre um link ("type": "cta_url"), botão que abre o discador ("type": "cta_call") e botão que copia um código ("type": "copy") — e os tipos convivem na mesma mensagem. O toque em resposta rápida volta no webhook; os outros três agem no aparelho de quem recebe.
chamar("/chat/send/buttons", { "Phone": "5511999999999", "Body": "A parcela 3 do contrato 4821 vence amanhã. Como podemos ajudar?", "Footer": "Financeiro — Empresa Exemplo", "Buttons": [ {"id": "cob4821|pix", "text": "Quero o Pix"}, # 12 caracteres {"id": "cob4821|segunda", "text": "Segunda via"}, {"id": "cob4821|negociar", "text": "Quero negociar"}, ], })
Note o desenho dos identificadores: prefixo com a chave do registro, separador e ação. No webhook, um split("|") devolve as duas partes e a resposta cai no lugar certo mesmo que o cliente tenha três cobranças abertas. Precisando de mais de três opções, use /chat/send/list, que abre um menu selecionável.
POST /user/check recebe uma lista no campo Phone e devolve, para cada item, o campo IsInWhatsapp. Vale rodar antes de um lote: base de cobrança costuma ter telefone fixo e número antigo, e uma sequência de envios para números inexistentes é exatamente o padrão que aumenta o risco do seu número.
from zapon import chamar candidatos = ["5511999999999", "5511888888888", "551133334444"] resposta = chamar("/user/check", {"Phone": candidatos}) validos = [u["Query"] for u in resposta["Users"] if u["IsInWhatsapp"]] fora = [u["Query"] for u in resposta["Users"] if not u["IsInWhatsapp"]] print(validos) # ['5511999999999'] marcar_sem_whatsapp(fora) # corrija o cadastro, não só o disparo de hoje
Guarde o resultado no cadastro em vez de reconsultar a cada rodada. Número reprovado quase sempre é dígito faltando ou telefone fixo cadastrado no campo de celular — problema de dado, que se resolve uma vez.
Cadastre no painel, por conexão, uma URL pública e escolha os eventos. Cada evento vira um POST para essa URL, com a mensagem inteira no corpo. No Python, a regra é responder 200 na própria função da rota e empurrar o processamento para uma fila ou uma thread — não faça consulta a banco antes de devolver a resposta.
O corpo chega em uma de duas formas: JSON puro ou formulário codificado, com o evento dentro do campo jsonData. Aceitar as duas custa três linhas e evita o sintoma de "o webhook não chega" — ele chega, e o request.get_json() é que levanta erro e devolve 400 sem você ver.
# webhook.py — pip install flask import json from queue import Queue from threading import Thread from flask import Flask, request from zapon import enviar_texto app = Flask(__name__) fila = Queue() vistos = set() # em produção: Redis ou uma tabela, não memória do processo @app.post("/whatsapp/eventos") def receber(): evento = request.get_json(silent=True) # silent=True: não estoura se não for JSON if evento is None and "jsonData" in request.form: evento = json.loads(request.form["jsonData"]) if evento: fila.put(evento) return "", 200 # responda já; o trabalho é do worker def tratar(evento): if evento.get("type") != "Message": return corpo = evento.get("event") or {} info = corpo.get("Info") or {} if info.get("IsFromMe") or info.get("IsGroup"): return # o que você enviou também gera evento if info.get("ID") in vistos: return vistos.add(info.get("ID")) telefone = str(info.get("Chat", "")).split("@")[0] msg = corpo.get("Message") or {} botao = (msg.get("buttonsResponseMessage") or {}).get("selectedButtonID", "") texto = (msg.get("conversation") or (msg.get("extendedTextMessage") or {}).get("text") or "").strip() if botao: cobranca, acao = botao.split("|") registrar_escolha(cobranca, acao) enviar_texto(telefone, "Certo! Já estamos providenciando.") return fila_de_atendimento(telefone, texto) # o resto é assunto de gente def trabalhador(): while True: evento = fila.get() try: tratar(evento) except Exception as e: # falha de um evento não pode parar a fila print("erro ao tratar evento:", repr(e), flush=True) finally: fila.task_done() Thread(target=trabalhador, daemon=True).start()
O corpo traz o tipo do evento e a mensagem completa. Assim chega quem respondeu por escrito:
{
"type": "Message",
"event": {
"Info": {
"ID": "3EB0C767D26A1B5F7C83",
"Chat": "[email protected]",
"Sender": "[email protected]",
"IsFromMe": false,
"IsGroup": false,
"PushName": "Ana Souza",
"Timestamp": "2026-09-03T18:04:22-03:00"
},
"Message": { "conversation": "manda a segunda via por favor" }
}
}
// quando ele toca num botão, muda só o bloco Message:
"Message": { "buttonsResponseMessage": {
"selectedButtonID": "cob4821|segunda",
"Response": { "SelectedDisplayText": "Segunda via" }
} }Não. Suba com gunicorn -w 4 webhook:app ou equivalente. E atenção a uma consequência disso: com quatro processos, o set de vistos e a Queue em memória existem quatro vezes, cada um enxergando só a sua fatia — a deduplicação some. Em produção, a chave Info.ID vai para Redis ou para uma tabela com índice único, e a fila vira Celery, RQ ou a fila do seu banco. Em Django, o equivalente é uma view com @csrf_exempt que devolve HttpResponse(status=200) e enfileira a tarefa; em FastAPI, um BackgroundTasks resolve o mesmo problema. As entregas chegam com o cabeçalho User-Agent: zapon-webhook/1, útil para filtrar nos seus registros.
400 é corpo inválido e se resolve corrigindo o código. 401 é token ausente, errado ou de outra conexão. 404 é caminho digitado errado. 500 costuma significar conexão fora do ar — é o único caso em que repetir ajuda, com espera crescente. Insistir num 400 ou num 401 só gasta chamada e polui o registro.
| Código | Significa | O que fazer em Python |
|---|---|---|
200 | Requisição aceita | Grave data["Id"] junto do seu registro e siga |
400 | Corpo inválido | Registre o payload: em geral falta um campo obrigatório ou o Phone veio com +, espaço ou traço |
401 | Token ausente ou inválido | Confirme que ZAPON_TOKEN foi carregada no ambiente do processo — o .env do editor não vale para o serviço |
404 | Endpoint inexistente | Confira a rota; um /chat/send/txt cai aqui |
500 | Erro ao processar | Repita com espera crescente e veja o estado da conexão no painel |
Com a exceção própria do cliente, a decisão fica em um lugar só:
from zapon import chamar, ErroZapon
def enviar_com_registro(cobranca_id, telefone, texto):
try:
dados = chamar("/chat/send/text", {"Phone": telefone, "Body": texto})
except ErroZapon as e:
if e.status == 401:
raise SystemExit("token inválido: nenhuma outra mensagem vai sair hoje")
if e.status in (0, 500):
pendencias.registrar(cobranca_id, "tentar de novo", e.detalhe)
else: # 400 e 404 são defeito nosso
pendencias.registrar(cobranca_id, "corrigir dados", e.detalhe)
return None
envios.gravar(cobranca_id, dados["Id"]) # só agora a cobrança conta como avisada
return dados["Id"]O SystemExit no 401 parece agressivo, mas é o comportamento certo numa rotina em lote: se o token está errado, as próximas mil chamadas também falharão. Melhor parar cedo e avisar do que encher a tabela de pendências com o mesmo erro.
A rotina em lote é o ponto de risco. Serialize os envios e coloque um intervalo com variação entre eles — dez minutos a mais na madrugada não custam nada perto de perder o número:
import random
import time
for cobranca in vencendo_amanha():
if cobranca.cliente.opt_out: # quem pediu para sair, sai de todas as rotinas
continue
enviar_com_registro(cobranca.id, cobranca.cliente.telefone, texto_de(cobranca))
time.sleep(random.uniform(3, 7)) # intervalo humano, não exatos 2 segundosThreadPoolExecutor com 20 workers resolve o tempo do lote e cria o problema: vinte mensagens por segundo não se parecem com uso humano.Sobre custo, não há variável escondida na conta: R$ 27 por mês por número conectado, sem cobrança por mensagem, respeitada a cota de 300 mensagens por dia por número conectado — acima do teto a resposta vem com HTTP 429 e a sua rotina em Python só precisa tratar o status. O teste é de 14 dias sem cartão e só começa a contar na primeira conexão — dá para escrever e testar a rotina inteira antes de gastar o primeiro dia.
Não existe SDK e não é preciso: a API é REST com JSON. Com requests instalado você já faz tudo o que esta página mostra, e o módulo zapon.py do exemplo tem menos de 50 linhas.
Funcionam. O parâmetro json= do requests serializa com ensure_ascii=True, então o acento viaja como sequência de escape no JSON e chega correto no WhatsApp — o formato é o mesmo. O cuidado é ao montar o corpo à mão com data=: aí você precisa de json.dumps(...).encode("utf-8") e do header Content-Type: application/json.
Troque requests por httpx.AsyncClient ou aiohttp, mantendo a mesma estrutura de cliente e o mesmo tratamento de erro. Lembre que assincronia serve para não bloquear o processo enquanto espera a rede, não para disparar tudo de uma vez: o intervalo entre envios continua necessário.
Dá. Em Django, uma view com @csrf_exempt que lê request.body ou request.POST["jsonData"], devolve HttpResponse(status=200) e enfileira a tarefa. Em FastAPI, uma rota async com BackgroundTasks. O importante é o mesmo em qualquer framework: responder 200 antes de processar.
Suba o Flask local, exponha a porta com um túnel HTTP e cadastre a URL do túnel na conexão. Para testar só a sua regra, faça um POST no seu próprio endpoint com o payload de exemplo desta página — é o mesmo formato que chega em produção.
Sim, guardando um token por conexão e passando-o como parâmetro em vez de fixá-lo na sessão. Cada número conectado custa R$ 27 por mês, sem cobrança por mensagem, e cada token tem a sua cota de 300 mensagens por dia — as contagens são isoladas entre si.
Não. O número é conectado por QR Code ou código de pareamento, sem fila de aprovação e sem cadastro prévio de mensagem. Em troca, a conduta do número fica por conta de quem integra — por isso a seção sobre ritmo de envio importa tanto quanto o código.
14 dias grátis, sem cartão — a contagem só começa na primeira conexão. R$ 27 por mês por número, sem cobrança por mensagem.