Agentmail

AgentMail SDK

AgentMail is an API-first email platform for AI agents. Install the SDK and initialize the client.

Installation

# TypeScript/Node
npm install agentmail

# Python
pip install agentmail

Setup

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

from agentmail import AgentMail
client = AgentMail(api_key="YOUR_API_KEY")

Inboxes

Create scalable inboxes on-demand. Each inbox has a unique email address.

// Create inbox (auto-generated address)
const autoInbox = await client.inboxes.create();

// Create with custom username and domain
const customInbox = await client.inboxes.create({
  username: "support",
  domain: "yourdomain.com",
});

// List, get, delete
const inboxes = await client.inboxes.list();
const fetchedInbox = await client.inboxes.get("inbox@agentmail.to");
await client.inboxes.delete("inbox@agentmail.to");

# Create inbox (auto-generated address)
inbox = client.inboxes.create()

# Create with custom username and domain
from agentmail.inboxes.types import CreateInboxRequest
inbox = client.inboxes.create(
    request=CreateInboxRequest(username="support", domain="yourdomain.com"),
)

# List, get, delete
inboxes = client.inboxes.list()
inbox = client.inboxes.get(inbox_id="inbox@agentmail.to")
client.inboxes.delete(inbox_id="inbox@agentmail.to")

Messages

Always send both text and html for best deliverability.

// Send message
await client.inboxes.messages.send("agent@agentmail.to", {
  to: "recipient@example.com",
  subject: "Hello",
  text: "Plain text version",
  html: "<p>HTML version</p>",
  labels: ["outreach"],
});

// Reply to message
await client.inboxes.messages.reply("agent@agentmail.to", "msg_123", {
  text: "Thanks for your email!",
});

// List and get messages
const messages = await client.inboxes.messages.list("agent@agentmail.to");
const message = await client.inboxes.messages.get("agent@agentmail.to", "msg_123");

// Update labels
await client.inboxes.messages.update("agent@agentmail.to", "msg_123", {
  addLabels: ["replied"],
  removeLabels: ["unreplied"],
});

# Send message
client.inboxes.messages.send(
    inbox_id="agent@agentmail.to",
    to="recipient@example.com",
    subject="Hello",
    text="Plain text version",
    html="<p>HTML version</p>",
    labels=["outreach"]
)

# Reply to message
client.inboxes.messages.reply(
    inbox_id="agent@agentmail.to",
    message_id="msg_123",
    text="Thanks for your email!"
)

# List and get messages
messages = client.inboxes.messages.list(inbox_id="agent@agentmail.to")
message = client.inboxes.messages.get(inbox_id="agent@agentmail.to", message_id="msg_123")

# Update labels
client.inboxes.messages.update(
    inbox_id="agent@agentmail.to",
    message_id="msg_123",
    add_labels=["replied"],
    remove_labels=["unreplied"]
)

Threads

Threads group related messages in a conversation.

// List threads (with optional label filter)
const threads = await client.inboxes.threads.list("agent@agentmail.to", {
  labels: ["unreplied"],
});

// Get thread details
const thread = await client.inboxes.threads.get("agent@agentmail.to", "thd_123");

// Org-wide thread listing
const allThreads = await client.threads.list();

# List threads (with optional label filter)
threads = client.inboxes.threads.list(inbox_id="agent@agentmail.to", labels=["unreplied"])

# Get thread details
thread = client.inboxes.threads.get(inbox_id="agent@agentmail.to", thread_id="thd_123")

# Org-wide thread listing
all_threads = client.threads.list()

Attachments

Send attachments with Base64 encoding. Retrieve from messages or threads.

// Send with attachment
const content = Buffer.from(fileBytes).toString("base64");
await client.inboxes.messages.send("agent@agentmail.to", {
  to: "recipient@example.com",
  subject: "Report",
  text: "See attached.",
  attachments: [
    { content, filename: "report.pdf", contentType: "application/pdf" },
  ],
});

// Get attachment
const fileData = await client.inboxes.messages.getAttachment(
  "agent@agentmail.to",
  "msg_123",
  "att_456",
);

