Agentmail

AgentMail

AgentMail is an API-first email platform for AI agents. Unlike traditional email services, it's designed for two-way conversations, allowing agents to send, receive, and reply to emails autonomously.

Quick Start

Installation

# Python
pip install agentmail

# Node.js
npm install agentmail

Initialize Client

from agentmail import AgentMail

client = AgentMail()  # Uses AGENTMAIL_API_KEY from environment

import { AgentMailClient } from "agentmail";

const client = new AgentMailClient();  // Uses AGENTMAIL_API_KEY from environment

Create Inbox and Send Email

# Create inbox (use client_id for idempotency)
inbox = client.inboxes.create(
    username="my-agent",
    client_id="my-agent-inbox"
)
print(f"Created: {inbox.inbox_id}")  # e.g., my-agent@agentmail.to

# Send email (always include both text and html)
client.inboxes.messages.send(
    inbox_id=inbox.inbox_id,
    to=["user@example.com"],
    subject="Hello from my agent",
    text="Plain text version",
    html="<p>HTML version</p>",
    labels=["outreach"]
)

const inbox = await client.inboxes.create({
  username: "my-agent",
  clientId: "my-agent-inbox"
});

await client.inboxes.messages.send(inbox.inboxId, {
  to: ["user@example.com"],
  subject: "Hello from my agent",
  text: "Plain text version",
  html: "<p>HTML version</p>",
  labels: ["outreach"]
});

Resource Hierarchy

Organization (top-level container)
└── Pod (optional, for multi-tenancy)
    └── Inbox (email account, e.g., agent@agentmail.to)
        └── Thread (conversation, auto-created)
            └── Message (individual email)
                └── Attachment (files)

Core Concepts

Common Workflows

1. Reply to Email

client.inboxes.messages.reply(
    inbox_id="agent@agentmail.to",
    message_id="msg_xxx",
    text="Thanks for your message!",
    html="<p>Thanks for your message!</p>"
)

await client.inboxes.messages.reply("agent@agentmail.to", "msg_xxx", {
  text: "Thanks for your message!",
  html: "<p>Thanks for your message!</p>"
});

2. List Messages with Label Filter

messages = client.inboxes.messages.list(
    inbox_id="agent@agentmail.to",
    labels=["unread", "important"]
)
for msg in messages.messages:
    print(f"{msg.subject} from {msg.from_}")

const messages = await client.inboxes.messages.list("agent@agentmail.to", {
  labels: ["unread", "important"]
});
for (const msg of messages.messages) {
  console.log(`${msg.subject} from ${msg.from}`);
}

3. Update Labels on Message

client.inboxes.messages.update(
    inbox_id="agent@agentmail.to",
    message_id="msg_xxx",
    add_labels=["processed"],
    remove_labels=["unread"]
)

await client.inboxes.messages.update("agent@agentmail.to", "msg_xxx", {
  addLabels: ["processed"],
  removeLabels: ["unread"]
});

4. List Threads Org-Wide

# Query all threads across all inboxes (for supervisor agents)
all_threads = client.threads.list()

# Or per inbox
inbox_threads = client.inboxes.threads.list(inbox_id="agent@agentmail.to")

// Query all threads across all inboxes (for supervisor agents)
const allThreads = await client.threads.list();

// Or per inbox
const inboxThreads = await client.inboxes.threads.list("agent@agentmail.to");

5. Send Attachment

import base64

with open("report.pdf", "rb") as f:
    content = base64.b64encode(f.read()).decode()

client.inboxes.messages.send(
    inbox_id="agent@agentmail.to",
    to=["user@example.com"],
    subject="Report attached",
    text="Please see attached.",
    attachments=[{
        "content": content,
        "filename": "report.pdf",
        "content_type": "application/pdf"
    }]
)

import * as fs from "fs";

const content = fs.readFileSync("report.pdf").toString("base64");

await client.inboxes.messages.send("agent@agentmail.to", {
  to: ["user@example.com"],
  subject: "Report attached",
  text: "Please see attached.",
  attachments: [{
    content,
    filename: "report.pdf",
    contentType: "application/pdf"
  }]
});

Webhook Setup (Flask + ngrok)

Webhooks notify your agent in real-time when events occur (e.g., new email received).

Complete Example

import os
from threading import Thread
from flask import Flask, request, Response
import ngrok
from agentmail import AgentMail

