Observability

Emit and subscribe to observability events from Agents

Overview

Agents emit observability events for state changes, RPC calls, connections, schedules, workflows, MCP operations, and emails. Events are published to diagnostic channels and can be consumed via subscribers or Tail Workers.

import { subscribe } from "agents/observability";

// Subscribe to RPC events
const unsubscribe = subscribe("rpc", (event) => {
  console.log(`RPC call: ${event.payload.method}`);
});

Event Channels

Events are published to named diagnostic channels:

state · Channel

State updates (state:update)

rpc · Channel

RPC method calls (rpc, rpc:error)

message · Channel

WebSocket messages, tool calls (message:*, tool:*)

schedule · Channel

Scheduled tasks and queues (schedule:*, queue:*)

lifecycle · Channel

Connection lifecycle (connect, disconnect, destroy)

workflow · Channel

Workflow events (workflow:*)

mcp · Channel

MCP operations (mcp:*)

email · Channel

Email routing (email:*)

subscribe()

Subscribe to a typed observability channel.

channelKey · keyof ChannelEventMap · required

Channel name (“rpc”, “state”, “lifecycle”, etc.)

callback · (event: ChannelEventMap[K]) => void · required

Callback to handle events

import { subscribe } from "agents/observability";

const unsubscribe = subscribe("rpc", (event) => {
  console.log(`RPC: ${event.payload.method}`);
  // event.payload is fully typed!
});

// Later: clean up
unsubscribe();

Returns: () => void - Function to unsubscribe

Event Types

State Events

state:update

Emitted when Agent state changes.

subscribe("state", (event) => {
  if (event.type === "state:update") {
    console.log("State updated");
  }
});

Payload: (none)

RPC Events

rpc

Emitted when an RPC method is called.

subscribe("rpc", (event) => {
  if (event.type === "rpc") {
    console.log(`Called: ${event.payload.method}`);
    console.log(`Streaming: ${event.payload.streaming}`);
  }
});

Payload:

• method: string - Method name
• streaming?: boolean - Whether the method is streaming

rpc:error

Emitted when an RPC call fails.

subscribe("rpc", (event) => {
  if (event.type === "rpc:error") {
    console.error(`RPC error in ${event.payload.method}: ${event.payload.error}`);
  }
});

Payload:

• method: string - Method name
• error: string - Error message

Lifecycle Events

connect

Emitted when a WebSocket connection is established.

subscribe("lifecycle", (event) => {
  if (event.type === "connect") {
    console.log(`Connection: ${event.payload.connectionId}`);
  }
});

Payload:

• connectionId: string - Connection ID

disconnect

Emitted when a WebSocket connection closes.

subscribe("lifecycle", (event) => {
  if (event.type === "disconnect") {
    console.log(`Disconnect: ${event.payload.connectionId}`);
    console.log(`Code: ${event.payload.code}, Reason: ${event.payload.reason}`);
  }
});

Payload:

• connectionId: string - Connection ID
• code: number - Close code
• reason: string - Close reason

Schedule Events

schedule:execute

Emitted when a scheduled task executes.

subscribe("schedule", (event) => {
  if (event.type === "schedule:execute") {
    console.log(`Executed: ${event.payload.callback}`);
  }
});

Payload:

• callback: string - Callback name
• scheduleId: string - Schedule ID
• type: string - Schedule type (“cron”, “delayed”, etc.)

queue:execute

Emitted when a queued task executes.

subscribe("schedule", (event) => {
  if (event.type === "queue:execute") {
    console.log(`Queue: ${event.payload.callback}`);
  }
});

Payload:

• callback: string - Callback name

Workflow Events

workflow:start

Emitted when a workflow is started.

subscribe("workflow", (event) => {
  if (event.type === "workflow:start") {
    console.log(`Started: ${event.payload.workflowName}`);
    console.log(`Instance: ${event.payload.instanceId}`);
  }
});

Payload:

• workflowName: string - Workflow binding name
• instanceId: string - Workflow instance ID

workflow:progress

Emitted when a workflow reports progress.

subscribe("workflow", (event) => {
  if (event.type === "workflow:progress") {
    console.log(`Progress: ${event.payload.workflowId}`);
  }
});

Payload:

• workflowId: string - Workflow instance ID
• progress: unknown - Progress data

workflow:complete

