Bot de WhatsApp con Claude: agente con la API de Anthropic

Un agente de Claude en WhatsApp es un webhook de FastAPI que verifica cada mensaje entrante, llama a la Messages API de Anthropic con el historial del chat y una herramienta, y devuelve el texto final con una llamada a la API. Este tutorial lo construye en Python con Claude Sonnet 5.5, una herramienta que consulta pedidos y prompt caching en el system prompt.

¿En qué se diferencia de la versión con OpenAI?

Este tutorial comparte la base con la versión del bot de WhatsApp con ChatGPT: el mismo webhook firmado, el mismo inbox guardado, la misma llamada de respuesta. Lo que cambia es el lado del modelo: el uso de herramientas, el prompt caching y el ajuste de esfuerzo de Claude. Para ver en qué se distingue un agente que usa herramientas de un bot basado en reglas, consulta la comparativa de chatbot de WhatsApp vs agente de IA. Si nunca has montado un receptor, empieza por el tutorial de OpenAI y lee la guía del webhook de WhatsApp para los reintentos, el orden y las pruebas.

¿Qué necesitas para ejecutarlo?

Este tutorial necesita Python 3.10 o superior, que es lo que exige el SDK de Anthropic, una clave de API de Anthropic y un número de whasp con su clave de API.

pip install anthropic fastapi "uvicorn[standard]" httpx standardwebhooks

Pon las tres claves en .env. Registra tu URL pública con PUT /v1/webhook como en el tutorial de OpenAI; la respuesta contiene el secreto whsec_.

ANTHROPIC_API_KEY=sk-ant-...
WHASP_API_KEY=wsk_...
WHASP_WEBHOOK_SECRET=whsec_...

¿Cómo es la aplicación completa?

El agente entero es un solo archivo, app.py. Ejecútalo con uvicorn app:app --port 3000 --env-file .env.

import asyncio
import json
import os
import sqlite3
from contextlib import asynccontextmanager

import anthropic
import httpx
from fastapi import FastAPI, Request, Response
from standardwebhooks import Webhook, WebhookVerificationError

WHASP = "https://api.whasp.ai/v1"
MODEL = "claude-sonnet-5-5"
# Aquí van prompts reales: por debajo de 512 tokens no se cachea nada.
SYSTEM = """Eres Ava, la asistente de Fernwood, una pequeña tienda online de plantas.
Chateas con clientes por WhatsApp: mensajes cortos en texto plano, sin Markdown.
Responde en el idioma del cliente. Para preguntas sobre pedidos, usa get_order_status."""

TOOLS = [
    {
        "name": "get_order_status",
        "description": "Consulta el estado y la fecha estimada de entrega de uno de los pedidos del propio "
        "cliente a partir del número de pedido (formato FW-1234). Úsala siempre que pregunte dónde está un pedido.",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "Número de pedido, p. ej. FW-1042"}},
            "required": ["order_id"],
        },
    }
]
ORDERS = {  # sustituto de tu base de datos o de la API de tu tienda
    "FW-1042": {"phone": "+15555550111", "status": "shipped", "eta": "2026-10-23"},
}

claude = anthropic.AsyncAnthropic()  # lee ANTHROPIC_API_KEY
whasp = httpx.AsyncClient(
    base_url=WHASP, timeout=15, headers={"Authorization": f"Bearer {os.environ['WHASP_API_KEY']}"}
)
wh = Webhook(os.environ["WHASP_WEBHOOK_SECRET"])  # whsec_...

db = sqlite3.connect("agent.db", isolation_level=None)
db.executescript("""
CREATE TABLE IF NOT EXISTS inbox (id TEXT PRIMARY KEY, body TEXT NOT NULL, done INTEGER NOT NULL DEFAULT 0);
CREATE TABLE IF NOT EXISTS turns (
  id INTEGER PRIMARY KEY, ref TEXT UNIQUE, chat TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL);
""")

locks: dict[str, asyncio.Lock] = {}  # un carril por chat
tasks: set[asyncio.Task] = set()


