Skip to content
machinemail

Realtime events

Consume MachineMail events live over Server-Sent Events or WebSocket, with filters, keepalives, reconnection, backfill from the event log and deduplication by event_id.

Webhooks need a public URL. When your agent runs on a laptop, in a sandbox or behind a firewall, open a stream instead. MachineMail offers two:

SSEWebSocket
EndpointGET /v0/events/streamGET /v0/ws (upgrade)
FiltersQuery parametersA subscribe message, changeable at any time
Resume after a dropLast-Event-ID header, replayed by the serverBackfill yourself from GET /v0/events
KeepaliveServer sends a comment every 15 secondsClient sends ping
Permissionmetrics_readAny of the read permissions below

Both carry the same event envelope as webhooks, and both only see events inside the key’s scope. A pod-scoped key streams its pod; an inbox-scoped key streams its inbox. Both also send an event only when the key holds its read permission: message.* needs message_read, inbox.* inbox_read, domain.* domain_read, draft.* draft_read and webhook.test webhook_read. See Events for the event types.

Authentication

Send Authorization: Bearer <key> when your client can. Browser EventSource and browser WebSocket cannot set headers, so both endpoints also accept the key as ?token=<key>.

Warning

A key in a URL can end up in proxy logs and browser history. In a browser, use an inbox-scoped key with only the permissions the page needs. See API keys.

Server-Sent Events

Terminal
curl -N "https://api.machinemail.dev/v0/events/stream?inbox_id=support-agent@machinemail.dev&event_types=message.received" \
  -H "Authorization: Bearer $MACHINEMAIL_API_KEY"
Stream
event: ready
data: {"type":"subscribed"}

event: message.received
data: {"type":"event","message":{...},"thread":{...},"event_id":"evt_01m4fy8hba0000449dcj0gmkc5","event_type":"message.received","timestamp":"2026-10-09T09:00:16.106Z","data":{"message":{...},"thread":{...}}}
id: evt_01m4fy8hba0000449dcj0gmkc5

: ping
  • The first frame is always event: ready. Wait for it before you treat the stream as live.
  • Each event frame has event set to the event type, id set to the event ID, and data set to the JSON envelope.
  • When nothing happens for 15 seconds the server writes a : ping comment. SSE parsers ignore it; it keeps proxies from closing the connection.

Filters

ParameterEffect
inbox_id (or inbox_ids)Only events for these inboxes.
pod_id (or pod_ids)Only events for these pods.
event_typesOnly these event types.

Each takes one value, repeated parameters (event_types=a&event_types=b), a comma-separated list (event_types=a,b) or a JSON array.

Resuming with Last-Event-ID

Reconnect with the header Last-Event-ID: <last evt_ id you processed> and the server first replays up to 500 events that happened after it, then continues live. EventSource sends this header for you on automatic reconnects.

Note

The replay honours event_types and the key’s scope, but not the inbox_id or pod_id query filters. After a reconnect you can receive events for other inboxes in scope. Filter on inbox_id in your handler, or use an inbox-scoped key.

TypeScript consumer

This consumer uses fetch, so it can send the Bearer header. It parses frames, reconnects with exponential backoff, resumes with Last-Event-ID and drops duplicates.

sse.ts
const URL_ = 'https://api.machinemail.dev/v0/events/stream?inbox_id=support-agent@machinemail.dev&event_types=message.received';
const seen = new Set<string>();
let lastEventId: string | undefined;

async function handle(event: any) {
  if (seen.has(event.event_id)) return;
  seen.add(event.event_id);
  const inboxId = event.data.message?.inbox_id ?? event.data.inbox?.inbox_id;
  if (inboxId && inboxId !== 'support-agent@machinemail.dev') return; // the replay is not inbox-filtered
  console.log(event.event_type, event.data.message?.message_id, event.data.message?.trust?.level);
}

