Cómo integrar un API REST de WhatsApp en una app Python (FastAPI)

¿Tus clientes escriben por WhatsApp y no reciben respuesta? Descubre cómo una solución multiagente evita mensajes perdidos, mejora tiempos de respuesta y convierte más ventas.
CONTÁCTANOS

1) Qué vas a construir

Un microservicio FastAPI que:
- Expone un webhook para recibir mensajes/eventos de WhatsApp.
- Ofrece un endpoint interno POST /send para enviar mensajes vía REST al proveedor (p.ej., WhatzMeAPI o Meta WhatsApp Cloud).
- Maneja verificación de firmas, reintentos, logs y plantillas (templates).

2) Requisitos

Python 3.10+
Cuenta y credenciales del proveedor:
- WhatzMeAPI (token y base URL) o
- WhatsApp Cloud API (token permanente, phone_number_id, app_secret, verify_token).
Un Webhook público (en local puedes usar ngrok o Cloudflare Tunnel).
Conocimiento básico de HTTP/JSON.

3) Estructura del proyecto

whatsapp-python/
├─ app.py
├─ service.py
├─ security.py
├─ models.py
├─ requirements.txt
└─ .env

requirements.txt

fastapi==0.115.0
uvicorn[standard]==0.30.6
python-dotenv==1.0.1
httpx==0.27.2
pydantic==2.9.2

.env (ejemplo WhatzMeAPI)

PROVIDER=WHATZMEAPI
WZ_API_BASE=https://api.whatzmeapi.com
WZ_TOKEN=tu_token_de_api
VERIFY_TOKEN=mi_verify_token_unico
APP_SECRET=opcional_si_el_proveedor_firma

.env (ejemplo Meta Cloud API)

PROVIDER=META
META_API_BASE=https://graph.facebook.com/v20.0
META_TOKEN=EAAG... (token)
PHONE_NUMBER_ID=123456789012345
VERIFY_TOKEN=mi_verify_token_unico
APP_SECRET=tu_app_secret

Nota: Las rutas exactas del proveedor pueden variar. En WhatzMeAPI, consulta su documentación para confirmar endpoints (envío de mensajes, subida de media, plantillas, etc.).

4) Modelos y helpers

models.py

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any, List

class SendTextRequest(BaseModel):
    to: str = Field(..., description="Número destino en formato E.164, ej. 5215551112233")
    text: str

class SendTemplateRequest(BaseModel):
    to: str
    template_name: str
    language: str = "es_MX"
    components: Optional[List[Dict[str, Any]]] = None  # parameters, headers, buttons, etc.

class ProviderResponse(BaseModel):
    ok: bool
    provider_id: Optional[str] = None
    status: str = "queued"
    raw: Optional[Dict[str, Any]] = None

security.py (verificación de firma; útil si el proveedor la envía, Meta usa X-Hub-Signature-256)

import hmac, hashlib, base64, os

APP_SECRET = os.getenv("APP_SECRET", "")

def verify_meta_signature(body: bytes, header_signature: str) -> bool:
    if not APP_SECRET or not header_signature:
        return True  # si no aplicas firma, no bloquees
    try:
        # header_signature: "sha256=..."
        method, sig = header_signature.split("=", 1)
        mac = hmac.new(APP_SECRET.encode(), msg=body, digestmod=hashlib.sha256)
        expected = mac.hexdigest()
        return hmac.compare_digest(expected, sig)
    except Exception:
        return False

5) Cliente del proveedor (WhatzMeAPI o Meta)

service.py

import os, httpx
from typing import Dict, Any
from models import ProviderResponse

PROVIDER = os.getenv("PROVIDER", "WHATZMEAPI").upper()

# WhatzMeAPI
WZ_API_BASE = os.getenv("WZ_API_BASE")
WZ_TOKEN = os.getenv("WZ_TOKEN")

# Meta Cloud
META_API_BASE = os.getenv("META_API_BASE")
META_TOKEN = os.getenv("META_TOKEN")
PHONE_NUMBER_ID = os.getenv("PHONE_NUMBER_ID")

