Agents Sdk

Cloudflare Agents SDK

Your knowledge of the Agents SDK may be outdated. Prefer retrieval over pre-training for any Agents SDK task.

Retrieval Sources

Cloudflare docs: https://developers.cloudflare.com/agents/

Capabilities

The Agents SDK provides:

• Persistent state — SQLite-backed, auto-synced to clients via setState
• Callable RPC — @callable() methods invoked over WebSocket
• Scheduling — One-time, recurring (scheduleEvery), and cron tasks
• Workflows — Durable multi-step background processing via AgentWorkflow
• Durable execution — runFiber() / stash() for work that survives DO eviction
• Queue — Built-in FIFO queue with retries via queue()
• Retries — this.retry() with exponential backoff and jitter
• MCP integration — Connect to MCP servers or build your own with McpAgent
• Email handling — Receive and reply to emails with secure routing
• Streaming chat — AIChatAgent with resumable streams, message persistence, tools
• Server-driven messages — saveMessages, waitUntilStable for proactive agent turns
• React hooks — useAgent, useAgentChat for client apps
• Observability — diagnostics_channel events for state, RPC, schedule, lifecycle
• Push notifications — Web Push + VAPID delivery from agents
• Webhooks — Receive and verify external webhooks
• Voice (experimental) — STT/TTS via @cloudflare/voice
• Browser tools (experimental) — CDP-powered browsing via agents/browser
• Think (experimental) — Higher-level chat agent via @cloudflare/think

FIRST: Verify Installation

npm ls agents  # Should show agents package

If not installed:

npm install agents

For chat agents:

npm install agents @cloudflare/ai-chat ai @ai-sdk/react

Wrangler Configuration

{
  "compatibility_flags": ["nodejs_compat"],
  "durable_objects": {
    "bindings": [{ "name": "MyAgent", "class_name": "MyAgent" }]
  },
  "migrations": [{ "tag": "v1", "new_sqlite_classes": ["MyAgent"] }]
}

Gotchas:

• Do NOT enable experimentalDecorators in tsconfig (breaks @callable)
• Never edit old migrations — always add new tags
• Each agent class needs its own DO binding + migration entry
• Add "ai": { "binding": "AI" } for Workers AI

Agent Class

import { Agent, routeAgentRequest, callable } from "agents";

type State = { count: number };

export class Counter extends Agent<Env, State> {
  initialState = { count: 0 };

  validateStateChange(nextState: State, source: Connection | "server") {
    if (nextState.count < 0) throw new Error("Count cannot be negative");
  }

  onStateUpdate(state: State, source: Connection | "server") {
    console.log("State updated:", state);
  }

  @callable()
  increment() {
    this.setState({ count: this.state.count + 1 });
    return this.state.count;
  }
}

export default {
  fetch: (req, env) => routeAgentRequest(req, env) ?? new Response("Not found", { status: 404 })
};

Routing

Requests route to /agents/{agent-name}/{instance-name}:

Client: useAgent({ agent: "Counter", name: "user-123" })

Custom routing: use getAgentByName(env.MyAgent, "instance-id") then agent.fetch(request).

Core APIs

React Client

import { useAgent } from "agents/react";

function App() {
  const [state, setLocalState] = useState({ count: 0 });

  const agent = useAgent({
    agent: "Counter",
    name: "my-instance",
    onStateUpdate: (newState) => setLocalState(newState),
    onIdentity: (name, agentType) => console.log(`Connected to ${name}`)
  });

  return (
    <button onClick={() => agent.setState({ count: state.count + 1 })}>
      Count: {state.count}
    </button>
  );
}

References

Core

• [references/state-scheduling.md](references/state-scheduling.md) — State persistence, scheduling, SQL
• [references/callable.md](references/callable.md) — RPC methods, streaming, timeouts
• [references/routing.md](references/routing.md) — URL patterns, custom routing, getAgentByName
• [references/configuration.md](references/configuration.md) — Wrangler config, bindings, Vite setup

Chat & Streaming

• [references/streaming-chat.md](references/streaming-chat.md) — AIChatAgent, resumable streams, tools
• [references/client-sdk.md](references/client-sdk.md) — useAgent, useAgentChat, AgentClient
• [references/server-driven-messages.md](references/server-driven-messages.md) — Trigger patterns, saveMessages
• [references/human-in-the-loop.md](references/human-in-the-loop.md) — Approval flows, needsApproval

Background Processing

• [references/workflows.md](references/workflows.md) — Durable Workflows integration
• [references/durable-execution.md](references/durable-execution.md) — runFiber, stash, surviving eviction
• [references/queue-retries.md](references/queue-retries.md) — Built-in queue, retry with backoff

Integrations

• [references/mcp.md](references/mcp.md) — MCP client and server, transports, securing
• [references/email.md](references/email.md) — Email routing and handling
• [references/webhooks-push.md](references/webhooks-push.md) — Webhooks, push notifications
• [references/observability.md](references/observability.md) — Diagnostics-channel events

Experimental

• [references/think.md](references/think.md) — @cloudflare/think higher-level chat agent
• [references/voice.md](references/voice.md) — @cloudflare/voice STT/TTS
• [references/codemode.md](references/codemode.md) — Code Mode for tool orchestration
• [references/browse-the-web.md](references/browse-the-web.md) — CDP browser tools

Browse the Web (Experimental)

Fetch https://developers.cloudflare.com/agents/api-reference/browse-the-web/ for complete documentation.

CDP-powered browser tools that let agents scrape, screenshot, and interact with web pages.

Setup

// wrangler.jsonc
{
  "browser": { "binding": "BROWSER" },
  "worker_loaders": [{ "binding": "LOADER" }],
  "compatibility_flags": ["nodejs_compat"]
}

Usage with AI SDK

import { createBrowserTools } from "agents/browser/ai";

export class MyAgent extends AIChatAgent<Env> {
  async onChatMessage(onFinish) {
    const browserTools = createBrowserTools({
      browser: this.env.BROWSER,
      loader: this.env.LOADER
    });

    const result = streamText({
      model: openai("gpt-4o"),
      messages: await convertToModelMessages(this.messages),
      tools: { ...myTools, ...browserTools },
      onFinish
    });
    return result.toUIMessageStreamResponse();
  }
}

