AI21 Labs handles generation — summarization, drafting, classification. MultiMail handles delivery, approvals, and audit trails so those outputs reach real inboxes safely.
AI21 Labs provides language models optimized for task-oriented text generation: summarization, structured completion, and chat. These capabilities map directly onto email workflows — classifying inbound messages, drafting context-aware replies, and condensing long threads into actionable summaries.
The gap AI21 doesn't fill is the send surface. Generating a well-structured reply is different from routing it through SMTP, enforcing per-mailbox identity, queuing it for human approval, or logging it for compliance. That's what MultiMail provides.
The integration pattern is straightforward: AI21 produces text, your agent calls MultiMail's REST API or MCP tools to act on it. You get the generation quality of AI21 with the governance guarantees — approval flows, webhook events, delivery receipts, audit logs — that production email requires.
Set oversight_mode to gated_send and every outbound message generated by your AI21 model sits in a pending queue until a human approves it. The decide_email endpoint lets your agent check queue status and handle approvals programmatically.
MultiMail binds each mailbox to a verified identity. Emails sent via send_email carry that identity in headers, preventing agents from spoofing sender addresses even when the generation model has no awareness of that constraint.
Configure a webhook on any MultiMail mailbox and your AI21-backed agent receives a POST the moment an email arrives — with headers, body, thread ID, and sender reputation signals already parsed.
Call get_thread to fetch a full conversation, pass it to AI21's summarization endpoint, then store the result or use it to inform a reply. The thread data includes timestamps, participants, and tags — not just raw body text.
MultiMail records every read, send, tag, and approval event with timestamps and actor identity. When a compliance audit asks what your agent sent and when, the answer is already there — no log scraping required.
Use mm_test_ keys during development. Emails are accepted, queued, and returned via API but never delivered externally. Switch to mm_live_ keys when you're ready to go to production — no code changes needed.
import os
import requests
from ai21 import AI21Client
from ai21.models.chat import ChatMessage
ai21_client = AI21Client(api_key=os.environ["AI21_API_KEY"])
MM_API_KEY = os.environ["MULTIMAIL_API_KEY"]
MM_BASE = "https://api.multimail.dev"
def classify_and_tag(email_id: str, subject: str, body: str) -> str:
response = ai21_client.chat.completions.create(
model="jamba-1.5-large",
messages=[
ChatMessage(
role="system",
content="Classify this email as one of: support, billing, partnership, spam, other. Reply with the single word."
),
ChatMessage(
role="user",
content=f"Subject: {subject}\n\n{body}"
)
]
)
label = response.choices[0].message.content.strip().lower()
requests.post(
f"{MM_BASE}/tag_email",
headers={"Authorization": f"Bearer {MM_API_KEY}"},
json={"email_id": email_id, "tags": [label]}
).raise_for_status()
return labelUse AI21's chat completion to classify an inbound email, then apply a tag via MultiMail's tag_email endpoint so downstream filters can route it correctly.
import os
import requests
from ai21 import AI21Client
from ai21.models.chat import ChatMessage
ai21_client = AI21Client(api_key=os.environ["AI21_API_KEY"])
MM_API_KEY = os.environ["MULTIMAIL_API_KEY"]
MM_BASE = "https://api.multimail.dev"
def draft_reply(thread_id: str, mailbox: str) -> dict:
"cm"># Fetch thread context
thread = requests.get(
f"{MM_BASE}/get_thread",
headers={"Authorization": f"Bearer {MM_API_KEY}"},
params={"thread_id": thread_id}
).json()
messages_text = "\n\n".join(
f"From: {m[&"cm">#039;from']}\n{m['body']}" for m in thread["messages"]
)
# Generate reply with AI21
response = ai21_client.chat.completions.create(
model="jamba-1.5-large",
messages=[
ChatMessage(
role="system",
content="You are a professional email assistant. Write a concise, helpful reply to this email thread."
),
ChatMessage(role="user", content=messages_text)
]
)
draft = response.choices[0].message.content.strip()
# Queue reply — oversight_mode gated_send holds it for approval
result = requests.post(
f"{MM_BASE}/reply_email",
headers={"Authorization": f"Bearer {MM_API_KEY}"},
json={
"thread_id": thread_id,
"from": mailbox,
"body": draft,
"oversight_mode": "gated_send"
}
).json()
return {"message_id": result["message_id"], "status": result["status"]}Fetch a thread with get_thread, ask AI21 to draft a reply, then send it via reply_email with gated_send oversight so it waits for human approval before delivery.
import os
import requests
from fastapi import FastAPI, Request
from ai21 import AI21Client
from ai21.models.chat import ChatMessage
app = FastAPI()
ai21_client = AI21Client(api_key=os.environ["AI21_API_KEY"])
MM_API_KEY = os.environ["MULTIMAIL_API_KEY"]
MM_BASE = "https://api.multimail.dev"
HEADERS = {"Authorization": f"Bearer {MM_API_KEY}"}
@app.post("/webhook/inbound")
async def handle_inbound(request: Request):
event = await request.json()
if event.get("type") != "email.received":
return {"ok": True}
email_id = event["data"]["email_id"]
thread_id = event["data"]["thread_id"]
"cm"># Get full thread
thread = requests.get(
f"{MM_BASE}/get_thread",
headers=HEADERS,
params={"thread_id": thread_id}
).json()
combined = "\n\n".join(
f"{m[&"cm">#039;from']} wrote:\n{m['body']}" for m in thread["messages"]
)
# Summarize with AI21
summary_resp = ai21_client.chat.completions.create(
model="jamba-1.5-large",
messages=[
ChatMessage(
role="system",
content="Summarize this email thread in two sentences. Focus on action items."
),
ChatMessage(role="user", content=combined)
]
)
summary = summary_resp.choices[0].message.content.strip()
# Store summary as a searchable tag
requests.post(
f"{MM_BASE}/tag_email",
headers=HEADERS,
json={"email_id": email_id, "tags": ["summarized"], "notes": summary}
)
return {"ok": True, "summary": summary}A FastAPI webhook handler that receives a MultiMail inbound event, fetches the full thread, summarizes it with AI21, then stores the summary as an email tag.
import os
import requests
MM_API_KEY = os.environ["MULTIMAIL_API_KEY"]
MM_BASE = "https://api.multimail.dev"
HEADERS = {"Authorization": f"Bearer {MM_API_KEY}"}
def process_approval_queue(mailbox: str) -> list[dict]:
pending = requests.get(
f"{MM_BASE}/list_pending",
headers=HEADERS,
params={"mailbox": mailbox}
).json()
results = []
for msg in pending.get("messages", []):
decision = requests.get(
f"{MM_BASE}/decide_email",
headers=HEADERS,
params={"message_id": msg["message_id"]}
).json()
results.append({
"message_id": msg["message_id"],
"subject": msg["subject"],
"decision": decision["decision"], "cm"># approved | rejected | pending
"decided_at": decision.get("decided_at")
})
return resultsPoll list_pending for messages awaiting human approval, then use decide_email to check individual approval status before releasing or cancelling.
Install the official AI21 client library and set your API key as an environment variable.
pip install ai21
export AI21_API_KEY=your_ai21_api_keySign up at multimail.dev, generate an API key, and create a mailbox. Use mm_test_ keys during development — emails are processed but never delivered externally.
curl -X POST https://api.multimail.dev/create_mailbox \
-H "Authorization: Bearer mm_test_your_key" \
-H "Content-Type: application/json" \
-d &"cm">#039;{"address": "[email protected]", "oversight_mode": "gated_send"}'Point a MultiMail webhook at your agent's endpoint so AI21-backed processing triggers automatically when email arrives. The payload includes the email ID, thread ID, subject, and parsed body.
curl -X POST https://api.multimail.dev/webhooks \
-H "Authorization: Bearer mm_test_your_key" \
-H "Content-Type: application/json" \
-d &"cm">#039;{"mailbox": "[email protected]", "url": "https://yourapp.com/webhook/inbound", "events": ["email.received"]}'In your webhook handler, call get_thread to fetch context, pass it to AI21 for classification or drafting, then call the appropriate MultiMail endpoint. Start with read_email and tag_email before enabling sends.
from ai21 import AI21Client
from ai21.models.chat import ChatMessage
import requests
client = AI21Client(api_key=os.environ["AI21_API_KEY"])
"cm"># Classify on arrival
response = client.chat.completions.create(
model="jamba-1.5-large",
messages=[ChatMessage(role="user", content=email_body)]
)
"cm"># Tag result in MultiMail
requests.post(
"https://api.multimail.dev/tag_email",
headers={"Authorization": f"Bearer {os.environ[&"cm">#039;MULTIMAIL_API_KEY']}"},
json={"email_id": email_id, "tags": [response.choices[0].message.content.strip()]}
)Once classification is working, enable reply drafting with oversight_mode set to gated_send. Check list_pending to review queued messages before they go out. Promote to monitored or autonomous only after reviewing several approval cycles.
"cm"># List messages awaiting approval
curl https://api.multimail.dev/list_pending \
-H "Authorization: Bearer $MULTIMAIL_API_KEY" \
-G -d [email protected]Email infrastructure built for AI agents. Verifiable identity, graduated oversight, and a 38-tool MCP server. Formally verified in Lean 4.