How to Build a WhatsApp AI Agent

1"""WhatsApp lead-qualification agent for a real estate listing.
2Run: uvicorn whatsapp_lead_agent:app --reload
3pip install fastapi uvicorn httpx anthropic
4"""
5import hashlib
6import hmac
7import os
8
9import anthropic
10import httpx
11from fastapi import FastAPI, Request, Response
12
13VERIFY_TOKEN = os.environ["WHATSAPP_VERIFY_TOKEN"]
14ACCESS_TOKEN = os.environ["WHATSAPP_ACCESS_TOKEN"]
15PHONE_NUMBER_ID = os.environ["WHATSAPP_PHONE_NUMBER_ID"]
16APP_SECRET = os.environ["WHATSAPP_APP_SECRET"]
17GRAPH_URL = f"https://graph.facebook.com/v26.0/{PHONE_NUMBER_ID}/messages"
18
19app = FastAPI()
20client = anthropic.Anthropic()
21CONVERSATIONS: dict[str, list[dict]] = {}
22
23SYSTEM_PROMPT = (
24    "You qualify real estate leads over WhatsApp for the 123 Maple listing. "
25    "Answer questions about the listing, check availability before offering a "
26    "showing time, and book it once confirmed. If the lead sounds ready to "
27    "make an offer, or asks anything about price negotiation, call "
28    "escalate_to_human instead of handling it yourself."
29)
30
31TOOLS = [
32    {"name": "check_availability", "description": "Check open showing slots for the listing.",
33     "input_schema": {"type": "object", "properties": {"date": {"type": "string"}}, "required": ["date"]}},
34    {"name": "book_showing", "description": "Book a confirmed showing slot.",
35     "input_schema": {"type": "object", "properties": {
36         "date": {"type": "string"}, "time": {"type": "string"}, "name": {"type": "string"}},
37         "required": ["date", "time", "name"]}},
38    {"name": "escalate_to_human", "description": "Hand off a hot lead to a human agent.",
39     "input_schema": {"type": "object", "properties": {"reason": {"type": "string"}}, "required": ["reason"]}},
40]
41
42
43def verify_signature(raw_body: bytes, signature_header: str) -> bool:
44    expected = hmac.new(APP_SECRET.encode(), raw_body, hashlib.sha256).hexdigest()
45    return hmac.compare_digest(f"sha256={expected}", signature_header or "")
46
47
48def get_history(sender: str) -> list[dict]:
49    return CONVERSATIONS.setdefault(sender, [])
50
51
52async def escalate(sender: str, reason: str) -> None:
53    # In production: POST to a Slack webhook or create a CRM task here.
54    print(f"[ESCALATION] {sender}: {reason}")
55
56
57def run_tool(name: str, tool_input: dict) -> dict:
58    if name == "check_availability":
59        return {"content": [{"type": "text", "text": "2:00 PM and 4:30 PM are open."}]}
60    if name == "book_showing":
61        return {"content": [{"type": "text", "text": f"Booked {tool_input['date']} at {tool_input['time']}."}]}
62    if name == "escalate_to_human":
63        return {"content": [{"type": "text", "text": "A human agent will follow up shortly."}]}
64    return {"content": [{"type": "text", "text": "Unknown tool."}], "is_error": True}
65
66
67async def ask_claude(sender: str) -> str:
68    history = get_history(sender)
69    response = client.messages.create(
70        model="claude-sonnet-5-5", max_tokens=1024,
71        system=SYSTEM_PROMPT, tools=TOOLS, messages=history,
72    )
73    history.append({"role": "assistant", "content": response.content})
74
75    tool_blocks = [b for b in response.content if b.type == "tool_use"]
76    if not tool_blocks:
77        return "".join(b.text for b in response.content if b.type == "text")
78
79    results = []
80    for block in tool_blocks:
81        if block.name == "escalate_to_human":
82            await escalate(sender, block.input.get("reason", ""))
83        result = run_tool(block.name, block.input)
84        results.append({"type": "tool_result", "tool_use_id": block.id, **result})
85    history.append({"role": "user", "content": results})
86
87    follow_up = client.messages.create(
88        model="claude-sonnet-5-5", max_tokens=1024,
89        system=SYSTEM_PROMPT, tools=TOOLS, messages=history,
90    )
91    history.append({"role": "assistant", "content": follow_up.content})
92    return "".join(b.text for b in follow_up.content if b.type == "text")
93
94
95async def send_whatsapp_message(to: str, body: str) -> None:
96    async with httpx.AsyncClient() as http:
97        await http.post(
98            GRAPH_URL,
99            headers={"Authorization": f"Bearer {ACCESS_TOKEN}"},
100            json={"messaging_product": "whatsapp", "to": to, "type": "text", "text": {"body": body}},
101        )
102
103
104@app.get("/webhook")
105async def verify(request: Request):
106    params = request.query_params
107    if params.get("hub.verify_token") == VERIFY_TOKEN:
108        return int(params["hub.challenge"])
109    return Response(status_code=403)
110
111
112@app.post("/webhook")
113async def receive(request: Request):
114    raw_body = await request.body()
115    if not verify_signature(raw_body, request.headers.get("X-Hub-Signature-256", "")):
116        return Response(status_code=403)
117
118    body = await request.json()
119    entry = body["entry"][0]["changes"][0]["value"]
120    if "messages" not in entry:
121        return {"status": "ignored"}
122
123    msg = entry["messages"][0]
124    sender, text = msg["from"], msg["text"]["body"]
125    get_history(sender).append({"role": "user", "content": text})
126
127    reply = await ask_claude(sender)
128    await send_whatsapp_message(sender, reply)
129    return {"status": "ok"}
130