HEADERS_WZ = {"Authorization": f"Bearer {WZ_TOKEN}"} if WZ_TOKEN else {}
HEADERS_META = {"Authorization": f"Bearer {META_TOKEN}"} if META_TOKEN else {}

async def send_text(to: str, text: str) -> ProviderResponse:
    async with httpx.AsyncClient(timeout=20) as client:
        if PROVIDER == "WHATZMEAPI":
            # ⚠️ Ajusta el endpoint conforme a la documentación de WhatzMeAPI
            url = f"{WZ_API_BASE}/messages"
            payload = {
                "to": to,
                "type": "text",
                "text": {"body": text}
            }
            r = await client.post(url, json=payload, headers=HEADERS_WZ)
            data = r.json()
            return ProviderResponse(ok=r.is_success, provider_id=data.get("id"), raw=data)

        elif PROVIDER == "META":
            # Meta: POST /{phone_number_id}/messages
            url = f"{META_API_BASE}/{PHONE_NUMBER_ID}/messages"
            payload = {
                "messaging_product": "whatsapp",
                "to": to,
                "type": "text",
                "text": {"preview_url": False, "body": text}
            }
            r = await client.post(url, json=payload, headers=HEADERS_META)
            data = r.json()
            msg_id = None
            try:
                msg_id = data["messages"][0]["id"]
            except Exception:
                pass
            return ProviderResponse(ok=r.is_success, provider_id=msg_id, raw=data)

        else:
            raise ValueError("Proveedor no soportado")

async def send_template(to: str, template_name: str, language: str = "es_MX", components=None) -> ProviderResponse:
    components = components or []
    async with httpx.AsyncClient(timeout=20) as client:
        if PROVIDER == "WHATZMEAPI":
            url = f"{WZ_API_BASE}/messages/templates"
            payload = {
                "to": to,
                "type": "template",
                "template": {
                    "name": template_name,
                    "language": language,
                    "components": components
                }
            }
            r = await client.post(url, json=payload, headers=HEADERS_WZ)
            data = r.json()
            return ProviderResponse(ok=r.is_success, provider_id=data.get("id"), raw=data)

        elif PROVIDER == "META":
            url = f"{META_API_BASE}/{PHONE_NUMBER_ID}/messages"
            payload = {
                "messaging_product": "whatsapp",
                "to": to,
                "type": "template",
                "template": {
                    "name": template_name,
                    "language": {"code": language.replace("_", "-")},
                    "components": components
                }
            }
            r = await client.post(url, json=payload, headers=HEADERS_META)
            data = r.json()
            msg_id = None
            try:
                msg_id = data["messages"][0]["id"]
            except Exception:
                pass
            return ProviderResponse(ok=r.is_success, provider_id=msg_id, raw=data)

        else:
            raise ValueError("Proveedor no soportado")

6) FastAPI: Webhook y endpoints internos

app.py

import os, json
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import PlainTextResponse, JSONResponse
from dotenv import load_dotenv
from models import SendTextRequest, SendTemplateRequest
from service import send_text, send_template
from security import verify_meta_signature

load_dotenv()

VERIFY_TOKEN = os.getenv("VERIFY_TOKEN", "secret")
PROVIDER = os.getenv("PROVIDER", "WHATZMEAPI").upper()

app = FastAPI(title="WhatsApp Integration (FastAPI)")

@app.get("/health")
def health():
    return {"status": "ok", "provider": PROVIDER}

# === Webhook verification (GET) ===
@app.get("/webhook", response_class=PlainTextResponse)
def verify(mode: str = "", challenge: str = "", verify_token: str = ""):
    # Meta envía: hub.mode, hub.challenge, hub.verify_token
    # Mapeamos parámetros para comodidad
    token = verify_token or ""
    if token == VERIFY_TOKEN:
        return challenge
    raise HTTPException(status_code=403, detail="Verification failed")

