# Claude WhatsApp-Bot bauen: KI-Agent mit der Anthropic API

> Claude-WhatsApp-Bot in Python mit FastAPI bauen: Webhook prüfen, Claude ein Tool geben, den System-Prompt cachen und die Antwort senden. Voller Code für Sonnet 5.5.

Source: https://whasp.ai/de/blog/claude-whatsapp-bot-bauen-anthropic-api · 2026-10-20

## Das Wichtigste

- Ein Claude-Agent auf WhatsApp besteht aus einem FastAPI-Webhook, einer Schleife um die Anthropic Messages API und einem POST, der die Antwort sendet, in rund 150 Zeilen Python.
- Tool Use ist eine Schleife: Solange stop_reason tool_use ist, führst du jeden tool_use-Block aus, gibst den Assistant-Turn unverändert zurück und antwortest in einer einzigen User-Nachricht, in der die tool_result-Blöcke an erster Stelle stehen.
- Die Autorisierung gehört in den Tool-Code: Das Tool bekommt die Telefonnummer des Kunden von deinem Server, nie vom Modell.
- Ein cache_control-Breakpoint auf dem System-Prompt cacht Tools und System-Prompt zusammen; bei Claude Sonnet 5.5 kostet ein Cache-Read $0,10 pro Million Tokens gegenüber $2 ohne Cache, aber nur Prompts ab 512 Tokens werden gecacht.
- whasp liefert Events mindestens einmal, und ein erneut abgespieltes Event führt die Schleife noch einmal aus, deshalb müssen Tools mit Seiteneffekten idempotent sein.

Ein Claude-Agent auf WhatsApp ist ein FastAPI-Webhook, der jede eingehende Nachricht prüft, die Anthropic Messages API mit Chatverlauf und einem Tool aufruft und den fertigen Text mit einem API-Call zurücksendet. Dieses Tutorial baut ihn in Python mit Claude Sonnet 5.5, einem Tool, das Bestellungen nachschlägt, und Prompt Caching auf dem System-Prompt.

## Was unterscheidet das von der OpenAI-Version?

