agent-email-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Agent Email Patterns

Agent 邮件模式

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.

用于构建可通过邮件通信的AI Agent的权威架构模式。本技能聚焦架构决策，不涉及SDK细节。如需了解AgentMail SDK的使用方法，请使用

agentmail

技能。

Pattern 1: one inbox per agent

模式1：每个Agent对应独立收件箱

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

python

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

每个Agent都拥有专属邮箱地址。绝对不要让多个Agent共享收件箱。

python

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

support-agent@agentmail.to is now live

support-agent@agentmail.to 现已启用


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.


优势：
- **身份标识**：收件人可明确识别发件方
- **隔离性**：Agent无法访问彼此的邮件
- **可审计性**：每条消息都可追溯到对应Agent
- **安全性**：单个Agent被攻陷不会波及其他Agent

反模式：多个Agent共用一个收件箱读取邮件。这会引发竞态条件，且完全无法调试。

Pattern 2: two-way conversation loops

模式2：双向对话循环

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:

python

undefined

Agent邮件的核心模式：Agent发送邮件，人类回复，Agent读取回复并做出响应。

Agent发送初始邮件
  -> 人类回复
    -> Agent读取回复（使用extracted_text去除引用历史）
      -> Agent决定下一步操作并回复
        -> 循环直至问题解决

实现方式：

python

undefined

1. Agent sends the opening message

1. Agent发送初始消息

client.inboxes.messages.send( inbox_id, to="user@example.com", subject="Your support ticket #1234", text="We received your request. Can you clarify the issue?", )

client.inboxes.messages.send( inbox_id, to="user@example.com", subject="您的支持工单 #1234", text="我们已收到您的请求。能否请您说明问题详情？", )

2. Later: agent reads the reply.

2. 后续：Agent读取回复。

messages.list() returns MessageItem objects (metadata only — NO body).

messages.list() 返回MessageItem对象（仅元数据 — 无邮件正文）。

Fetch the full Message with .get() to access .text / .extracted_text.

调用.get()获取完整Message以访问.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.

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会去除引用历史和签名 new_content = msg.extracted_text or msg.text # 将new_content传入LLM以生成下一条响应


关键规则：
- 处理 inbound 回复时务必使用`extracted_text` / `extracted_html`，避免处理完整的引用链
- 在数据库中跟踪对话状态，而非存储在邮件正文中
- 如需将消息归为同一线程，调用`client.inboxes.messages.reply(inbox_id, message_id, ...)`并传入父级`message_id` — AgentMail会自动将回复路由至现有线程。回复调用中没有`thread_id`参数。

Pattern 3: human-in-the-loop drafts

模式3：人工介入的草稿审批流程

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

python

undefined

针对高风险邮件，先由Agent生成草稿，经人工审批后再发送。

python

undefined

Agent drafts

Agent生成草稿

draft = client.inboxes.drafts.create( inbox_id, to="important-client@example.com", subject="Contract proposal", text=agent_generated_text, )

draft = client.inboxes.drafts.create( inbox_id, to="important-client@example.com", subject="合同提案", text=agent_generated_text, )

Human reviews in console or via API, then:

人工通过控制台或API审核后，执行：

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)

client.inboxes.drafts.send(inbox_id, draft.draft_id)


适用场景：
- 邮件涉及法律或财务影响
- 收件人为VIP或外部利益相关方
- Agent为新部署，该工作流下尚未验证可靠性

直接发送的场景：
- 常规通知（收据、确认信息）
- Agent已被证明可靠
- 对速度有要求（OTP转发、自动化告警）

Pattern 4: event-driven architecture

模式4：事件驱动架构

Never poll for new emails. Use WebSockets or webhooks.

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

python

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):

python

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

绝对不要轮询新邮件。使用WebSocket或webhook。

WebSocket（最适合Agent，无需公网URL）：

python

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)

Webhook（适用于拥有公网端点的服务器）：

python

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

决策指南：

考量因素	WebSocket	Webhook
是否需要公网URL	否	是
最佳适用场景	Agent、机器人、本地开发	服务器、无服务器架构
延迟	最低（持久连接）	HTTP往返延迟
重连机制	需自行处理	AgentMail自动重试

Pattern 5: multi-agent topologies

模式5：多Agent拓扑结构

For systems with multiple agents, assign clear roles:

support@agentmail.to     -> customer support
sales@agentmail.to       -> sales inquiries
billing@agentmail.to     -> invoices and payments
router@agentmail.to      -> intake, routes to correct agent

Agents can email each other for internal coordination:

python

undefined

针对包含多个Agent的系统，需分配清晰的角色：

support@agentmail.to     -> 客户支持
sales@agentmail.to       -> 销售咨询
billing@agentmail.to     -> 发票与支付
router@agentmail.to      -> 收件分流，路由至对应Agent

Agent之间可通过邮件进行内部协作：

python

undefined

Support agent escalates to sales

支持Agent将线索转介给销售

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`.

client.inboxes.messages.send( support_inbox_id, to=sales_inbox.email, subject="线索移交：Acme Corp", text="客户希望了解企业版定价。完整对话如下。", )


使用允许列表（`references/security.md`）限制可联系各Agent的外部发件人。如需了解 hub-and-spoke、peer-to-peer 以及分层升级模式，请查看`references/multi-agent-topologies.md`。

Pattern 6: OTP and verification flows

模式6：OTP与验证流程

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

python

import re

inbox = client.inboxes.create()

需注册服务的Agent需要接收并提取验证码。

python

import re

inbox = client.inboxes.create()

Use inbox.email to sign up for a service

使用inbox.email注册服务

Listen for OTP via WebSocket

通过WebSocket监听OTP

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


最佳实践：
- 每个注册流程创建独立的收件箱以实现隔离
- 设置超时时间（不要无限等待OTP）
- 若为一次性使用，流程完成后删除收件箱

Pattern 7: labels for workflow state

模式7：用标签追踪工作流状态

Use labels to track message processing state within an inbox:

python

undefined

使用标签追踪收件箱内消息的处理状态：

python

undefined

When agent processes a message

Agent处理消息时

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)

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


常见标签方案：
- `unread` / `processed` / `archived`（已读/已处理/已归档）
- `needs-reply` / `replied` / `escalated`（需回复/已回复/已升级）
- `billing` / `support` / `sales`（分类路由）

Security essentials

安全要点

See

references/security.md

for full coverage. Critical rules:

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.
Use allow lists on production agent inboxes to restrict senders.
Verify webhook signatures to prevent spoofed events.
Never put API keys or secrets in email bodies or subjects.
Separate agent credentials from human credentials -- each agent gets its own API key.

完整内容请查看

references/security.md

。关键规则：

将 inbound 邮件清洗后再传入LLM -- 通过邮件进行提示注入是真实存在的攻击手段。绝对不要将原始邮件内容直接作为系统提示词传入。
生产环境的Agent收件箱使用允许列表限制发件人。
验证webhook签名以防止伪造事件。
绝对不要在邮件正文或主题中放入API密钥或机密信息。
将Agent凭据与人类凭据分离 -- 每个Agent拥有独立的API密钥。

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

```
references/multi-agent-topologies.md
```
-- hub-and-spoke、peer-to-peer 以及分层Agent邮件架构
```
references/security.md
```
-- 提示注入防护、发件人验证、凭据隔离