# === Webhook receiver (POST) ===
@app.post("/webhook")
async def webhook(request: Request):
    raw = await request.body()
    # Verifica firma si aplica (Meta: X-Hub-Signature-256)
    signature = request.headers.get("X-Hub-Signature-256", "")
    if PROVIDER == "META":
        if not verify_meta_signature(raw, signature):
            raise HTTPException(status_code=401, detail="Invalid signature")

    payload = json.loads(raw.decode("utf-8") or "{}")

    # LOG BÁSICO
    print("📥 Webhook payload:", json.dumps(payload, ensure_ascii=False))

    # Normaliza eventos (ejemplo Meta)
    # En WhatzMeAPI la estructura puede ser similar (mensajes entrantes y statuses)
    try:
        entries = payload.get("entry", [])
        for entry in entries:
            changes = entry.get("changes", [])
            for ch in changes:
                value = ch.get("value", {})
                messages = value.get("messages", [])
                statuses = value.get("statuses", [])

                for m in messages:
                    from_ = m.get("from")
                    msg_type = m.get("type")
                    if msg_type == "text":
                        text = m.get("text", {}).get("body", "")
                        print(f"👤 {from_}: {text}")
                        # Ejemplo de auto-reply
                        # await send_text(from_, "¡Gracias! Recibimos tu mensaje 🤖")

                for st in statuses:
                    print("📦 Status:", st.get("status"), "MsgID:", st.get("id"))

    except Exception as e:
        print("⚠️ Error parseando webhook:", e)

    return JSONResponse({"ok": True})

# === Endpoint interno para enviar texto ===
@app.post("/send")
async def api_send(req: SendTextRequest):
    res = await send_text(req.to, req.text)
    if not res.ok:
        raise HTTPException(status_code=502, detail=res.raw or "Send failed")
    return res

# === Endpoint interno para enviar plantilla ===
@app.post("/send-template")
async def api_send_template(req: SendTemplateRequest):
    res = await send_template(req.to, req.template_name, req.language, req.components)
    if not res.ok:
        raise HTTPException(status_code=502, detail=res.raw or "Send failed")
    return res

7) Levantar el servicio

python -m venv .venv
source .venv/bin/activate  # (Windows: .venv\Scripts\activate)
pip install -r requirements.txt
uvicorn app:app --reload --port 8000

Prueba salud: GET http://localhost:8000/health
Expón tu webhook con ngrok http 8000 y registra la URL pública /webhook en tu proveedor.

8) Enviar mensajes (pruebas rápidas)

Texto

curl -X POST http://localhost:8000/send \
  -H "Content-Type: application/json" \
  -d '{"to": "5215551112233", "text": "Hola desde Python 👋"}'

Plantilla (Meta)

curl -X POST http://localhost:8000/send-template \
  -H "Content-Type: application/json" \
  -d '{
    "to": "5215551112233",
    "template_name": "hello_world",
    "language": "es_MX",
    "components": [
      {"type":"body", "parameters":[{"type":"text","text":"Gisselle"}]}
    ]
  }'

En WhatzMeAPI, ajusta template_name, language y el formato de components según su documentación.

9) Buenas prácticas (producción)

Seguridad
- Verifica firmas del proveedor (Meta: X-Hub-Signature-256).
- Valida verify_token en handshake.
- Cifra/rota tokens (Vault, AWS Secrets Manager, GCP Secret Manager).
Idempotencia & reintentos
- WhatsApp puede reenviar eventos. Guarda message_id/status_id y evita duplicados.
- Implementa backoff exponencial en reintentos 429/5xx.
Observabilidad
- Log JSON con messageId, from, type, status.
- Trazas distribuidas (trace-id) si integras con más microservicios.
Plantillas & Opt-ins
- Usa plantillas aprobadas para iniciar conversación fuera de la ventana de 24h.
- Respeta políticas de la plataforma (consentimiento, marketing).
Media & Adjuntos
- Soporta imágenes, documentos, audio. Subida previa (media) y luego id en el mensaje.
Escalabilidad
- Colas (RabbitMQ/SQS) para procesar webhooks sin bloquear.
- Desacopla respuestas automáticas con workers.
Testing
- Crea fixtures de webhooks y pruebas unitarias para el parser.
- Postman/Insomnia collections para endpoints de envío.

