Bot WhatsApp Claude : créer un agent avec l'API Anthropic

Un agent Claude sur WhatsApp est un webhook FastAPI qui vérifie chaque message entrant, appelle l’API Messages d’Anthropic avec l’historique de la conversation et un outil, puis renvoie le texte final avec un seul appel d’API. Ce tutoriel le construit en Python avec Claude Sonnet 5.5, un outil qui consulte les commandes et le prompt caching sur le system prompt.

En quoi cela diffère-t-il de la version OpenAI ?

Ce tutoriel partage sa plomberie avec l’intégration OpenAI WhatsApp : le même webhook signé, la même boîte de réception stockée, le même appel de réponse. Ce qui change, c’est le côté modèle : l’utilisation d’outils, le prompt caching et le réglage d’effort de Claude. Pour voir en quoi un agent qui appelle des outils diffère d’un bot à règles, lisez Chatbot WhatsApp ou agent IA. Si vous n’avez jamais construit de récepteur, commencez par le tutoriel OpenAI, et lisez le guide du webhook pour les nouvelles tentatives, l’ordre des événements et les tests.

De quoi avez-vous besoin pour le lancer ?

Ce tutoriel demande Python 3.10 ou plus récent, version exigée par le SDK Anthropic, une clé d’API Anthropic et un numéro whasp avec sa clé d’API.

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

Mettez les trois clés dans .env. Enregistrez votre URL publique avec PUT /v1/webhook comme dans le tutoriel OpenAI ; la réponse contient le secret whsec_.

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

À quoi ressemble l’application complète ?

L’agent tient dans un seul fichier, app.py. Lancez-le avec 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"
# Mettez ici de vrais prompts : en dessous de 512 tokens, rien n'est mis en cache.
SYSTEM = """Tu es Ava, l'assistante de Fernwood, une petite boutique de plantes en ligne.
Tu discutes avec les clients sur WhatsApp : messages courts en texte brut, sans Markdown.
Réponds dans la langue du client. Pour les questions sur une commande, utilise get_order_status."""

TOOLS = [
    {
        "name": "get_order_status",
        "description": "Cherche le statut et la date de livraison estimée d'une commande du client, "
        "à partir du numéro de commande (format FW-1234). À utiliser dès qu'il demande où en est une commande.",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "Numéro de commande, par ex. FW-1042"}},
            "required": ["order_id"],
        },
    }
]
ORDERS = {  # remplace votre base de données ou l'API de votre boutique
    "FW-1042": {"phone": "+15555550111", "status": "shipped", "eta": "2026-10-23"},
}

claude = anthropic.AsyncAnthropic()  # lit 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] = {}  # une file par conversation
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))  # rejoue ce qu'un plantage a laissé en suspens
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/whasp/webhook")
async def webhook(request: Request):
    body = await request.body()  # octets bruts : à vérifier avant d'analyser
    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 : cet identifiant d'événement est déjà connu
        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 envoie une chaîne
    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:  # reste à done = 0, rejoué au prochain démarrage
            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  # média seul : rien à répondre

    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 fenêtre doit commencer par un tour de l'utilisateur
    return [{"role": role, "content": content} for role, content in rows]