import base64

# Send with attachment
content = base64.b64encode(file_bytes).decode()
client.inboxes.messages.send(
    inbox_id="agent@agentmail.to",
    to="recipient@example.com",
    subject="Report",
    text="See attached.",
    attachments=[{"content": content, "filename": "report.pdf", "content_type": "application/pdf"}]
)

# Get attachment
file_data = client.inboxes.messages.get_attachment(
    inbox_id="agent@agentmail.to",
    message_id="msg_123",
    attachment_id="att_456"
)

Drafts

Create drafts for human-in-the-loop approval before sending.

// Create draft
const draft = await client.inboxes.drafts.create("agent@agentmail.to", {
  to: "recipient@example.com",
  subject: "Pending approval",
  text: "Draft content",
});

// Send draft (converts to message)
await client.inboxes.drafts.send("agent@agentmail.to", draft.draftId, {});

# Create draft
draft = client.inboxes.drafts.create(
    inbox_id="agent@agentmail.to",
    to="recipient@example.com",
    subject="Pending approval",
    text="Draft content"
)

# Send draft (converts to message)
client.inboxes.drafts.send(inbox_id="agent@agentmail.to", draft_id=draft.draft_id)

Pods

Multi-tenant isolation for SaaS platforms. Each customer gets isolated inboxes.

// Create pod for a customer
const pod = await client.pods.create({ clientId: "customer_123" });

// Create inbox within pod
const inbox = await client.pods.inboxes.create(pod.podId, {});

// List inboxes scoped to pod
const inboxes = await client.pods.inboxes.list(pod.podId);

# Create pod for a customer
pod = client.pods.create(client_id="customer_123")

# Create inbox within pod (pods.inboxes.create accepts flat kwargs)
inbox = client.pods.inboxes.create(pod_id=pod.pod_id)

# List inboxes scoped to pod
inboxes = client.pods.inboxes.list(pod_id=pod.pod_id)

Idempotency

Use clientId for safe retries on create operations.

const inbox = await client.inboxes.create({
  clientId: "unique-idempotency-key",
});
// Retrying with same clientId returns the original inbox, not a duplicate

from agentmail.inboxes.types import CreateInboxRequest

inbox = client.inboxes.create(
    request=CreateInboxRequest(client_id="unique-idempotency-key"),
)
# Retrying with same client_id returns the original inbox, not a duplicate

Real-Time Events

For real-time notifications, see the reference files:

• webhooks.md - HTTP-based notifications (requires public URL)
• websockets.md - Persistent connection (no public URL needed)

Webhooks

Webhooks provide real-time HTTP notifications when email events occur. Use webhooks when you have a public URL endpoint.

When to Use

• Production applications with public endpoints
• Event-driven architectures
• When you need to process events on your server

For local development without a public URL, use websockets.md instead.

Setup

Register a webhook endpoint to receive events.

eventTypes / event_types is required — you must pass the list of events the webhook should receive.

import { AgentMailClient } from "agentmail";

const client = new AgentMailClient({ apiKey: "YOUR_API_KEY" });

// Create webhook
const webhook = await client.webhooks.create({
  url: "https://your-server.com/webhooks",
  eventTypes: ["message.received"],
});

// List webhooks
const webhooks = await client.webhooks.list();

// Delete webhook
await client.webhooks.delete(webhook.webhookId);

from agentmail import AgentMail
client = AgentMail(api_key="YOUR_API_KEY")

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

# List webhooks
webhooks = client.webhooks.list()

# Delete webhook
client.webhooks.delete(webhook_id=webhook.webhook_id)

Event Types

Payload Structure

All webhook payloads follow this structure:

{
  "type": "event",
  "event_type": "message.received",
  "event_id": "evt_123abc",
  "message": {
    "inbox_id": "inbox_456def",
    "thread_id": "thd_789ghi",
    "message_id": "msg_123abc",
    "from": "Jane Doe <jane@example.com>",
    "to": ["Agent <agent@agentmail.to>"],
    "subject": "Question about my account",
    "text": "Full text body",
    "html": "<html>...</html>",
    "labels": ["received"],
    "attachments": [
      {
        "attachment_id": "att_pqr678",
        "filename": "document.pdf",
        "content_type": "application/pdf",
        "size": 123456
      }
    ],
    "created_at": "2023-10-27T10:00:00Z"
  },
  "thread": {}
}

Handling Webhooks

Your endpoint should:

1. Return 200 OK immediately 2. Process the payload asynchronously

Express (TypeScript)

import express from "express";

const app = express();
app.use(express.json());

app.post("/webhooks", (req, res) => {
  const payload = req.body;

  if (payload.event_type === "message.received") {
    // Queue for async processing
    processEmail(payload.message);
  }

  res.status(200).send("OK"); // Return immediately
});

Flask (Python)

from flask import Flask, request

app = Flask(__name__)

@app.route("/webhooks", methods=["POST"])
def handle_webhook():
    payload = request.json

    if payload["event_type"] == "message.received":
        # Queue for async processing
        process_email.delay(payload["message"])

    return "OK", 200  # Return immediately

Webhook Verification

Verify webhook signatures to ensure requests are from AgentMail.

TypeScript

import crypto from "crypto";
import express from "express";

function verifySignature(
  payload: Buffer,
  signature: string,
  secret: string
): boolean {
  const expected = crypto
    .createHmac("sha256", secret)
    .update(payload)
    .digest("hex");
  const expectedBuf = Buffer.from(expected, "hex");
  const signatureBuf = Buffer.from(signature, "hex");
  // timingSafeEqual throws RangeError on mismatched lengths;
  // return false for any malformed header instead of crashing.
  if (expectedBuf.length !== signatureBuf.length) return false;
  return crypto.timingSafeEqual(expectedBuf, signatureBuf);
}

app.post("/webhooks", express.raw({ type: "application/json" }), (req, res) => {
  const signature = req.headers["x-agentmail-signature"];
  if (typeof signature !== "string") {
    return res.status(401).send("Missing signature");
  }

  const payload = req.body;
  if (!verifySignature(payload, signature, WEBHOOK_SECRET)) {
    return res.status(401).send("Invalid signature");
  }
  const event = JSON.parse(payload.toString("utf8"));
  // Process event...
  res.status(200).send("OK");
});

Python

import hmac
import hashlib

def verify_signature(payload: bytes, signature, secret: str) -> bool:
    # compare_digest raises TypeError on None, bytes, or any non-str value.
    # Reject anything that isn't a string up front.
    if not isinstance(signature, str) or not signature:
        return False
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.route("/webhooks", methods=["POST"])
def handle_webhook():
    signature = request.headers.get("X-AgentMail-Signature")
    if not verify_signature(request.data, signature, WEBHOOK_SECRET):
        return "Invalid signature", 401
    # Process payload...

Local Development

Use ngrok to expose your local server:

ngrok http 5000
# Use the ngrok URL when creating the webhook

WebSockets

WebSockets provide real-time, low-latency email event streaming over a persistent connection. No public URL required.

When to Use

• Local development (no ngrok needed)
• Client-side applications
• When you need bidirectional communication
• Lower latency than webhooks

For production with public endpoints, webhooks.md may be simpler.

Comparison

TypeScript SDK

Basic Usage

import { AgentMailClient, AgentMail } from "agentmail";

const client = new AgentMailClient({ apiKey: process.env.AGENTMAIL_API_KEY });

async function main() {
  const socket = await client.websockets.connect();

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

  socket.on("message", (event: AgentMail.MessageReceivedEvent) => {
    if (event.eventType === "message.received") {
      console.log("From:", event.message.from);
      console.log("Subject:", event.message.subject);
    }
  });

  socket.on("close", (event) => console.log("Disconnected:", event.code));
  socket.on("error", (error) => console.error("Error:", error));
}

main();

React Hook

import { useEffect, useState } from "react";
import { AgentMailClient, AgentMail } from "agentmail";