app = Flask(__name__)
client = AgentMail()
port = 8080

# Start ngrok tunnel
listener = ngrok.forward(port, authtoken_from_env=True)
webhook_url = f"{listener.url()}/webhooks"

# Create inbox and webhook idempotently
client.inboxes.create(username="webhook-agent", client_id="webhook-agent-inbox")
client.webhooks.create(
    url=webhook_url,
    event_types=["message.received"],
    client_id="webhook-agent-webhook"
)

@app.route("/webhooks", methods=["POST"])
def receive_webhook():
    # Return 200 immediately, process in background
    Thread(target=process_webhook, args=(request.json,)).start()
    return Response(status=200)

def process_webhook(payload):
    event_type = payload["event_type"]
    if event_type == "message.received":
        message = payload["message"]
        print(f"New email from {message['from']}: {message['subject']}")
        
        # Reply to the message
        client.inboxes.messages.reply(
            inbox_id=message["inbox_id"],
            message_id=message["message_id"],
            text="Thanks for your email! I'll get back to you soon."
        )

if __name__ == "__main__":
    print(f"Webhook URL: {webhook_url}")
    app.run(port=port)

Webhook Event Types

• message.received - New email arrived (includes full Thread + Message data)
• message.sent - Email was sent
• message.delivered - Email was delivered to recipient's server
• message.bounced - Email failed to deliver
• message.complained - Recipient marked as spam
• message.rejected - Email rejected before sending
• domain.verified - Custom domain verified

See references/webhook-events.md for payload structures.

WebSocket Real-time Events

WebSockets provide real-time events without needing a public URL.

Async Pattern

import asyncio
from agentmail import AsyncAgentMail, Subscribe, MessageReceivedEvent

client = AsyncAgentMail()

async def main():
    async with client.websockets.connect() as socket:
        await socket.send_subscribe(Subscribe(
            inbox_ids=["agent@agentmail.to"]
        ))
        
        async for event in socket:
            if isinstance(event, MessageReceivedEvent):
                print(f"New email: {event.message.subject}")

asyncio.run(main())

const socket = await client.websockets.connect();

socket.on("open", () => {
  socket.sendSubscribe({
    type: "subscribe",
    inboxIds: ["agent@agentmail.to"]
  });
});

socket.on("message", (event) => {
  if (event.type === "message_received") {
    console.log(`New email: ${event.message.subject}`);
  }
});

AI Agent Integration

Use agentmail-toolkit to give AI agents email capabilities.

Installation

pip install agentmail-toolkit

OpenAI Agents

from agentmail import AgentMail
from agentmail_toolkit.openai import AgentMailToolkit
from agents import Agent, Runner

client = AgentMail()
toolkit = AgentMailToolkit(client)

agent = Agent(
    name="Email Agent",
    instructions=f"""You are an email agent. Your inbox is agent@agentmail.to.
    You can send, receive, and reply to emails.""",
    tools=toolkit.get_tools()
)

response = Runner.run(agent, [{"role": "user", "content": "Send a hello email to user@example.com"}])

Langchain

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from agentmail_toolkit.langchain import AgentMailToolkit

agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=AgentMailToolkit().get_tools()
)

Best Practices

Idempotency

Always use client_id on create operations to prevent duplicates:

# Safe to run multiple times - won't create duplicates
inbox = client.inboxes.create(
    username="my-agent",
    client_id="user-123-primary-inbox"
)

webhook = client.webhooks.create(
    url="https://example.com/webhooks",
    event_types=["message.received"],
    client_id="user-123-webhook"
)

Webhook Handling

Always return 200 immediately and process in background:

@app.route("/webhooks", methods=["POST"])
def webhook():
    Thread(target=process, args=(request.json,)).start()
    return Response(status=200)  # Return immediately!

Reply Extraction

Use extracted_text / extracted_html fields for clean reply content (removes quoted text):

message = client.inboxes.messages.get(inbox_id, message_id)
clean_reply = message.extracted_text  # Just the new content
full_email = message.text  # Includes quoted replies

Or use Talon library for more control:

from talon import quotations
clean = quotations.extract_from_plain(email_text)

Critical Gotchas

1. Bounced/complained addresses are permanently blocked - AgentMail prevents sending to them to protect your reputation

2. Keep bounce rate < 4% - Or your account goes under review