async function connect() {
  const headers: Record<string, string> = { Authorization: `Bearer ${process.env.MACHINEMAIL_API_KEY}` };
  if (lastEventId) headers['Last-Event-ID'] = lastEventId;
  const res = await fetch(URL_, { headers });
  if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);

  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
  let buffer = '';
  for (;;) {
    const { value, done } = await reader.read();
    if (done) throw new Error('stream closed');
    buffer += value;
    let end: number;
    while ((end = buffer.indexOf('\n\n')) !== -1) {
      const frame = buffer.slice(0, end);
      buffer = buffer.slice(end + 2);
      let type = 'message';
      let id: string | undefined;
      const data: string[] = [];
      for (const line of frame.split('\n')) {
        if (line.startsWith(':')) continue; // keepalive comment
        if (line.startsWith('event: ')) type = line.slice(7);
        else if (line.startsWith('id: ')) id = line.slice(4);
        else if (line.startsWith('data: ')) data.push(line.slice(6));
      }
      if (type === 'ready' || !data.length) continue;
      await handle(JSON.parse(data.join('\n')));
      if (id) lastEventId = id;
    }
  }
}

let delay = 1000;
for (;;) {
  try {
    await connect();
  } catch (err) {
    console.warn('stream dropped:', (err as Error).message);
  }
  await new Promise((r) => setTimeout(r, delay + Math.random() * 500));
  delay = Math.min(delay * 2, 60_000);
}

Reset delay to 1 second once a connection has delivered ready, if you want fast reconnects after long healthy sessions. In a browser, new EventSource(url + '&token=' + key) gives you reconnection and Last-Event-ID for free.

Python consumer

sse.py
import json, os, random, time, requests

URL = "https://api.machinemail.dev/v0/events/stream"
PARAMS = {"inbox_id": "support-agent@machinemail.dev", "event_types": "message.received"}
seen, last_event_id = set(), None


def handle(event):
    if event["event_id"] in seen:
        return
    seen.add(event["event_id"])
    msg = event["data"].get("message") or {}
    print(event["event_type"], msg.get("message_id"), (msg.get("trust") or {}).get("level"))


def connect():
    global last_event_id
    headers = {"Authorization": f"Bearer {os.environ['MACHINEMAIL_API_KEY']}"}
    if last_event_id:
        headers["Last-Event-ID"] = last_event_id
    with requests.get(URL, params=PARAMS, headers=headers, stream=True, timeout=(10, 60)) as res:
        res.raise_for_status()
        kind, ident, data = "message", None, []
        for line in res.iter_lines(decode_unicode=True):
            if line == "":
                if kind != "ready" and data:
                    handle(json.loads("\n".join(data)))
                    if ident:
                        last_event_id = ident
                kind, ident, data = "message", None, []
            elif line.startswith(":"):
                continue
            elif line.startswith("event: "):
                kind = line[7:]
            elif line.startswith("id: "):
                ident = line[4:]
            elif line.startswith("data: "):
                data.append(line[6:])


delay = 1
while True:
    try:
        connect()
    except Exception as err:
        print("stream dropped:", err)
    time.sleep(delay + random.random() / 2)
    delay = min(delay * 2, 60)

The 60-second read timeout is safe because the server pings every 15 seconds; a silent minute means the connection is dead.

WebSocket

Connect to wss://api.machinemail.dev/v0/ws. On open you are subscribed to every event in the key’s scope. Narrow it with a subscribe message, which you can send again at any time to replace the filter:

Client to server
{ "type": "subscribe", "inbox_ids": ["support-agent@machinemail.dev"], "event_types": ["message.received"] }
Server to client
{ "type": "subscribed", "inbox_ids": ["support-agent@machinemail.dev"], "event_types": ["message.received"] }

Then every event arrives as a text frame with the envelope ("type": "event"). The other messages:

Client sendsServer answers
{ "type": "ping" }{ "type": "pong" }
Invalid JSON{ "type": "error", "message": "Invalid JSON" }
Any other type{ "type": "error", "message": "Unknown message type: ..." }

The server closes connections that are idle for 120 seconds. Send a ping every 30 seconds or so.

Note

The WebSocket does not replay missed events. After a reconnect, backfill from the event log with GET /v0/events?after=<last event_id>, which returns newer events oldest first. That endpoint needs metrics_read.

TypeScript consumer

ws.ts
const API = 'https://api.machinemail.dev/v0';
const KEY = process.env.MACHINEMAIL_API_KEY!;
const seen = new Set<string>();
let lastEventId: string | undefined;

function handle(event: any) {
  if (event.type !== 'event' || seen.has(event.event_id)) return;
  seen.add(event.event_id);
  lastEventId = event.event_id;
  console.log(event.event_type, event.data.message?.message_id);
}

