Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/cloudflare/agents/llms.txt

Use this file to discover all available pages before exploring further.

​
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);
  }
});

Build docs developers (and LLMs) love

Get started for freeTalk to us