@asynccontextmanager
async def lifespan(app: FastAPI):
    for (body,) in db.execute("SELECT body FROM inbox WHERE done = 0 ORDER BY rowid").fetchall():
        spawn(json.loads(body))  # reprocesa lo que dejó pendiente una caída
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/whasp/webhook")
async def webhook(request: Request):
    body = await request.body()  # bytes en bruto: verifica antes de parsear
    try:
        event = wh.verify(body, dict(request.headers))
    except WebhookVerificationError:
        return Response("Invalid signature", status_code=401)
    cur = db.execute("INSERT OR IGNORE INTO inbox (id, body) VALUES (?, ?)", (event["id"], body.decode()))
    if cur.rowcount:  # 0 significa que ya tenemos este id de evento
        spawn(event)
    return Response(status_code=204)


def spawn(event: dict):
    task = asyncio.create_task(handle(event))
    tasks.add(task)
    task.add_done_callback(tasks.discard)


async def handle(event: dict):
    message = event["data"].get("message")
    message = message if isinstance(message, dict) else {}  # webhook.test envía una cadena
    async with locks.setdefault(message.get("chat", "system"), asyncio.Lock()):
        try:
            if event["type"] == "message.received":
                await reply(message)
            db.execute("UPDATE inbox SET done = 1 WHERE id = ?", (event["id"],))
        except Exception as err:  # queda con done = 0, se reprocesa en el siguiente arranque
            print(event["id"], repr(err))


async def reply(m: dict):
    chat, key = m["chat"], f"reply-{m['id']}"
    text = (m.get("text") or "").strip()
    if not text:
        return  # solo multimedia: nada que responder

    save_turn(m["id"], chat, "user", text)
    answer = await run_agent(chat, load_history(chat))
    await send(chat, answer[:4096], key)
    save_turn(key, chat, "assistant", answer)


def save_turn(ref: str, chat: str, role: str, content: str):
    db.execute("INSERT OR IGNORE INTO turns (ref, chat, role, content) VALUES (?, ?, ?, ?)", (ref, chat, role, content))


def load_history(chat: str) -> list[dict]:
    rows = db.execute(
        "SELECT role, content FROM (SELECT * FROM turns WHERE chat = ? ORDER BY id DESC LIMIT 20) ORDER BY id",
        (chat,),
    ).fetchall()
    while rows and rows[0][0] != "user":
        rows.pop(0)  # la ventana debe empezar con un turno de user
    return [{"role": role, "content": content} for role, content in rows]


async def run_agent(chat: str, messages: list[dict]) -> str:
    for _ in range(5):  # tope de rondas de herramientas
        response = await claude.messages.create(
            model=MODEL,
            max_tokens=2048,  # los tokens de thinking cuentan para este límite
            output_config={"effort": "medium"},
            system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],
            tools=TOOLS,
            messages=messages,
        )
        u = response.usage
        print(f"tokens: cache write {u.cache_creation_input_tokens}, read {u.cache_read_input_tokens}, plain {u.input_tokens}")
        if response.stop_reason != "tool_use":
            text = "".join(b.text for b in response.content if b.type == "text").strip()
            return text or "Lo siento, no pude responder a eso."
        # Devuelve intacto el turno del asistente (thinking incluido) y después los resultados.
        messages.append({"role": "assistant", "content": response.content})
        results = [run_tool(chat, b) for b in response.content if b.type == "tool_use"]
        messages.append({"role": "user", "content": results})
    return "Lo siento, esto requirió demasiados pasos."


def run_tool(chat: str, block) -> dict:
    result = {"type": "tool_result", "tool_use_id": block.id}
    if block.name != "get_order_status":
        return {**result, "content": f"Unknown tool {block.name}", "is_error": True}
    order = ORDERS.get(block.input.get("order_id", ""))
    if order and order["phone"] == chat:  # el modelo elige el id; tu código comprueba quién pregunta
        return {**result, "content": json.dumps({k: v for k, v in order.items() if k != "phone"})}
    return {**result, "content": "No hay ningún pedido con ese número para este cliente. Pídele que lo revise.", "is_error": True}


async def send(to: str, text: str, key: str):
    for attempt in range(1, 5):
        res = await whasp.post("/messages", json={"to": to, "text": text}, headers={"Idempotency-Key": key})
        if res.is_success:
            return
        if res.status_code in (403, 409):  # baja, chat no abierto, número no disponible
            print("whasp refused the reply:", res.status_code, res.text)
            return
        if (res.status_code != 429 and res.status_code < 500) or attempt == 4:
            raise RuntimeError(f"whasp {res.status_code}")
        await asyncio.sleep(2**attempt)  # 2 s, 4 s, 8 s