async function backfill() {
  if (!lastEventId) return;
  let pageToken: string | undefined;
  do {
    const q = new URLSearchParams({ after: lastEventId, event_types: 'message.received', limit: '100' });
    if (pageToken) q.set('page_token', pageToken);
    const page = await fetch(`${API}/events?${q}`, { headers: { Authorization: `Bearer ${KEY}` } }).then((r) => r.json());
    for (const e of page.events) handle({ type: 'event', ...e });
    pageToken = page.next_page_token;
  } while (pageToken);
}

let delay = 1000;
function connect() {
  const ws = new WebSocket(`wss://api.machinemail.dev/v0/ws?token=${encodeURIComponent(KEY)}`);
  let pinger: ReturnType<typeof setInterval>;
  ws.onopen = async () => {
    delay = 1000;
    ws.send(JSON.stringify({ type: 'subscribe', inbox_ids: ['support-agent@machinemail.dev'], event_types: ['message.received'] }));
    pinger = setInterval(() => ws.send(JSON.stringify({ type: 'ping' })), 30_000);
    await backfill(); // events live on the socket and in the backfill are deduplicated by event_id
  };
  ws.onmessage = (m) => handle(JSON.parse(String(m.data)));
  ws.onclose = () => {
    clearInterval(pinger);
    setTimeout(connect, delay + Math.random() * 500);
    delay = Math.min(delay * 2, 60_000);
  };
}
connect();

GET /v0/events items have the same event_id, event_type, timestamp and data as stream events, without the top-level copies of the data keys, so read from data in your handler.

Python consumer

ws.py
import asyncio, json, os, random
import requests, websockets

KEY = os.environ["MACHINEMAIL_API_KEY"]
seen, last_event_id = set(), None


def handle(event):
    global last_event_id
    if event["event_id"] in seen:
        return
    seen.add(event["event_id"])
    last_event_id = event["event_id"]
    print(event["event_type"], (event["data"].get("message") or {}).get("message_id"))


def backfill():
    if not last_event_id:
        return
    params = {"after": last_event_id, "event_types": "message.received", "limit": 100}
    while True:
        page = requests.get("https://api.machinemail.dev/v0/events", params=params, headers={"Authorization": f"Bearer {KEY}"}).json()
        for e in page["events"]:
            handle(e)
        if not page.get("next_page_token"):
            return
        params["page_token"] = page["next_page_token"]


async def run():
    delay = 1
    while True:
        try:
            async with websockets.connect(
                "wss://api.machinemail.dev/v0/ws",
                additional_headers={"Authorization": f"Bearer {KEY}"},
                ping_interval=None,
            ) as ws:
                delay = 1
                await ws.send(json.dumps({"type": "subscribe", "inbox_ids": ["support-agent@machinemail.dev"], "event_types": ["message.received"]}))
                await asyncio.to_thread(backfill)

                async def keepalive():
                    while True:
                        await asyncio.sleep(30)
                        await ws.send(json.dumps({"type": "ping"}))

                pinger = asyncio.create_task(keepalive())
                try:
                    async for raw in ws:
                        msg = json.loads(raw)
                        if msg.get("type") == "event":
                            handle(msg)
                finally:
                    pinger.cancel()
        except Exception as err:
            print("socket dropped:", err)
        await asyncio.sleep(delay + random.random() / 2)
        delay = min(delay * 2, 60)


asyncio.run(run())

additional_headers is the argument name in websockets 14 and later; older versions call it extra_headers.

Backfill and deduplication rules

  • Store the last event_id you processed, durably if missing an event matters.
  • Event IDs sort in creation order, so GET /v0/events?after=<event_id> returns exactly what came later, oldest first, 50 per page by default and up to 500 with limit.
  • after also accepts an ISO 8601 timestamp, for example when you start a consumer for the first time.
  • Always deduplicate by event_id. Backfill and the live stream overlap by design.
  • The event log also contains event types that are never streamed or sent to webhooks, such as webhook.test. Pass event_types when you backfill.
EventsWebhook eventsVerify webhooksGive an agent an inbox

On this page

AuthenticationServer-Sent EventsFiltersResuming with Last-Event-IDTypeScript consumerPython consumerWebSocketTypeScript consumerPython consumerBackfill and deduplication rulesRelated
0% read