Available Tools

The LLM writes async JavaScript IIFEs that run in a fresh browser session.

When to Use

• Need a real browser (JS rendering, screenshots, interaction) → browser tools
• Just need HTML/API data → use fetch() instead (faster, cheaper)

Low-Level API

import { connectBrowser, CdpSession } from "agents/browser";

const browser = await connectBrowser(this.env.BROWSER);
const cdp = new CdpSession(browser);
await cdp.send("Page.navigate", { url: "https://example.com" });

Callable Methods

Fetch https://developers.cloudflare.com/agents/api-reference/callable-methods/ for complete documentation.

Overview

@callable() exposes agent methods to clients via WebSocket RPC.

import { Agent, callable } from "agents";

export class MyAgent extends Agent<Env, State> {
  @callable()
  async greet(name: string): Promise<string> {
    return `Hello, ${name}!`;
  }

  @callable()
  async processData(data: unknown): Promise<Result> {
    // Long-running work
    return result;
  }
}

Client Usage

// Basic call
const greeting = await agent.call("greet", ["World"]);

// With timeout
const result = await agent.call("processData", [data], {
  timeout: 5000  // 5 second timeout
});

Streaming Responses

import { Agent, callable, StreamingResponse } from "agents";

export class MyAgent extends Agent<Env, State> {
  @callable({ streaming: true })
  async streamResults(stream: StreamingResponse, query: string) {
    for await (const item of fetchResults(query)) {
      stream.send(JSON.stringify(item));
    }
    stream.close();
  }

  @callable({ streaming: true })
  async streamWithError(stream: StreamingResponse) {
    try {
      // ... work
    } catch (error) {
      stream.error(error.message);  // Signal error to client
      return;
    }
    stream.close();
  }
}

Client with streaming:

await agent.call("streamResults", ["search term"], {
  stream: {
    onChunk: (data) => console.log("Chunk:", data),
    onDone: () => console.log("Complete"),
    onError: (error) => console.error("Error:", error)
  }
});

Introspection

// Get list of callable methods on an agent
const methods = await agent.call("getCallableMethods", []);
// Returns: ["greet", "processData", "streamResults", ...]

When to Use

Client SDK

Fetch https://developers.cloudflare.com/agents/api-reference/client-sdk/ for complete documentation.

React: useAgent

import { useAgent } from "agents/react";

function App() {
  const [state, setState] = useState({ count: 0 });

  const agent = useAgent({
    agent: "Counter",
    name: "my-instance",
    onStateUpdate: (newState) => setState(newState),
    onIdentity: (name, agentType) => console.log(`Connected to ${name}`)
  });

  return <button onClick={() => agent.setState({ count: state.count + 1 })}>
    {state.count}
  </button>;
}

Typed RPC via stub

const agent = useAgent<typeof MyAgent>({
  agent: "MyAgent",
  name: "default"
});

const result = await agent.stub.myMethod(arg1, arg2);

Auth via Query Params

useAgent({
  agent: "MyAgent",
  name: "default",
  query: async () => `token=${await getToken()}`,
  queryDeps: [tokenVersion]
});

React: useAgentChat

import { useAgent } from "agents/react";
import { useAgentChat } from "@cloudflare/ai-chat/react";

function Chat() {
  const agent = useAgent({ agent: "ChatAgent", name: "session-1" });

  const { messages, input, handleInputChange, handleSubmit, status } =
    useAgentChat({ agent });

  return (
    <div>
      {messages.map((m) => <div key={m.id}>{m.role}: {m.content}</div>)}
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
      </form>
    </div>
  );
}

Vanilla JS: AgentClient

import { AgentClient } from "agents/client";

const client = new AgentClient({
  agent: "MyAgent",
  name: "default",
  host: "https://my-worker.workers.dev"
});

client.addEventListener("stateUpdate", (e) => console.log(e.state));
const result = await client.call("myMethod", [arg]);
client.close();

agentFetch for HTTP-only

import { agentFetch } from "agents/client";

const response = await agentFetch({
  agent: "MyAgent",
  name: "default",
  host: "https://my-worker.workers.dev",
  path: "/api/data"
});

Streaming RPC

await agent.call("streamResults", ["query"], {
  stream: {
    onChunk: (data) => console.log(data),
    onDone: () => console.log("done"),
    onError: (err) => console.error(err)
  }
});

Codemode (Experimental)

Fetch https://developers.cloudflare.com/agents/api-reference/codemode/ for complete documentation.

Codemode lets LLMs write and execute code that orchestrates your tools, instead of calling them one at a time. The LLM gets a single "write code" tool; generated JavaScript runs in an isolated Worker sandbox.

When to Use

Setup

Wrangler Config

{
  "worker_loaders": [{ "binding": "LOADER" }],
  "compatibility_flags": ["nodejs_compat"]
}

Install

npm install @cloudflare/codemode ai zod

Usage

import { createCodeTool } from "@cloudflare/codemode/ai";
import { DynamicWorkerExecutor } from "@cloudflare/codemode";
import { streamText, tool, convertToModelMessages } from "ai";
import { z } from "zod";

const tools = {
  getWeather: tool({
    description: "Get weather for a location",
    inputSchema: z.object({ location: z.string() }),
    execute: async ({ location }) => `Weather: ${location} 72°F`
  }),
  sendEmail: tool({
    description: "Send an email",
    inputSchema: z.object({ to: z.string(), subject: z.string(), body: z.string() }),
    execute: async ({ to, subject, body }) => `Email sent to ${to}`
  })
};

export class MyAgent extends Agent<Env, State> {
  async onChatMessage() {
    const executor = new DynamicWorkerExecutor({
      loader: this.env.LOADER
    });

    const codemode = createCodeTool({ tools, executor });

    const result = streamText({
      model,
      system: "You are a helpful assistant.",
      messages: await convertToModelMessages(this.messages),
      tools: { codemode }
    });

    return result.toUIMessageStreamResponse();
  }
}

With MCP Tools