¿Cómo funciona la parte del webhook en FastAPI?

La parte del webhook de app.py es el receptor de TypeScript llevado a FastAPI. request.body() devuelve los bytes en bruto, Webhook.verify lanza WebhookVerificationError ante una firma incorrecta o una marca de tiempo a más de cinco minutos, y INSERT OR IGNORE con cur.rowcount convierte una entrega duplicada en una operación vacía que aun así recibe un 204. Cada evento nuevo se ejecuta como su propia tarea de asyncio, así que el manejador responde 204 sin esperar al modelo y el límite de 10 segundos de whasp nunca importa. El conjunto tasks guarda una referencia para que el recolector de basura no elimine la tarea a mitad de camino, un asyncio.Lock por chat pone en cola en orden los mensajes de una persona, y lifespan reprocesa las filas sin terminar tras una caída.

Dos simplificaciones: no trata STOP como caso especial (el siguiente envío a un chat dado de baja recibe 403 y send lo descarta, así que un STOP cuesta una llamada al modelo; el tutorial de OpenAI muestra la salida anticipada), y se salta los mensajes sin texto.

¿Cómo llama Claude a una herramienta?

El uso de herramientas es un bucle alrededor de la Messages API, y run_agent es ese bucle escrito a mano. Describes una herramienta con un nombre, una descripción y un JSON Schema. Cuando Claude la quiere, la respuesta trae stop_reason: "tool_use" y uno o más bloques tool_use, cada uno con un id, un name y un input. Tu código ejecuta la herramienta y devuelve el resultado como un bloque tool_result en un mensaje de usuario. Claude continúa entonces con otra llamada o con el texto final.

Un turno de la aplicación de ejemplo es así. El cliente escribe “¿Dónde está mi pedido FW-1042?”. La primera petición termina con un bloque tool_use de get_order_status con {"order_id": "FW-1042"}. run_tool devuelve {"status": "shipped", "eta": "2026-10-23"}. La segunda petición termina con end_turn y un texto como “Ya salió y debería llegar el 23 de octubre.”. Son dos llamadas a la API y un mensaje de WhatsApp.

El bucle sigue cinco reglas de la documentación de llamadas a herramientas y de la documentación de thinking de Anthropic:

1. Devuelve intacto el turno del asistente. Sonnet 5.5 piensa por defecto, y dentro de un turno de uso de herramientas sus bloques de thinking deben volver completos y sin modificar. Añadir response.content tal cual lo consigue.
2. Responde a cada bloque tool_use, con los resultados primero. Los bloques tool_result van primero en el mensaje de usuario. Claude puede llamar a varias herramientas a la vez, así que el código recorre todos los bloques.
3. Informa de los fallos con is_error. Una consulta que no encuentra nada devuelve un mensaje instructivo con is_error activado, y Claude se lo dice al cliente en lugar de inventar un estado.
4. Pon un tope al bucle. Cinco rondas sobran para un chat. Después, el agente se disculpa en lugar de gastar tokens.
5. Mantén el bucle en memoria. Solo se guardan el texto del cliente y la respuesta final de Claude, que es todo lo que necesitan los turnos posteriores. Anthropic permite omitir el thinking de los turnos anteriores fuera del uso de herramientas.

La autorización vive en tu código. El modelo elige order_id, pero run_tool toma el número de teléfono del webhook y lo compara con el propietario del pedido, así que preguntar por el pedido de otra persona da el mismo error que un número equivocado. Trata las entradas de las herramientas como entradas de usuario. Trata también como no fiables los resultados de herramientas que traigan texto externo, porque las instrucciones ocultas en ellos pueden intentar manipular a Claude.

Los efectos secundarios deben ser idempotentes. whasp entrega los eventos al menos una vez, y una repetición vuelve a ejecutar el bucle entero, llamadas a herramientas incluidas. Una consulta de solo lectura es inofensiva. Un reembolso o una reserva necesitan una clave construida con el id del mensaje entrante, para que la segunda ejecución encuentre el resultado de la primera.

Deja tool_choice con su valor por defecto: Anthropic incluye a Sonnet 5.5 entre los modelos que no admiten el uso forzado de herramientas con thinking activado, así que el system prompt le indica a Claude cuándo usar la herramienta.