3. AWS Route 53 DKIM records - Must split into two quoted strings with NO space:

   Correct: "first-part""second-part"
   Wrong:   "first-part" "second-part"  (space breaks it)

4. Only one SPF record per domain - Merge multiple services:

   v=spf1 include:spf.agentmail.to include:other.com ~all

5. `message.received` is the only webhook with full Thread + Message data - Other events have minimal metadata

6. Pods cannot be deleted with existing resources - Delete all inboxes/domains in the pod first

7. Inboxes cannot be moved between pods - Create new inbox in target pod

IMAP/SMTP Access

For email client integration:

Additional Resources

• API Reference - Complete method signatures
• Webhook Events - Event payloads
• Advanced Examples - Agent patterns
• Official Docs
• Console

AgentMail Claude Skill

A Claude Skill that teaches Claude how to build email agents using the AgentMail API.

What is a Claude Skill?

A Claude Skill is a "skill pack" that helps Claude learn to use specific APIs/tools. The core is a SKILL.md file containing essential information, best practices, and code examples.

Structure

agentmail-claude-skill/
├── SKILL.md              # Core skill file (~400 lines)
└── references/
    ├── api-reference.md  # Complete API method signatures
    ├── webhook-events.md # Webhook event types and payloads
    └── examples.md       # Advanced agent patterns

What's Included

SKILL.md

• Quick start (installation, initialization, first email)
• Resource hierarchy (Organization → Pod → Inbox → Thread → Message)
• Common workflows (send, reply, webhooks, websockets)
• AI agent integration (agentmail-toolkit)
• Best practices (idempotency, deliverability)
• Critical gotchas

references/api-reference.md

• All methods for: Inboxes, Messages, Threads, Webhooks, WebSockets, Domains, Pods, Drafts, API Keys, Metrics, Organizations
• Parameter types and return values
• Error types

references/webhook-events.md

• All 7 event types with full payload examples
• message.received, message.sent, message.delivered, message.bounced, message.complained, message.rejected, domain.verified

references/examples.md

• Event-driven agent (Flask + ngrok)
• WebSocket real-time agent
• Multi-step workflow with state
• Human-in-the-loop with drafts
• Label-based workflow
• Multi-tenant with pods
• Attachment processing
• TypeScript examples

Installation

For Personal Use

Copy the skill to your Cursor skills directory:

cp -r agentmail-claude-skill ~/.cursor/skills/agentmail

For Project Use

Copy to your project's .cursor/skills/ directory:

cp -r agentmail-claude-skill .cursor/skills/agentmail

Usage

Once installed, Claude will automatically use this skill when:

• Building email automation
• Creating AI email agents
• Setting up webhooks for email notifications
• Integrating email into AI workflows
• Working with AgentMail, inboxes, or email agents

Resources

• AgentMail Documentation
• AgentMail Console
• Python SDK
• Node.js SDK

License

MIT

AgentMail API Reference

Complete method signatures for all AgentMail resources.

Client Initialization

# Python - Sync
from agentmail import AgentMail
client = AgentMail(api_key="...")  # or uses AGENTMAIL_API_KEY env

# Python - Async
from agentmail import AsyncAgentMail
client = AsyncAgentMail()

// TypeScript
import { AgentMailClient } from "agentmail";
const client = new AgentMailClient({ apiKey: "..." });

Note: Method signatures below use Python naming conventions (snake_case). In TypeScript, use camelCase for method names and request properties:

- Methods: replyAll, getAttachment, getRaw, getZoneFile (not reply_all, get_attachment, etc.)

- Properties: clientId, displayName, pageToken, eventTypes, inboxId (not client_id, display_name, etc.)

- List/query params are passed as a request object: client.inboxes.list({ limit: 10, pageToken: "..." })

- Create/update params are also a request object: client.inboxes.create({ username: "...", clientId: "..." })

- Path IDs remain positional: client.inboxes.get("inbox_id"), client.inboxes.messages.get("inbox_id", "message_id")

- Accessor: client.apiKeys (not client.api_keys)

Inboxes

Create Inbox Parameters

Inbox Object

{
    "inbox_id": "agent@agentmail.to",
    "pod_id": "pod_xxx",
    "display_name": "My Agent",
    "client_id": "my-client-id",
    "created_at": "2024-01-01T00:00:00Z",
    "updated_at": "2024-01-01T00:00:00Z"
}

Messages

Send Message Parameters

Attachment Object (for sending)

