Claude WhatsApp bot: build an agent with the Anthropic API

A Claude agent on WhatsApp is a FastAPI webhook that verifies each incoming message, calls the Anthropic Messages API with the chat history and a tool, and sends the final text back with one API call. This tutorial builds it in Python with Claude Sonnet 5.5, a tool that looks up orders, and prompt caching on the system prompt.

How is this different from the OpenAI version?

This tutorial shares its plumbing with the OpenAI version: the same signed webhook, the same stored inbox, the same reply call. What changes is the model side: tool use, prompt caching and Claude’s effort setting. For how a tool-calling agent differs from a rules-based bot, see chatbot vs agent. If you have not built a receiver before, start with the OpenAI tutorial, and read the webhook guide for retries, ordering and testing.

What do you need to run it?

This tutorial needs Python 3.10 or newer, which the Anthropic SDK requires, an Anthropic API key, and a whasp number with an API key.

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

Put the three keys in .env. Register your public URL with PUT /v1/webhook as in the OpenAI tutorial; the response holds the whsec_ secret.

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

What does the complete app look like?

The whole agent is one file, app.py. Run it with 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"
# Real prompts go here: under 512 tokens nothing is cached.
SYSTEM = """You are Ava, the assistant of Fernwood, a small online plant shop.
You chat with customers on WhatsApp: short plain-text messages, no Markdown.
Answer in the customer's language. For order questions, use get_order_status."""

TOOLS = [
    {
        "name": "get_order_status",
        "description": "Look up the status and delivery estimate of one of the customer's own "
        "orders by order number (format FW-1234). Use it whenever they ask where an order is.",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "Order number, e.g. FW-1042"}},
            "required": ["order_id"],
        },
    }
]
ORDERS = {  # stand-in for your database or shop API
    "FW-1042": {"phone": "+15555550111", "status": "shipped", "eta": "2026-10-23"},
}

claude = anthropic.AsyncAnthropic()  # reads 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] = {}  # one lane 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))  # replay what a crash left behind
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/whasp/webhook")
async def webhook(request: Request):
    body = await request.body()  # raw bytes: verify before parsing
    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 means we already have this event id
        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 sends a 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:  # stays done = 0, replayed at the next start
            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  # media only: nothing to answer

    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)  # the window must start on a user turn
    return [{"role": role, "content": content} for role, content in rows]


