WhatsApp — Single-WABA Inbound + Human Handoff

pip install maximem-synap maximem-synap-openai-agents openai-agents heyoo fastapi uvicorn

# .env
SYNAP_API_KEY=...
SYNAP_SERVER_URL=<maximem-server>
OPENAI_API_KEY=...
WABA_PHONE_NUMBER_ID=...
WABA_ACCESS_TOKEN=...
WABA_WEBHOOK_VERIFY_TOKEN=...

# Set when AI hands off; cleared when human releases. Use Redis in production.
AI_HANDOFF: set[str] = set()

def is_handoff(phone: str) -> bool:
    return phone in AI_HANDOFF

def start_handoff(phone: str) -> None:
    AI_HANDOFF.add(phone)

def end_handoff(phone: str) -> None:
    AI_HANDOFF.discard(phone)

@function_tool
async def hand_off_to_human(phone: str, reason: str, summary: str) -> dict:
    """Bring a human into the thread. Reason: 'customer_asked' | 'frustration' | 'out_of_scope' | 'sensitive'."""
    start_handoff(phone)
    await sdk.memories.create(
        document=f"AI handed off to human. Reason: {reason}. Summary: {summary}",
        document_type="support-handoff",
        user_id=phone,
        customer_id=CUSTOMER_ID,
        metadata={"handoff_reason": reason},
    )
    await notify_human_console(phone, reason, summary)  # your Slack/console push
    return {"status": "handed_off"}

from fastapi import FastAPI, Request, BackgroundTasks
from heyoo import WhatsApp

app = FastAPI()
wa = WhatsApp(token=os.environ["WABA_ACCESS_TOKEN"], phone_number_id=os.environ["WABA_PHONE_NUMBER_ID"])
CUSTOMER_ID = "your-business"

@app.post("/webhook/whatsapp")
async def webhook(request: Request, bg: BackgroundTasks):
    body = await request.json()
    msg = wa.get_message(body)
    phone = wa.get_mobile(body)
    if msg and phone:
        bg.add_task(handle_inbound, phone, msg)
    return {"ok": True}

async def handle_inbound(phone: str, text: str):
    # Always ingest the inbound — even during handoff
    await sdk.memories.create(
        document=f"Customer: {text}",
        document_type="ai-chat-conversation",
        user_id=phone,
        customer_id=CUSTOMER_ID,
        metadata={"channel": "whatsapp", "direction": "inbound"},
    )

    if is_handoff(phone):
        return  # human owns the thread; AI stays quiet

    reply = await run_ai(phone, text)
    wa.send_message(reply, phone)

    await sdk.memories.create(
        document=f"Agent: {reply}",
        document_type="ai-chat-conversation",
        user_id=phone,
        customer_id=CUSTOMER_ID,
        metadata={"channel": "whatsapp", "direction": "outbound", "tier": "ai"},
    )

SYSTEM = """You are a WhatsApp business agent.

- Be concise — WhatsApp users want short replies.
- Recall the customer's history and prior preferences from memory.
- Hand off to a human when: the customer explicitly asks, they're frustrated, the issue is sensitive
  (financial, medical, legal), or you've tried twice without resolution. Use hand_off_to_human.
- Don't promise anything you can't deliver via tools."""

async def run_ai(phone: str, text: str) -> str:
    synap_search = create_search_tool(sdk=sdk, user_id=phone, customer_id=CUSTOMER_ID)
    synap_store  = create_store_tool(sdk=sdk, user_id=phone, customer_id=CUSTOMER_ID)

    agent = Agent(
        name="wa_agent",
        instructions=SYSTEM,
        tools=[
            FunctionTool(synap_search, name_override="synap_search"),
            FunctionTool(synap_store,  name_override="synap_store"),
            hand_off_to_human,
            # ...your business tools
        ],
    )
    result = await Runner.run(agent, input=text)
    return result.final_output

async def human_reply(phone: str, text: str):
    wa.send_message(text, phone)
    await sdk.memories.create(
        document=f"Human agent: {text}",
        document_type="ai-chat-conversation",
        user_id=phone,
        customer_id=CUSTOMER_ID,
        metadata={"channel": "whatsapp", "direction": "outbound", "tier": "human"},
    )

async def human_release(phone: str):
    end_handoff(phone)
    await sdk.memories.create(
        document="Human released thread back to AI.",
        document_type="support-handoff",
        user_id=phone,
        customer_id=CUSTOMER_ID,
        metadata={"handoff_event": "released"},
    )