WhatsApp — Single-WABA Inbound + Outbound Campaign

pip install maximem-synap maximem-synap-openai-agents openai-agents heyoo celery redis

from celery import Celery
celery = Celery("campaigns", broker=os.environ["REDIS_URL"])

@celery.task
async def send_campaign(campaign_id: str, recipient_phone: str, vars: dict):
    if await is_opted_out(recipient_phone):
        return {"skipped": "opted_out"}

    # 1. Send the template via WABA
    wa_response = wa.send_template(
        recipient_id=recipient_phone,
        template=campaign_id,            # e.g. "may_relaunch_v3"
        components=template_components(vars),
    )

    # 2. Ingest the outbound into Synap so the inbound agent has context
    await sdk.memories.create(
        document=f"Outbound campaign: {campaign_id}\nVariables: {vars}",
        document_type="campaign-send",
        user_id=recipient_phone,
        customer_id=CUSTOMER_ID,
        metadata={
            "campaign_id": campaign_id,
            "wa_message_id": wa_response["messages"][0]["id"],
            "direction": "outbound",
        },
    )
    return {"sent": wa_response["messages"][0]["id"]}

@app.post("/webhook/whatsapp")
async def webhook(request: Request, bg: BackgroundTasks):
    body = await request.json()

    # Delivery / read receipts
    for status in iter_statuses(body):
        bg.add_task(record_delivery, status)

    # Inbound messages
    for inbound in iter_inbound(body):
        if is_optout(inbound["text"]):
            bg.add_task(mark_opted_out, inbound["from"])
            wa.send_message("You're opted out. Reply START to opt back in.", inbound["from"])
            continue
        bg.add_task(handle_reply, inbound["from"], inbound["text"], inbound["context"])

    return {"ok": True}

async def record_delivery(status: dict):
    await sdk.memories.create(
        document=f"Delivery status: {status['status']}",
        document_type="campaign-delivery",
        user_id=status["recipient_id"],
        customer_id=CUSTOMER_ID,
        metadata={"wa_message_id": status["id"], "status": status["status"]},
    )

SYSTEM = """You are a WhatsApp business agent. Your replies are short.

- The recipient may be responding to a recent campaign. Search memory for the most recent
  `campaign-send` document — if it's within the last 7 days, treat their reply as a
  response to that campaign and act on the campaign's intent.
- If they're asking a generic support question unrelated to recent campaigns, respond
  from general memory.
- Always honor opt-out (STOP, UNSUBSCRIBE, REMOVE) — but the webhook handles those before
  you ever see them, so if you receive the message it's not an opt-out.
- Keep replies under 2 sentences when possible."""

async def handle_reply(phone: str, text: str, ctx: dict):
    # Ingest the inbound first
    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",
            "in_reply_to_wa_id": ctx.get("id"),  # WA threads replies to specific msg
        },
    )

    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_reply_agent",
        instructions=SYSTEM,
        tools=[
            FunctionTool(synap_search, name_override="synap_search"),
            FunctionTool(synap_store,  name_override="synap_store"),
            # ...your business tools
        ],
    )
    result = await Runner.run(agent, input=text)
    reply = result.final_output

    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"},
    )

OPT_OUT_KEYWORDS = {"STOP", "UNSUBSCRIBE", "REMOVE", "QUIT"}

def is_optout(text: str) -> bool:
    return text.strip().upper() in OPT_OUT_KEYWORDS

async def mark_opted_out(phone: str):
    await redis.set(f"optout:{phone}", "1")
    await sdk.memories.create(
        document="Recipient opted out of campaigns.",
        document_type="campaign-optout",
        user_id=phone,
        customer_id=CUSTOMER_ID,
    )

async def is_opted_out(phone: str) -> bool:
    return bool(await redis.get(f"optout:{phone}"))