> ## 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.

# 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.

```typescript theme={null}
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:

<ResponseField name="state" type="Channel">
  State updates (`state:update`)
</ResponseField>

<ResponseField name="rpc" type="Channel">
  RPC method calls (`rpc`, `rpc:error`)
</ResponseField>

<ResponseField name="message" type="Channel">
  WebSocket messages, tool calls (`message:*`, `tool:*`)
</ResponseField>

<ResponseField name="schedule" type="Channel">
  Scheduled tasks and queues (`schedule:*`, `queue:*`)
</ResponseField>

<ResponseField name="lifecycle" type="Channel">
  Connection lifecycle (`connect`, `disconnect`, `destroy`)
</ResponseField>

<ResponseField name="workflow" type="Channel">
  Workflow events (`workflow:*`)
</ResponseField>

<ResponseField name="mcp" type="Channel">
  MCP operations (`mcp:*`)
</ResponseField>

<ResponseField name="email" type="Channel">
  Email routing (`email:*`)
</ResponseField>

## subscribe()

Subscribe to a typed observability channel.

<ParamField path="channelKey" type="keyof ChannelEventMap" required>
  Channel name ("rpc", "state", "lifecycle", etc.)
</ParamField>

<ParamField path="callback" type="(event: ChannelEventMap[K]) => void" required>
  Callback to handle events
</ParamField>

```typescript theme={null}
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.

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

**Payload:** (none)

### RPC Events

#### rpc

Emitted when an RPC method is called.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

```typescript theme={null}
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.

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

### Email Events

#### email:receive

Emitted when an email is received.

```typescript theme={null}
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.

```typescript theme={null}
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:

```typescript theme={null}
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`:

```typescript theme={null}
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:

```typescript theme={null}
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

```typescript theme={null}
// ✅ Good - subscribe before any operations
const unsubscribe = subscribe("rpc", handleRpcEvent);

// ... use Agent ...

// Clean up
unsubscribe();
```

### Filter Events

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

### Aggregate Metrics

```typescript theme={null}
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

```typescript theme={null}
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](/api/agent-class) - Observability property
* [Node.js diagnostics\_channel](https://nodejs.org/api/diagnostics_channel.html) - Underlying API