async def run_agent(chat: str, messages: list[dict]) -> str:
    for _ in range(5):  # plafond sur les tours d'outils
        response = await claude.messages.create(
            model=MODEL,
            max_tokens=2048,  # les tokens de thinking comptent dans cette 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 "Désolée, je n'ai pas pu répondre à cela."
        # On renvoie le tour de l'assistant tel quel (thinking compris), puis les résultats.
        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 "Désolée, cela a pris trop d'étapes."


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"Outil inconnu {block.name}", "is_error": True}
    order = ORDERS.get(block.input.get("order_id", ""))
    if order and order["phone"] == chat:  # le modèle choisit l'identifiant ; votre code vérifie qui demande
        return {**result, "content": json.dumps({k: v for k, v in order.items() if k != "phone"})}
    return {**result, "content": "Aucune commande avec ce numéro pour ce client. Demande-lui de vérifier.", "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):  # désinscrit, conversation non ouverte, numéro indisponible
            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

Comment fonctionne la moitié webhook avec FastAPI ?

La moitié webhook de app.py est le récepteur TypeScript porté sur FastAPI. request.body() renvoie les octets bruts, Webhook.verify lève WebhookVerificationError pour une signature invalide ou un horodatage décalé de plus de cinq minutes, et INSERT OR IGNORE avec cur.rowcount transforme une livraison en double en opération sans effet, qui reçoit quand même un 204. Chaque nouvel événement s’exécute dans sa propre tâche asyncio : le gestionnaire répond 204 sans attendre le modèle, et la limite de 10 secondes de whasp ne compte jamais. L’ensemble tasks garde une référence pour que le garbage collector ne ramasse pas la tâche en plein vol, un asyncio.Lock par conversation met en file les messages d’une personne dans l’ordre, et lifespan rejoue les lignes inachevées après un plantage.

Deux simplifications : le code ne traite pas STOP à part (l’envoi suivant vers une conversation désinscrite reçoit 403 et send l’abandonne, donc un STOP coûte un appel au modèle ; le tutoriel OpenAI montre la sortie anticipée), et il ignore les messages sans texte.

Comment Claude appelle-t-il un outil ?

L’utilisation d’outils est une boucle autour de l’API Messages, et run_agent est cette boucle écrite à la main. Vous décrivez un outil avec un nom, une description et un JSON Schema. Quand Claude veut s’en servir, la réponse contient stop_reason: "tool_use" et un ou plusieurs blocs tool_use, chacun avec un id, un name et un input. Votre code exécute l’outil et renvoie sa sortie sous forme de bloc tool_result dans un message user. Claude poursuit alors avec un autre appel ou avec le texte final.

Un tour de l’application d’exemple se déroule ainsi. Le client écrit « Où en est ma commande FW-1042 ? ». La première requête se termine par un bloc tool_use pour get_order_status avec {"order_id": "FW-1042"}. run_tool renvoie {"status": "shipped", "eta": "2026-10-23"}. La seconde requête se termine par end_turn et un texte comme « Elle a été expédiée et devrait arriver le 23 octobre. » Cela fait deux appels d’API et un message WhatsApp.

La boucle suit cinq règles tirées de la documentation d’Anthropic sur les appels d’outils et de sa documentation sur le thinking :

1. Renvoyez le tour de l’assistant tel quel. Sonnet 5.5 réfléchit par défaut, et dans un tour avec outils ses blocs thinking doivent revenir complets et non modifiés. Ajouter response.content tel quel y suffit.
2. Répondez à chaque bloc tool_use, résultats en premier. Les blocs tool_result viennent en premier dans le message user. Claude peut appeler plusieurs outils à la fois : le code parcourt donc tous les blocs.
3. Signalez les échecs avec is_error. Une recherche qui ne trouve rien renvoie un message explicite avec is_error activé, et Claude le dit au client au lieu d’inventer un statut.
4. Plafonnez la boucle. Cinq tours suffisent largement pour un chat. Au-delà, l’agent s’excuse au lieu de dépenser des tokens.
5. Gardez la boucle en mémoire. Seuls le texte du client et la réponse finale de Claude sont enregistrés, ce qui suffit aux tours suivants. Anthropic permet d’omettre le thinking des tours précédents en dehors de l’utilisation d’outils.

L’autorisation se joue dans votre code. Le modèle choisit order_id, mais run_tool prend le numéro de téléphone dans le webhook et le compare au propriétaire de la commande : demander la commande de quelqu’un d’autre produit donc la même erreur qu’un mauvais numéro. Traitez les entrées d’un outil comme une saisie utilisateur. Traitez aussi comme non fiables les résultats d’outil qui contiennent du texte venu de l’extérieur, car les instructions qui s’y cachent peuvent tenter d’orienter Claude.

Les effets de bord doivent être idempotents. whasp livre les événements au moins une fois, et un rejeu relance toute la boucle, appels d’outils compris. Une simple lecture est sans danger. Un remboursement ou une réservation demande une clé construite à partir de l’identifiant du message entrant, pour que la seconde exécution retrouve le résultat de la première.

Laissez tool_choice à sa valeur par défaut : Anthropic range Sonnet 5.5 parmi les modèles qui ne prennent pas en charge l’appel d’outil forcé quand le thinking est activé ; le system prompt dit donc à Claude quand utiliser l’outil.

Deux raccourcis existent. Mettre strict: true et additionalProperties: false sur un outil fait que les entrées de Claude respectent exactement votre schéma. Le tool runner du SDK, encore en bêta, écrit la boucle à votre place. La boucle écrite à la main reste ici parce que l’autorisation, le signalement des erreurs et le plafond de tours sont les parties que vous voulez contrôler.

Comment le prompt caching réduit-il le coût ?

Le prompt caching permet à Claude de réutiliser le début d’un prompt qu’il a déjà traité. L’ordre de rendu est outils, puis system, puis messages : un seul point d’arrêt cache_control sur le dernier bloc system met donc en cache les définitions d’outils et le system prompt ensemble.

Sur Claude Sonnet 5.5, l’entrée coûte $2 par million de tokens, une écriture en cache $2,50 et une lecture du cache $0,10. Un prompt de 3 000 tokens coûte $0,006 hors cache et $0,0003 en lecture du cache. Trois règles décident si vous obtenez cette remise :

• Taille minimale. Sonnet 5.5 met en cache les prompts d’au moins 512 tokens. Les prompts plus courts sont traités normalement et l’API ne renvoie aucune erreur : le prompt d’exemple de app.py ne sera donc pas mis en cache. Mettez-y vos vraies règles et votre FAQ, puis surveillez les trois nombres que le code affiche : écriture en cache, lecture du cache et entrée simple.
• Durée de vie. Une entrée de cache dure 5 minutes par défaut et chaque lecture la prolonge ; "ttl": "1h" coûte plus cher. Une écriture à 5 minutes coûte 25 % de plus que l’entrée simple, donc le cache est rentable dès la première lecture. Un system prompt partagé par tous les utilisateurs reste chaud tant que des messages arrivent. L’historique d’un seul utilisateur l’est rarement si les gens répondent à plusieurs minutes d’écart : ce code ne met donc en cache que la partie commune. L’option cache_control de niveau supérieur déplace le point d’arrêt vers l’avant à mesure que la conversation s’allonge et convient aux échanges rapides.
• Correspondance exacte. Tout changement avant le point d’arrêt ouvre un nouveau cache : outils modifiés, autre tool_choice, nouveau réglage d’effort ou de thinking, ou horodatage dans le system prompt. Mettez les détails propres à l’utilisateur dans un second bloc system placé après le bloc en cache.

Quel modèle Claude et quel effort pour un agent WhatsApp ?

Claude Sonnet 5.5 (claude-sonnet-5-5) est le modèle par défaut de app.py. Au 19 octobre 2026, l’aperçu des modèles d’Anthropic le liste à $2 en entrée et $10 en sortie par million de tokens, le présente comme la meilleure combinaison de vitesse et d’intelligence, et s’engage à le maintenir disponible au moins jusqu’au 28 septembre 2027. claude-haiku-5-5, à partir de $0,10 et $0,50, est le modèle le plus rapide et convient aux réponses simples à gros volume. Changez la constante MODEL pour l’essayer.

L’effort et le thinking déterminent la latence. Sonnet 5.5 est par défaut à l’effort high, et Anthropic suggère de commencer les charges de chat à medium ou low ; le code utilise medium. Le thinking est activé par défaut et compte dans max_tokens même quand son texte n’est pas renvoyé, ce qui explique que max_tokens vaille 2 048 et non 300. Pour supprimer la réflexion initiale, envoyez thinking={"type": "between_tools"} : Sonnet 5.5 rejette disabled, et between_tools fonctionne à l’effort high ou en dessous. Réglez l’effort une fois pour toutes, car le changer ouvre un nouveau cache.

Le SDK retente deux fois par défaut les erreurs de connexion, 408, 409, 429 et 5xx. Si un appel échoue malgré tout, l’événement garde done = 0 et est rejoué au prochain démarrage. Une réponse qui s’arrête sur max_tokens ou sur un refus peut revenir sans texte : run_agent envoie donc une courte phrase de repli plutôt qu’un message vide.

Que construire ensuite ?

Donnez à l’agent une mémoire au-delà de 20 tours avec des résumés et des faits dans Agent IA WhatsApp avec mémoire, travaillez sa voix avec Prompt pour agent IA WhatsApp, ou donnez à Claude un numéro WhatsApp comme outil avec un serveur MCP.