WhatsApp: Single-WABA Inbound + Outbound Campaign

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

uv add maximem-synap maximem-synap-openai-agents openai-agents heyoo celery redis
# pip-compatible (existing venv): uv pip install maximem-synap maximem-synap-openai-agents openai-agents heyoo celery redis

npm install @maximem/synap-js-sdk @maximem/synap-vercel-adk ai @ai-sdk/openai bullmq ioredis

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"]}

// Worker (BullMQ)
import { Worker } from "bullmq";

new Worker("campaigns", async (job) => {
  const { campaignId, recipientPhone, vars } = job.data;
  if (await isOptedOut(recipientPhone)) return { skipped: "opted_out" };

  const waResponse = await sendWaTemplate(recipientPhone, campaignId, vars);

  await synap.sdk.memories.create({
    document: `Outbound campaign: ${campaignId}\nVariables: ${JSON.stringify(vars)}`,
    documentType: "campaign-send",
    userId: recipientPhone,
    customerId: CUSTOMER_ID,
    metadata: {
      campaignId,
      waMessageId: waResponse.messages[0].id,
      direction: "outbound",
    },
  });
  return { sent: waResponse.messages[0].id };
}, { connection: redis });

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

export const runtime = "nodejs";

export async function POST(req: Request) {
  const body = await req.json();
  for (const status of iterStatuses(body)) {
    recordDelivery(status).catch(console.error);
  }
  for (const inbound of iterInbound(body)) {
    if (isOptout(inbound.text)) {
      markOptedOut(inbound.from);
      sendWaMessage(inbound.from, "You're opted out. Reply START to opt back in.");
      continue;
    }
    handleReply(inbound.from, inbound.text, inbound.context).catch(console.error);
  }
  return Response.json({ ok: true });
}

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

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

- The recipient may be responding to a recent campaign. Check memory for a recent
  campaign-send document; if it's within the last 7 days, treat the reply as a response
  to that campaign.
- Otherwise respond from general memory.
- Keep replies under 2 sentences when possible.`;

async function handleReply(phone: string, text: string, ctx: any) {
  await synap.sdk.memories.create({
    document: `Customer: ${text}`,
    documentType: "ai-chat-conversation",
    userId: phone,
    customerId: CUSTOMER_ID,
    metadata: {
      channel: "whatsapp",
      direction: "inbound",
      inReplyToWaId: ctx?.id,
    },
  });

  const model = synap.wrap(openai("gpt-4o-mini"), {
    userId: phone,
    customerId: CUSTOMER_ID,
    conversationId: phone,
  });

  const { text: reply } = await generateText({
    model,
    system: SYSTEM,
    prompt: text,
    // tools: { ...your business tools }
  });

  await sendWaMessage(phone, reply);
  await synap.sdk.memories.create({
    document: `Agent: ${reply}`,
    documentType: "ai-chat-conversation",
    userId: phone,
    customerId: 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}"))

const OPT_OUT_KEYWORDS = new Set(["STOP", "UNSUBSCRIBE", "REMOVE", "QUIT"]);
const isOptout = (text: string) => OPT_OUT_KEYWORDS.has(text.trim().toUpperCase());

async function markOptedOut(phone: string) {
  await redis.set(`optout:${phone}`, "1");
  await synap.sdk.memories.create({
    document: "Recipient opted out of campaigns.",
    documentType: "campaign-optout",
    userId: phone,
    customerId: CUSTOMER_ID,
  });
}

const isOptedOut = async (phone: string) => Boolean(await redis.get(`optout:${phone}`));