const codemode = createCodeTool({
  tools: {
    ...myTools,
    ...this.mcp.getAITools()
  },
  executor
});

How It Works

1. createCodeTool generates TypeScript type definitions from your tools 2. The LLM writes an async arrow function calling codemode.toolName(args) 3. Code runs in an isolated Worker sandbox via DynamicWorkerExecutor 4. Tool calls route back to the host via Workers RPC 5. External fetch() is blocked by default — sandbox can only call your tools

Network Isolation

const executor = new DynamicWorkerExecutor({
  loader: env.LOADER,
  globalOutbound: null           // default — fully isolated
  // globalOutbound: env.MY_SERVICE  // route through a Fetcher
});

Limitations

• Experimental — API may change
• needsApproval tools execute immediately in sandbox (no approval pause yet)
• JavaScript execution only
• Requires worker_loaders binding

Configuration

Fetch https://developers.cloudflare.com/agents/api-reference/configuration/ for complete documentation.

Wrangler Config (wrangler.jsonc)

{
  "name": "my-agent",
  "main": "src/index.ts",
  "compatibility_date": "2025-01-28",
  "compatibility_flags": ["nodejs_compat"],
  "durable_objects": {
    "bindings": [
      { "name": "MyAgent", "class_name": "MyAgent" },
      { "name": "ChatAgent", "class_name": "ChatAgent" }
    ]
  },
  "migrations": [
    { "tag": "v1", "new_sqlite_classes": ["MyAgent", "ChatAgent"] }
  ],
  "ai": { "binding": "AI" },
  "assets": {
    "directory": "./dist/client",
    "binding": "ASSETS",
    "not_found_handling": "single-page-application",
    "run_worker_first": true
  }
}

Key Rules

• Every agent class needs a DO binding AND a new_sqlite_classes migration entry
• nodejs_compat is required
• Never edit old migrations — add a new tag (e.g. v2) for new classes
• Do NOT enable experimentalDecorators in tsconfig — it breaks @callable
• For Workers AI locally, set "ai": { "binding": "AI", "remote": true } in .dev.vars or config
• Use wrangler secret put for secrets, never hardcode them

Vite Setup

import { defineConfig } from "vite";
import react from "@vitejs/plugin-react";
import { cloudflare } from "@cloudflare/vite-plugin";
import { agents } from "agents/vite";

export default defineConfig({
  plugins: [react(), cloudflare(), agents()]
});

Type Generation

npx wrangler types

This generates env.d.ts with typed bindings. Regenerate after changing wrangler.jsonc.

tsconfig

Extend the agents tsconfig for correct settings:

{
  "extends": ["agents/tsconfig"],
  "include": ["src/**/*.ts", "src/**/*.tsx"],
  "compilerOptions": { "paths": { "~/*": ["./src/*"] } }
}

Durable Execution

Fetch https://developers.cloudflare.com/agents/api-reference/durable-execution/ for complete documentation.

Fibers let agent work survive Durable Object eviction. Progress is checkpointed to SQLite; on recovery, you decide what to do.

runFiber

export class MyAgent extends Agent<Env, State> {
  async onRequest(request: Request) {
    await this.runFiber("process-data", async (ctx) => {
      const step1 = await fetchData();
      ctx.stash({ step: 1, data: step1 });

      const step2 = await transform(step1);
      ctx.stash({ step: 2, result: step2 });

      this.setState({ result: step2 });
    });
    return new Response("Started");
  }

  async onFiberRecovered(ctx) {
    const checkpoint = ctx.stash;
    if (checkpoint.step === 1) {
      const step2 = await transform(checkpoint.data);
      this.setState({ result: step2 });
    }
  }
}

Key APIs

Important

• stash replaces the entire checkpoint — not a merge
• The lambda is NOT restored on recovery — only the stash data is. You must re-derive what to do in onFiberRecovered
• No auto-retry on throw — handle errors yourself
• For long-running pipelines with automatic retries, use Workflows instead
• Filter concurrent fibers by ctx.name in onFiberRecovered

Email Handling

Fetch https://developers.cloudflare.com/agents/api-reference/email/ for complete documentation.

Overview

Agents receive and reply to emails via Cloudflare Email Routing.

Wrangler Configuration

{
  "durable_objects": {
    "bindings": [{ "name": "EmailAgent", "class_name": "EmailAgent" }]
  },
  "migrations": [{ "tag": "v1", "new_sqlite_classes": ["EmailAgent"] }],
  "send_email": [
    { "name": "SEB", "destination_address": "reply@yourdomain.com" }
  ]
}

Basic Email Handler

import { Agent } from "agents";
import { type AgentEmail } from "agents/email";
import PostalMime from "postal-mime";

export class EmailAgent extends Agent<Env, State> {
  async onEmail(email: AgentEmail) {
    const raw = await email.getRaw();
    const parsed = await PostalMime.parse(raw);

    console.log("From:", email.from);
    console.log("Subject:", parsed.subject);

    await this.replyToEmail(email, {
      fromName: "My Agent",
      subject: `Re: ${parsed.subject}`,
      body: "Thanks for your email!"
    });
  }
}

Routing Emails

import { routeAgentRequest, routeAgentEmail } from "agents";
import { createAddressBasedEmailResolver } from "agents/email";

export default {
  async email(message, env) {
    await routeAgentEmail(message, env, {
      resolver: createAddressBasedEmailResolver("EmailAgent")
    });
  },

  async fetch(request, env) {
    return routeAgentRequest(request, env) ?? new Response("Not found", { status: 404 });
  }
};

Resolvers

Address-Based (Inbound Mail)

Routes based on recipient address:

import { createAddressBasedEmailResolver } from "agents/email";

const resolver = createAddressBasedEmailResolver("EmailAgent");
// support@example.com → EmailAgent, instance "support"
// NotificationAgent+user123@example.com → NotificationAgent, instance "user123"

Secure Reply (Reply Flows)

Verifies replies are authentic using HMAC-SHA256 signatures:

import { createSecureReplyEmailResolver } from "agents/email";