10) Errores comunes y cómo evitarlos

403 al verificar webhook: VERIFY_TOKEN no coincide.
401 al enviar: token inválido o expirado. Revisa cabeceras Authorization.
400 por plantilla: nombre/idioma/variables no coinciden con lo aprobado.
Duplicados: no idempotencia en el consumo del webhook.
429 (rate limit): implementa backoff y colas.

11) ¿Qué cambia entre WhatzMeAPI y Meta?

Base URL y rutas: cambia la URL y el path (/messages, /media, /templates).
Firmas/headers: Meta usa firma HMAC; tu gateway puede variar.
Plantillas: Meta exige plantillas aprobadas; un gateway puede simplificar el manejo.
Manejo de sesión: respeta siempre la ventana de 24 horas para mensajes iniciados por el usuario; fuera de ventana usa template.

Con WhatzMeAPI tienes la ventaja de una capa de simplificación y soporte local para integrarte rápido. Revisa su documentación para endpoints exactos de envío de texto, media, plantillas y estatus.

12) Extensiones útiles

Auto-replies inteligentes: conecta tu bot NLU o LLM (OpenAI, etc.) cuando llega un texto.
CRM: persiste conversaciones (contactos, tags, embudos).
Panel de control: métricas (mensajes enviados, entregados, leídos, errores).

13) Checklist de despliegue

Variables en .env correctamente establecidas.
Webhook accesible y validado.
Firma de proveedor activa (si aplica).
Logs estructurados y captura de errores.
Pruebas de envío de texto y plantilla exitosas.
Documentación interna (Postman + README).

Conoce WhatzMeAPI (API REST + Chatbot): WhatzMeApi

Optimiza tu entorno de trabajo: Extensiones VS Code que todo desarrollador necesita

Qué es WhatsApp Multiagente y Cómo Puede Duplicar las Ventas de tu Equipo con WhatzMeAPI

Automatiza tus tareas de desarrollo con scripts y herramientas DevOps

Microservicios: Buenas Prácticas que Todo Programador Debe Aplicar en Proyectos Reales

Cómo integrar un API REST de WhatsApp en una app Python (FastAPI)

1) Qué vas a construir

2) Requisitos

3) Estructura del proyecto

4) Modelos y helpers

5) Cliente del proveedor (WhatzMeAPI o Meta)

6) FastAPI: Webhook y endpoints internos

7) Levantar el servicio

8) Enviar mensajes (pruebas rápidas)

9) Buenas prácticas (producción)

10) Errores comunes y cómo evitarlos

11) ¿Qué cambia entre WhatzMeAPI y Meta?

12) Extensiones útiles

13) Checklist de despliegue

Tags:

About Author

Giss Trejo

Pérdida de información crítica: un problema común al usar WhatsApp sin control

Cómo usar el API de WhatsApp para confirmar pedidos en tiempo real

Buscar

Entradas recientes

Related Posts

Cómo integrar un API REST de WhatsApp en una app Python (FastAPI)

1) Qué vas a construir

2) Requisitos

3) Estructura del proyecto

4) Modelos y helpers

5) Cliente del proveedor (WhatzMeAPI o Meta)

6) FastAPI: Webhook y endpoints internos

7) Levantar el servicio

8) Enviar mensajes (pruebas rápidas)

9) Buenas prácticas (producción)

10) Errores comunes y cómo evitarlos

11) ¿Qué cambia entre WhatzMeAPI y Meta?

12) Extensiones útiles

13) Checklist de despliegue

Tags:

About Author

Giss Trejo

Buscar

Entradas recientes

Etiquetas

Related Posts