{
    "content": "base64-encoded-content",
    "filename": "report.pdf",
    "content_type": "application/pdf"
}

Message Object

{
    "inbox_id": "agent@agentmail.to",
    "thread_id": "thd_xxx",
    "message_id": "msg_xxx",
    "from": "sender@example.com",
    "to": ["recipient@example.com"],
    "cc": [],
    "bcc": [],
    "subject": "Hello",
    "text": "Full email text with quotes",
    "html": "<p>Full email HTML</p>",
    "extracted_text": "Just the new reply content",
    "extracted_html": "<p>Just the new reply</p>",
    "preview": "Short preview...",
    "labels": ["received"],
    "attachments": [...],
    "in_reply_to": "msg_parent",
    "references": ["msg_1", "msg_2"],
    "timestamp": "2024-01-01T00:00:00Z",
    "created_at": "2024-01-01T00:00:00Z",
    "updated_at": "2024-01-01T00:00:00Z"
}

Threads

Thread Object

{
    "thread_id": "thd_xxx",
    "inbox_id": "agent@agentmail.to",
    "subject": "Conversation subject",
    "messages": [...],  # List of MessageItem
    "labels": ["important"],
    "created_at": "2024-01-01T00:00:00Z",
    "updated_at": "2024-01-01T00:00:00Z"
}

Webhooks

Create Webhook Parameters

Event Types

• message.received
• message.sent
• message.delivered
• message.bounced
• message.complained
• message.rejected
• domain.verified

WebSockets

# Sync
with client.websockets.connect() as socket:
    socket.send_subscribe(Subscribe(inbox_ids=["agent@agentmail.to"]))
    for event in socket:
        if isinstance(event, MessageReceivedEvent):
            print(event.message.subject)

# Async
async with client.websockets.connect() as socket:
    await socket.send_subscribe(Subscribe(inbox_ids=["agent@agentmail.to"]))
    async for event in socket:
        ...

Subscribe Options

Event Types

Domains

Domain Object

{
    "domain_id": "example.com",
    "pod_id": "pod_xxx",
    "status": "VERIFIED",  # NOT_STARTED, PENDING, INVALID, FAILED, VERIFYING, VERIFIED
    "feedback_enabled": True,
    "records": [
        {"type": "TXT", "name": "_dmarc.example.com", "value": "...", "status": "VALID"},
        {"type": "TXT", "name": "mail.example.com", "value": "...", "status": "VALID"},
        {"type": "MX", "name": "example.com", "value": "...", "priority": 10, "status": "VALID"}
    ],
    "created_at": "2024-01-01T00:00:00Z"
}

Pods

Drafts

API Keys

Metrics

Organizations

Error Types

Pagination

All list endpoints support pagination:

# First page
response = client.inboxes.list(limit=10)

# Next page
if response.next_page_token:
    next_page = client.inboxes.list(limit=10, page_token=response.next_page_token)

AgentMail Advanced Examples

Advanced patterns for building email agents.

Pattern 1: Event-Driven Agent (Webhook)

A complete agent that responds to incoming emails using webhooks.

import os
import asyncio
from threading import Thread
from flask import Flask, request, Response
import ngrok
from agentmail import AgentMail
from agentmail_toolkit.openai import AgentMailToolkit
from agents import Agent, Runner

app = Flask(__name__)
client = AgentMail()
port = 8080

inbox_username = "support-agent"
inbox = f"{inbox_username}@agentmail.to"

# Setup infrastructure idempotently
client.inboxes.create(username=inbox_username, client_id=f"{inbox_username}-inbox")

listener = ngrok.forward(port, authtoken_from_env=True)
webhook_url = f"{listener.url()}/webhooks"
client.webhooks.create(
    url=webhook_url,
    event_types=["message.received"],
    client_id=f"{inbox_username}-webhook"
)

# Define agent
agent = Agent(
    name="Support Agent",
    instructions=f"""You are a helpful support agent. Your email is {inbox}.
    When you receive an email, respond helpfully and professionally.
    Always be concise and address the user's question directly.""",
    tools=AgentMailToolkit(client).get_tools()
)

messages = []

@app.route("/webhooks", methods=["POST"])
def receive_webhook():
    Thread(target=process_webhook, args=(request.json,)).start()
    return Response(status=200)