const resolver = createSecureReplyEmailResolver(env.EMAIL_SECRET, {
  maxAge: 7 * 24 * 60 * 60, // 7 days (default: 30 days)
  onInvalidSignature: (email, reason) => {
    console.warn(`Invalid signature from ${email.from}: ${reason}`);
  }
});

Sign outbound emails to enable secure reply routing:

await this.replyToEmail(email, {
  fromName: "My Agent",
  body: "Thanks!",
  secret: this.env.EMAIL_SECRET  // Signs headers for secure reply routing
});

Catch-All (Single Instance)

Routes all emails to one agent instance:

import { createCatchAllEmailResolver } from "agents/email";

const resolver = createCatchAllEmailResolver("EmailAgent", "default");

Combining Resolvers

async email(message, env) {
  const secureReply = createSecureReplyEmailResolver(env.EMAIL_SECRET);
  const addressBased = createAddressBasedEmailResolver("EmailAgent");

  await routeAgentEmail(message, env, {
    resolver: async (email, env) => {
      // Try secure reply first
      const result = await secureReply(email, env);
      if (result) return result;
      // Fall back to address-based
      return addressBased(email, env);
    }
  });
}

Utilities

import { isAutoReplyEmail } from "agents/email";

async onEmail(email: AgentEmail) {
  if (isAutoReplyEmail(email.headers)) {
    // Skip auto-replies (vacation, out-of-office, etc.)
    return;
  }
  // Process email...
}

Human-in-the-Loop

Fetch https://developers.cloudflare.com/agents/concepts/human-in-the-loop/ for complete documentation.

Multiple patterns for adding human approval to agent actions.

Decision Guide

Workflow Approvals

// In AgentWorkflow:
const approved = await step.waitForEvent<{ approved: boolean }>("approval", {
  timeout: "7d"
});
if (!approved.approved) throw new Error("Rejected");

// From agent:
await this.approveWorkflow(workflowId);
await this.rejectWorkflow(workflowId);

Chat Tool Approvals (needsApproval)

const tools = {
  deleteItem: tool({
    description: "Delete an item",
    parameters: z.object({ id: z.string() }),
    execute: async ({ id }) => { /* delete */ },
    needsApproval: true  // or a function: (toolCall) => boolean
  })
};

Client handles approval:

const { addToolApprovalResponse, addToolOutput } = useAgentChat({
  agent,
  onToolCall: async ({ toolCall }) => {
    if (confirm(`Allow ${toolCall.toolName}?`)) {
      return { approve: true };
    }
    return { approve: false };
  }
});

To deny with a custom message:

addToolOutput(toolCallId, "output-error", "User rejected this action");

Important

• waitForApproval may return undefined on timeout — handle it
• addToolOutput with output-error does NOT auto-continue the LLM — you may need sendMessage after
• For OpenAI Agents SDK, use needsApproval on the tool definition (same pattern)

MCP Integration

Fetch https://developers.cloudflare.com/agents/api-reference/mcp-client-api/ and https://developers.cloudflare.com/agents/api-reference/mcp-agent-api/ for complete documentation.

Agents include a multi-server MCP client for connecting to external MCP servers, and McpAgent for building MCP servers.

Add an MCP Server

import { Agent, callable } from "agents";

export class MyAgent extends Agent<Env, State> {
  @callable()
  async addServer(name: string, url: string) {
    // Options-based API (recommended)
    const result = await this.addMcpServer(name, url, {
      callbackHost: "https://my-worker.workers.dev",
      transport: { headers: { Authorization: "Bearer ..." } }
    });

    if (result.state === "authenticating") {
      // OAuth required - redirect user to result.authUrl
      return { needsAuth: true, authUrl: result.authUrl };
    }

    return { ready: true, id: result.id };
  }
}

Use MCP Tools

async onChatMessage() {
  // Get AI-compatible tools from all connected MCP servers
  const mcpTools = this.mcp.getAITools();
  
  const allTools = {
    ...localTools,
    ...mcpTools
  };

  const result = streamText({
    model: openai("gpt-4o"),
    messages: await convertToModelMessages(this.messages),
    tools: allTools
  });
  
  return result.toUIMessageStreamResponse();
}

List MCP Resources

// List all registered servers
const servers = this.mcp.listServers();

// List tools from all servers
const tools = this.mcp.listTools();

// List resources
const resources = this.mcp.listResources();

// List prompts
const prompts = this.mcp.listPrompts();

Remove Server

await this.removeMcpServer(serverId);

Building an MCP Server

Use McpAgent from the SDK to create an MCP server.

Install dependencies:

npm install @modelcontextprotocol/sdk zod

Wrangler config:

{
  "durable_objects": {
    "bindings": [{ "name": "MyMCP", "class_name": "MyMCP" }]
  },
  "migrations": [{ "tag": "v1", "new_sqlite_classes": ["MyMCP"] }]
}

Server implementation:

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { McpAgent } from "agents/mcp";
import { z } from "zod";

type State = { counter: number };

export class MyMCP extends McpAgent<Env, State, {}> {
  server = new McpServer({
    name: "MyMCPServer",
    version: "1.0.0"
  });

  initialState = { counter: 0 };

  async init() {
    // Register a resource
    this.server.resource("counter", "mcp://resource/counter", (uri) => ({
      contents: [{ text: String(this.state.counter), uri: uri.href }]
    }));

    // Register a tool
    this.server.registerTool(
      "increment",
      {
        description: "Increment the counter",
        inputSchema: { amount: z.number().default(1) }
      },
      async ({ amount }) => {
        this.setState({ counter: this.state.counter + amount });
        return {
          content: [{ text: `Counter: ${this.state.counter}`, type: "text" }]
        };
      }
    );
  }
}

Serve MCP Server

export default {
  fetch(request: Request, env: Env, ctx: ExecutionContext) {
    const url = new URL(request.url);

    // Streamable HTTP transport (recommended)
    if (url.pathname.startsWith("/mcp")) {
      return MyMCP.serve("/mcp", { binding: "MyMCP" }).fetch(request, env, ctx);
    }

    // SSE transport (legacy, deprecated)
    if (url.pathname.startsWith("/sse")) {
      return MyMCP.serveSSE("/sse", { binding: "MyMCP" }).fetch(request, env, ctx);
    }

    return new Response("Not found", { status: 404 });
  }
};

