Wire Cohere tool use, classification, and reranking to MultiMail's REST API to give your agents production-ready email capabilities with approval gates and audit trails.
Cohere's platform covers the full range of enterprise AI tasks — generation, embeddings, reranking, and classification — making it a strong foundation for agents that need more than a single model's capabilities. Many teams reach for Cohere specifically when they want provider flexibility, on-premises deployment, or retrieval pipelines that combine generation with semantic search.
Connecting Cohere to email introduces the same set of production concerns it always does: approval flows, deliverability, rate limits, and audit trails. Cohere's API doesn't enforce any of these — it generates text and returns tool calls. MultiMail handles the email layer, including gated send modes that require human sign-off before any message leaves your infrastructure.
The integration follows Cohere's standard tool-use loop. You define MultiMail endpoints as tools, run the chat loop, and execute the function calls against MultiMail's REST API. Classification and reranking can run before or after inbox retrieval to filter, prioritize, or route messages without the model needing to reason about every email in a large inbox.
Cohere generates a tool call to send_email and stops. MultiMail's gated_send mode holds that message in a pending queue until a human approves it via the list_pending and decide_email endpoints. Your agent code doesn't change — the gate is enforced server-side.
MultiMail exposes check_inbox, send_email, reply_email, read_email, tag_email, and decide_email as stable REST endpoints. You can register these directly as Cohere tools and rely on consistent schemas across model versions.
Cohere's rerank model can prioritize emails returned by check_inbox before passing them to the generation model. This reduces context length and prevents the LLM from burying high-priority messages when the inbox is large.
Cohere's classify endpoint can label incoming emails by category, urgency, or sender type using a small example set. Pass those labels to MultiMail's tag_email endpoint to route messages before the generation model ever sees them.
Rate limits, send-time windows, and domain allowlists are enforced by MultiMail regardless of what Cohere's model outputs. An agent that hallucinates a send call to an off-hours recipient gets queued, not delivered.
import cohere
import requests
import json
co = cohere.ClientV2(api_key="YOUR_COHERE_API_KEY")
MM_TOKEN = "mm_live_your_token_here"
MM_BASE = "https://api.multimail.dev"
MM_HEADERS = {"Authorization": f"Bearer {MM_TOKEN}", "Content-Type": "application/json"}
tools = [
{
"type": "function",
"function": {
"name": "check_inbox",
"description": "Retrieve recent emails from a MultiMail mailbox.",
"parameters": {
"type": "object",
"properties": {
"mailbox": {"type": "string", "description": "Mailbox address, e.g. [email protected]"},
"limit": {"type": "integer", "description": "Number of emails to fetch, max 50"}
},
"required": ["mailbox"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "Send an email via MultiMail. Subject to gated_send approval if configured.",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string"},
"subject": {"type": "string"},
"body": {"type": "string"},
"from_mailbox": {"type": "string"}
},
"required": ["to", "subject", "body", "from_mailbox"]
}
}
}
]
def dispatch_tool(name: str, args: dict) -> dict:
if name == "check_inbox":
r = requests.get(f"{MM_BASE}/check_inbox", headers=MM_HEADERS, params=args)
return r.json()
if name == "send_email":
r = requests.post(f"{MM_BASE}/send_email", headers=MM_HEADERS, json={
"to": args["to"],
"subject": args["subject"],
"body": args["body"],
"from": args["from_mailbox"]
})
return r.json()
return {"error": f"Unknown tool: {name}"}
messages = [{"role": "user", "content": "Check [email protected] for any unread messages and draft replies to anything marked urgent."}]
while True:
response = co.chat(
model="command-r-plus-08-2024",
messages=messages,
tools=tools
)
if response.message.tool_calls:
messages.append({"role": "assistant", "tool_calls": response.message.tool_calls, "content": ""})
tool_results = []
for tc in response.message.tool_calls:
result = dispatch_tool(tc.function.name, json.loads(tc.function.arguments))
tool_results.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)})
messages.extend(tool_results)
else:
print(response.message.content[0].text)
break
Register MultiMail's check_inbox and send_email as Cohere tools and run the standard agentic loop. The agent reads the inbox, decides whether to reply, and MultiMail holds outbound messages for human approval when oversight_mode is gated_send.
import cohere
import requests
co = cohere.ClientV2(api_key="YOUR_COHERE_API_KEY")
MM_TOKEN = "mm_live_your_token_here"
MM_BASE = "https://api.multimail.dev"
MM_HEADERS = {"Authorization": f"Bearer {MM_TOKEN}"}
"cm"># Fetch inbox
inbox_resp = requests.get(
f"{MM_BASE}/check_inbox",
headers=MM_HEADERS,
params={"mailbox": "[email protected]", "limit": 30}
)
emails = inbox_resp.json().get("emails", [])
if not emails:
print("Inbox empty")
exit()
"cm"># Build classification inputs from subject + snippet
inputs = [f"{e[&"cm">#039;subject']} — {e.get('snippet', '')[:120]}" for e in emails]
# Few-shot examples for classification
examples = [
cohere.ClassifyExample(text="Invoice "cm">#4821 overdue — payment required", label="billing"),
cohere.ClassifyExample(text="Your account has been suspended", label="billing"),
cohere.ClassifyExample(text="Bug report: API returns 500 on /send_email", label="support"),
cohere.ClassifyExample(text="Integration not working after update", label="support"),
cohere.ClassifyExample(text="Partnership proposal for Q3", label="sales"),
cohere.ClassifyExample(text="Interested in your enterprise plan", label="sales"),
cohere.ClassifyExample(text="Out of office until Monday", label="auto-reply"),
cohere.ClassifyExample(text="Delivery notification: your package shipped", label="auto-reply"),
]
classify_resp = co.classify(
inputs=inputs,
examples=examples,
model="embed-english-v3.0"
)
"cm"># Apply tags via MultiMail tag_email endpoint
for email, classification in zip(emails, classify_resp.classifications):
label = classification.prediction
confidence = classification.confidence
if confidence < 0.6:
label = "review" "cm"># low-confidence emails need human review
requests.post(
f"{MM_BASE}/tag_email",
headers={**MM_HEADERS, "Content-Type": "application/json"},
json={"email_id": email["id"], "tag": label}
)
print(f"[{label}] ({confidence:.0%}) {email[&"cm">#039;subject']}")
Use Cohere's classify endpoint to categorize emails retrieved from MultiMail, then apply tags via the tag_email endpoint so downstream agents or humans can filter by category without reading every message.
import cohere
import requests
co = cohere.ClientV2(api_key="YOUR_COHERE_API_KEY")
MM_TOKEN = "mm_live_your_token_here"
MM_BASE = "https://api.multimail.dev"
MM_HEADERS = {"Authorization": f"Bearer {MM_TOKEN}"}
"cm"># Retrieve a larger batch from MultiMail
inbox_resp = requests.get(
f"{MM_BASE}/check_inbox",
headers=MM_HEADERS,
params={"mailbox": "[email protected]", "limit": 50}
)
emails = inbox_resp.json().get("emails", [])
if not emails:
print("No emails to process")
exit()
"cm"># Build document strings for reranking
documents = [
f"From: {e[&"cm">#039;from']}\nSubject: {e['subject']}\n{e.get('snippet', '')}"
for e in emails
]
# Rerank against the task the agent is performing
rerank_resp = co.rerank(
model="rerank-english-v3.0",
query="urgent contract renewal or signature required",
documents=documents,
top_n=5,
return_documents=True
)
# Process only the top-ranked emails
top_emails = [emails[r.index] for r in rerank_resp.results]
print(f"Top {len(top_emails)} emails by relevance:")
for rank, (result, email) in enumerate(zip(rerank_resp.results, top_emails), 1):
print(f" {rank}. [{result.relevance_score:.2f}] {email[&"cm">#039;subject']} — {email['from']}")
# Read full content for top emails
read_resp = requests.get(
f"{MM_BASE}/read_email",
headers=MM_HEADERS,
params={"email_id": email["id"]}
)
email["body"] = read_resp.json().get("body", "")
"cm"># Pass top_emails to your Cohere generation step
"cm"># The model now sees 5 emails instead of 50
When an inbox contains many messages, use Cohere's rerank endpoint to surface the most relevant emails before passing them to the generation model. This keeps context length manageable and ensures high-priority messages aren't buried.
import cohere
import requests
import time
co = cohere.ClientV2(api_key="YOUR_COHERE_API_KEY")
MM_TOKEN = "mm_live_your_token_here"
MM_BASE = "https://api.multimail.dev"
MM_HEADERS = {"Authorization": f"Bearer {MM_TOKEN}", "Content-Type": "application/json"}
"cm"># Generate a reply with Cohere
original_email = {
"id": "em_01abc123",
"from": "[email protected]",
"subject": "Contract renewal — action required",
"body": "Hi, our contract expires on May 1st. Can you confirm renewal terms?"
}
draft_response = co.chat(
model="command-r-plus-08-2024",
messages=[
{"role": "system", "content": "You are an account manager assistant. Draft professional, concise replies."},
{"role": "user", "content": f"Draft a reply to this email:\n\nFrom: {original_email[&"cm">#039;from']}\nSubject: {original_email['subject']}\n\n{original_email['body']}"}
]
)
draft_body = draft_response.message.content[0].text
# Submit reply to MultiMail — gated_send holds it for approval
send_resp = requests.post(
f"{MM_BASE}/reply_email",
headers=MM_HEADERS,
json={
"email_id": original_email["id"],
"body": draft_body,
"from": "[email protected]"
}
)
queued = send_resp.json()
print(f"Message queued: {queued.get(&"cm">#039;message_id')} — status: {queued.get('status')}")
# Check pending queue (a human would normally review this in a dashboard)
pending_resp = requests.get(f"{MM_BASE}/list_pending", headers=MM_HEADERS)
pending = pending_resp.json().get("pending", [])
for msg in pending:
print(f"Pending: {msg[&"cm">#039;message_id']} to {msg['to']} — awaiting approval")
# To approve programmatically (e.g. after automated policy check):
# requests.post(f"{MM_BASE}/decide_email", headers=MM_HEADERS,
# json={"message_id": msg["message_id"], "decision": "approve"})
After Cohere generates a draft, submit it to MultiMail with oversight_mode gated_send. Poll list_pending for the queued message and use decide_email to approve or reject it programmatically — or leave the decision to a human reviewer.
Install the Cohere Python client. Create a MultiMail account and provision a mailbox — use a custom domain or a @multimail.dev address for testing.
pip install cohere requests
"cm"># Create a mailbox via MultiMail REST API
curl -X POST https://api.multimail.dev/create_mailbox \
-H "Authorization: Bearer $MULTIMAIL_API_KEY" \
-H "Content-Type: application/json" \
-d &"cm">#039;{"address": "[email protected]", "oversight_mode": "gated_send"}'Register check_inbox, send_email, and any other MultiMail endpoints your agent needs as tools in the Cohere tool format. Keep parameter descriptions specific — Cohere's model uses them to decide when and how to call each tool.
import cohere
co = cohere.ClientV2(api_key="YOUR_COHERE_API_KEY")
tools = [
{
"type": "function",
"function": {
"name": "check_inbox",
"description": "Fetch recent emails from a MultiMail mailbox. Returns sender, subject, snippet, and email_id for each message.",
"parameters": {
"type": "object",
"properties": {
"mailbox": {"type": "string"},
"limit": {"type": "integer"}
},
"required": ["mailbox"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "Send an email via MultiMail. May be held for human approval depending on mailbox oversight_mode.",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string"},
"subject": {"type": "string"},
"body": {"type": "string"},
"from": {"type": "string"}
},
"required": ["to", "subject", "body", "from"]
}
}
}
]Call co.chat with your tools list, detect tool_calls in the response, dispatch each call to the corresponding MultiMail endpoint, and feed results back into the message history until the model returns a final text response.
import json, requests
MM_HEADERS = {"Authorization": "Bearer $MULTIMAIL_API_KEY", "Content-Type": "application/json"}
MM_BASE = "https://api.multimail.dev"
messages = [{"role": "user", "content": "Check [email protected] and reply to any support requests."}]
while True:
resp = co.chat(model="command-r-plus-08-2024", messages=messages, tools=tools)
if resp.message.tool_calls:
messages.append({"role": "assistant", "tool_calls": resp.message.tool_calls, "content": ""})
for tc in resp.message.tool_calls:
args = json.loads(tc.function.arguments)
result = requests.post(f"{MM_BASE}/{tc.function.name}", headers=MM_HEADERS, json=args).json()
messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)})
else:
print(resp.message.content[0].text)
breakUpdate the mailbox oversight_mode as you build confidence in agent behavior. Start with gated_send so every outbound message is reviewed, then move to monitored once the agent's output quality is established. You can change this via the MultiMail dashboard or the create_mailbox API.
"cm"># Switch an existing mailbox to monitored mode via MultiMail API
curl -X PATCH https://api.multimail.dev/mailbox/[email protected] \
-H "Authorization: Bearer $MULTIMAIL_API_KEY" \
-H "Content-Type: application/json" \
-d &"cm">#039;{"oversight_mode": "monitored"}'Email infrastructure built for AI agents. Verifiable identity, graduated oversight, and a 38-tool MCP server. Formally verified in Lean 4.