# How to Build a WhatsApp AI Agent - Tool: WhatsApp Cloud API - Difficulty: Intermediate - Time required: 60-75 min - Compatibility: Python 3.10+, FastAPI, a Meta developer account with WhatsApp Cloud API access, Claude API key - Last updated: November 2026 ## TL;DR A WhatsApp AI agent is a webhook: Meta's Cloud API posts incoming messages to your server, and you reply through the same API within a 24-hour messaging window. This guide builds one with FastAPI and Claude that qualifies real estate leads, answers listing questions, books a showing, and escalates serious buyers to a human agent. Budget 60-75 minutes and a Meta developer account to test it. ## What you'll build You'll build a WhatsApp agent for a real estate brokerage that talks to inbound leads: it answers questions about a listing, checks whether a showing slot is open, books it, and the moment a lead sounds like a serious, ready-to-buy prospect, hands them off to a human agent instead of trying to close anything itself. Unlike a chat widget or an SDK-based agent, a WhatsApp agent has no persistent connection to maintain. Meta's Cloud API posts each incoming message to a webhook URL you host, and you reply by making your own POST request back to a different endpoint on the same API, inside a 24-hour window from the user's last message. There's no streaming session to keep alive; the whole architecture is just two HTTP endpoints and whatever state you keep between messages yourself. That statelessness is also the main engineering problem: nothing about WhatsApp tracks the conversation for you. You keep the message history per sender, feed it to Claude along with the leasing tools, and send the reply back through the API, all inside a single webhook handler that Meta expects to get a fast 200 response from regardless of how long the actual reasoning takes. ## Before you start - A Meta developer account with a WhatsApp Business app, a test phone number, and a permanent access token - A publicly reachable HTTPS URL for your webhook (a tunnel like ngrok is fine for testing) - An Anthropic API key (ANTHROPIC_API_KEY) with available credits - pip install fastapi uvicorn httpx anthropic ## Step-by-step guide ### 1. Register your webhook and verify it In Meta's developer console, point your WhatsApp app's webhook at your server's URL and choose a verify token, any string you make up yourself. Meta calls your endpoint once with a GET request containing that token and a challenge value; your server must echo the challenge back exactly, or the webhook registration fails before you ever receive a real message. ``` VERIFY_TOKEN = "a-secret-you-choose" @app.get("/webhook") async def verify(request: Request): params = request.query_params if params.get("hub.verify_token") == VERIFY_TOKEN: return int(params["hub.challenge"]) return Response(status_code=403) ``` **Expected result:** Meta's console shows the webhook as verified, with a green check next to the callback URL. ### 2. Parse the incoming message payload A real incoming text message arrives as a POST with a nested structure: entry, then changes, then a value object containing a messages array. Pull the sender's WhatsApp ID and the message text out of that structure, and return a 200 immediately regardless of what you do next; Meta expects a fast acknowledgment and will retry delivery if your webhook is slow to respond. ``` @app.post("/webhook") async def receive(request: Request): body = await request.json() entry = body["entry"][0]["changes"][0]["value"] if "messages" not in entry: return {"status": "ignored"} msg = entry["messages"][0] sender, text = msg["from"], msg["text"]["body"] # hand off to the agent, see next step return {"status": "received"} ``` **Expected result:** Sending yourself a test WhatsApp message to the test number logs the correct sender ID and message text. ### 3. Keep per-sender conversation history WhatsApp's API has no concept of a session; every incoming message is just a POST with no memory attached. Keep a dict keyed by the sender's WhatsApp ID holding that lead's message history, and swap it for Redis or a database before going to production, since an in-memory dict resets on every deploy and doesn't survive multiple server instances. ``` CONVERSATIONS: dict[str, list[dict]] = {} def get_history(sender: str) -> list[dict]: return CONVERSATIONS.setdefault(sender, []) ``` **Expected result:** A dict that grows one entry per unique lead, each holding that lead's full message history so far. ### 4. Call Claude with the leasing tools Append the incoming message to that sender's history, call the Messages API with check_availability, book_showing, and escalate_to_human defined as tools, and process any tool_use blocks the same way you would in any other agent. The system prompt sets the one hard rule: escalate instead of guessing the moment a lead sounds ready to make an offer. ``` SYSTEM_PROMPT = ( "You qualify real estate leads over WhatsApp for a 123 Maple listing. " "Answer questions about the listing, check availability before offering a " "showing time, and book it once confirmed. If the lead sounds ready to make " "an offer, or asks anything about price negotiation, call escalate_to_human." ) def ask_claude(sender: str, text: str) -> str: history = get_history(sender) history.append({"role": "user", "content": text}) response = client.messages.create( model="claude-sonnet-5-5", max_tokens=1024, system=SYSTEM_PROMPT, tools=TOOLS, messages=history, ) history.append({"role": "assistant", "content": response.content}) return handle_response(sender, response) ``` **Expected result:** A function that returns the final text reply after resolving any tool calls, ready to send back to WhatsApp. ### 5. Send the reply back through the Cloud API Reply by making a separate POST request to the Cloud API's /messages endpoint for that phone number, with the recipient's WhatsApp ID and your reply text in the body, authenticated with your access token as a bearer token. This is a completely different call than receiving the message; there's no 'respond' field on the webhook itself. ``` import httpx GRAPH_URL = f"https://graph.facebook.com/v26.0/{PHONE_NUMBER_ID}/messages" async def send_whatsapp_message(to: str, body: str): async with httpx.AsyncClient() as http: await http.post(GRAPH_URL, headers={"Authorization": f"Bearer {ACCESS_TOKEN}"}, json={"messaging_product": "whatsapp", "to": to, "type": "text", "text": {"body": body}}) ``` **Expected result:** The reply shows up in the test WhatsApp conversation within a second or two of your webhook processing the message. ### 6. Wire up escalation to a real human When escalate_to_human fires, don't just tell the lead someone will follow up and stop there; actually notify a human, through a Slack webhook, an email, or a CRM task, with the lead's phone number and the reason for the handoff included. The whole point of this tool is that a hot lead gets a real human's attention quickly, not that the bot politely ends the conversation. ``` async def escalate(sender: str, reason: str): # In production: POST to a Slack webhook or create a CRM task here. print(f"[ESCALATION] {sender}: {reason}") ``` **Expected result:** Asking the test number something like 'I want to make an offer today' triggers a logged (or posted) escalation, not a vague bot reply. ### 7. Verify webhook signatures before trusting anything Meta signs every webhook POST with an X-Hub-Signature-256 header, an HMAC over the raw request body using your app secret. Check it before parsing anything from the payload; without this, anyone who finds your webhook URL can POST a fake message and get your agent to act on it. ``` import hashlib, hmac def verify_signature(raw_body: bytes, signature_header: str, app_secret: str) -> bool: expected = hmac.new(app_secret.encode(), raw_body, hashlib.sha256).hexdigest() return hmac.compare_digest(f"sha256={expected}", signature_header) ``` **Expected result:** A function that rejects any webhook POST whose signature doesn't match, before the payload is parsed at all. ## Complete code example File: `whatsapp_lead_agent.py` ```python """WhatsApp lead-qualification agent for a real estate listing. Run: uvicorn whatsapp_lead_agent:app --reload pip install fastapi uvicorn httpx anthropic """ import hashlib import hmac import os import anthropic import httpx from fastapi import FastAPI, Request, Response VERIFY_TOKEN = os.environ["WHATSAPP_VERIFY_TOKEN"] ACCESS_TOKEN = os.environ["WHATSAPP_ACCESS_TOKEN"] PHONE_NUMBER_ID = os.environ["WHATSAPP_PHONE_NUMBER_ID"] APP_SECRET = os.environ["WHATSAPP_APP_SECRET"] GRAPH_URL = f"https://graph.facebook.com/v26.0/{PHONE_NUMBER_ID}/messages" app = FastAPI() client = anthropic.Anthropic() CONVERSATIONS: dict[str, list[dict]] = {} SYSTEM_PROMPT = ( "You qualify real estate leads over WhatsApp for the 123 Maple listing. " "Answer questions about the listing, check availability before offering a " "showing time, and book it once confirmed. If the lead sounds ready to " "make an offer, or asks anything about price negotiation, call " "escalate_to_human instead of handling it yourself." ) TOOLS = [ {"name": "check_availability", "description": "Check open showing slots for the listing.", "input_schema": {"type": "object", "properties": {"date": {"type": "string"}}, "required": ["date"]}}, {"name": "book_showing", "description": "Book a confirmed showing slot.", "input_schema": {"type": "object", "properties": { "date": {"type": "string"}, "time": {"type": "string"}, "name": {"type": "string"}}, "required": ["date", "time", "name"]}}, {"name": "escalate_to_human", "description": "Hand off a hot lead to a human agent.", "input_schema": {"type": "object", "properties": {"reason": {"type": "string"}}, "required": ["reason"]}}, ] def verify_signature(raw_body: bytes, signature_header: str) -> bool: expected = hmac.new(APP_SECRET.encode(), raw_body, hashlib.sha256).hexdigest() return hmac.compare_digest(f"sha256={expected}", signature_header or "") def get_history(sender: str) -> list[dict]: return CONVERSATIONS.setdefault(sender, []) async def escalate(sender: str, reason: str) -> None: # In production: POST to a Slack webhook or create a CRM task here. print(f"[ESCALATION] {sender}: {reason}") def run_tool(name: str, tool_input: dict) -> dict: if name == "check_availability": return {"content": [{"type": "text", "text": "2:00 PM and 4:30 PM are open."}]} if name == "book_showing": return {"content": [{"type": "text", "text": f"Booked {tool_input['date']} at {tool_input['time']}."}]} if name == "escalate_to_human": return {"content": [{"type": "text", "text": "A human agent will follow up shortly."}]} return {"content": [{"type": "text", "text": "Unknown tool."}], "is_error": True} async def ask_claude(sender: str) -> str: history = get_history(sender) response = client.messages.create( model="claude-sonnet-5-5", max_tokens=1024, system=SYSTEM_PROMPT, tools=TOOLS, messages=history, ) history.append({"role": "assistant", "content": response.content}) tool_blocks = [b for b in response.content if b.type == "tool_use"] if not tool_blocks: return "".join(b.text for b in response.content if b.type == "text") results = [] for block in tool_blocks: if block.name == "escalate_to_human": await escalate(sender, block.input.get("reason", "")) result = run_tool(block.name, block.input) results.append({"type": "tool_result", "tool_use_id": block.id, **result}) history.append({"role": "user", "content": results}) follow_up = client.messages.create( model="claude-sonnet-5-5", max_tokens=1024, system=SYSTEM_PROMPT, tools=TOOLS, messages=history, ) history.append({"role": "assistant", "content": follow_up.content}) return "".join(b.text for b in follow_up.content if b.type == "text") async def send_whatsapp_message(to: str, body: str) -> None: async with httpx.AsyncClient() as http: await http.post( GRAPH_URL, headers={"Authorization": f"Bearer {ACCESS_TOKEN}"}, json={"messaging_product": "whatsapp", "to": to, "type": "text", "text": {"body": body}}, ) @app.get("/webhook") async def verify(request: Request): params = request.query_params if params.get("hub.verify_token") == VERIFY_TOKEN: return int(params["hub.challenge"]) return Response(status_code=403) @app.post("/webhook") async def receive(request: Request): raw_body = await request.body() if not verify_signature(raw_body, request.headers.get("X-Hub-Signature-256", "")): return Response(status_code=403) body = await request.json() entry = body["entry"][0]["changes"][0]["value"] if "messages" not in entry: return {"status": "ignored"} msg = entry["messages"][0] sender, text = msg["from"], msg["text"]["body"] get_history(sender).append({"role": "user", "content": text}) reply = await ask_claude(sender) await send_whatsapp_message(sender, reply) return {"status": "ok"} ``` ## Common mistakes - **Forgetting that WhatsApp expects your webhook to respond quickly** — undefined Fix: Return a fast 200 and do slow work (LLM calls, tool execution) inside the request handler asynchronously, or hand it to a background task. A slow webhook response triggers Meta's retry logic and can duplicate messages. - **Trusting the webhook payload without checking X-Hub-Signature-256** — undefined Fix: Verify the HMAC signature against your app secret before parsing anything. An unverified webhook URL can be POSTed to by anyone who finds it. - **Keeping conversation history only in an in-memory dict in production** — undefined Fix: Move it to Redis or a database before going live. An in-memory dict loses every lead's history on every deploy and doesn't work across multiple server instances. - **Trying to send a free-form reply more than 24 hours after the lead's last message** — undefined Fix: Use a pre-approved template message to re-engage outside the 24-hour window. A plain text reply outside that window will be rejected by the API. - **Having the agent try to close or negotiate instead of escalating** — undefined Fix: Keep escalate_to_human in the system prompt's explicit scope boundary, and make sure it actually notifies a person, not just a logged print statement left in from testing. ## Best practices - Verify every webhook's signature before parsing its body, no exceptions. - Acknowledge the webhook fast and do the real work asynchronously so Meta's retry logic doesn't fire. - Store per-sender conversation history somewhere durable and shared across server instances. - Make escalate_to_human actually reach a human, through Slack, email, or a CRM task, not just a log line. - Track the 24-hour messaging window per sender and fall back to a template message once it's expired. - Keep the system prompt's scope boundary explicit: what the agent can decide versus what it must hand off. ## Use cases ### undefined ### undefined ### undefined ## Frequently asked questions ### Do I need the WhatsApp Business app, or does the Cloud API work differently? This guide uses Meta's WhatsApp Cloud API, which is a REST API your server calls directly, separate from the consumer WhatsApp Business app. You don't need the app installed anywhere; you need a Meta developer account, a test or production phone number, and an access token. ### What's the 24-hour messaging window, exactly? You can send free-form replies to a user for 24 hours after their last message to you. After that window closes, you can only re-engage them with a pre-approved template message, not a plain text reply, until they message you again. ### How do I test this before I have a production phone number? Meta's developer console provides a free test phone number tied to your app. Use a tunnel like ngrok to expose your local FastAPI server's webhook URL, register that URL in the console, and message the test number directly from your own WhatsApp. ### Why verify X-Hub-Signature-256 instead of just trusting the webhook? Your webhook URL is a public HTTPS endpoint. Without signature verification, anyone who discovers it can POST a fake message payload and have your agent act on it, including triggering tools like book_showing with made-up data. ### Can this agent send images or quick-reply buttons, not just text? Yes, the Cloud API supports interactive messages, images, documents, and location sharing in addition to plain text; this guide builds the text-only path first since it covers the harder architectural pieces (webhooks, history, tools) that every message type shares. ### Can someone build and host this WhatsApp agent for my business? Yes. RapidDev builds production WhatsApp agents, including signature verification, durable conversation storage, and real escalation routing. A free 30-minute consultation is the fastest way to scope your version. --- Source: https://www.rapidevelopers.com/build-ai-agent/how-to-build-a-whatsapp-ai-agent © RapidDev — https://www.rapidevelopers.com/build-ai-agent/how-to-build-a-whatsapp-ai-agent