# Collegare WhatsApp a Claude: agente Python con API Anthropic

> Collega WhatsApp a Claude in Python con FastAPI: verifica il webhook, dai un tool a Claude, usa la cache sul system prompt e invia la risposta. Codice per Sonnet 5.5.

Source: https://whasp.ai/it/blog/collegare-whatsapp-claude-api-anthropic · 2026-10-20

## In breve

- Un agente Claude su WhatsApp è un webhook FastAPI, un ciclo attorno alla Messages API di Anthropic e una POST che invia la risposta, in circa 150 righe di Python.
- Il tool use è un ciclo: finché stop_reason è tool_use, esegui ogni blocco tool_use, rimanda intatto il turno dell'assistente e rispondi con i blocchi tool_result messi per primi in un unico messaggio user.
- L'autorizzazione sta nel codice del tool: il tool riceve il numero di telefono del cliente dal tuo server, mai dal modello.
- Un breakpoint cache_control sul system prompt mette in cache insieme tool e system prompt; su Claude Sonnet 5.5 una lettura dalla cache costa $0,10 per milione di token contro $2 senza cache, ma si mettono in cache solo prompt di almeno 512 token.
- whasp consegna gli eventi almeno una volta e un evento rigiocato rifà girare il ciclo, quindi i tool con effetti collaterali devono essere idempotenti.

Un agente Claude su WhatsApp è un webhook FastAPI che verifica ogni messaggio in arrivo, chiama la Messages API di Anthropic con la cronologia della chat e un tool, e rimanda il testo finale con una sola chiamata API. Questa guida lo costruisce in Python con Claude Sonnet 5.5, un tool che cerca gli ordini e il prompt caching sul system prompt.

## In cosa si differenzia dalla versione con OpenAI?

