Bot de WhatsApp com Claude: agente com a API da Anthropic

Um agente Claude no WhatsApp é um webhook FastAPI que verifica cada mensagem recebida, chama a Messages API da Anthropic com o histórico do chat e uma ferramenta, e devolve o texto final com uma chamada de API. Este tutorial o monta em Python com o Claude Sonnet 5.5, uma ferramenta que consulta pedidos e prompt caching no system prompt.

Qual a diferença para a versão com a OpenAI?

Este tutorial divide a infraestrutura com a versão com a OpenAI: o mesmo webhook assinado, a mesma caixa de entrada gravada, a mesma chamada de resposta. O que muda é o lado do modelo: tool use, prompt caching e o ajuste de effort do Claude. Para ver como um agente que chama ferramentas se diferencia de um bot baseado em regras, leia chatbot vs agente de IA. Se você nunca montou um receptor, comece pelo tutorial da OpenAI e leia o guia de webhooks para ver novas tentativas, ordem e testes.

O que você precisa para rodar?

Este tutorial precisa do Python 3.10 ou mais novo, exigido pelo SDK da Anthropic, de uma chave da API da Anthropic e de um número da whasp com chave de API.

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

Coloque as três chaves no .env. Registre a sua URL pública com PUT /v1/webhook, como no tutorial da OpenAI; a resposta traz o segredo whsec_.

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

Como fica o app completo?

O agente inteiro é um arquivo, app.py. Rode com 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"
# Os prompts reais entram aqui: abaixo de 512 tokens, nada entra no cache.
SYSTEM = """Você é a Ava, a assistente da Fernwood, uma pequena loja virtual de plantas.
Você conversa com os clientes no WhatsApp: mensagens curtas em texto simples, sem Markdown.
Responda no idioma do cliente. Para dúvidas sobre pedidos, use get_order_status."""

TOOLS = [
    {
        "name": "get_order_status",
        "description": "Consulta o status e a previsão de entrega de um dos pedidos do próprio cliente "
        "pelo número do pedido (formato FW-1234). Use sempre que ele perguntar onde está um pedido.",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "Número do pedido, ex.: FW-1042"}},
            "required": ["order_id"],
        },
    }
]
ORDERS = {  # no lugar do seu banco de dados ou da API da loja
    "FW-1042": {"phone": "+15555550111", "status": "shipped", "eta": "2026-10-23"},
}

claude = anthropic.AsyncAnthropic()  # lê 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] = {}  # uma fila 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))  # reprocessa o que uma queda deixou pendente
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/whasp/webhook")
async def webhook(request: Request):
    body = await request.body()  # bytes brutos: verifique antes de fazer o parse
    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 já temos 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 {}  # o webhook.test envia uma string
    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:  # continua com done = 0 e é reprocessado na próxima inicialização
            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  # só mídia: nada a 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)  # a janela precisa começar em um turno do usuário
    return [{"role": role, "content": content} for role, content in rows]


