Enviar WhatsApp com Python: exemplo de API REST (requests e httpx)

cURL

curl -X POST https://api.zapzapapi.com/api/v1/SEU_INSTANCE_ID/send/text \
  -H "Content-Type: application/json" \
  -H "x-api-key: SUA_API_KEY" \
  -H "x-api-secret: SEU_API_SECRET" \
  -d '{"number":"5511999999999","text":"Olá! Teste de envio pela API."}'

Python (requests)

import os
import requests

BASE = "https://api.zapzapapi.com/api/v1/" + os.environ["ZAPZAP_INSTANCE_ID"]
sessao = requests.Session()
sessao.headers.update({
    "x-api-key": os.environ["ZAPZAP_API_KEY"],
    "x-api-secret": os.environ["ZAPZAP_API_SECRET"],
})

def enviar_texto(numero: str, texto: str) -> dict:
    r = sessao.post(
        BASE + "/send/text",
        json={"number": numero, "text": texto},
        timeout=15,
    )
    r.raise_for_status()
    return r.json()  # {"messageId": "...", "status": "sent"}

if __name__ == "__main__":
    print(enviar_texto("5511999999999", "Olá! Seu pedido saiu para entrega."))

Python (requests)

def enviar_pdf(numero: str, url: str, nome: str, legenda: str = "") -> dict:
    r = sessao.post(
        BASE + "/send/media",
        json={"number": numero, "type": "document", "file": url,
              "docName": nome, "text": legenda},
        timeout=60,
    )
    r.raise_for_status()
    return r.json()

Python (httpx)

import asyncio, os, random
import httpx

BASE = "https://api.zapzapapi.com/api/v1/" + os.environ["ZAPZAP_INSTANCE_ID"]
HEADERS = {
    "x-api-key": os.environ["ZAPZAP_API_KEY"],
    "x-api-secret": os.environ["ZAPZAP_API_SECRET"],
}
limite = asyncio.Semaphore(3)

async def enviar(cliente: httpx.AsyncClient, numero: str, texto: str) -> dict:
    for tentativa in range(4):
        async with limite:
            r = await cliente.post(BASE + "/send/text",
                                   json={"number": numero, "text": texto})
        if r.status_code < 400:
            return r.json()
        if r.status_code not in (429, 500, 502, 503, 504):
            r.raise_for_status()  # erro do pedido: não adianta repetir
        await asyncio.sleep(2 ** tentativa + random.random())
    r.raise_for_status()

async def main(contatos: dict[str, str]):
    async with httpx.AsyncClient(headers=HEADERS, timeout=15) as c:
        for numero, texto in contatos.items():
            print(await enviar(c, numero, texto))
            await asyncio.sleep(random.uniform(2, 6))  # intervalo entre mensagens

asyncio.run(main({"5511999999999": "Olá! Confirmando sua consulta de amanhã."}))

Python (FastAPI)

from fastapi import BackgroundTasks, FastAPI, Request

app = FastAPI()

def tratar(evento: dict):
    msg = (evento.get("data") or {}).get("message") or {}
    if evento.get("event") != "messages" or msg.get("fromMe"):
        return
    print("mensagem de", msg.get("chatid"), ":", msg.get("text"))

@app.post("/webhooks/zapzap")
async def webhook(req: Request, tarefas: BackgroundTasks):
    evento = await req.json()
    tarefas.add_task(tratar, evento)  # trabalha depois de responder
    return {"ok": True}