agent-email-patterns

name	agent-email-patterns
description	Architecture patterns and best practices for giving AI agents email capabilities. Use when designing how agents send, receive, and manage email conversations, building two-way communication loops, implementing human-in-the-loop approval with drafts, choosing between WebSockets and webhooks, setting up multi-agent email topologies, handling OTP and verification flows, or securing agent email against prompt injection.
license	MIT
metadata	{"author":"agentmail-to","version":"1.0"}

Agent Email Patterns

Opinionated patterns for building AI agents that communicate over email. This skill covers architecture decisions, not SDK specifics. For AgentMail SDK usage, use the agentmail skill.

Pattern 1: one inbox per agent

Every agent gets its own email address. Never share inboxes between agents.

from agentmail import AgentMail
from agentmail.inboxes.types import CreateInboxRequest

client = AgentMail()

support_inbox = client.inboxes.create(
    request=CreateInboxRequest(
        username="support-agent",
        display_name="Acme Support",
        client_id="support-v1",  # idempotent
    ),
)
# [e-mail] is now live

Why:

• Identity: recipients see a clear sender
• Isolation: agents cannot access each other's email
• Auditability: every message is traceable to one agent
• Security: compromising one agent does not expose others

Anti-pattern: one shared inbox with multiple agents reading from it. This creates race conditions and makes debugging impossible.

Pattern 2: two-way conversation loops

The core agent email pattern: agent sends, human replies, agent reads the reply and responds.

Agent sends initial email
  -> Human replies
    -> Agent reads reply (use extracted_text to strip quoted history)
      -> Agent decides next action and responds
        -> Loop continues until resolved

Implementation:

# 1. Agent sends the opening message
client.inboxes.messages.send(
    inbox_id,
    to="[e-mail]",
    subject="Your support ticket #1234",
    text="We received your request. Can you clarify the issue?",
)

# 2. Later: agent reads the reply.
# messages.list() returns MessageItem objects (metadata only — NO body).
# Fetch the full Message with .get() to access .text / .extracted_text.
response = client.inboxes.messages.list(inbox_id, limit=5)
for item in response.messages:
    msg = client.inboxes.messages.get(
        inbox_id=item.inbox_id,
        message_id=item.message_id,
    )
    # extracted_text strips quoted history and signatures
    new_content = msg.extracted_text or msg.text
    # Feed new_content to your LLM for next response

Key rules:

• Always use extracted_text / extracted_html for inbound replies to avoid processing the entire quoted chain
• Track conversation state in your database, not in the email body
• To keep messages grouped in the same thread, call client.inboxes.messages.reply(inbox_id, message_id, ...) with the parent message_id — AgentMail routes the reply into the existing thread automatically. There is no thread_id parameter on the reply call.

Pattern 3: human-in-the-loop drafts

For high-stakes emails, let the agent draft and a human approve before sending.

# Agent drafts
draft = client.inboxes.drafts.create(
    inbox_id,
    to="[e-mail]",
    subject="Contract proposal",
    text=agent_generated_text,
)
# Human reviews in console or via API, then:
client.inboxes.drafts.send(inbox_id, draft.draft_id)

Use drafts when:

• Email has legal or financial implications
• Recipient is a VIP or external stakeholder
• Agent is new and untrusted for this workflow

Send directly when:

• Routine notification (receipts, confirmations)
• Agent has proven reliability
• Speed matters (OTP forwarding, automated alerts)

Pattern 4: event-driven architecture

Never poll for new emails. Use WebSockets or webhooks.

WebSockets (best for agents, no public URL needed):

from agentmail import AgentMail, Subscribe, MessageReceivedEvent

client = AgentMail()
with client.websockets.connect() as socket:
    socket.send_subscribe(Subscribe(inbox_ids=[inbox_id]))
    for event in socket:
        if isinstance(event, MessageReceivedEvent):
            process_email(event.message)

Webhooks (for servers with public endpoints):

webhook = client.webhooks.create(
    url="https://your-server.com/agent/email",
    event_types=["message.received"],
)

Decision guide:

Factor	WebSockets	Webhooks
Public URL needed	No	Yes
Best for	Agents, bots, local dev	Servers, serverless
Latency	Lowest (persistent)	HTTP round-trip
Reconnection	You handle it	AgentMail retries

Pattern 5: multi-agent topologies

For systems with multiple agents, assign clear roles:

[e-mail]     -> customer support
[e-mail]       -> sales inquiries
[e-mail]     -> invoices and payments
[e-mail]      -> intake, routes to correct agent

Agents can email each other for internal coordination:

# Support agent escalates to sales
client.inboxes.messages.send(
    support_inbox_id,
    to=sales_inbox.email,
    subject="Lead handoff: Acme Corp",
    text="Customer wants enterprise pricing. Full thread below.",
)

Use allow lists (references/security.md) to restrict which external senders can reach each agent. For hub-and-spoke, peer-to-peer, and hierarchical escalation patterns, see references/multi-agent-topologies.md.

Pattern 6: OTP and verification flows

Agents that sign up for services need to receive and extract verification codes.

import re

inbox = client.inboxes.create()
# Use inbox.email to sign up for a service

# Listen for OTP via WebSocket
with client.websockets.connect() as socket:
    socket.send_subscribe(Subscribe(inbox_ids=[inbox.inbox_id]))
    for event in socket:
        if isinstance(event, MessageReceivedEvent):
            text = event.message.text or ""
            match = re.search(r"\b(\d{4,8})\b", text)
            if match:
                otp = match.group(1)
                break

Best practices:

• Create a fresh inbox per sign-up flow for isolation
• Set a timeout (do not wait indefinitely for OTP)
• Delete the inbox after the flow completes if it is single-use

Pattern 7: labels for workflow state

Use labels to track message processing state within an inbox:

# When agent processes a message
client.inboxes.messages.update(
    inbox_id, message_id,
    add_labels=["processed", "needs-followup"],
    remove_labels=["unread"],
)

# Query by label
unprocessed = client.inboxes.messages.list(inbox_id, labels=["unread"])

Common label schemes:

• unread / processed / archived
• needs-reply / replied / escalated
• billing / support / sales (category routing)

Security essentials

See references/security.md for full coverage. Critical rules:

1. Sanitize inbound email before passing to LLM -- prompt injection via email is a real attack vector. Never pass raw email content directly as a system prompt.
2. Use allow lists on production agent inboxes to restrict senders.
3. Verify webhook signatures to prevent spoofed events.
4. Never put API keys or secrets in email bodies or subjects.
5. Separate agent credentials from human credentials -- each agent gets its own API key.

Reference files

• references/multi-agent-topologies.md -- hub-and-spoke, peer-to-peer, and hierarchical agent email architectures
• references/security.md -- prompt injection defense, sender validation, credential isolation