async def run_agent(chat: str, messages: list[dict]) -> str:
    for _ in range(5):  # limite de rodadas de ferramenta
        response = await claude.messages.create(
            model=MODEL,
            max_tokens=2048,  # os tokens de thinking contam neste limite
            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 "Desculpe, não consegui responder a isso."
        # O turno do assistente volta intacto (thinking incluído), depois os 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 "Desculpe, isso exigiu passos demais."


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"Ferramenta desconhecida {block.name}", "is_error": True}
    order = ORDERS.get(block.input.get("order_id", ""))
    if order and order["phone"] == chat:  # o modelo escolhe o id; o seu código confere quem está perguntando
        return {**result, "content": json.dumps({k: v for k, v in order.items() if k != "phone"})}
    return {**result, "content": "Nenhum pedido com esse número para este cliente. Peça para ele conferir.", "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):  # descadastrado, chat não aberto, número indisponível
            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

Como funciona a metade do webhook no FastAPI?

A metade do webhook do app.py é o receptor em TypeScript portado para o FastAPI. request.body() devolve os bytes brutos, Webhook.verify lança WebhookVerificationError quando a assinatura é inválida ou o timestamp está a mais de cinco minutos de diferença, e INSERT OR IGNORE com cur.rowcount transforma uma entrega duplicada em uma operação nula que ainda recebe 204. Cada evento novo roda como uma tarefa asyncio própria, então o handler responde 204 sem esperar o modelo e o limite de 10 segundos da whasp nunca entra em jogo. O conjunto tasks mantém uma referência para que a tarefa não seja coletada pelo garbage collector no meio do caminho, um asyncio.Lock por chat coloca em fila, em ordem, as mensagens de uma pessoa, e o lifespan reprocessa as linhas não concluídas depois de uma queda.

Duas simplificações: ele não trata o STOP como caso especial (o próximo envio para um chat descadastrado recebe 403 e o send o descarta, então um STOP custa uma chamada ao modelo; o tutorial da OpenAI mostra a saída antecipada) e ignora mensagens sem texto.

Como o Claude chama uma ferramenta?

Tool use é um loop em torno da Messages API, e run_agent é esse loop escrito à mão. Você descreve uma ferramenta com um nome, uma descrição e um JSON Schema. Quando o Claude a quer, a resposta traz stop_reason: "tool_use" e um ou mais blocos tool_use, cada um com um id, um name e um input. Seu código executa a ferramenta e devolve a saída como um bloco tool_result em uma mensagem de usuário. O Claude então continua com outra chamada ou com o texto final.

Um turno do app de exemplo funciona assim. O cliente escreve “Onde está meu pedido FW-1042?”. A primeira requisição termina com um bloco tool_use para get_order_status com {"order_id": "FW-1042"}. O run_tool devolve {"status": "shipped", "eta": "2026-10-23"}. A segunda requisição termina com end_turn e um texto como “Ele foi enviado e deve chegar em 23 de outubro.” São duas chamadas de API e uma mensagem de WhatsApp.

O loop segue cinco regras da documentação de tool calls e da documentação de thinking da Anthropic:

1. Devolva o turno do assistente sem alterações. O Sonnet 5.5 pensa por padrão, e dentro de um turno de tool use os blocos de thinking precisam voltar completos e sem modificação. Acrescentar response.content como está faz isso.
2. Responda a cada bloco tool_use, com os resultados primeiro. Os blocos tool_result vêm primeiro na mensagem de usuário. O Claude pode chamar várias ferramentas de uma vez, então o código percorre todos os blocos.
3. Reporte falhas com is_error. Uma consulta que não acha nada devolve uma mensagem instrutiva com is_error ativado, e o Claude avisa o cliente em vez de inventar um status.
4. Limite o loop. Cinco rodadas bastam para chat. Depois disso, o agente pede desculpas em vez de gastar tokens.
5. Mantenha o loop só na memória. Só o texto do cliente e a resposta final do Claude são salvos, o que é tudo de que os turnos seguintes precisam. A Anthropic permite deixar de fora o thinking dos turnos anteriores quando não há tool use.

A autorização mora no seu código. O modelo escolhe o order_id, mas o run_tool pega o número de telefone do webhook e o compara com o dono do pedido, então perguntar sobre o pedido de outra pessoa gera o mesmo erro de um número errado. Trate as entradas das ferramentas como entrada de usuário. Trate também como não confiáveis os resultados de ferramentas que trazem texto de fora, porque instruções escondidas neles podem tentar desviar o Claude.

Efeitos colaterais precisam ser idempotentes. A whasp entrega eventos pelo menos uma vez, e um replay roda o loop inteiro de novo, chamadas de ferramenta incluídas. Uma consulta somente de leitura é inofensiva. Um reembolso ou uma reserva precisa de uma chave montada com o id da mensagem recebida, para que a segunda execução encontre o resultado da primeira.

Deixe o tool_choice no padrão: a Anthropic lista o Sonnet 5.5 entre os modelos que não aceitam tool use forçado com thinking ligado, então o system prompt diz ao Claude quando usar a ferramenta.

Existem dois atalhos. Definir strict: true e additionalProperties: false em uma ferramenta faz as entradas do Claude corresponderem exatamente ao seu schema. O tool runner do SDK, ainda em beta, escreve o loop para você. O loop escrito à mão fica aqui porque a autorização, o reporte de erros e o limite de rodadas são as partes que você quer controlar.

Como o prompt caching reduz o custo?

O prompt caching permite que o Claude reaproveite o começo de um prompt que já processou. A ordem de renderização é ferramentas, depois system, depois mensagens, então um ponto cache_control no último bloco de system faz o cache das definições das ferramentas e do system prompt juntos.

No Claude Sonnet 5.5, a entrada custa $2 por milhão de tokens, uma escrita de cache $2,50 e uma leitura de cache $0,10. Um prompt de 3.000 tokens custa $0,006 sem cache e $0,0003 como leitura de cache. Três regras decidem se você leva esse desconto:

• Tamanho mínimo. O Sonnet 5.5 faz cache de prompts de pelo menos 512 tokens. Os mais curtos são processados normalmente e a API não devolve erro, então o prompt de exemplo do app.py não entra no cache. Ponha nele as suas políticas e o FAQ reais e depois confira os três números que o código imprime: escrita de cache, leitura de cache e entrada comum.
• Duração. Uma entrada de cache dura 5 minutos por padrão e cada acerto a renova; "ttl": "1h" custa mais. Uma escrita de 5 minutos custa 25% a mais que a entrada comum, então o cache se paga depois de uma leitura. Um system prompt compartilhado por todos os usuários continua quente enquanto as mensagens chegam. O histórico de um usuário raramente se mantém quente se as pessoas respondem com minutos de intervalo, e por isso este código faz cache só da parte compartilhada. A opção cache_control de nível superior move o ponto para a frente conforme a conversa cresce e serve para trocas rápidas.
• Correspondência exata. Qualquer mudança antes do ponto inicia um cache novo: ferramentas editadas, um tool_choice diferente, um novo ajuste de effort ou de thinking, ou um timestamp no system prompt. Ponha os detalhes de cada usuário em um segundo bloco de system, depois do que vai para o cache.

Qual modelo e qual effort do Claude usar em um agente de WhatsApp?

O Claude Sonnet 5.5 (claude-sonnet-5-5) é o padrão no app.py. Em 19 de outubro de 2026, a visão geral de modelos da Anthropic o lista a $2 de entrada e $10 de saída por milhão de tokens, o descreve como a melhor combinação de velocidade e inteligência e se compromete a mantê-lo disponível até pelo menos 28 de setembro de 2027. O claude-haiku-5-5, a partir de $0,10 e $0,50, é o modelo mais rápido e serve para respostas simples e de alto volume. Troque a constante MODEL para testá-lo.

Effort e thinking decidem a latência. O Sonnet 5.5 usa effort high por padrão, e a Anthropic sugere começar cargas de chat em medium ou low; o código usa medium. O thinking vem ligado por padrão e conta para o max_tokens mesmo quando o texto não é devolvido, e por isso o max_tokens é 2.048 e não 300. Para pular o thinking inicial, envie thinking={"type": "between_tools"}: o Sonnet 5.5 rejeita disabled, e between_tools funciona com effort high ou menor. Defina o effort uma vez e mantenha, porque mudá-lo inicia um cache novo.

O SDK tenta de novo erros de conexão, 408, 409, 429 e respostas 5xx duas vezes por padrão. Se uma chamada ainda assim falhar, o evento continua com done = 0 e é reprocessado na próxima inicialização. Uma resposta que termina em max_tokens ou em uma recusa pode voltar sem texto, então o run_agent envia uma linha curta de contingência em vez de uma mensagem vazia.

O que construir em seguida?

Dê ao agente memória além de 20 turnos com resumos e fatos, afine a voz dele com prompts escritos para o WhatsApp ou entregue um número de WhatsApp ao Claude como ferramenta com um servidor MCP.