def process_webhook(payload):
    global messages
    email = payload["message"]
    
    prompt = f"""
    New email received:
    From: {email["from"]}
    Subject: {email["subject"]}
    Body: {email["text"]}
    """
    
    response = asyncio.run(Runner.run(agent, messages + [{"role": "user", "content": prompt}]))
    
    # Reply to the email
    client.inboxes.messages.reply(
        inbox_id=inbox,
        message_id=email["message_id"],
        html=response.final_output
    )
    
    messages = response.to_input_list()

if __name__ == "__main__":
    print(f"Inbox: {inbox}")
    print(f"Webhook: {webhook_url}")
    app.run(port=port)

Pattern 2: WebSocket Real-time Agent

No public URL required - uses persistent WebSocket connection.

import asyncio
from agentmail import AsyncAgentMail, Subscribe, MessageReceivedEvent
from openai import OpenAI

client = AsyncAgentMail()
openai = OpenAI()
inbox = "realtime-agent@agentmail.to"

async def handle_email(event: MessageReceivedEvent):
    message = event.message
    
    # Generate response with OpenAI
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a helpful email assistant."},
            {"role": "user", "content": f"Reply to this email:\n\n{message.text}"}
        ]
    )
    
    reply_text = response.choices[0].message.content
    
    # Send reply
    await client.inboxes.messages.reply(
        inbox_id=inbox,
        message_id=message.message_id,
        text=reply_text
    )
    print(f"Replied to {message.from_}")

async def main():
    # Create inbox if needed
    await client.inboxes.create(username="realtime-agent", client_id="realtime-agent-inbox")
    
    async with client.websockets.connect() as socket:
        await socket.send_subscribe(Subscribe(inbox_ids=[inbox]))
        print(f"Listening for emails to {inbox}...")
        
        async for event in socket:
            if isinstance(event, MessageReceivedEvent):
                await handle_email(event)

asyncio.run(main())

Pattern 3: Multi-Step Workflow with State

State management for complex workflows (e.g., collecting RSVPs).

from flask import Flask, request, Response
from threading import Thread
from agentmail import AgentMail

app = Flask(__name__)
client = AgentMail()

# In-memory state (use database in production)
events = {}

def create_event(organizer_email, details):
    event_id = f"event_{len(events)}"
    events[event_id] = {
        "state": "collecting",  # collecting | ready | completed
        "organizer": organizer_email,
        "details": details,
        "rsvps": [],
        "target_count": 5
    }
    return event_id

def handle_rsvp(event_id, from_email, response):
    event = events.get(event_id)
    if not event or event["state"] != "collecting":
        return
    
    event["rsvps"].append({"email": from_email, "response": response})
    
    # Check if we hit the target
    yes_count = len([r for r in event["rsvps"] if r["response"] == "yes"])
    if yes_count >= event["target_count"]:
        event["state"] = "ready"
        notify_organizer(event)

def notify_organizer(event):
    client.inboxes.messages.send(
        inbox_id="event-agent@agentmail.to",
        to=[event["organizer"]],
        subject="Your event is ready!",
        text=f"You have {len(event['rsvps'])} RSVPs."
    )

@app.route("/webhooks", methods=["POST"])
def webhook():
    Thread(target=process, args=(request.json,)).start()
    return Response(status=200)

def process(payload):
    message = payload["message"]
    subject = message.get("subject", "")
    text = message.get("text", "").lower()
    from_email = message["from"]
    
    # Parse event ID from subject (e.g., "Re: [event_123] Dinner Invitation")
    if "[event_" in subject:
        event_id = subject.split("[")[1].split("]")[0]
        
        if "yes" in text:
            handle_rsvp(event_id, from_email, "yes")
        elif "no" in text:
            handle_rsvp(event_id, from_email, "no")

if __name__ == "__main__":
    app.run(port=8080)

Pattern 4: Human-in-the-Loop with Drafts

Agent creates drafts for human approval before sending.

from agentmail import AgentMail

client = AgentMail()
inbox = "sales-agent@agentmail.to"

def create_outreach_draft(prospect_email, personalization):
    """Agent creates draft, human reviews and sends."""
    
    draft = client.inboxes.drafts.create(
        inbox_id=inbox,
        to=[prospect_email],
        subject="Quick question about your company",
        text=f"""Hi,

{personalization}

Would you be open to a quick call this week?

Best,
Sales Agent""",
        labels=["pending-review", "outreach"]
    )
    
    print(f"Draft created: {draft.draft_id}")
    return draft.draft_id

