WebSockets

WebSockets

Real-time, low-latency email event streaming

WebSockets provide a persistent, bidirectional connection to AgentMail for receiving email events in real-time. Unlike webhooks, WebSockets don’t require a public URL or external tools like ngrok.

Why Use WebSockets?

FeatureWebhookWebSocket
SetupRequires public URL + ngrokNo external tools needed
ConnectionHTTP request per eventPersistent connection
DirectionAgentMail → Your serverBidirectional
FirewallMust expose portOutbound only
LatencyHTTP round-tripInstant streaming

Python SDK

The Python SDK provides both synchronous and asynchronous WebSocket clients.

Async Usage

import asyncio
from agentmail import AsyncAgentMail, Subscribe, Subscribed, MessageReceivedEvent
client = AsyncAgentMail(api_key="YOUR_API_KEY")
async def main():
    async with client.websockets.connect() as socket:
        # Subscribe to inboxes
        await socket.send_subscribe(Subscribe(inbox_ids=["agent@agentmail.to"]))
        # Process events as they arrive
        async for event in socket:
            if isinstance(event, Subscribed):
                print(f"Subscribed to: {event.inbox_ids}")
            elif isinstance(event, MessageReceivedEvent):
                print(f"New email from: {event.message.from_}")
                print(f"Subject: {event.message.subject}")
asyncio.run(main())

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 as they arrive
    for event in socket:
        if isinstance(event, Subscribed):
            print(f"Subscribed to: {event.inbox_ids}")
        elif isinstance(event, MessageReceivedEvent):
            print(f"New email from: {event.message.from_}")
            print(f"Subject: {event.message.subject}")

Event Handler Pattern

You can also use event handlers instead of iterating:

import asyncio
from agentmail import AsyncAgentMail, Subscribe, EventType
client = AsyncAgentMail(api_key="YOUR_API_KEY")
async def main():
    async with client.websockets.connect() as socket:
        # Register event handlers
        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))
        # Subscribe and start listening
        await socket.send_subscribe(Subscribe(inbox_ids=["agent@agentmail.to"]))
        await socket.start_listening()
asyncio.run(main())

For sync usage with event handlers, run the listener in a background thread:

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"]))
    # Start listening in background thread
    listener = threading.Thread(target=socket.start_listening, daemon=True)
    listener.start()
    listener.join()

TypeScript SDK

The TypeScript SDK provides a WebSocket client with automatic reconnection.

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();
  // Handle events
  socket.on("open", () => {
    console.log("Connected");
    // Subscribe to inboxes after connection is open
    socket.sendSubscribe({
      type: "subscribe",
      inboxIds: ["agent@agentmail.to"],
    });
  });
  socket.on("message", (event: AgentMail.Subscribed | AgentMail.MessageReceivedEvent) => {
    if (event.type === "subscribed") {
      console.log("Subscribed to:", event.inboxIds);
    } else if (event.type === "message_received") {
      console.log("New email from:", event.message.from_);
      console.log("Subject:", event.message.subject);
    }
  });
  socket.on("close", (event) => {
    console.log("Disconnected:", event.code, event.reason);
  });
  socket.on("error", (error) => {
    console.error("Error:", error);
  });
}
main();

React/Next.js Usage

Using the SDK with React:

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.type === "message_received") {
          setLastMessage(event);
        }
      });
      socket.on("close", () => setIsConnected(false));
    }
    connect();
    return () => socket?.close();
  }, [apiKey, inboxIds.join(",")]);
  return { lastMessage, isConnected };
}

Subscribe Options

When subscribing to events, you can filter by inbox, pod, or event type:

Python:

from agentmail import Subscribe
# Subscribe to specific inboxes
Subscribe(inbox_ids=["inbox1@agentmail.to", "inbox2@agentmail.to"])
# Subscribe to pods
Subscribe(pod_ids=["pod-id-1", "pod-id-2"])
# Subscribe to specific event types
Subscribe(
    inbox_ids=["agent@agentmail.to"],
    event_types=["message.received", "message.sent"]
)
# Subscribe to filtered inbound events
Subscribe(
    inbox_ids=["agent@agentmail.to"],
    event_types=[
        "message.received",
        "message.received.spam",
        "message.received.blocked",
        "message.received.unauthenticated",
    ]
)

TypeScript:

// Subscribe to specific inboxes
socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["inbox1@agentmail.to", "inbox2@agentmail.to"],
});
// Subscribe to pods
socket.sendSubscribe({
  type: "subscribe",
  podIds: ["pod-id-1", "pod-id-2"],
});
// Subscribe to specific event types
socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["agent@agentmail.to"],
  eventTypes: ["message.received", "message.sent"],
});
// Subscribe to filtered inbound events
socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["agent@agentmail.to"],
  eventTypes: [
    "message.received",
    "message.received.spam",
    "message.received.blocked",
    "message.received.unauthenticated",
  ],
});

By default (when no event_types are specified), spam, blocked, and unauthenticated events are excluded from the subscription. To receive them, explicitly include message.received.spam, message.received.blocked, or message.received.unauthenticated in event_types. These events also require the matching label visibility permission: label_spam_read, label_blocked_read, or label_unauthenticated_read.

Event Types

Connection Events

EventPythonTypeScriptDescription
subscribedSubscribedAgentMail.SubscribedSubscription confirmed

Message Events

EventPythonTypeScriptDescription
message_receivedMessageReceivedEventAgentMail.MessageReceivedEventEmail received. Check event_type for message.received, message.received.spam, message.received.blocked, or message.received.unauthenticated
message_sentMessageSentEventAgentMail.MessageSentEventEmail was sent
message_deliveredMessageDeliveredEventAgentMail.MessageDeliveredEventEmail was delivered
message_bouncedMessageBouncedEventAgentMail.MessageBouncedEventEmail bounced
message_complainedMessageComplainedEventAgentMail.MessageComplainedEventEmail marked as spam
message_rejectedMessageRejectedEventAgentMail.MessageRejectedEventEmail was rejected
message_openedMessageOpenedEventAgentMail.MessageOpenedEventTracked email was opened for the first time

Domain Events

EventPythonTypeScriptDescription
domain_verifiedDomainVerifiedEventAgentMail.DomainVerifiedEventDomain verification completed

Message Properties

The event.message object contains:

PythonTypeScriptDescription
inbox_idinboxIdInbox that received the email
message_idmessageIdUnique message ID
thread_idthreadIdConversation thread ID
from_from_Sender email address
totoRecipients list
subjectsubjectSubject line
texttextPlain text body
htmlhtmlHTML body (if present)
attachmentsattachmentsList of attachments

Error Handling

Python:

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}")

TypeScript:

import { AgentMailClient, AgentMail, AgentMailError } from "agentmail";
const client = new AgentMailClient({
  apiKey: process.env.AGENTMAIL_API_KEY,
});
async function main() {
  try {
    const socket = await client.websockets.connect();
    socket.on("open", () => {
      socket.sendSubscribe({
        type: "subscribe",
        inboxIds: ["agent@agentmail.to"],
      });
    });
    socket.on("message", (event: AgentMail.MessageReceivedEvent) => {
      if (event.type === "message_received") {
        processEmail(event.message);
      }
    });
    socket.on("error", (error) => {
      console.error("WebSocket error:", error);
    });
    socket.on("close", (event) => {
      console.log("Disconnected:", event.code, event.reason);
    });
  } catch (err) {
    if (err instanceof AgentMailError) {
      console.error(`API error: ${err.statusCode} - ${err.message}`);
    } else {
      console.error("Connection error:", err);
    }
  }
}
main();

Copy for Cursor / Claude

Copy one of the blocks below into Cursor or Claude for WebSockets in one shot.

"""
AgentMail WebSockets — copy into Cursor/Claude. Real-time events, no public URL needed.
Sync: with client.websockets.connect() as socket: socket.send_subscribe(Subscribe(inbox_ids=[...])); for event in socket: ...
Async: async with client.websockets.connect() as socket: await socket.send_subscribe(...); async for event in socket: ...
Subscribe(inbox_ids=[...], pod_ids=[...], event_types=[...])
Event types: Subscribed, MessageReceivedEvent, MessageSentEvent, MessageDeliveredEvent, MessageBouncedEvent, MessageComplainedEvent, MessageRejectedEvent, MessageOpenedEvent, DomainVerifiedEvent
Spam, blocked, and unauthenticated events require explicit opt-in via event_types, plus the matching label_spam_read / label_blocked_read / label_unauthenticated_read permission.
"""
from agentmail import AgentMail, Subscribe, Subscribed, MessageReceivedEvent
client = AgentMail(api_key="YOUR_API_KEY")
with client.websockets.connect() as socket:
  socket.send_subscribe(Subscribe(inbox_ids=["agent@agentmail.to"]))
  for event in socket:
    if isinstance(event, Subscribed): print(event.inbox_ids)
    elif isinstance(event, MessageReceivedEvent): print(event.message.subject)