Emitted when a workflow completes.

subscribe("workflow", (event) => {
  if (event.type === "workflow:complete") {
    console.log(`Complete: ${event.payload.workflowId}`);
  }
});

Payload:

• workflowId: string - Workflow instance ID
• result: unknown - Workflow result

workflow:error

Emitted when a workflow errors.

subscribe("workflow", (event) => {
  if (event.type === "workflow:error") {
    console.error(`Error: ${event.payload.workflowId}`);
    console.error(event.payload.error);
  }
});

Payload:

• workflowId: string - Workflow instance ID
• error: string - Error message

MCP Events

mcp:client:connect

Emitted when connecting to an MCP server.

subscribe("mcp", (event) => {
  if (event.type === "mcp:client:connect") {
    console.log(`Connecting to: ${event.payload.url}`);
    console.log(`Transport: ${event.payload.transport}`);
    console.log(`State: ${event.payload.state}`);
  }
});

Payload:

• url: string - Server URL
• transport: string - Transport type
• state: string - Connection state
• error?: string - Error message (if failed)

mcp:client:discover

Emitted when discovering MCP server capabilities.

subscribe("mcp", (event) => {
  if (event.type === "mcp:client:discover") {
    console.log("Discovering MCP server...");
  }
});

Email Events

email:receive

Emitted when an email is received.

subscribe("email", (event) => {
  if (event.type === "email:receive") {
    console.log(`Email from: ${event.payload.from}`);
    console.log(`To: ${event.payload.to}`);
    console.log(`Subject: ${event.payload.subject}`);
  }
});

Payload:

• from: string - Sender address
• to: string - Recipient address
• subject?: string - Email subject

email:reply

Emitted when a reply is sent.

subscribe("email", (event) => {
  if (event.type === "email:reply") {
    console.log(`Reply from: ${event.payload.from}`);
    console.log(`To: ${event.payload.to}`);
  }
});

Payload:

• from: string - Sender address
• to: string - Recipient address
• subject?: string - Reply subject

Custom Observability

Override observability on the Agent to use a custom implementation:

import type { Observability, ObservabilityEvent } from "agents/observability";

class LoggingObservability implements Observability {
  emit(event: ObservabilityEvent): void {
    console.log(`[${event.type}]`, event.payload);
  }
}

class MyAgent extends Agent {
  observability = new LoggingObservability();
}

Tail Workers

In production, events are automatically forwarded to Tail Workers via event.diagnosticsChannelEvents:

export default {
  async tail(events: TraceItem[]) {
    for (const event of events) {
      if (event.diagnosticsChannelEvents) {
        for (const dcEvent of event.diagnosticsChannelEvents) {
          console.log(`[${dcEvent.channel}]`, dcEvent.message);
        }
      }
    }
  }
};

Event Structure

All events have the same base structure:

type ObservabilityEvent = {
  type: string;           // Event type (e.g., "rpc", "state:update")
  agent: string;          // Agent class name
  name: string;           // Agent instance name
  payload: unknown;       // Event-specific data
  timestamp: number;      // Unix timestamp (ms)
};

Best Practices

Subscribe Early

// ✅ Good - subscribe before any operations
const unsubscribe = subscribe("rpc", handleRpcEvent);

// ... use Agent ...

// Clean up
unsubscribe();

Filter Events

subscribe("rpc", (event) => {
  // Filter by method name
  if (event.type === "rpc" && event.payload.method === "increment") {
    console.log("Increment called");
  }
});

Aggregate Metrics

const metrics = {
  rpcCalls: 0,
  rpcErrors: 0,
  connections: 0
};

subscribe("rpc", (event) => {
  if (event.type === "rpc") metrics.rpcCalls++;
  if (event.type === "rpc:error") metrics.rpcErrors++;
});

subscribe("lifecycle", (event) => {
  if (event.type === "connect") metrics.connections++;
});

setInterval(() => {
  console.log("Metrics:", metrics);
}, 60000);

Use Type Narrowing

subscribe("rpc", (event) => {
  if (event.type === "rpc") {
    // event.payload.method is typed!
    console.log(event.payload.method);
  } else if (event.type === "rpc:error") {
    // event.payload.error is typed!
    console.error(event.payload.error);
  }
});

Related

• Agent Class - Observability property
• Node.js diagnostics_channel - Underlying API