Events & streaming — the live wire

Events are how mails.ai tells you something happened — a send delivered, a reply arrived, a draft sent. There are three ways to consume them: list them on demand, stream them live over SSE, or have them pushed to a webhook. This guide covers the first two.

List events

GET /v1/events (scope read) returns typed events, newest first. Filter by event_type, agent_id, or since (an ISO 8601 time or a date), and page with limit + cursor:

curl "https://api.mails.ai/v1/events?event_type=message.received&limit=50" \
  -H "Authorization: Bearer $MAILS_API_KEY"

Each event carries the signal computed for it:

{
  "id": "evt_01J9X2K7Q8...",
  "type": "message.received",
  "workspace_id": "wsk_01J9...",
  "agent_id": "agt_01J9...",
  "source_message_id": "rcv_01J9...",
  "intent": "schedule_demo",
  "urgency": 0.8,
  "injection_score": 0.02,
  "sender_reputation": 0.91,
  "entities": { "date": "2026-05-14", "time": "10:00" },
  "test_mode": false,
  "data": { "...": "type-specific payload" },
  "created_at": "2026-06-19T18:02:44.000Z"
}

Stream events (SSE)

GET /v1/events/stream (scope read) is a Server-Sent Events live tail. Pass ?since=<event_id> (or a Last-Event-ID header) to resume from where you left off, and ?event_types= with a comma-separated list to filter. The connection sends a : connected comment, then events as they arrive with keepalive comments between, and self-closes after about 50 seconds — your client reconnects with the last id it saw.

The frame format is standard SSE:

id: evt_01J9X2K7Q8...
event: message.received
data: {"id":"evt_01J9...","type":"message.received","agent_id":"agt_01J9...","data":{...}}

Because auth is a bearer header, use a streaming HTTP client (not the browser EventSource, which can’t set headers). A reconnect-safe consumer in Node:

let lastId: string | undefined;

async function stream() {
  const url = new URL("https://api.mails.ai/v1/events/stream");
  url.searchParams.set("event_types", "message.received,reply.received");

  const res = await fetch(url, {
    headers: {
      Authorization: `Bearer ${process.env.MAILS_API_KEY}`,
      ...(lastId ? { "Last-Event-ID": lastId } : {}),
    },
  });

  const reader = res.body!.pipeThrough(new TextDecoderStream()).getReader();
  let buffer = "";
  for (;;) {
    const { value, done } = await reader.read();
    if (done) break; // server closed at ~50s — reconnect below
    buffer += value;
    const frames = buffer.split("\n\n");
    buffer = frames.pop() ?? "";
    for (const frame of frames) {
      const id = frame.match(/^id: (.+)$/m)?.[1];
      const data = frame.match(/^data: (.+)$/m)?.[1];
      if (id) lastId = id;
      if (data) handleEvent(JSON.parse(data));
    }
  }
}

// Reconnect forever, resuming from lastId.
for (;;) {
  try { await stream(); } catch (e) { /* log */ }
  await new Promise((r) => setTimeout(r, 1000));
}

Event types

The events mails.ai emits, grouped by what they describe:

• Send lifecycle: message.sent, message.scheduled, message.delivered, message.delayed, message.bounced, message.complained, message.failed.
• Inbound: message.received for first contact, reply.received when the inbound matches one of your sends, and message.received.unauthenticated when the sender looks forged: no DKIM signature aligned with its domain verifies, and that domain’s DMARC policy is quarantine or reject (treat that sender with suspicion). message.received.over_limit reports mail received while the workspace was over a limit.
• Drafts: draft.scheduled, draft.sending, draft.sent, draft.failed.
• Threads and suppression: thread.created, thread.updated, thread.deleted, suppression.added.
• Diagnostics: webhook.test (fired by the webhook test endpoint).

The inbound payload

message.received and reply.received are the two you build on — same payload shape, so one handler serves both. The data carries the sender, the parsed content, the authentication results, and the security signal mails.ai computes on every inbound within your plan’s limits:

{
  "agent_id": "agt_01J9...",
  "agent_email": "hello@acme.mails.ai",
  "thread_id": "thrd_01J9...",
  "source_message_id": "rcv_01J9...",
  "is_thread_reply": false,
  "from": { "address": "user@example.com", "name": "Jordan Lee" },
  "subject": "Demo next week?",
  "injection_score": 0.02,
  "sender_reputation": 0.91,
  "spf_pass": null,
  "dkim_pass": true,
  "intent": "schedule_demo",
  "entities": { "date": "2026-05-14", "time": "10:00" },
  "urgency": 0.8,
  "body_text_excerpt": "Could we see a demo Tuesday at 10am?",
  "raw_url": "/v1/messages/rcv_01J9.../raw"
}

agent_email is the agent’s canonical sender identity. The message itself arrived on the agent’s inbound address, <name>.<slug>@in.mails.ai (here hello.acme@in.mails.ai) — that inbound form, not the sender identity, is the address to give anyone who needs to email the agent.

injection_score and sender_reputation are present on every inbound within your plan’s limits. The intent, entities, and urgency fields are read from the message only when it is first contact and the receiving agent has classify_inbound enabled; otherwise intent is a fixed value (general_reply for a reply in a thread, unknown or unclassified) and entities is empty. A robust handler reads the signal before acting on the body — for example, hold a high-injection-score message for a person instead of following its instructions:

function handleEvent(evt) {
  // reply.received fires when the inbound matches one of your sends; message.received
  // is first contact. Handle BOTH or you silently ignore every reply.
  if (evt.type !== "message.received" && evt.type !== "reply.received") return;
  const d = evt.data;
  // Hold for a person: quarantined, not scanned (no score), or 0.5 and up
  if (d.quarantined || typeof d.injection_score !== "number" || d.injection_score >= 0.5) {
    return flagForReview(evt);
  }
  if (d.intent === "schedule_demo") return scheduleDemo(d.entities);
  return llm.handle(d.body_text_excerpt);
}