Transports

Fetch https://developers.cloudflare.com/agents/api-reference/mcp-transports/ for complete documentation.

RPC Transport (Same Worker)

async onStart() {
  await this.addMcpServer("internal-tools", this.env.MyMCPBinding, {
    props: { userId: this.name }
  });
}

Retry on MCP Connections

await this.addMcpServer("tools", url, {
  retry: { maxAttempts: 3, baseDelayMs: 500 }
});

Securing MCP Servers

Fetch https://developers.cloudflare.com/agents/api-reference/securing-mcp-servers/ for complete documentation.

Use @cloudflare/workers-oauth-provider to add OAuth in front of your MCP server. See the securing docs for proxy patterns and redirect_uri validation.

Observability

Fetch https://developers.cloudflare.com/agents/api-reference/observability/ for complete documentation.

Agents emit structured events via Node.js diagnostics_channel. Subscribe in development or forward via Tail Workers in production.

Subscribe to Events

import { subscribe } from "agents/observability";

subscribe("agents:rpc", (event) => {
  console.log(`RPC call: ${event.payload.method}`);
});

subscribe("agents:state", (event) => {
  console.log(`State change on ${event.agent}`);
});

Available Channels

Per-Agent Override

export class MyAgent extends Agent<Env, State> {
  observability = undefined; // disable for this agent
}

Production: Tail Workers

In production, events appear as diagnosticsChannelEvents on the Tail Worker event object. Attach a Tail Worker to your agent's Worker to forward events to your observability platform.

Queue & Retries

Fetch https://developers.cloudflare.com/agents/api-reference/queue-tasks/ and https://developers.cloudflare.com/agents/api-reference/retries/ for complete documentation.

Built-in Queue

FIFO queue persisted in SQLite. Sequential processing, one item at a time.

export class MyAgent extends Agent<Env, State> {
  async onRequest(request: Request) {
    this.queue("processItem", { id: "abc", data: "..." });
    this.queue("processItem", { id: "def", data: "..." }, { retry: { maxAttempts: 5 } });
    return new Response("Queued");
  }

  async processItem(payload: { id: string; data: string }, queueItem: QueueItem) {
    await doWork(payload);
  }
}

Queue Management

const items = this.getQueue();
const byCallback = this.getQueues("processItem");
this.dequeue(itemId);
this.dequeueAll();
this.dequeueAllByCallback("processItem");

Retries

Exponential backoff with full jitter. Defaults: 3 attempts, 100ms base, 3000ms max.

const result = await this.retry(
  async () => {
    const res = await fetch("https://api.example.com/data");
    if (!res.ok) throw new Error(`HTTP ${res.status}`);
    return res.json();
  },
  {
    maxAttempts: 5,
    baseDelayMs: 200,
    maxDelayMs: 5000,
    shouldRetry: (err, nextAttempt) => {
      if (err.message.includes("429")) return true;
      if (err.message.includes("401")) return false;
      return nextAttempt <= 3;
    }
  }
);

Retry on Schedules and Queue

await this.schedule(60, "task", payload, { retry: { maxAttempts: 3 } });
await this.scheduleEvery(30, "poll", undefined, { retry: { maxAttempts: 2 } });
this.queue("handler", payload, { retry: { maxAttempts: 5 } });

Class-level Defaults

export class MyAgent extends Agent<Env, State> {
  static options = {
    retry: { maxAttempts: 5, baseDelayMs: 200, maxDelayMs: 10000 }
  };
}

Important

