WhatsApp Assistant

This guide shows how to create a WhatsApp assistant with Timbal that can:

Respond to messages in real time via webhooks
Persist conversations using Timbal Knowledge Bases or JSONL
Handle text messages and preserve metadata for audio, images, and documents

The runnable example answers text messages. Media IDs and filenames are placeholders until you add downloading and transcription or file analysis. Install timbal[server] and httpx; add pyngrok only if you use the optional tunnel helper. This demonstration processes the agent inside the webhook request. For production, authenticate webhooks, deduplicate message IDs, acknowledge promptly, and process through a durable queue.

Setup Requirements

Meta WhatsApp App and Webhook

Create or open your WhatsApp App

Go to the Meta console and open your app’s WhatsApp section: Meta WhatsApp App Dashboard.

Set Callback URL and Verify Token

Set the Callback URL to your endpoint, e.g., https://yourdomain.com/
Set a Verify Token (keep it secret)
In your app code, use the same token (see WHATSAPP_VERIFY_TOKEN in the example below)
Click Verify and Save

Subscribe to fields

In the WhatsApp product settings, subscribe to message-related events (e.g., messages, statuses).

Environment Variables

Create a .env file and set:

.env

# Timbal platform (if using KB persistence)
ORG_ID=your_org_id
KB_ID=your_kb_id
TIMBAL_API_HOST=api.timbal.ai
TIMBAL_API_KEY=your_timbal_key

# WhatsApp API
WHATSAPP_ACCESS_TOKEN=your_long_lived_token
WHATSAPP_PHONE_NUMBER_ID=your_phone_number_id
# Set a Graph API version supported by your Meta app, including the v prefix
WHATSAPP_GRAPH_API_VERSION=your_supported_version

# Webhook verification
WHATSAPP_VERIFY_TOKEN=your_verify_token

# Server
HOST=0.0.0.0
PORT=4343

# Optional: enable ngrok tunnel
ENABLE_NGROK=true

The main example stores JSONL. The optional KB section replaces that storage; setting environment variables alone does not switch it. Replace your_supported_version with your app’s Graph API version.

Minimal Webhook App (FastAPI)

Use a simple FastAPI app to receive the webhook and pass the full payload to the agent. The agent will process everything in pre_hook and send responses in post_hook.

app.py

from fastapi import FastAPI, Request, Response
import os
import uvicorn
from dotenv import load_dotenv

load_dotenv()
from agent import agent

app = FastAPI()

VERIFY_TOKEN = os.getenv("WHATSAPP_VERIFY_TOKEN")

@app.get("/")
async def verify(request: Request):
    token = request.query_params.get("hub.verify_token")
    challenge = request.query_params.get("hub.challenge")
    if VERIFY_TOKEN and token == VERIFY_TOKEN and challenge:
        return Response(content=challenge, media_type="text/plain")
    return Response(content="Verification failed", status_code=403)

@app.post("/")
async def webhook(request: Request):
    _webhook = await request.json()
    await agent(_webhook=_webhook).collect()
    return {"status": "ok"}

def _maybe_start_ngrok(port: int) -> None:
    enable = os.getenv("ENABLE_NGROK", "false")
    if enable != "true":
        return
    try:
        from pyngrok import ngrok
        public_url = ngrok.connect(port, "http")
        print(f"ngrok public url: {public_url}")
    except Exception as e:
        print(f"Failed to start ngrok: {e}")

if __name__ == "__main__":
    host = os.getenv("HOST", "0.0.0.0")
    try:
        port = int(os.getenv("PORT", "4343"))
    except Exception:
        port = 4343
    _maybe_start_ngrok(port)
    uvicorn.run("app:app", host=host, port=port, log_level="info")

Expose locally (optional):

ngrok http 4343

Then set the ngrok HTTPS URL as the Callback URL in the Meta console.

Creating the Agent

agent.py

import os
import json
from datetime import datetime
import httpx
from timbal import Agent
from timbal.errors import bail
from timbal.state import get_run_context

async def send_whatsapp_message(to: str, message: str) -> dict:
    version = os.environ["WHATSAPP_GRAPH_API_VERSION"]
    phone_id = os.environ["WHATSAPP_PHONE_NUMBER_ID"]
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"https://graph.facebook.com/{version}/{phone_id}/messages",
            headers={"Authorization": f"Bearer {os.environ['WHATSAPP_ACCESS_TOKEN']}"},
            json={"messaging_product": "whatsapp", "to": to,
                  "type": "text", "text": {"body": message}},
            timeout=10,
        )
        response.raise_for_status()
        return response.json()

def _value(payload: dict) -> dict | None:
    try:
        return payload["entry"][0]["changes"][0]["value"]
    except Exception:
        return None