function useAgentMailWebSocket(apiKey: string, inboxIds: string[]) {
  const [lastMessage, setLastMessage] =
    useState<AgentMail.MessageReceivedEvent | null>(null);
  const [isConnected, setIsConnected] = useState(false);

  useEffect(() => {
    const client = new AgentMailClient({ apiKey });
    let socket: Awaited<ReturnType<typeof client.websockets.connect>>;

    async function connect() {
      socket = await client.websockets.connect();

      socket.on("open", () => {
        setIsConnected(true);
        socket.sendSubscribe({ type: "subscribe", inboxIds });
      });

      socket.on("message", (event) => {
        if (event.eventType === "message.received") {
          setLastMessage(event);
        }
      });

      socket.on("close", () => setIsConnected(false));
    }

    connect();
    return () => socket?.close();
  }, [apiKey, inboxIds.join(",")]);

  return { lastMessage, isConnected };
}

Python SDK

Sync Usage

from agentmail import AgentMail, Subscribe, Subscribed, MessageReceivedEvent

client = AgentMail(api_key="YOUR_API_KEY")

with client.websockets.connect() as socket:
    # Subscribe to inboxes
    socket.send_subscribe(Subscribe(inbox_ids=["agent@agentmail.to"]))

    # Process events
    for event in socket:
        if isinstance(event, Subscribed):
            print(f"Subscribed to: {event.inbox_ids}")
        elif isinstance(event, MessageReceivedEvent):
            print(f"From: {event.message.from_}")
            print(f"Subject: {event.message.subject}")

Async Usage

import asyncio
from agentmail import AsyncAgentMail, Subscribe, MessageReceivedEvent

client = AsyncAgentMail(api_key="YOUR_API_KEY")

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: {event.message.subject}")

asyncio.run(main())

Event Handler Pattern

import threading
from agentmail import AgentMail, Subscribe, EventType

client = AgentMail(api_key="YOUR_API_KEY")

with client.websockets.connect() as socket:
    socket.on(EventType.OPEN, lambda _: print("Connected"))
    socket.on(EventType.MESSAGE, lambda msg: print("Received:", msg))
    socket.on(EventType.CLOSE, lambda _: print("Disconnected"))
    socket.on(EventType.ERROR, lambda err: print("Error:", err))

    socket.send_subscribe(Subscribe(inbox_ids=["agent@agentmail.to"]))

    # Run listener in background thread
    listener = threading.Thread(target=socket.start_listening, daemon=True)
    listener.start()
    listener.join()

Subscribe Options

Filter events by inbox, pod, or event type.

socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["agent@agentmail.to"],
  eventTypes: ["message.received", "message.sent"],
});

// By pods
socket.sendSubscribe({
  type: "subscribe",
  podIds: ["pod_123", "pod_456"],
});

from agentmail import Subscribe

# By inboxes
Subscribe(inbox_ids=["inbox1@agentmail.to", "inbox2@agentmail.to"])

# By pods
Subscribe(pod_ids=["pod_123", "pod_456"])

# By event types
Subscribe(
    inbox_ids=["agent@agentmail.to"],
    event_types=["message.received", "message.sent"]
)

Event Types

Message Properties

The event.message object contains (Python snake_case / TypeScript camelCase):

Python uses from_ because from is a reserved keyword; TypeScript uses from directly.

Error Handling

import { AgentMailClient, AgentMailError } from "agentmail";

try {
  const socket = await client.websockets.connect();
  // ...
} catch (err) {
  if (err instanceof AgentMailError) {
    console.error(`API error: ${err.statusCode} - ${err.message}`);
  } else {
    console.error("Connection error:", err);
  }
}

from agentmail import AsyncAgentMail, Subscribe, MessageReceivedEvent
from agentmail.core.api_error import ApiError

client = AsyncAgentMail(api_key="YOUR_API_KEY")

async def main():
    try:
        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):
                    await process_email(event.message)

    except ApiError as e:
        print(f"API error: {e.status_code} - {e.body}")
    except Exception as e:
        print(f"Connection error: {e}")