Dieses Tutorial teilt seinen Unterbau mit [der OpenAI-Version](https://whasp.ai/de/blog/chatgpt-whatsapp-verbinden-openai-api): derselbe signierte Webhook, dieselbe gespeicherte Inbox, derselbe Antwort-Call. Was sich ändert, ist die Modellseite: Tool Use, Prompt Caching und Claudes Effort-Einstellung. Wie sich ein Agent mit Tool-Aufrufen von einem regelbasierten Bot unterscheidet, steht in [WhatsApp-Chatbot oder KI-Agent](https://whasp.ai/de/blog/whatsapp-chatbot-oder-ki-agent). Hast du noch nie einen Empfänger gebaut, beginne mit dem OpenAI-Tutorial und lies den [Webhook-Leitfaden](https://whasp.ai/de/blog/whatsapp-webhook-signatur-retries-duplikate) für Retries, Reihenfolge und Tests.

## Was brauchst du, um ihn zu starten?

Dieses Tutorial braucht Python 3.10 oder neuer, was [das Anthropic SDK voraussetzt](https://platform.claude.com/docs/en/cli-sdks-libraries/sdks/python), einen Anthropic-API-Key und eine whasp-Nummer mit API-Key.

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

Trage die drei Keys in `.env` ein. Registriere deine öffentliche URL wie im OpenAI-Tutorial mit `PUT /v1/webhook`; die Antwort enthält das `whsec_`-Secret.

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

## Wie sieht die komplette App aus?

Der ganze Agent steckt in einer Datei, `app.py`. Starte sie mit `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"
# Hier gehören echte Prompts hin: unter 512 Tokens wird nichts gecacht.
SYSTEM = """Du bist Ava, die Assistentin von Fernwood, einem kleinen Online-Pflanzenshop.
Du chattest mit Kunden auf WhatsApp: kurze Nachrichten in einfachem Text, kein Markdown.
Antworte in der Sprache des Kunden. Nutze bei Fragen zu Bestellungen get_order_status."""

TOOLS = [
    {
        "name": "get_order_status",
        "description": "Ruft Status und voraussichtliche Lieferung einer eigenen Bestellung des Kunden "
        "anhand der Bestellnummer ab (Format FW-1234). Nutze das Tool immer, wenn der Kunde fragt, wo seine Bestellung ist.",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "Bestellnummer, z. B. FW-1042"}},
            "required": ["order_id"],
        },
    }
]
ORDERS = {  # Platzhalter für deine Datenbank oder Shop-API
    "FW-1042": {"phone": "+15555550111", "status": "shipped", "eta": "2026-10-23"},
}

claude = anthropic.AsyncAnthropic()  # liest 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] = {}  # eine Spur pro 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))  # Reste eines Absturzes erneut verarbeiten
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/whasp/webhook")
async def webhook(request: Request):
    body = await request.body()  # rohe Bytes: erst prüfen, dann parsen
    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 heißt: Diese Event-ID haben wir schon
        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 sendet einen 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:  # bleibt done = 0, wird beim nächsten Start erneut verarbeitet
            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  # nur Medien: nichts zu beantworten

    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)  # das Fenster muss mit einem User-Turn beginnen
    return [{"role": role, "content": content} for role, content in rows]


async def run_agent(chat: str, messages: list[dict]) -> str:
    for _ in range(5):  # Obergrenze für Tool-Runden
        response = await claude.messages.create(
            model=MODEL,
            max_tokens=2048,  # Thinking-Tokens zählen dazu
            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 "Entschuldige, das konnte ich nicht beantworten."
        # Den Assistant-Turn unverändert zurückgeben (Thinking inklusive), danach die Ergebnisse.
        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 "Entschuldige, dafür waren zu viele Schritte nötig."


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"Unbekanntes Tool {block.name}", "is_error": True}
    order = ORDERS.get(block.input.get("order_id", ""))
    if order and order["phone"] == chat:  # das Modell wählt die ID; dein Code prüft, wer fragt
        return {**result, "content": json.dumps({k: v for k, v in order.items() if k != "phone"})}
    return {**result, "content": "Keine Bestellung mit dieser Nummer für diesen Kunden. Bitte ihn, die Nummer zu prüfen.", "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):  # abgemeldet, Chat nicht offen, Nummer nicht verfügbar
            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
```

## Wie funktioniert die Webhook-Hälfte in FastAPI?

Die Webhook-Hälfte von `app.py` ist der TypeScript-Empfänger aus dem OpenAI-Tutorial, nach FastAPI portiert. `request.body()` liefert die rohen Bytes, `Webhook.verify` wirft bei einer falschen Signatur oder einem Zeitstempel, der mehr als fünf Minuten abweicht, einen `WebhookVerificationError`, und `INSERT OR IGNORE` mit `cur.rowcount` macht aus einer doppelten Zustellung ein No-op, das trotzdem ein 204 bekommt. Jedes neue Event läuft als eigener asyncio-Task, der Handler antwortet also mit 204, ohne auf das Modell zu warten, und das 10-Sekunden-Limit von whasp spielt nie eine Rolle. Das Set `tasks` hält eine Referenz, damit der Task nicht mitten im Lauf vom Garbage Collector eingesammelt wird, ein `asyncio.Lock` pro Chat reiht die Nachrichten einer Person der Reihe nach ein, und `lifespan` verarbeitet nach einem Absturz unfertige Zeilen erneut.

Zwei Vereinfachungen: Die App behandelt STOP nicht gesondert (der nächste Versand an einen abgemeldeten Chat bekommt 403 und `send` verwirft ihn, ein STOP kostet also einen Modellaufruf; das OpenAI-Tutorial zeigt den frühen Ausstieg), und sie überspringt Nachrichten ohne Text.

## Wie ruft Claude ein Tool auf?

Tool Use ist eine Schleife um die Messages API, und `run_agent` ist diese Schleife von Hand geschrieben. Du beschreibst ein Tool mit Name, Beschreibung und einem JSON Schema. Will Claude es nutzen, enthält die Antwort `stop_reason: "tool_use"` und einen oder mehrere `tool_use`-Blöcke mit je `id`, `name` und `input`. Dein Code führt das Tool aus und gibt die Ausgabe als `tool_result`-Block in einer User-Nachricht zurück. Claude macht dann mit einem weiteren Aufruf oder mit dem fertigen Text weiter.

Ein Durchlauf der Beispiel-App sieht so aus. Der Kunde schreibt „Wo ist meine Bestellung FW-1042?“. Die erste Anfrage endet mit einem `tool_use`-Block für `get_order_status` mit `{"order_id": "FW-1042"}`. `run_tool` gibt `{"status": "shipped", "eta": "2026-10-23"}` zurück. Die zweite Anfrage endet mit `end_turn` und einem Text wie „Sie wurde versendet und sollte am 23. Oktober ankommen.“ Das sind zwei API-Calls und eine WhatsApp-Nachricht.

Die Schleife folgt fünf Regeln aus Anthropics [Doku zu Tool-Aufrufen](https://platform.claude.com/docs/en/agents-and-tools/tool-use/handle-tool-calls) und [Thinking-Doku](https://platform.claude.com/docs/en/build-with-claude/thinking):

1. **Gib den Assistant-Turn unverändert zurück.** Sonnet 5.5 denkt standardmäßig, und innerhalb eines Tool-Use-Turns müssen seine Thinking-Blöcke vollständig und unverändert zurückgehen. `response.content` unverändert anzuhängen erledigt das.
2. **Beantworte jeden `tool_use`-Block, Ergebnisse zuerst.** Die `tool_result`-Blöcke stehen in der User-Nachricht an erster Stelle. Claude kann mehrere Tools gleichzeitig aufrufen, deshalb läuft der Code über alle Blöcke.
3. **Melde Fehler mit `is_error`.** Eine Suche ohne Treffer gibt eine erklärende Meldung mit gesetztem `is_error` zurück, und Claude sagt es dem Kunden, statt einen Status zu erfinden.
4. **Begrenze die Schleife.** Für einen Chat reichen fünf Runden locker. Danach entschuldigt sich der Agent, statt Tokens zu verbrauchen.
5. **Halte die Schleife im Speicher.** Gespeichert werden nur der Text des Kunden und Claudes endgültige Antwort, mehr brauchen spätere Turns nicht. Anthropic erlaubt es, das Thinking früherer Turns außerhalb von Tool Use wegzulassen.

**Die Autorisierung liegt in deinem Code.** Das Modell wählt `order_id`, aber `run_tool` nimmt die Telefonnummer aus dem Webhook und vergleicht sie mit dem Besitzer der Bestellung. Wer nach der Bestellung einer anderen Person fragt, bekommt also denselben Fehler wie bei einer falschen Nummer. Behandle Tool-Eingaben wie Nutzereingaben. Behandle auch Tool-Ergebnisse, die fremden Text enthalten, als nicht vertrauenswürdig, denn darin versteckte Anweisungen können versuchen, Claude zu steuern.

**Seiteneffekte müssen idempotent sein.** whasp liefert Events mindestens einmal, und ein Replay führt die ganze Schleife erneut aus, Tool-Aufrufe eingeschlossen. Eine reine Abfrage ist harmlos. Eine Erstattung oder eine Buchung braucht einen Key, der aus der ID der eingehenden Nachricht gebildet wird, damit der zweite Lauf das Ergebnis des ersten findet.

Lass `tool_choice` auf dem Standard: Anthropic führt Sonnet 5.5 unter den Modellen, die [bei aktiviertem Thinking keine erzwungene Tool-Nutzung unterstützen](https://platform.claude.com/docs/en/build-with-claude/thinking), deshalb sagt der System-Prompt Claude, wann es das Tool nutzen soll.

Zwei Abkürzungen gibt es. Mit `strict: true` und `additionalProperties: false` an einem Tool [passen Claudes Eingaben exakt zu deinem Schema](https://platform.claude.com/docs/en/agents-and-tools/tool-use/strict-tool-use). Der [Tool Runner](https://platform.claude.com/docs/en/agents-and-tools/tool-use/tool-runner) des SDK, noch in der Beta, schreibt die Schleife für dich. Die handgeschriebene Schleife bleibt hier, weil Autorisierung, Fehlermeldung und Rundenbegrenzung die Teile sind, die du selbst kontrollieren willst.

## Wie senkt Prompt Caching die Kosten?

Prompt Caching lässt Claude den Anfang eines Prompts wiederverwenden, den es schon verarbeitet hat. Die Reihenfolge beim Rendern ist Tools, dann System, dann Messages, ein `cache_control`-Breakpoint am letzten System-Block cacht also die Tool-Definitionen und den System-Prompt zusammen.

Bei Claude Sonnet 5.5 kostet [Input $2 pro Million Tokens, ein Cache-Write $2,50 und ein Cache-Read $0,10](https://platform.claude.com/docs/en/build-with-claude/prompt-caching). Ein Prompt mit 3.000 Tokens kostet ohne Cache $0,006 und als Cache-Read $0,0003. Drei Regeln entscheiden, ob du diesen Rabatt bekommst:

- **Mindestgröße.** Sonnet 5.5 cacht Prompts ab 512 Tokens. Kürzere werden normal verarbeitet und die API gibt keinen Fehler zurück, der Beispiel-Prompt in `app.py` wird also nicht gecacht. Trage deine echten Richtlinien und FAQ ein und prüfe dann die drei Zahlen, die der Code ausgibt: Cache-Write, Cache-Read und normaler Input.
- **Lebensdauer.** Ein Cache-Eintrag hält standardmäßig 5 Minuten, und jeder Treffer verlängert ihn; `"ttl": "1h"` kostet mehr. Ein 5-Minuten-Write kostet 25 % mehr als normaler Input, das Caching lohnt sich also nach einem Read. Ein System-Prompt, den alle Nutzer teilen, bleibt warm, solange Nachrichten ankommen. Der Verlauf eines einzelnen Nutzers bleibt selten warm, wenn Leute im Minutenabstand antworten, deshalb cacht dieser Code nur den gemeinsamen Teil. Die `cache_control`-Option auf oberster Ebene schiebt den Breakpoint weiter, während ein Gespräch wächst, und passt zu schnellem Hin und Her.
- **Exakte Übereinstimmung.** Jede Änderung vor dem Breakpoint startet einen neuen Cache: geänderte Tools, ein anderes `tool_choice`, eine neue Effort- oder Thinking-Einstellung oder ein Zeitstempel im System-Prompt. Setze Angaben pro Nutzer in einen zweiten System-Block nach dem gecachten.

## Welches Claude-Modell und welchen Effort sollte ein WhatsApp-Agent nutzen?

Claude Sonnet 5.5 (`claude-sonnet-5-5`) ist das Standardmodell in `app.py`. Stand 19. Oktober 2026 führt [Anthropics Modellübersicht](https://platform.claude.com/docs/en/about-claude/models/overview) es mit $2 Input und $10 Output pro Million Tokens, nennt es die beste Kombination aus Geschwindigkeit und Intelligenz und sagt zu, es mindestens bis zum 28. September 2027 verfügbar zu halten. `claude-haiku-5-5`, ab $0,10 und $0,50, ist das schnellste Modell und passt zu einfachen Antworten in hohem Volumen. Ändere die Konstante `MODEL`, um es auszuprobieren.

Effort und Thinking bestimmen die Latenz. Sonnet 5.5 nutzt standardmäßig den Effort `high`, und [Anthropic empfiehlt](https://platform.claude.com/docs/en/build-with-claude/effort), Chat-Workloads mit `medium` oder `low` zu beginnen; der Code nutzt `medium`. Thinking ist standardmäßig an und zählt zu `max_tokens`, auch wenn sein Text nicht zurückgegeben wird, deshalb steht `max_tokens` auf 2048 und nicht auf 300. Um das Thinking vorab zu überspringen, sende `thinking={"type": "between_tools"}`: Sonnet 5.5 lehnt `disabled` ab, und `between_tools` funktioniert bei Effort `high` oder niedriger. Lege den Effort einmal fest und behalte ihn, denn eine Änderung startet einen neuen Cache.

[Das SDK wiederholt](https://platform.claude.com/docs/en/cli-sdks-libraries/sdks/python) Verbindungsfehler sowie Antworten mit 408, 409, 429 und 5xx standardmäßig zweimal. Schlägt ein Aufruf trotzdem fehl, behält das Event `done = 0` und wird beim nächsten Start erneut verarbeitet. Eine Antwort, die mit `max_tokens` oder einer Ablehnung endet, kann ohne Text zurückkommen, deshalb sendet `run_agent` eine kurze Ersatzzeile statt einer leeren Nachricht.

## Was baust du als Nächstes?

Gib dem Agenten ein [Konversationsgedächtnis aus Zusammenfassungen und Fakten](https://whasp.ai/de/blog/whatsapp-ki-agent-gedaechtnis), das über 20 Turns hinausreicht, schärfe seine Stimme, indem du ihn [wie einen Menschen schreiben lässt](https://whasp.ai/de/blog/ki-agent-whatsapp-menschlich-schreiben), oder gib Claude eine WhatsApp-Nummer als Tool mit [einem MCP-Server](https://whasp.ai/de/blog/whatsapp-mcp-server-claude-chatgpt).

## Fragen

### Wie baue ich einen Claude-WhatsApp-Bot?

Empfange WhatsApp-Nachrichten an einem Webhook, rufe die Anthropic Messages API mit dem Chatverlauf auf, führe jedes Tool aus, das Claude anfordert, und sende den fertigen Text über eine Messaging-API. Dieses Tutorial macht das in Python mit FastAPI und claude-sonnet-5-5.

### Welches Claude-Modell sollte ein WhatsApp-Agent nutzen?

Beginne mit claude-sonnet-5-5, das am 19. Oktober 2026 $2 Input und $10 Output pro Million Tokens kostet und das Anthropic als beste Kombination aus Geschwindigkeit und Intelligenz beschreibt. claude-haiku-5-5, ab $0,10 und $0,50, ist die schnellere und günstigere Wahl für einfache Antworten.

### Funktioniert Prompt Caching bei einem WhatsApp-Bot?

Ja, für den Teil des Prompts, den alle Nutzer teilen. Ein Cache-Eintrag hält 5 Minuten und jeder Treffer verlängert ihn, ein gemeinsamer System-Prompt bleibt also warm, der Verlauf eines einzelnen Nutzers selten. Claude Sonnet 5.5 braucht mindestens 512 Tokens zum Cachen.

### Wie funktioniert Tool Use bei Claude?

Claude antwortet mit stop_reason tool_use und einem oder mehreren tool_use-Blöcken. Dein Code führt sie aus und gibt tool_result-Blöcke in einer User-Nachricht zurück, danach macht Claude mit einem weiteren Aufruf oder dem fertigen Text weiter.