def _prompt_from_message(msg: dict) -> str:
    t = msg.get("type")
    if t == "text":
        return (msg.get("text") or {}).get("body", "").strip()
    if t == "image":
        mid = (msg.get("image") or {}).get("id")
        cap = (msg.get("image") or {}).get("caption", "")
        return f"[image:{mid}] {cap}".strip()
    if t == "audio":
        mid = (msg.get("audio") or {}).get("id")
        return f"[audio:{mid}]"
    if t == "document":
        fn = (msg.get("document") or {}).get("filename", "document")
        return f"[document:{fn}]"
    return f"[{t or 'unknown'}]"

def _append_jsonl(path: str, rec: dict) -> None:
    try:
        with open(path, "a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    except Exception:
        pass

def _recent_history(path: str, user_phone: str, limit: int = 10) -> list[str]:
    out = []
    try:
        if not os.path.exists(path):
            return []
        with open(path, "r", encoding="utf-8") as f:
            rows = [json.loads(l) for l in f if l.strip()]
            for r in reversed(rows):
                if r.get("user_phone") == user_phone:
                    try:
                        payload = json.loads(r.get("message") or "{}")
                        out.append(f"{r.get('direction')}: {payload.get('content','')}")
                    except Exception:
                        continue
                    if len(out) >= limit:
                        break
        return list(reversed(out))
    except Exception:
        return []

def pre_hook():
    """Build professional prompt from WhatsApp metadata, message, and optional history."""
    span = get_run_context().current_span()
    payload = span.input.get("_webhook")
    if not payload:
        bail("No payload found in webhook")

    value = _value(payload)
    if not value:
        bail("No value found in webhook")

    messages = value.get("messages", [])
    if not messages:
        bail("No messages found in webhook")

    contacts = value.get("contacts", [])
    profile = (contacts[0].get("profile") or {}) if contacts else {}
    name = profile.get("name", "")
    msg = messages[0]
    from_number = msg.get("from")
    message_id = msg.get("id")
    phone_number_id = (value.get("metadata") or {}).get("phone_number_id")
    prompt = _prompt_from_message(msg)
    msg_type = msg.get("type")

    rec = {
        "id": msg.get("id"),
        "user_phone": from_number,
        "direction": "inbound",
        "message_type": msg_type,
        "message": json.dumps({"type": msg_type, "content": prompt}, ensure_ascii=False),
        "timestamp": msg.get("timestamp"),
        "conversation_id": f"{phone_number_id}_{from_number}",
        "user_name": name,
    }
    _append_jsonl("whatsapp_messages.jsonl", rec)

    span.input["prompt"] = _recent_history("whatsapp_messages.jsonl", user_phone=from_number, limit=10)
    span.input["whatsapp_from_number"] = from_number
    span.input["name"] = name

The pre_hook runs before invoking the agent. It extracts metadata from the webhook, builds a prompt from the incoming message, persists the inbound event, and sets values in span.input for the agent to use during inference.

agent.py

async def post_hook():
    """Send the agent response via WhatsApp and persist outbound to JSONL."""
    span = get_run_context().current_span()
    from_number = span.input.get("whatsapp_from_number")
    if not from_number:
        return
    try:
        response_text = (span.output.collect_text() or "").strip()
    except Exception:
        return
    if not response_text:
        return
    await send_whatsapp_message(to=from_number, message=response_text)
    _append_jsonl("whatsapp_messages.jsonl", {
        "id": f"outbound_{from_number}_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
        "user_phone": from_number,
        "direction": "outbound",
        "message_type": "agent",
        "message": json.dumps({"type": "text", "content": response_text}, ensure_ascii=False),
        "timestamp": datetime.now().isoformat(),
        "conversation_id": "",
        "user_name": span.input.get("name", ""),
    })

The post_hook runs after obtaining the agent’s response. It sends the text via WhatsApp and persists the outbound event to JSONL.

agent.py

agent = Agent(
    name="WhatsAppAgent",
    description="A minimal WhatsApp agent for Timbal",
    model="openai/gpt-4.1-mini",
    pre_hook=pre_hook,
    post_hook=post_hook,
)

Optional: system prompt example

Show example system_prompt

title: "WHATSAPP AGENT from {{YOUR_BRAND}}",
description: "You are a WhatsApp agent from {{YOUR_BRAND}}. Your goal is to help users find the perfect vehicle and manage all their automotive needs.",

## WhatsApp message format
You are responding to messages in WhatsApp. Use WhatsApp's text formatting syntax when appropriate:

Text format:
- *text* = bold (for important information)
- _text_ = italic (for clarifications or soft emphasis)
- ~text~ = strikethrough (for corrections or obsolete information)
- `text` = monospace (for code, commands or technical text)
- Use triple backticks for code blocks (for longer fragments)
- > text = quote (for references or highlighted information)
- Important: Never use [View here](url) or ![Image](url). Send the direct link instead; WhatsApp does not render these markdown forms.

Important rules:
- Symbols must be attached to the text (no spaces)
- Every opening symbol must have its closing symbol
- You can combine formats: *_text_* for bold+italic
- Use sparingly; only when it adds value
- For lists use bullets (•), numbers (1.), or hyphens (-)

Optional: Timbal KB persistence and history

If you prefer Timbal Knowledge Bases instead of JSONL, use the async SQL helper below and replace the storage calls in async pre/post hooks. Create the whatsapp_messages table with the listed columns first. The helper targets K2; pass legacy=True for a legacy KB. User values are SQL parameters.

import json
import os
from timbal.platform.knowledge_bases import query

ORG_ID = os.environ["ORG_ID"]
KB_ID = os.environ["KB_ID"]

async def save_record_to_kb(record: dict) -> None:
    await query(
        """INSERT INTO whatsapp_messages
           (id, user_phone, direction, message_type, message, timestamp, conversation_id, user_name)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8)""",
        params=[record.get(key) for key in (
            "id", "user_phone", "direction", "message_type", "message",
            "timestamp", "conversation_id", "user_name",
        )],
        org_id=ORG_ID, kb_id=KB_ID,
    )

async def fetch_recent_history_from_kb(user_phone: str, limit: int = 10) -> list[str]:
    result = await query("""
        SELECT direction, message
        FROM whatsapp_messages
        WHERE user_phone = $1
        ORDER BY timestamp DESC
        LIMIT $2
    """, params=[user_phone, limit], org_id=ORG_ID, kb_id=KB_ID)
    lines: list[str] = []
    for r in reversed(result.get("rows", [])):
        try:
            payload = json.loads(r.get("message") or "{}")
            lines.append(f"{r.get('direction')}: {payload.get('content','')}")
        except Exception:
            continue
    return lines

Media Handling Notes

Images: The webhook includes an image id. Use the Graph API to fetch the media URL or content using your WHATSAPP_ACCESS_TOKEN. Provide the URL or a File object to the agent if needed.
Audio: The webhook includes an audio id. You can fetch and transcribe audio, then pass the transcript as the prompt.
Documents: You receive basic metadata (e.g., filename). Decide whether to fetch and analyze content or treat as a reference.

Keep the pre_hook minimal: extract the message type, create a concise prompt, and persist the event.

Complete implementation

For additional webhook and utility examples, see the Timbal repository. These are separate examples and may use different configuration from this guide:

timbal/examples/whatsapp

Possible Improvements

Delay response to batch context (debounce)

Each WhatsApp webhook starts a separate agent run. Persist every inbound message immediately (JSONL or KB), then wait a short debounce window. After that window, if storage shows a newer message for the same user, bail this run so the most recent run answers with full context. No in-memory buffer is required.

Why it helps: avoids replying to partial context when the user is still typing or sending media.
Trade-offs: adds small latency; choose a short window (e.g., 800–1500 ms).
Example approach (JSONL):

import asyncio, os, json

DEBOUNCE_MS = int(os.getenv("WHATSAPP_DEBOUNCE_MS", "1200"))

def _message_ts(msg: dict) -> int:
    try:
        return int(msg.get("timestamp") or 0)
    except Exception:
        return 0

def _newer_exists_in_jsonl(path: str, user_phone: str, after_ts: int) -> bool:
    try:
        if not os.path.exists(path):
            return False
        with open(path, "r", encoding="utf-8") as f:
            rows = [json.loads(l) for l in f if l.strip()]
        for rec in reversed(rows):
            if rec.get("user_phone") != user_phone:
                continue
            if rec.get("direction") != "inbound":
                continue
            try:
                ts = int(rec.get("timestamp") or 0)
            except Exception:
                continue
            if ts > after_ts:
                return True
        return False
    except Exception:
        return False

async def pre_hook():
    span = get_run_context().current_span()
    payload = span.input.get("_webhook") or {}
    value = (payload.get("entry") or [{}])[0].get("changes", [{}])[0].get("value", {})
    msg = (value.get("messages") or [None])[0] or {}
    from_number = msg.get("from", "")
    current_ts = _message_ts(msg)

    # 1) Persist inbound immediately (example: JSONL)
    _append_jsonl("whatsapp_messages.jsonl", {
        "id": msg.get("id"),
        "user_phone": from_number,
        "direction": "inbound",
        "message_type": msg.get("type"),
        "message": json.dumps({"type": msg.get("type"), "content": _prompt_from_message(msg)}, ensure_ascii=False),
        "timestamp": str(current_ts),
        "conversation_id": (value.get("metadata") or {}).get("phone_number_id", ""),
        "user_name": (value.get("contacts") or [{}])[0].get("profile", {}).get("name", ""),
    })

    # 2. Wait a small window, then check for a newer message
    await asyncio.sleep(DEBOUNCE_MS / 1000.0)
    if _newer_exists_in_jsonl("whatsapp_messages.jsonl", from_number, current_ts):
        bail("Newer message detected; aborting this run")

    # 3. Proceed with prompt building from recent history
    span.input["prompt"] = _recent_history("whatsapp_messages.jsonl", user_phone=from_number, limit=10)

Mark inbound message as read

Use the WhatsApp Cloud API to mark messages as read. This improves UX and clears unread indicators for the user.

import os, requests

def mark_as_read(message_id: str) -> None:
    token = os.getenv("WHATSAPP_ACCESS_TOKEN")
    phone_id = os.getenv("WHATSAPP_PHONE_NUMBER_ID")
    try:
        requests.post(
            f"https://graph.facebook.com/{os.environ['WHATSAPP_GRAPH_API_VERSION']}/{phone_id}/messages",
            headers={"Authorization": f"Bearer {token}"},
            json={
                "messaging_product": "whatsapp",
                "status": "read",
                "message_id": message_id,
            },
            timeout=10,
        )
    except Exception:
        pass

Call mark_as_read(message_id) right after parsing the webhook.

Progress updates (typing indicator and ack)

You can mark the message as read and briefly show a typing indicator while the agent thinks or uses tools. Optionally send a short acknowledgement, then follow up with the final answer.This request follows Meta’s typing indicator example. The synchronous helper below should run in a worker, rather than block an async webhook handler.

import os, httpx

def send_typing_and_read(message_id: str) -> dict:
    token = os.getenv("WHATSAPP_ACCESS_TOKEN")
    phone_id = os.getenv("WHATSAPP_PHONE_NUMBER_ID")
    if not token or not phone_id:
        return {}

    url = f"https://graph.facebook.com/{os.environ['WHATSAPP_GRAPH_API_VERSION']}/{phone_id}/messages"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }

    with httpx.Client() as client:
        # Mark as read and display the typing indicator in one request
        resp = client.post(
            url,
            headers=headers,
            json={
                "messaging_product": "whatsapp",
                "status": "read",
                "message_id": message_id,
                "typing_indicator": {"type": "text"},
            },
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

Call this right after parsing the webhook and before long operations.
Keep interim notifications minimal to avoid noise.

Persist and reuse media in the KB

Persist metadata for images, audio, videos, and documents so the agent can reuse them as context later (e.g., to re-attach an image or analyze it again).Suggested fields to store: id, user_phone, direction, message_type, media_id, media_url (or object store key), mime_type, caption, timestamp.

async def save_media_example(record: dict) -> None:
    # Merge this idea with your existing save flow
    await save_record_to_kb({
        **record,
        "message_type": "image",  # or "video", "audio", "document"
        # e.g. values fetched via Graph API media URL endpoint
        "media_id": record.get("media_id"),
        "media_url": record.get("media_url"),
        "mime_type": record.get("mime_type"),
        "caption": record.get("caption", ""),
    })

Later, query recent media by user_phone or conversation_id and inject URLs or file handles into the agent context as needed.

Running

python app.py

Expose with ngrok, then verify the webhook in the Meta console. New messages should invoke your agent and send responses via WhatsApp.

Troubleshooting

Webhook verification fails

Verify token: Ensure the value in Meta matches VERIFY_TOKEN in your app
Public URL: Your callback must be publicly reachable (use ngrok)
HTTPS: Required in production

Messages not sent

Tokens: Confirm WHATSAPP_ACCESS_TOKEN and WHATSAPP_PHONE_NUMBER_ID are set
Phone ID: Ensure your app uses the correct phone number ID
Permissions: Check token permissions in Access Token Debugger

No context/history

Check ORG_ID and KB_ID if using Timbal KB
If using JSONL, verify the JSONL file path and write permissions

Examples

Guides

Agents

Workflows

WhatsApp Assistant

Setup Requirements

Minimal Webhook App (FastAPI)

Creating the Agent

Optional: system prompt example

Optional: Timbal KB persistence and history

Media Handling Notes

Complete implementation

Possible Improvements

Running

Troubleshooting

​Setup Requirements

​Minimal Webhook App (FastAPI)

​Creating the Agent

​Optional: system prompt example

​Optional: Timbal KB persistence and history

​Media Handling Notes

​Complete implementation

​Possible Improvements

​Running

​Troubleshooting

Setup Requirements

Minimal Webhook App (FastAPI)

Creating the Agent

Optional: system prompt example

Optional: Timbal KB persistence and history

Media Handling Notes

Complete implementation

Possible Improvements

Running

Troubleshooting