def list_pending_drafts():
    """Human reviews drafts."""
    drafts = client.drafts.list(labels=["pending-review"])
    for draft in drafts.drafts:
        print(f"To: {draft.to}, Subject: {draft.subject}")
    return drafts

def approve_and_send(draft_id):
    """Human approves, agent sends."""
    # Update labels
    client.inboxes.drafts.update(
        inbox_id=inbox,
        draft_id=draft_id,
        add_labels=["approved"],
        remove_labels=["pending-review"]
    )
    
    # Send the draft
    result = client.inboxes.drafts.send(inbox_id=inbox, draft_id=draft_id)
    print(f"Sent as message: {result.message_id}")

Pattern 5: Label-Based Workflow

Using labels for state machine and filtering.

from agentmail import AgentMail

client = AgentMail()
inbox = "workflow@agentmail.to"

# Label conventions:
# - status:new, status:in-progress, status:resolved
# - priority:high, priority:low
# - category:billing, category:technical, category:general

def process_new_tickets():
    """Process all new tickets."""
    new_messages = client.inboxes.messages.list(
        inbox_id=inbox,
        labels=["status:new"]
    )
    
    for msg in new_messages.messages:
        # Classify and update
        category = classify_email(msg.text)
        priority = assess_priority(msg.text)
        
        client.inboxes.messages.update(
            inbox_id=inbox,
            message_id=msg.message_id,
            add_labels=[f"category:{category}", f"priority:{priority}", "status:in-progress"],
            remove_labels=["status:new"]
        )

def get_high_priority():
    """Get high priority items."""
    return client.inboxes.threads.list(
        inbox_id=inbox,
        labels=["priority:high", "status:in-progress"]
    )

def resolve_ticket(message_id, resolution):
    """Mark ticket as resolved."""
    # Send resolution
    client.inboxes.messages.reply(
        inbox_id=inbox,
        message_id=message_id,
        text=resolution
    )
    
    # Update status
    client.inboxes.messages.update(
        inbox_id=inbox,
        message_id=message_id,
        add_labels=["status:resolved"],
        remove_labels=["status:in-progress"]
    )

def classify_email(text):
    # Use AI or rules to classify
    if "invoice" in text.lower() or "payment" in text.lower():
        return "billing"
    elif "error" in text.lower() or "bug" in text.lower():
        return "technical"
    return "general"

def assess_priority(text):
    if "urgent" in text.lower() or "asap" in text.lower():
        return "high"
    return "low"

Pattern 6: Multi-Tenant with Pods

Isolating resources per customer.

from agentmail import AgentMail

client = AgentMail()

def onboard_customer(customer_id, domain):
    """Set up isolated email infrastructure for a customer."""
    
    # Create pod for isolation
    pod = client.pods.create(
        name=f"Customer {customer_id}",
        client_id=f"pod-{customer_id}"
    )
    
    # Create inbox in pod
    inbox = client.pods.inboxes.create(
        pod_id=pod.pod_id,
        username="support",
        domain=domain,
        client_id=f"inbox-{customer_id}-support"
    )
    
    # Add custom domain (optional)
    domain_obj = client.pods.domains.create(
        pod_id=pod.pod_id,
        domain=domain,
        feedback_enabled=True
    )
    
    return {
        "pod_id": pod.pod_id,
        "inbox": inbox.inbox_id,
        "domain_status": domain_obj.status
    }

def get_customer_threads(pod_id):
    """Get all threads for a customer."""
    return client.pods.threads.list(pod_id=pod_id)

def offboard_customer(pod_id):
    """Clean up customer resources."""
    # Must delete resources before pod
    
    # Delete inboxes
    inboxes = client.pods.inboxes.list(pod_id=pod_id)
    for inbox in inboxes.inboxes:
        client.inboxes.delete(inbox.inbox_id)
    
    # Delete domains
    domains = client.pods.domains.list(pod_id=pod_id)
    for domain in domains.domains:
        client.domains.delete(domain.domain_id)
    
    # Now delete pod
    client.pods.delete(pod_id)

Pattern 7: Attachment Processing

Download and process email attachments.

from agentmail import AgentMail
import base64

client = AgentMail()
inbox = "processor@agentmail.to"