Existen dos atajos. Poner strict: true y additionalProperties: false en una herramienta hace que las entradas de Claude coincidan exactamente con tu esquema. El tool runner del SDK, todavía en beta, escribe el bucle por ti. Aquí se mantiene el bucle escrito a mano porque la autorización, el informe de errores y el tope de rondas son las partes que quieres controlar.

¿Cómo reduce el coste el prompt caching?

El prompt caching permite a Claude reutilizar el inicio de un prompt que ya ha procesado. El orden de composición es herramientas, luego system, luego mensajes, así que un punto de corte cache_control en el último bloque de system cachea juntas las definiciones de herramientas y el system prompt.

En Claude Sonnet 5.5, la entrada cuesta $2 por millón de tokens, una escritura en caché $2,50 y una lectura de caché $0,10. Un prompt de 3.000 tokens cuesta $0,006 sin caché y $0,0003 como lectura de caché. Tres reglas deciden si consigues ese descuento:

• Tamaño mínimo. Sonnet 5.5 cachea prompts de al menos 512 tokens. Los más cortos se procesan con normalidad y la API no devuelve ningún error, así que el prompt de ejemplo de app.py no se cacheará. Pon en él tus políticas y preguntas frecuentes reales y luego mira los tres números que imprime el código: cache write (escritura en caché), read (lectura de caché) y plain (entrada normal).
• Duración. Una entrada de caché dura 5 minutos por defecto y cada acierto la renueva; "ttl": "1h" cuesta más. Una escritura de 5 minutos cuesta un 25% más que la entrada normal, así que el caché compensa tras una sola lectura. Un system prompt compartido por todos los usuarios se mantiene caliente mientras sigan llegando mensajes. El historial de un solo usuario rara vez lo está si la gente responde con minutos de diferencia, por eso este código cachea solo la parte compartida. La opción cache_control de nivel superior mueve el punto de corte hacia delante a medida que crece una conversación y encaja con los intercambios rápidos.
• Coincidencia exacta. Cualquier cambio anterior al punto de corte crea un caché nuevo: herramientas editadas, otro tool_choice, un ajuste distinto de esfuerzo o de thinking, o una marca de tiempo en el system prompt. Pon los datos de cada usuario en un segundo bloque de system, después del cacheado.

¿Qué modelo y qué esfuerzo de Claude conviene usar en un agente de WhatsApp?

Claude Sonnet 5.5 (claude-sonnet-5-5) es el modelo por defecto en app.py. Al 19 de octubre de 2026, la descripción general de modelos de Anthropic lo lista a $2 de entrada y $10 de salida por millón de tokens, lo describe como la mejor combinación de velocidad e inteligencia y se compromete a mantenerlo disponible al menos hasta el 28 de septiembre de 2027. claude-haiku-5-5, desde $0,10 y $0,50, es el modelo más rápido y encaja con respuestas simples de alto volumen. Cambia la constante MODEL para probarlo.

El esfuerzo y el thinking deciden la latencia. Sonnet 5.5 usa por defecto el esfuerzo high, y Anthropic sugiere empezar las cargas de trabajo de chat en medium o low; el código usa medium. El thinking viene activado por defecto y cuenta para max_tokens aunque su texto no se devuelva, y por eso max_tokens es 2048 y no 300. Para saltarte el thinking inicial, envía thinking={"type": "between_tools"}: Sonnet 5.5 rechaza disabled, y between_tools funciona con esfuerzo high o inferior. Fija el esfuerzo una vez y no lo cambies, porque cambiarlo crea un caché nuevo.

El SDK reintenta por defecto, dos veces, los errores de conexión y las respuestas 408, 409, 429 y 5xx. Si una llamada sigue fallando, el evento conserva done = 0 y se reprocesa en el siguiente arranque. Una respuesta que termina en max_tokens o en una negativa puede volver sin texto, así que run_agent envía una línea de respaldo corta en lugar de un mensaje vacío.

¿Qué construir después?

Dale al agente memoria más allá de 20 turnos con historial, resúmenes y datos, afina su voz con prompts para un agente de IA en WhatsApp que suene humano, o dale a Claude un número de WhatsApp como herramienta con un servidor MCP.