• shouldRetry only works on this.retry() — not on schedule/queue (callbacks aren't serializable)
• Queue retries block head-of-line; long delays keep the DO awake — use schedule for long waits instead
• No dead-letter queue — failed items are removed after retries exhausted

Routing

Fetch https://developers.cloudflare.com/agents/api-reference/routing/ for complete documentation.

Default URL Pattern

/agents/{kebab-class-name}/{instance-name}

import { routeAgentRequest } from "agents";

export default {
  fetch: (req, env) =>
    routeAgentRequest(req, env) ?? new Response("Not found", { status: 404 })
};

Subpaths after the instance name (e.g. /agents/my-agent/default/api/data) route to onRequest.

Custom Routing with getAgentByName

import { getAgentByName } from "agents";

export default {
  async fetch(req, env) {
    const url = new URL(req.url);
    if (url.pathname.startsWith("/api/")) {
      const agent = getAgentByName(env.MyAgent, "singleton");
      return agent.fetch(req);
    }
    return routeAgentRequest(req, env);
  }
};

Options

routeAgentRequest(req, env, {
  cors: true,
  prefix: "/api/agents",
  locationHint: "enam",
  jurisdiction: "eu",
  props: { userId: "123" },
  onBeforeConnect: async (req) => { /* auth check */ },
  onBeforeRequest: async (req) => { /* auth check */ }
});

props are delivered to onStart(props) on first access.

Client Side

useAgent({
  agent: "MyAgent",
  name: "instance-1",
  host: "https://my-worker.workers.dev",
  basePath: "/api/agents",
  path: "/custom-subpath"
});

Common Mistakes

• Class name MyAgent becomes kebab my-agent in URLs — match exactly
• "Namespace not found" error = the class_name in wrangler doesn't match your exported class
• If sendIdentityOnConnect: false, the ready promise on the client may never resolve — use state sync instead

Server-Driven Messages (Trigger Patterns)

Fetch https://developers.cloudflare.com/agents/api-reference/trigger-patterns/ for complete documentation.

Patterns for server-initiated LLM turns in AIChatAgent — from schedules, webhooks, email, or other agents.

saveMessages — Trigger an LLM Turn

await this.saveMessages((existingMessages) => [
  ...existingMessages,
  { role: "user", content: "Check for new notifications" }
]);

saveMessages persists the messages AND triggers onChatMessage.

persistMessages — Save Without Triggering

await this.persistMessages([
  ...this.messages,
  { role: "assistant", content: "System note: checked at " + new Date() }
]);

waitUntilStable

Always call before `saveMessages` from non-chat contexts (schedules, webhooks, email):

async checkNotifications(payload: unknown, schedule: Schedule) {
  await this.waitUntilStable({ timeout: 30_000 });
  await this.saveMessages((msgs) => [
    ...msgs,
    { role: "user", content: "Run scheduled notification check" }
  ]);
}

onChatResponse

Runs after each LLM turn completes. Use for chaining:

async onChatResponse(result: ChatResponseResult) {
  if (result.type === "finish" && needsFollowUp(result)) {
    await this.saveMessages((msgs) => [
      ...msgs,
      { role: "user", content: "Continue with next step" }
    ]);
  }
}

Client Status

const { isStreaming, isServerStreaming } = useAgentChat({ agent });

• isStreaming — true during any streaming (user-initiated or server-initiated)
• isServerStreaming — true only during server-initiated streams

State & Scheduling

Fetch https://developers.cloudflare.com/agents/api-reference/store-and-sync-state/ and https://developers.cloudflare.com/agents/api-reference/schedule-tasks/ for complete documentation.

State Management

State persists to SQLite and broadcasts to connected clients automatically.

Define Typed State

type State = { 
  count: number;
  items: string[];
};

export class MyAgent extends Agent<Env, State> {
  initialState: State = { count: 0, items: [] };
}

Read and Update

// Read (lazy-loaded from SQLite)
const count = this.state.count;

// Write (sync, persists, broadcasts)
this.setState({ count: this.state.count + 1 });

Validation Hook

validateStateChange() runs synchronously before state persists. Throw to reject the update.

validateStateChange(nextState: State, source: Connection | "server") {
  if (nextState.count < 0) {
    throw new Error("Count cannot be negative");
  }
}

Execution Order

1. validateStateChange(nextState, source) - sync, gating 2. State persisted to SQLite 3. State broadcast to connected clients 4. onStateUpdate(nextState, source) - async via ctx.waitUntil, non-gating

Client-Side Sync (React)

import { useAgent } from "agents/react";

function App() {
  const [state, setLocalState] = useState<State>({ count: 0 });
  
  const agent = useAgent<State>({
    agent: "MyAgent",
    name: "instance-1",
    onStateUpdate: (newState) => setLocalState(newState)
  });

  return <button onClick={() => agent.setState({ count: state.count + 1 })}>
    Count: {state.count}
  </button>;
}

SQL API

Direct SQLite access for custom queries:

// Create table
this.sql`
  CREATE TABLE IF NOT EXISTS items (
    id TEXT PRIMARY KEY,
    name TEXT,
    created_at INTEGER DEFAULT (unixepoch())
  )
`;

// Insert
this.sql`INSERT INTO items (id, name) VALUES (${id}, ${name})`;

// Query with types
const items = this.sql<{ id: string; name: string }>`
  SELECT * FROM items WHERE name LIKE ${`%${search}%`}
`;

Scheduling

Schedule Types

Examples

// Delay (seconds)
await this.schedule(60, "checkStatus", { id: "abc123" });

// Specific date
await this.schedule(new Date("2025-12-25T00:00:00Z"), "sendGreeting", { to: "user" });

// Cron (recurring)
await this.schedule("0 9 * * 1-5", "weekdayReport", {});

// Fixed interval (every 30 seconds, overlap prevention built-in)
await this.scheduleEvery(30, "pollUpdates");
await this.scheduleEvery(300, "syncData", { source: "api" });

Handler

async sendGreeting(payload: { to: string }, schedule: Schedule) {
  console.log(`Sending greeting to ${payload.to}`);
  // Cron schedules auto-reschedule; one-time schedules are deleted
}

Manage Schedules

const schedules = this.getSchedules();
const crons = this.getSchedules({ type: "cron" });
await this.cancelSchedule(schedule.id);

Retry on Schedules

await this.schedule(60, "task", payload, { retry: { maxAttempts: 3 } });
await this.scheduleEvery(30, "poll", undefined, { retry: { maxAttempts: 2 } });

Lifecycle Callbacks

export class MyAgent extends Agent<Env, State> {
  async onStart() {
    // Agent started or woke from hibernation
  }

  onConnect(conn: Connection, ctx: ConnectionContext) {
    // WebSocket connected
  }

  onMessage(conn: Connection, message: WSMessage) {
    // WebSocket message (non-RPC)
  }

  onStateUpdate(state: State, source: Connection | "server") {
    // State changed (async, non-blocking)
  }

  onError(error: unknown) {
    // Error handler
    throw error; // Re-throw to propagate
  }
}

Streaming Chat with AIChatAgent

Fetch https://developers.cloudflare.com/agents/api-reference/chat-agents/ for complete documentation.

AIChatAgent from @cloudflare/ai-chat provides streaming chat with automatic message persistence and resumable streams.

Basic Chat Agent

import { AIChatAgent } from "@cloudflare/ai-chat";
import { streamText, convertToModelMessages } from "ai";
import { openai } from "@ai-sdk/openai";

export class Chat extends AIChatAgent<Env> {
  async onChatMessage(onFinish, options) {
    const result = streamText({
      model: openai("gpt-4o"),
      system: "You are a helpful assistant.",
      messages: await convertToModelMessages(this.messages),
      abortSignal: options?.abortSignal,
      onFinish
    });
    return result.toUIMessageStreamResponse();
  }
}

Important: Always pass abortSignal and onFinish — they enable proper cleanup and message persistence.

With Tools

import { tool } from "ai";
import { z } from "zod";

const tools = {
  getWeather: tool({
    description: "Get weather for a location",
    parameters: z.object({ location: z.string() }),
    execute: async ({ location }) => `Weather in ${location}: 72°F, sunny`
  })
};

export class Chat extends AIChatAgent<Env> {
  async onChatMessage(onFinish, options) {
    const result = streamText({
      model: openai("gpt-4o"),
      messages: await convertToModelMessages(this.messages),
      tools,
      abortSignal: options?.abortSignal,
      onFinish
    });
    return result.toUIMessageStreamResponse();
  }
}

With Workers AI (no API keys)

import { createWorkersAI } from "workers-ai-provider";

export class Chat extends AIChatAgent<Env> {
  async onChatMessage(onFinish, options) {
    const workersai = createWorkersAI({ binding: this.env.AI });
    const result = streamText({
      model: workersai("@cf/meta/llama-4-scout-17b-16e-instruct"),
      messages: await convertToModelMessages(this.messages),
      abortSignal: options?.abortSignal,
      onFinish
    });
    return result.toUIMessageStreamResponse();
  }
}

Custom UI Message Stream

For more control, use createUIMessageStream:

import { createUIMessageStream, createUIMessageStreamResponse } from "ai";

export class Chat extends AIChatAgent<Env> {
  async onChatMessage(onFinish) {
    const stream = createUIMessageStream({
      execute: async ({ writer }) => {
        const result = streamText({
          model: openai("gpt-4o"),
          messages: await convertToModelMessages(this.messages),
          onFinish
        });
        writer.merge(result.toUIMessageStream());
      }
    });
    return createUIMessageStreamResponse({ stream });
  }
}

Resumable Streaming

Streams automatically resume if client disconnects and reconnects:

1. Chunks buffered to SQLite during streaming 2. On reconnect, buffered chunks sent immediately 3. Live streaming continues from where it left off

Enabled by default. To disable:

const { messages } = useAgentChat({ agent, resume: false });

React Client

import { useAgent } from "agents/react";
import { useAgentChat } from "@cloudflare/ai-chat/react";

function ChatUI() {
  const agent = useAgent({
    agent: "Chat",
    name: "my-chat-session"
  });

  const { 
    messages, 
    input, 
    handleInputChange, 
    handleSubmit, 
    status 
  } = useAgentChat({ agent });

  return (
    <div>
      {messages.map((m) => (
        <div key={m.id}>
          <strong>{m.role}:</strong> {m.content}
        </div>
      ))}
      
      <form onSubmit={handleSubmit}>
        <input 
          value={input} 
          onChange={handleInputChange}
          disabled={status === "streaming"}
        />
        <button type="submit">Send</button>
      </form>
    </div>
  );
}

Streaming RPC Methods

For non-chat streaming, use @callable({ streaming: true }):

import { Agent, callable, StreamingResponse } from "agents";

export class MyAgent extends Agent<Env> {
  @callable({ streaming: true })
  async streamData(stream: StreamingResponse, query: string) {
    for (let i = 0; i < 10; i++) {
      stream.send(`Result ${i}: ${query}`);
      await sleep(100);
    }
    stream.close();
  }
}

Client receives streamed messages via WebSocket RPC.

Key Properties

Status Values

useAgentChat status:

Also: isStreaming, isServerStreaming for distinguishing user vs server-initiated streams.

Think (Experimental)

Fetch https://developers.cloudflare.com/agents/api-reference/think/ for complete documentation.

@cloudflare/think — a higher-level chat agent class that handles the streamText loop, tool execution, and message persistence for you. You provide getModel() and getSystemPrompt(); Think handles the rest.

npm install @cloudflare/think

Minimal Agent

import { Think } from "@cloudflare/think";
import { createWorkersAI } from "workers-ai-provider";
import { routeAgentRequest } from "agents";

export class MyAgent extends Think<Env> {
  getModel() {
    return createWorkersAI({ binding: this.env.AI })("@cf/meta/llama-4-scout-17b-16e-instruct");
  }

  getSystemPrompt() {
    return "You are a helpful assistant.";
  }
}

export default {
  fetch: (req, env) => routeAgentRequest(req, env)
};

Wrangler Config

{
  "compatibility_flags": ["nodejs_compat", "experimental"],
  "durable_objects": {
    "bindings": [{ "name": "MyAgent", "class_name": "MyAgent" }]
  },
  "migrations": [{ "tag": "v1", "new_sqlite_classes": ["MyAgent"] }],
  "ai": { "binding": "AI" }
}

Note: Think requires the experimental compatibility flag.

Custom Tools

import { tool } from "ai";
import { z } from "zod";

export class MyAgent extends Think<Env> {
  getTools() {
    return {
      getWeather: tool({
        description: "Get weather",
        parameters: z.object({ city: z.string() }),
        execute: async ({ city }) => `72°F in ${city}`
      })
    };
  }
}

Lifecycle Hooks

async beforeTurn(ctx: TurnContext): Promise<TurnConfig> {
  if (ctx.continuation) {
    return { model: cheaperModel };
  }
  return {};
}

Sub-Agents

const child = this.subAgent(SpecialistAgent, "specialist-1");
await child.chat("Analyze this data...", (chunk) => {
  // stream callback
});

Client

Same React hooks as AIChatAgent:

const agent = useAgent({ agent: "MyAgent", name: "session-1" });
const { messages, input, handleInputChange, handleSubmit } = useAgentChat({ agent });

Think vs AIChatAgent

Voice (Experimental)

Fetch https://developers.cloudflare.com/agents/api-reference/voice/ for complete documentation.

@cloudflare/voice — real-time speech-to-text and text-to-speech for agents. Audio streams over WebSocket.

npm install @cloudflare/voice

Server

import { Agent } from "agents";
import { withVoice, WorkersAITTS, WorkersAINova3STT } from "@cloudflare/voice";

export class VoiceAgent extends withVoice(Agent)<Env> {
  transcriber = new WorkersAINova3STT(this);
  tts = new WorkersAITTS(this);

  async onTurn(transcript: string, context: VoiceTurnContext) {
    const result = streamText({
      model: createWorkersAI({ binding: this.env.AI })("@cf/meta/llama-4-scout-17b-16e-instruct"),
      messages: [
        { role: "system", content: "You are a voice assistant." },
        ...context.conversationHistory,
        { role: "user", content: transcript }
      ]
    });

    for await (const chunk of result.textStream) {
      if (context.signal.aborted) break;
      context.speak(chunk);
    }
  }
}

Lifecycle Hooks

Client (React)

import { useVoiceAgent } from "@cloudflare/voice/react";

function VoiceUI() {
  const { isConnected, isSpeaking, connect, disconnect } = useVoiceAgent({
    agent: "VoiceAgent",
    name: "session-1"
  });

  return <button onClick={isConnected ? disconnect : connect}>
    {isConnected ? "End Call" : "Start Call"}
  </button>;
}

STT/TTS Providers

Workers AI (default), Deepgram, ElevenLabs — install the provider package and swap the transcriber/tts properties.

Webhooks & Push Notifications

Webhooks

Fetch https://developers.cloudflare.com/agents/api-reference/webhooks/ for complete documentation.

Route external webhooks to agent instances via onRequest:

export default {
  async fetch(req: Request, env: Env) {
    const url = new URL(req.url);
    if (url.pathname.startsWith("/webhooks/")) {
      const entityId = url.pathname.split("/")[2];
      const agent = getAgentByName(env.MyAgent, entityId);
      return agent.fetch(req);
    }
    return routeAgentRequest(req, env);
  }
};

In the agent:

export class MyAgent extends Agent<Env, State> {
  async onRequest(request: Request) {
    const signature = request.headers.get("X-Signature");
    if (!verifySignature(signature, await request.text(), this.env.WEBHOOK_SECRET)) {
      return new Response("Unauthorized", { status: 401 });
    }
    const payload = JSON.parse(await request.text());
    this.queue("processWebhook", payload);
    return new Response("OK", { status: 202 });
  }
}

Tips: Respond quickly (200/202), verify signatures, deduplicate with stored event IDs, use queue() for async processing.

Push Notifications

Fetch https://developers.cloudflare.com/agents/api-reference/push-notifications/ for complete documentation.

Web Push via VAPID from agents. Store subscriptions in agent state, send via web-push.

npm install web-push

import webpush from "web-push";

export class NotifyAgent extends Agent<Env, State> {
  @callable()
  async subscribe(subscription: PushSubscription) {
    this.setState({
      ...this.state,
      subscriptions: [...this.state.subscriptions, subscription]
    });
  }

  async sendReminder(payload: { message: string }, schedule: Schedule) {
    for (const sub of this.state.subscriptions) {
      try {
        await webpush.sendNotification(sub, JSON.stringify({
          title: "Reminder",
          body: payload.message
        }), {
          vapidDetails: {
            subject: "mailto:you@example.com",
            publicKey: this.env.VAPID_PUBLIC_KEY,
            privateKey: this.env.VAPID_PRIVATE_KEY
          }
        });
      } catch (err) {
        if (err.statusCode === 404 || err.statusCode === 410) {
          // Remove expired subscription
        }
      }
    }
  }
}

VAPID keys: generate with npx web-push generate-vapid-keys, store as secrets.

Workflows Integration

Fetch https://developers.cloudflare.com/agents/api-reference/run-workflows/ for complete documentation.

Overview

Agents handle real-time communication; Workflows handle durable execution. Together they enable:

• Long-running background tasks with automatic retries
• Human-in-the-loop approval flows
• Multi-step pipelines that survive failures

AgentWorkflow Base Class

import { AgentWorkflow } from "agents/workflows";
import type { AgentWorkflowEvent, AgentWorkflowStep } from "agents/workflows";

type TaskParams = { taskId: string; data: string };

export class ProcessingWorkflow extends AgentWorkflow<MyAgent, TaskParams> {
  async run(event: AgentWorkflowEvent<TaskParams>, step: AgentWorkflowStep) {
    const params = event.payload;

    // Durable step - retries on failure
    const result = await step.do("process", async () => {
      return processData(params.data);
    });

    // Non-durable: progress reporting
    await this.reportProgress({ step: "process", percent: 0.5 });

    // Non-durable: broadcast to connected clients
    this.broadcastToClients({ type: "update", taskId: params.taskId });

    // Durable: merge state via step
    await step.mergeAgentState({ lastProcessed: params.taskId });

    // Durable: report completion
    await step.reportComplete(result);

    return result;
  }
}

Wrangler Configuration

{
  "workflows": [
    { "name": "processing-workflow", "binding": "PROCESSING_WORKFLOW", "class_name": "ProcessingWorkflow" }
  ],
  "durable_objects": {
    "bindings": [{ "name": "MyAgent", "class_name": "MyAgent" }]
  },
  "migrations": [{ "tag": "v1", "new_sqlite_classes": ["MyAgent"] }]
}

Agent Methods for Workflows

// Start a workflow
const instance = await this.runWorkflow("ProcessingWorkflow", { taskId: "123", data: "..." });

// Send event to waiting workflow
await this.sendWorkflowEvent("ProcessingWorkflow", workflowId, { type: "approve" });

// Query workflows
const workflow = await this.getWorkflow(workflowId);
const workflows = await this.getWorkflows({ status: "running" });

// Control workflows
await this.approveWorkflow(workflowId);
await this.rejectWorkflow(workflowId);
await this.terminateWorkflow(workflowId);
await this.pauseWorkflow(workflowId);
await this.resumeWorkflow(workflowId);

// Delete workflows
await this.deleteWorkflow(workflowId);
await this.deleteWorkflows({ status: "complete", before: new Date(...) });

Lifecycle Callbacks

export class MyAgent extends Agent<Env, State> {
  async onWorkflowProgress(workflowName: string, workflowId: string, progress: unknown) {
    // Workflow reported progress via this.reportProgress()
    this.broadcast({ type: "progress", workflowId, progress });
  }

  async onWorkflowComplete(workflowName: string, workflowId: string, result?: unknown) {
    // Workflow finished successfully
  }

  async onWorkflowError(workflowName: string, workflowId: string, error: Error) {
    // Workflow failed
  }

  async onWorkflowEvent(workflowName: string, workflowId: string, event: unknown) {
    // Workflow received an event via sendWorkflowEvent()
  }
}

Human-in-the-Loop

// In workflow: wait for approval
const approved = await step.waitForEvent<{ approved: boolean }>("approval", {
  timeout: "7d"
});

if (!approved.approved) {
  throw new Error("Rejected");
}

// From agent: approve or reject
await this.approveWorkflow(workflowId);  // Sends { approved: true }
await this.rejectWorkflow(workflowId);   // Sends { approved: false }