def process_incoming_attachments(message_id):
    """Download and process all attachments from a message."""
    
    message = client.inboxes.messages.get(inbox_id=inbox, message_id=message_id)
    
    results = []
    for attachment in message.attachments or []:
        # Download attachment
        content = client.inboxes.messages.get_attachment(
            inbox_id=inbox,
            message_id=message_id,
            attachment_id=attachment.attachment_id
        )
        
        # Save to file
        with open(attachment.filename, "wb") as f:
            for chunk in content:
                f.write(chunk)
        
        results.append({
            "filename": attachment.filename,
            "size": attachment.size,
            "type": attachment.content_type
        })
        
        print(f"Downloaded: {attachment.filename}")
    
    return results

def send_with_multiple_attachments(to, files):
    """Send email with multiple attachments."""
    
    attachments = []
    for filepath in files:
        with open(filepath, "rb") as f:
            content = base64.b64encode(f.read()).decode()
        
        filename = filepath.split("/")[-1]
        attachments.append({
            "content": content,
            "filename": filename,
            "content_type": guess_mime_type(filename)
        })
    
    client.inboxes.messages.send(
        inbox_id=inbox,
        to=[to],
        subject="Files attached",
        text="Please see the attached files.",
        attachments=attachments
    )

def guess_mime_type(filename):
    ext = filename.split(".")[-1].lower()
    types = {
        "pdf": "application/pdf",
        "png": "image/png",
        "jpg": "image/jpeg",
        "csv": "text/csv",
        "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    }
    return types.get(ext, "application/octet-stream")

Pattern 8: TypeScript Examples

Webhook Handler (Express)

import express from "express";
import { AgentMailClient } from "agentmail";

const app = express();
const client = new AgentMailClient();

app.use(express.json());

app.post("/webhooks", async (req, res) => {
  res.status(200).send();  // Return immediately
  
  const { event_type, message } = req.body;
  
  if (event_type === "message.received") {
    await client.inboxes.messages.reply(message.inbox_id, message.message_id, {
      text: "Thanks for your email!"
    });
  }
});

app.listen(8080);

WebSocket Listener

import { AgentMailClient, AgentMail } from "agentmail";

const client = new AgentMailClient();

async function main() {
  const socket = await client.websockets.connect();
  
  socket.on("open", () => {
    socket.sendSubscribe({
      type: "subscribe",
      inboxIds: ["agent@agentmail.to"]
    });
  });
  
  socket.on("message", async (event: AgentMail.MessageReceivedEvent) => {
    if (event.type === "message_received") {
      console.log(`New email: ${event.message.subject}`);
      
      await client.inboxes.messages.reply(
        event.message.inboxId,
        event.message.messageId,
        { text: "Got it!" }
      );
    }
  });
}

main();

AgentMail Webhook Events

Complete webhook event types and payload structures.

Event Structure

All webhook payloads follow this structure:

{
  "type": "event",
  "event_type": "message.received",
  "event_id": "evt_xxx"
  // ... event-specific data
}

Message Events

message.received

Triggered when a new email is received. This is the only event that includes full Thread + Message data.

{
  "type": "event",
  "event_type": "message.received",
  "event_id": "evt_xxx",
  "message": {
    "inbox_id": "agent@agentmail.to",
    "thread_id": "thd_xxx",
    "message_id": "msg_xxx",
    "from_": ["sender@example.com"],
    "to": ["agent@agentmail.to"],
    "cc": [],
    "bcc": [],
    "reply_to": [],
    "subject": "Hello",
    "preview": "Short preview...",
    "text": "Full email text",
    "html": "<p>Full email HTML</p>",
    "labels": ["received"],
    "attachments": [
      {
        "attachment_id": "att_xxx",
        "filename": "document.pdf",
        "content_type": "application/pdf",
        "size": 123456,
        "inline": false
      }
    ],
    "in_reply_to": "msg_parent",
    "references": ["msg_1", "msg_2"],
    "timestamp": "2024-01-01T10:00:00Z",
    "created_at": "2024-01-01T10:00:00Z"
  },
  "thread": {
    "thread_id": "thd_xxx",
    "inbox_id": "agent@agentmail.to",
    "subject": "Hello",
    "created_at": "2024-01-01T10:00:00Z"
  }
}

message.sent

Triggered when a message is successfully sent.

{
  "type": "event",
  "event_type": "message.sent",
  "event_id": "evt_xxx",
  "send": {
    "inbox_id": "agent@agentmail.to",
    "thread_id": "thd_xxx",
    "message_id": "msg_xxx",
    "timestamp": "2024-01-01T10:05:00Z",
    "recipients": ["recipient@example.com"]
  }
}

message.delivered

Triggered when recipient's mail server accepts the message.

{
  "type": "event",
  "event_type": "message.delivered",
  "event_id": "evt_xxx",
  "delivery": {
    "inbox_id": "agent@agentmail.to",
    "thread_id": "thd_xxx",
    "message_id": "msg_xxx",
    "timestamp": "2024-01-01T10:06:00Z",
    "recipients": ["recipient@example.com"]
  }
}

Note: message.delivered means the receiving server accepted it, NOT that it landed in inbox (could still go to spam).

message.bounced

Triggered when a message fails to deliver.

{
  "type": "event",
  "event_type": "message.bounced",
  "event_id": "evt_xxx",
  "bounce": {
    "inbox_id": "agent@agentmail.to",
    "thread_id": "thd_xxx",
    "message_id": "msg_xxx",
    "timestamp": "2024-01-01T10:07:00Z",
    "type": "Permanent",
    "sub_type": "General",
    "recipients": [
      {
        "address": "invalid@example.com",
        "status": "bounced"
      }
    ]
  }
}

Warning: Bounced addresses are permanently blocked. Keep bounce rate < 4%.

message.complained

Triggered when recipient marks email as spam.

{
  "type": "event",
  "event_type": "message.complained",
  "event_id": "evt_xxx",
  "complaint": {
    "inbox_id": "agent@agentmail.to",
    "thread_id": "thd_xxx",
    "message_id": "msg_xxx",
    "timestamp": "2024-01-01T10:08:00Z",
    "type": "abuse",
    "sub_type": "spam",
    "recipients": ["complainer@example.com"]
  }
}

Warning: Complained addresses are permanently blocked.

message.rejected

Triggered when a message is rejected before sending (e.g., validation error, blocked address).

{
  "type": "event",
  "event_type": "message.rejected",
  "event_id": "evt_xxx",
  "reject": {
    "inbox_id": "agent@agentmail.to",
    "thread_id": "thd_xxx",
    "message_id": "msg_xxx",
    "timestamp": "2024-01-01T10:09:00Z",
    "reason": "Recipient address is blocked"
  }
}

Domain Events

domain.verified

Triggered when a custom domain is successfully verified.

{
  "type": "event",
  "event_type": "domain.verified",
  "event_id": "evt_xxx",
  "domain": {
    "domain_id": "example.com",
    "status": "verified",
    "feedback_enabled": true,
    "records": [
      {
        "type": "TXT",
        "name": "_dmarc.example.com",
        "value": "v=DMARC1; p=none",
        "status": "VALID"
      }
    ],
    "created_at": "2024-01-01T09:00:00Z",
    "updated_at": "2024-01-01T10:00:00Z"
  }
}

Creating Webhooks

Subscribe to Specific Events

client.webhooks.create(
    url="https://example.com/webhooks",
    event_types=["message.received", "message.bounced"],
    client_id="my-webhook"
)

Filter by Inbox

client.webhooks.create(
    url="https://example.com/webhooks",
    event_types=["message.received"],
    inbox_ids=["agent@agentmail.to"],
    client_id="inbox-specific-webhook"
)

Filter by Pod

client.webhooks.create(
    url="https://example.com/webhooks",
    event_types=["message.received"],
    pod_ids=["pod_xxx"],
    client_id="pod-specific-webhook"
)

Handling Webhooks

Best Practice Pattern

from flask import Flask, request, Response
from threading import Thread

app = Flask(__name__)

@app.route("/webhooks", methods=["POST"])
def webhook():
    # Return 200 immediately
    Thread(target=process, args=(request.json,)).start()
    return Response(status=200)

def process(payload):
    event_type = payload["event_type"]
    
    if event_type == "message.received":
        handle_received(payload["message"])
    elif event_type == "message.bounced":
        handle_bounce(payload["bounce"])
    elif event_type == "message.complained":
        handle_complaint(payload["complaint"])

def handle_received(message):
    print(f"New email from {message['from_']}")

def handle_bounce(bounce):
    print(f"Bounce for {bounce['recipients']}")
    # Remove bounced addresses from your list

def handle_complaint(complaint):
    print(f"Complaint from {complaint['recipients']}")
    # Remove complaining addresses from your list

Field Reference

Common Fields

Attachment Fields