Loading...
Loading...
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.
npx skill4agent add agentmail-to/agentmail-skills agent-email-patternsagentmailfrom 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 liveAgent 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# 1. Agent sends the opening message
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?",
)
# 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 responseextracted_textextracted_htmlclient.inboxes.messages.reply(inbox_id, message_id, ...)message_idthread_id# Agent drafts
draft = client.inboxes.drafts.create(
inbox_id,
to="important-client@example.com",
subject="Contract proposal",
text=agent_generated_text,
)
# Human reviews in console or via API, then:
client.inboxes.drafts.send(inbox_id, draft.draft_id)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 = client.webhooks.create(
url="https://your-server.com/agent/email",
event_types=["message.received"],
)| 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 |
support@agentmail.to -> customer support
sales@agentmail.to -> sales inquiries
billing@agentmail.to -> invoices and payments
router@agentmail.to -> intake, routes to correct agent# 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.",
)references/security.mdreferences/multi-agent-topologies.mdimport 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# 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"])unreadprocessedarchivedneeds-replyrepliedescalatedbillingsupportsalesreferences/security.mdreferences/multi-agent-topologies.mdreferences/security.md