Turn Inbound Email into Structured, Actionable Events

import express from 'express';
import crypto from 'crypto';

const app = express();
app.use(express.json());

const WEBHOOK_SECRET = process.env.MULTIMAIL_WEBHOOK_SECRET!;

app.post('/webhooks/inbound', async (req, res) => {
  const sig = req.headers['x-multimail-signature'] as string;
  const body = JSON.stringify(req.body);
  const expected = crypto
    .createHmac('sha256', WEBHOOK_SECRET)
    .update(body)
    .digest('hex');

  if (!crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected))) {
    return res.status(401).json({ error: 'invalid signature' });
  }

  const { event, data } = req.body;
  if (event !== 'inbound_email') return res.sendStatus(204);

  const { message_id, thread_id, from, subject } = data;
  console.log(`Inbound: ${message_id} thread=${thread_id} from=${from.email} subject="${subject}"`);

  "cm">// Hand off to agent pipeline asynchronously
  processInbound(message_id, thread_id).catch(console.error);

  res.sendStatus(202);
});

async function processInbound(messageId: string, threadId: string) {
  "cm">// Fetch full thread, classify, route — see next samples
}

app.listen(3000);

import os
import anthropic
import httpx

MM_API_KEY = os.environ["MULTIMAIL_API_KEY"]
MM_BASE = "https://api.multimail.dev"
headers = {"Authorization": f"Bearer {MM_API_KEY}"}

client = anthropic.Anthropic()

def classify_and_tag(message_id: str, thread_id: str):
    "cm"># Fetch full thread
    thread_resp = httpx.get(
        f"{MM_BASE}/get_thread",
        params={"thread_id": thread_id},
        headers=headers,
    )
    thread_resp.raise_for_status()
    thread = thread_resp.json()

    "cm"># Build context from thread messages
    thread_text = "\n---\n".join(
        f"From: {m[&"cm">#039;from']['email']}\nDate: {m['date']}\n\n{m['body_plain']}"
        for m in thread["messages"]
    )

    # Classify with Claude
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=256,
        system=(
            "Classify this email thread into exactly one category: "
            "support_request, billing_inquiry, feature_request, spam, or escalation_required. "
            "Reply with only the category label."
        ),
        messages=[{"role": "user", "content": thread_text}],
    )
    intent = resp.content[0].text.strip()

    # Write classification back to MultiMail
    tag_resp = httpx.post(
        f"{MM_BASE}/tag_email",
        headers=headers,
        json={"message_id": message_id, "tags": [intent, "ai-classified"]},
    )
    tag_resp.raise_for_status()
    print(f"Tagged {message_id} as {intent}")
    return intent

def respond_or_escalate(message_id: str, thread_id: str, intent: str):
    AUTONOMOUS_INTENTS = {"support_request", "billing_inquiry"}
    ESCALATE_INTENTS = {"escalation_required"}

    if intent in ESCALATE_INTENTS:
        resp = httpx.post(
            f"{MM_BASE}/decide_email",
            headers=headers,
            json={
                "message_id": message_id,
                "action": "escalate",
                "reason": f"Classified as {intent} — routing to human reviewer",
            },
        )
        resp.raise_for_status()
        print(f"Escalated {message_id}")
        return

    if intent not in AUTONOMOUS_INTENTS:
        print(f"No action configured for intent &"cm">#039;{intent}', skipping")
        return

    # Fetch thread for reply context
    thread = httpx.get(
        f"{MM_BASE}/get_thread",
        params={"thread_id": thread_id},
        headers=headers,
    ).json()

    latest = thread["messages"][-1]
    thread_text = "\n---\n".join(m["body_plain"] for m in thread["messages"])

    draft_resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        system="Draft a concise, helpful email reply. Do not include a subject line.",
        messages=[{"role": "user", "content": thread_text}],
    )
    reply_body = draft_resp.content[0].text.strip()

    send_resp = httpx.post(
        f"{MM_BASE}/reply_email",
        headers=headers,
        json={
            "message_id": message_id,
            "body": reply_body,
            "from_mailbox": "[email protected]",
        },
    )
    send_resp.raise_for_status()
    print(f"Replied to {message_id}")

import time

def poll_and_process(mailbox: str, interval_seconds: int = 30):
    """
    Fallback poller for environments where webhook delivery is unreliable.
    Prefer webhooks in production — polling introduces latency.
    """
    seen: set[str] = set()

    while True:
        resp = httpx.get(
            f"{MM_BASE}/check_inbox",
            headers=headers,
            params={
                "mailbox": mailbox,
                "unread_only": True,
                "limit": 50,
            },
        )
        resp.raise_for_status()
        messages = resp.json().get("messages", [])

        for msg in messages:
            if msg["message_id"] in seen:
                continue
            seen.add(msg["message_id"])

            intent = classify_and_tag(msg["message_id"], msg["thread_id"])
            respond_or_escalate(msg["message_id"], msg["thread_id"], intent)

        time.sleep(interval_seconds)