async def run_agent(chat: str, messages: list[dict]) -> str:
    for _ in range(5):  # cap on tool rounds
        response = await claude.messages.create(
            model=MODEL,
            max_tokens=2048,  # thinking tokens count against this
            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 "Sorry, I could not answer that."
        # Assistant turn back untouched (thinking included), then the results.
        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 "Sorry, that took too many steps."


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:  # the model picks the id; your code checks who is asking
        return {**result, "content": json.dumps({k: v for k, v in order.items() if k != "phone"})}
    return {**result, "content": "No order with that number for this customer. Ask them to check it.", "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):  # opted out, chat not open, number unavailable
            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

How does the webhook half work in FastAPI?

The webhook half of app.py is the TypeScript receiver ported to FastAPI. request.body() returns the raw bytes, Webhook.verify raises WebhookVerificationError on a bad signature or a timestamp more than five minutes off, and INSERT OR IGNORE with cur.rowcount turns a duplicate delivery into a no-op that still gets a 204. Each new event runs as its own asyncio task, so the handler answers 204 without waiting for the model and whasp’s 10-second limit never matters. The set tasks keeps a reference so the task is not garbage-collected mid-flight, a per-chat asyncio.Lock queues one person’s messages in order, and lifespan replays unfinished rows after a crash.

Two simplifications: it does not special-case STOP (the next send to an opted-out chat gets 403 and send drops it, so a STOP costs one model call; the OpenAI tutorial shows the early exit), and it skips messages without text.

How does Claude call a tool?

Tool use is a loop around the Messages API, and run_agent is that loop written by hand. You describe a tool with a name, a description and a JSON Schema. When Claude wants it, the response has stop_reason: "tool_use" and one or more tool_use blocks, each with an id, a name and an input. Your code runs the tool and returns the output as a tool_result block in a user message. Claude then continues with another call or with the final text.

One turn of the sample app goes like this. The customer writes “Where is my order FW-1042?”. The first request ends with a tool_use block for get_order_status with {"order_id": "FW-1042"}. run_tool returns {"status": "shipped", "eta": "2026-10-23"}. The second request ends with end_turn and a text such as “It shipped and should arrive on October 23.” That is two API calls and one WhatsApp message.

The loop follows five rules from Anthropic’s tool call docs and thinking docs:

1. Pass the assistant turn back untouched. Sonnet 5.5 thinks by default, and within a tool-use turn its thinking blocks must return complete and unmodified. Appending response.content as is does that.
2. Answer every tool_use block, results first. The tool_result blocks come first in the user message. Claude can call several tools at once, so the code loops over all blocks.
3. Report failures with is_error. A lookup that finds nothing returns an instructive message with is_error set, and Claude tells the customer instead of inventing a status.
4. Cap the loop. Five rounds is plenty for chat. After that the agent apologizes instead of spending tokens.
5. Keep the loop in memory. Only the customer’s text and Claude’s final answer are saved, which is all later turns need. Anthropic allows leaving earlier turns’ thinking out outside of tool use.

Authorization lives in your code. The model chooses order_id, but run_tool takes the phone number from the webhook and compares it with the order’s owner, so asking about someone else’s order gets the same error as a wrong number. Treat tool inputs like user input. Treat tool results that carry outside text as untrusted too, because instructions hidden in them can try to steer Claude.

Side effects must be idempotent. whasp delivers events at least once, and a replay re-runs the whole loop, tool calls included. A read-only lookup is harmless. A refund or a booking needs a key built from the inbound message id, so the second run finds the first run’s result.

Leave tool_choice on its default: Anthropic lists Sonnet 5.5 among the models that do not support forced tool use with thinking on, so the system prompt tells Claude when to use the tool.

Two shortcuts exist. Setting strict: true and additionalProperties: false on a tool makes Claude’s inputs match your schema exactly. The SDK’s tool runner, still in beta, writes the loop for you. The hand-written loop stays here because the authorization, the error reporting and the round cap are the parts you want to control.

How does prompt caching cut the cost?

Prompt caching lets Claude reuse the start of a prompt it has already processed. The render order is tools, then system, then messages, so one cache_control breakpoint on the last system block caches the tool definitions and the system prompt together.

On Claude Sonnet 5.5, input costs $2 per million tokens, a cache write $2.50 and a cache read $0.10. A 3,000-token prompt costs $0.006 uncached and $0.0003 as a cache read. Three rules decide whether you get that discount:

• Minimum size. Sonnet 5.5 caches prompts of at least 512 tokens. Shorter ones are processed normally and the API returns no error, so the sample prompt in app.py will not cache. Put your real policies and FAQ in it, then check the three numbers the code prints: cache write, cache read and plain input.
• Lifetime. A cache entry lasts 5 minutes by default and each hit refreshes it; "ttl": "1h" costs more. A 5-minute write costs 25% more than plain input, so caching pays off after one read. A system prompt shared by all users stays warm while messages keep arriving. One user’s history rarely does if people reply minutes apart, so this code caches only the shared part. The top-level cache_control option moves the breakpoint forward as a conversation grows and suits fast back-and-forth.
• Exact match. Any change before the breakpoint starts a new cache: edited tools, a different tool_choice, a new effort or thinking setting, or a timestamp in the system prompt. Put per-user details in a second system block after the cached one.

Which Claude model and effort should a WhatsApp agent use?

Claude Sonnet 5.5 (claude-sonnet-5-5) is the default in app.py. As of 19 October 2026, Anthropic’s model overview lists it at $2 input and $10 output per million tokens, calls it the best combination of speed and intelligence, and commits to keeping it available until at least 28 September 2027. claude-haiku-5-5, from $0.10 and $0.50, is the fastest model and fits simple, high-volume replies. Change the MODEL constant to try it.

Effort and thinking decide the latency. Sonnet 5.5 defaults to high effort, and Anthropic suggests starting chat workloads at medium or low; the code uses medium. Thinking is on by default and counts toward max_tokens even when its text is not returned, which is why max_tokens is 2048 and not 300. To skip up-front thinking, send thinking={"type": "between_tools"}: Sonnet 5.5 rejects disabled, and between_tools works at effort high or below. Set effort once and keep it, because changing it starts a new cache.

The SDK retries connection errors, 408, 409, 429 and 5xx responses twice by default. If a call still fails, the event keeps done = 0 and replays at the next start. A reply that ends on max_tokens or a refusal can come back without text, so run_agent sends a short fallback line instead of an empty message.

What should you build next?

Give the agent memory beyond 20 turns with summaries and facts, tune its voice with prompts written for WhatsApp, or hand Claude a WhatsApp number as a tool with an MCP server.