Questa guida condivide l'impianto con [la guida su come collegare WhatsApp a ChatGPT](https://whasp.ai/it/blog/collegare-whatsapp-chatgpt-api-openai): lo stesso webhook firmato, la stessa inbox salvata, la stessa chiamata di risposta. Cambia il lato del modello: tool use, prompt caching e l'impostazione effort di Claude. Per capire in cosa un agente che usa i tool differisce da un bot a regole, vedi [chatbot WhatsApp o agente AI](https://whasp.ai/it/blog/chatbot-whatsapp-agente-ai-differenze). Se non hai mai costruito un receiver, parti dalla guida OpenAI e leggi la [guida ai webhook WhatsApp](https://whasp.ai/it/blog/webhook-whatsapp-firma-retry-duplicati) per retry, ordine e test.

## Cosa serve per farlo girare?

Questa guida richiede Python 3.10 o più recente ([la versione minima dell'SDK di Anthropic](https://platform.claude.com/docs/en/cli-sdks-libraries/sdks/python)), una chiave API di Anthropic e un numero whasp con la sua chiave API.

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

Metti le tre chiavi in `.env`. Registra il tuo URL pubblico con `PUT /v1/webhook` come nella guida OpenAI; la risposta contiene il segreto `whsec_`.

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

## Com'è fatta l'app completa?

L'intero agente sta in un solo file, `app.py`. Avvialo con `uvicorn app:app --port 3000 --env-file .env`.

```python
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"
# Qui vanno i prompt veri: sotto i 512 token non viene messo in cache nulla.
SYSTEM = """Sei Ava, l'assistente di Fernwood, un piccolo negozio online di piante.
Chatti con i clienti su WhatsApp: messaggi brevi in testo semplice, senza Markdown.
Rispondi nella lingua del cliente. Per le domande sugli ordini, usa get_order_status."""

TOOLS = [
    {
        "name": "get_order_status",
        "description": "Cerca lo stato e la data di consegna stimata di un ordine del cliente "
        "tramite il numero d'ordine (formato FW-1234). Usalo ogni volta che chiede dov'è un ordine.",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "Numero d'ordine, es. FW-1042"}},
            "required": ["order_id"],
        },
    }
]
ORDERS = {  # al posto del tuo database o dell'API del negozio
    "FW-1042": {"phone": "+15555550111", "status": "shipped", "eta": "2026-10-23"},
}

claude = anthropic.AsyncAnthropic()  # legge 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] = {}  # una corsia per 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))  # rigioca ciò che un crash ha lasciato in sospeso
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/whasp/webhook")
async def webhook(request: Request):
    body = await request.body()  # byte grezzi: verifica prima di fare il 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 vuol dire che abbiamo già questo id 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 invia una stringa
    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:  # resta done = 0, rigiocato al prossimo avvio
            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 contenuto multimediale: niente a cui rispondere

    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 finestra deve iniziare con un turno 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):  # tetto ai giri di tool
        response = await claude.messages.create(
            model=MODEL,
            max_tokens=2048,  # i token di thinking contano qui dentro
            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 "Scusa, non ho saputo rispondere."
        # Il turno dell'assistente torna indietro intatto (thinking incluso), poi i risultati.
        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 "Scusa, ci sono voluti troppi passaggi."


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:  # il modello sceglie l'id; il tuo codice controlla chi lo chiede
        return {**result, "content": json.dumps({k: v for k, v in order.items() if k != "phone"})}
    return {**result, "content": "Nessun ordine con questo numero per questo cliente. Chiedi al cliente di ricontrollare.", "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):  # disiscritto, chat non aperta, numero non disponibile
            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
```

## Come funziona la parte webhook in FastAPI?

La parte webhook di `app.py` è il receiver TypeScript della guida OpenAI portato in FastAPI. `request.body()` restituisce i byte grezzi, `Webhook.verify` solleva `WebhookVerificationError` con una firma sbagliata o un timestamp sfasato di oltre cinque minuti, e `INSERT OR IGNORE` con `cur.rowcount` trasforma una consegna duplicata in un'operazione nulla che riceve comunque un 204. Ogni nuovo evento gira come un task asyncio a sé, quindi il gestore risponde 204 senza aspettare il modello e il limite di 10 secondi di whasp non conta mai. L'insieme `tasks` tiene un riferimento perché il task non venga raccolto dal garbage collector a metà strada, un `asyncio.Lock` per chat mette in coda in ordine i messaggi di una persona e `lifespan` rigioca le righe non finite dopo un crash.

Due semplificazioni: non tratta STOP come caso speciale (l'invio successivo a una chat disiscritta riceve 403 e `send` lo scarta, quindi uno STOP costa una chiamata al modello; la guida OpenAI mostra l'uscita anticipata) e salta i messaggi senza testo.

## Come chiama un tool Claude?

Il tool use è un ciclo attorno alla Messages API, e `run_agent` è quel ciclo scritto a mano. Descrivi un tool con un nome, una descrizione e uno JSON Schema. Quando Claude lo vuole, la risposta ha `stop_reason: "tool_use"` e uno o più blocchi `tool_use`, ciascuno con un `id`, un `name` e un `input`. Il tuo codice esegue il tool e restituisce l'output come blocco `tool_result` in un messaggio user. Claude poi continua con un'altra chiamata o con il testo finale.

Un turno dell'app di esempio va così. Il cliente scrive «Dov'è il mio ordine FW-1042?». La prima richiesta termina con un blocco `tool_use` per `get_order_status` con `{"order_id": "FW-1042"}`. `run_tool` restituisce `{"status": "shipped", "eta": "2026-10-23"}`. La seconda richiesta termina con `end_turn` e un testo come «È stato spedito e dovrebbe arrivare il 23 ottobre.» Sono due chiamate API e un messaggio WhatsApp.

Il ciclo segue cinque regole dalla documentazione di Anthropic sulle [chiamate ai tool](https://platform.claude.com/docs/en/agents-and-tools/tool-use/handle-tool-calls) e sul [thinking](https://platform.claude.com/docs/en/build-with-claude/thinking):

1. **Rimanda intatto il turno dell'assistente.** Sonnet 5.5 ragiona per impostazione predefinita, e dentro un turno di tool use i suoi blocchi di thinking devono tornare completi e non modificati. Aggiungere `response.content` così com'è fa proprio questo.
2. **Rispondi a ogni blocco `tool_use`, con i risultati per primi.** I blocchi `tool_result` vengono per primi nel messaggio user. Claude può chiamare più tool insieme, quindi il codice scorre tutti i blocchi.
3. **Segnala i fallimenti con `is_error`.** Una ricerca che non trova nulla restituisce un messaggio esplicativo con `is_error` impostato, e Claude lo dice al cliente invece di inventare uno stato.
4. **Metti un tetto al ciclo.** Cinque giri bastano e avanzano per una chat. Dopo, l'agente si scusa invece di consumare token.
5. **Tieni il ciclo in memoria.** Si salvano solo il testo del cliente e la risposta finale di Claude, che è tutto ciò che serve ai turni successivi. Anthropic consente di lasciare fuori il thinking dei turni precedenti quando non c'è tool use.

**L'autorizzazione sta nel tuo codice.** Il modello sceglie `order_id`, ma `run_tool` prende il numero di telefono dal webhook e lo confronta con il proprietario dell'ordine, quindi chiedere dell'ordine di un altro dà lo stesso errore di un numero sbagliato. Tratta gli input dei tool come input dell'utente. Tratta come non fidati anche i risultati dei tool che portano testo dall'esterno, perché le istruzioni nascoste al loro interno possono provare a pilotare Claude.

**Gli effetti collaterali devono essere idempotenti.** whasp consegna gli eventi almeno una volta, e un replay rifà girare l'intero ciclo, chiamate ai tool comprese. Una ricerca in sola lettura è innocua. Un rimborso o una prenotazione richiedono una chiave costruita con l'id del messaggio in arrivo, così la seconda esecuzione trova il risultato della prima.

Lascia `tool_choice` al valore predefinito: Anthropic elenca Sonnet 5.5 tra i modelli che [non supportano l'uso forzato dei tool](https://platform.claude.com/docs/en/build-with-claude/thinking) con il thinking attivo, quindi è il system prompt a dire a Claude quando usare il tool.

Esistono due scorciatoie. Impostare `strict: true` e `additionalProperties: false` su un tool fa sì che [gli input di Claude corrispondano esattamente al tuo schema](https://platform.claude.com/docs/en/agents-and-tools/tool-use/strict-tool-use). Il [tool runner](https://platform.claude.com/docs/en/agents-and-tools/tool-use/tool-runner) dell'SDK, ancora in beta, scrive il ciclo al posto tuo. Qui il ciclo resta scritto a mano perché l'autorizzazione, la segnalazione degli errori e il tetto ai giri sono le parti che vuoi controllare tu.

## Come fa il prompt caching a ridurre il costo?

Il prompt caching permette a Claude di riusare l'inizio di un prompt che ha già elaborato. L'ordine di rendering è tool, poi system, poi messaggi, quindi un solo breakpoint `cache_control` sull'ultimo blocco di system mette in cache insieme le definizioni dei tool e il system prompt.

Su Claude Sonnet 5.5, [l'input costa $2 per milione di token, una scrittura in cache $2,50 e una lettura dalla cache $0,10](https://platform.claude.com/docs/en/build-with-claude/prompt-caching). Un prompt da 3.000 token costa $0,006 senza cache e $0,0003 come lettura dalla cache. Tre regole decidono se ottieni lo sconto:

- **Dimensione minima.** Sonnet 5.5 mette in cache prompt di almeno 512 token. Quelli più corti vengono elaborati normalmente e l'API non restituisce alcun errore, quindi il prompt di esempio in `app.py` non finirà in cache. Mettici le tue vere policy e le FAQ, poi controlla i tre numeri che il codice stampa: cache write, cache read e plain input.
- **Durata.** Una voce di cache dura 5 minuti per impostazione predefinita e ogni lettura la rinnova; `"ttl": "1h"` costa di più. Una scrittura da 5 minuti costa il 25% in più dell'input normale, quindi la cache si ripaga dopo una lettura. Un system prompt condiviso da tutti gli utenti resta caldo finché continuano ad arrivare messaggi. La cronologia di un singolo utente di rado resta calda, se le persone rispondono a distanza di minuti, quindi questo codice mette in cache solo la parte condivisa. L'opzione `cache_control` di primo livello sposta il breakpoint in avanti man mano che la conversazione cresce e va bene per i botta e risposta veloci.
- **Corrispondenza esatta.** Qualsiasi modifica prima del breakpoint avvia una nuova cache: tool modificati, un `tool_choice` diverso, una nuova impostazione di effort o thinking, o un timestamp nel system prompt. Metti i dettagli per utente in un secondo blocco di system dopo quello in cache.

## Quale modello Claude e quale effort usare per un agente WhatsApp?

Claude Sonnet 5.5 (`claude-sonnet-5-5`) è il modello predefinito in `app.py`. Al 19 ottobre 2026 [la panoramica dei modelli di Anthropic](https://platform.claude.com/docs/en/about-claude/models/overview) lo indica a $2 in input e $10 in output per milione di token, lo definisce la migliore combinazione di velocità e intelligenza e si impegna a mantenerlo disponibile almeno fino al 28 settembre 2027. `claude-haiku-5-5`, da $0,10 e $0,50, è il modello più veloce e va bene per risposte semplici e ad alto volume. Cambia la costante `MODEL` per provarlo.

Effort e thinking decidono la latenza. Sonnet 5.5 usa per impostazione predefinita l'effort `high`, e [Anthropic suggerisce](https://platform.claude.com/docs/en/build-with-claude/effort) di iniziare i carichi di chat con `medium` o `low`; il codice usa `medium`. Il thinking è attivo per impostazione predefinita e conta nel `max_tokens` anche quando il suo testo non viene restituito, ed è per questo che `max_tokens` vale 2048 e non 300. Per saltare il thinking iniziale, invia `thinking={"type": "between_tools"}`: Sonnet 5.5 rifiuta `disabled`, e `between_tools` funziona con effort `high` o inferiore. Imposta l'effort una volta e non toccarlo più, perché cambiarlo avvia una nuova cache.

[L'SDK ritenta](https://platform.claude.com/docs/en/cli-sdks-libraries/sdks/python) due volte, per impostazione predefinita, gli errori di connessione e le risposte 408, 409, 429 e 5xx. Se una chiamata fallisce comunque, l'evento resta con `done = 0` e viene rigiocato al prossimo avvio. Una risposta che termina su `max_tokens` o con un rifiuto può tornare senza testo, quindi `run_agent` invia una breve riga di ripiego invece di un messaggio vuoto.

## Cosa costruire dopo?

Dai all'agente una [memoria della conversazione](https://whasp.ai/it/blog/memoria-conversazione-agente-ai-whatsapp) oltre i 20 turni, con riassunti e fatti, affina la sua voce seguendo la guida su [come far scrivere un agente AI come una persona](https://whasp.ai/it/blog/agente-ai-whatsapp-prompt-formattazione), oppure dai a Claude un numero WhatsApp come tool con [un server MCP](https://whasp.ai/it/blog/server-mcp-whatsapp-claude-chatgpt).

## Domande

### Come costruisco un bot WhatsApp con Claude?

Ricevi i messaggi WhatsApp su un webhook, chiama la Messages API di Anthropic con la cronologia della chat, esegui ogni tool che Claude chiede e invia il testo finale con un'API di messaggistica. Questa guida lo fa in Python con FastAPI e claude-sonnet-5-5.

### Quale modello Claude conviene usare per un agente WhatsApp?

Parti da claude-sonnet-5-5, che al 19 ottobre 2026 costa $2 in input e $10 in output per milione di token e che Anthropic descrive come la migliore combinazione di velocità e intelligenza. claude-haiku-5-5, da $0,10 e $0,50, è la scelta più veloce ed economica per risposte semplici.

### Il prompt caching funziona per un bot WhatsApp?

Sì, per la parte del prompt che tutti gli utenti condividono. Una voce di cache dura 5 minuti e ogni lettura la rinnova, quindi un system prompt condiviso resta caldo mentre la cronologia di un singolo utente di rado lo è. Claude Sonnet 5.5 richiede almeno 512 token per usare la cache.

### Come funziona il tool use di Claude?

Claude risponde con stop_reason tool_use e uno o più blocchi tool_use. Il tuo codice li esegue e restituisce blocchi tool_result in un messaggio user, poi Claude continua con un'altra chiamata o con il testo finale.
