From ac6ab726653e0fc4bcef07de8363c552fb9c50cb Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Tue, 25 Nov 2025 16:50:23 +0000 Subject: [PATCH 1/2] Add resumable streaming documentation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Synced from cloudflare/agents PR #673: added resumable streaming with minimal setup This documentation covers the automatic resumable streaming feature in AIChatAgent that allows chat streams to seamlessly resume after disconnections. Source PR: https://github.com/cloudflare/agents/pull/673 🤖 Generated with Claude Code Co-Authored-By: Claude --- .../agents/guides/resumable-streaming.mdx | 102 ++++++++++++++++++ 1 file changed, 102 insertions(+) create mode 100644 src/content/docs/agents/guides/resumable-streaming.mdx diff --git a/src/content/docs/agents/guides/resumable-streaming.mdx b/src/content/docs/agents/guides/resumable-streaming.mdx new file mode 100644 index 00000000000..c99a781d05d --- /dev/null +++ b/src/content/docs/agents/guides/resumable-streaming.mdx @@ -0,0 +1,102 @@ +--- +pcx_content_type: concept +title: Resumable Streaming +sidebar: + order: 10 +--- + +import { TypeScriptExample, WranglerConfig } from "~/components"; + +The `AIChatAgent` class provides **automatic resumable streaming** out of the box. When a client disconnects and reconnects during an active stream, the response automatically resumes from where it left off. + +## How it works + +When you use `AIChatAgent` with `useAgentChat`: + +1. **During streaming**: All chunks are automatically persisted to SQLite +2. **On disconnect**: The stream continues server-side, buffering chunks +3. **On reconnect**: Client receives all buffered chunks and continues streaming + +It just works! + +## Example + +### Server + + +```ts +import { AIChatAgent } from "agents/ai-chat-agent"; +import { streamText } from "ai"; +import { openai } from "@ai-sdk/openai"; + +export class ChatAgent extends AIChatAgent { + async onChatMessage() { + const result = streamText({ + model: openai("gpt-4o"), + messages: this.messages + }); + + // Automatic resumable streaming - no extra code needed! + return result.toUIMessageStreamResponse(); + } +} +``` + + +### Client + + +```tsx +import { useAgent } from "agents/react"; +import { useAgentChat } from "agents/ai-react"; + +function Chat() { + const agent = useAgent({ + agent: "ChatAgent", + name: "my-chat" + }); + + const { messages, input, handleInputChange, handleSubmit, status } = + useAgentChat({ + agent + // resume: true is the default - streams automatically resume on reconnect + }); + + // ... render your chat UI +} +``` + + +## Under the hood + +### Server-side (AIChatAgent) + +- Creates SQLite tables for stream chunks and metadata on startup +- Each stream gets a unique ID and tracks chunk indices +- Chunks are buffered and flushed to SQLite every 100ms for performance +- On client connect, checks for active streams and sends `CF_AGENT_STREAM_RESUMING` +- Old completed streams are cleaned up after 24 hours + +### Client-side (useAgentChat) + +- Listens for `CF_AGENT_STREAM_RESUMING` notification +- Sends `CF_AGENT_STREAM_RESUME_ACK` when ready +- Receives all buffered chunks and reconstructs the message +- Continues receiving live chunks as they arrive + +## Disabling resume + +If you do not want automatic resume (for example, for short responses), disable it: + + +```tsx +const { messages } = useAgentChat({ + agent, + resume: false // Disable automatic stream resumption +}); +``` + + +## Try it + +Refer to the [resumable-stream-chat example](https://github.com/cloudflare/agents/tree/main/examples/resumable-stream-chat) for a complete working example. Start a long response, refresh the page mid-stream, and watch it resume automatically. From cb42c4d2eeec88d793ef82b161258e7d81f0758b Mon Sep 17 00:00:00 2001 From: katereznykova Date: Thu, 27 Nov 2025 19:28:35 +0000 Subject: [PATCH 2/2] adding resumable streaming to the docs --- .../docs/agents/api-reference/agents-api.mdx | 1133 +++++++++-------- .../agents/guides/resumable-streaming.mdx | 102 -- 2 files changed, 607 insertions(+), 628 deletions(-) delete mode 100644 src/content/docs/agents/guides/resumable-streaming.mdx diff --git a/src/content/docs/agents/api-reference/agents-api.mdx b/src/content/docs/agents/api-reference/agents-api.mdx index 8551fbd8097..b58194fcb79 100644 --- a/src/content/docs/agents/api-reference/agents-api.mdx +++ b/src/content/docs/agents/api-reference/agents-api.mdx @@ -5,14 +5,20 @@ sidebar: order: 1 --- -import { MetaInfo, Render, Type, TypeScriptExample, WranglerConfig } from "~/components"; +import { + MetaInfo, + Render, + Type, + TypeScriptExample, + WranglerConfig, +} from "~/components"; This page provides an overview of the Agent SDK API, including the `Agent` class, methods and properties built-in to the Agents SDK. The Agents SDK exposes two main APIs: -* The server-side `Agent` class. An Agent encapsulates all of the logic for an Agent, including how clients can connect to it, how it stores state, the methods it exposes, how to call AI models, and any error handling. -* The client-side `AgentClient` class, which allows you to connect to an Agent instance from a client-side application. The client APIs also include React hooks, including `useAgent` and `useAgentChat`, and allow you to automatically synchronize state between each unique Agent (running server-side) and your client applications. +- The server-side `Agent` class. An Agent encapsulates all of the logic for an Agent, including how clients can connect to it, how it stores state, the methods it exposes, how to call AI models, and any error handling. +- The client-side `AgentClient` class, which allows you to connect to an Agent instance from a client-side application. The client APIs also include React hooks, including `useAgent` and `useAgentChat`, and allow you to automatically synchronize state between each unique Agent (running server-side) and your client applications. :::note @@ -20,7 +26,6 @@ Agents require [Cloudflare Durable Objects](/durable-objects/), see [Configurati ::: - You can also find more specific usage examples for each API in the [Agents API Reference](/agents/api-reference/). @@ -68,61 +73,66 @@ interface Env { // complex AI workflows, schedule tasks, and interact with users and other // Agents. class MyAgent extends Agent { - // Optional initial state definition - initialState = { - counter: 0, - messages: [], - lastUpdated: null - }; - - // Called when a new Agent instance starts or wakes from hibernation - async onStart() { - console.log('Agent started with state:', this.state); - } - - // Handle HTTP requests coming to this Agent instance - // Returns a Response object - async onRequest(request: Request): Promise { - return new Response("Hello from Agent!"); - } - - // Called when a WebSocket connection is established - // Access the original request via ctx.request for auth etc. - async onConnect(connection: Connection, ctx: ConnectionContext) { - // Connections are automatically accepted by the SDK. - // You can also explicitly close a connection here with connection.close() - // Access the Request on ctx.request to inspect headers, cookies and the URL - } - - // Called for each message received on a WebSocket connection - // Message can be string, ArrayBuffer, or ArrayBufferView - async onMessage(connection: Connection, message: WSMessage) { - // Handle incoming messages - connection.send("Received your message"); - } - - // Handle WebSocket connection errors - async onError(connection: Connection, error: unknown): Promise { - console.error(`Connection error:`, error); - } - - // Handle WebSocket connection close events - async onClose(connection: Connection, code: number, reason: string, wasClean: boolean): Promise { - console.log(`Connection closed: ${code} - ${reason}`); - } - - // Called when the Agent's state is updated from any source - // source can be "server" or a client Connection - onStateUpdate(state: State, source: "server" | Connection) { - console.log("State updated:", state, "Source:", source); - } - - // You can define your own custom methods to be called by requests, - // WebSocket messages, or scheduled tasks - async customProcessingMethod(data: any) { - // Process data, update state, schedule tasks, etc. - this.setState({ ...this.state, lastUpdated: new Date() }); - } + // Optional initial state definition + initialState = { + counter: 0, + messages: [], + lastUpdated: null, + }; + + // Called when a new Agent instance starts or wakes from hibernation + async onStart() { + console.log("Agent started with state:", this.state); + } + + // Handle HTTP requests coming to this Agent instance + // Returns a Response object + async onRequest(request: Request): Promise { + return new Response("Hello from Agent!"); + } + + // Called when a WebSocket connection is established + // Access the original request via ctx.request for auth etc. + async onConnect(connection: Connection, ctx: ConnectionContext) { + // Connections are automatically accepted by the SDK. + // You can also explicitly close a connection here with connection.close() + // Access the Request on ctx.request to inspect headers, cookies and the URL + } + + // Called for each message received on a WebSocket connection + // Message can be string, ArrayBuffer, or ArrayBufferView + async onMessage(connection: Connection, message: WSMessage) { + // Handle incoming messages + connection.send("Received your message"); + } + + // Handle WebSocket connection errors + async onError(connection: Connection, error: unknown): Promise { + console.error(`Connection error:`, error); + } + + // Handle WebSocket connection close events + async onClose( + connection: Connection, + code: number, + reason: string, + wasClean: boolean, + ): Promise { + console.log(`Connection closed: ${code} - ${reason}`); + } + + // Called when the Agent's state is updated from any source + // source can be "server" or a client Connection + onStateUpdate(state: State, source: "server" | Connection) { + console.log("State updated:", state, "Source:", source); + } + + // You can define your own custom methods to be called by requests, + // WebSocket messages, or scheduled tasks + async customProcessingMethod(data: any) { + // Process data, update state, schedule tasks, etc. + this.setState({ ...this.state, lastUpdated: new Date() }); + } } ``` @@ -135,34 +145,34 @@ class MyAgent extends Agent { import { Agent } from "agents"; interface MyState { - counter: number; - lastUpdated: Date | null; + counter: number; + lastUpdated: Date | null; } class MyAgent extends Agent { - initialState = { - counter: 0, - lastUpdated: null - }; - - async onRequest(request: Request) { - if (request.method === "POST") { - await this.incrementCounter(); - return new Response(JSON.stringify(this.state), { - headers: { "Content-Type": "application/json" } - }); - } - return new Response(JSON.stringify(this.state), { - headers: { "Content-Type": "application/json" } - }); - } - - async incrementCounter() { - this.setState({ - counter: this.state.counter + 1, - lastUpdated: new Date() - }); - } + initialState = { + counter: 0, + lastUpdated: null, + }; + + async onRequest(request: Request) { + if (request.method === "POST") { + await this.incrementCounter(); + return new Response(JSON.stringify(this.state), { + headers: { "Content-Type": "application/json" }, + }); + } + return new Response(JSON.stringify(this.state), { + headers: { "Content-Type": "application/json" }, + }); + } + + async incrementCounter() { + this.setState({ + counter: this.state.counter + 1, + lastUpdated: new Date(), + }); + } } ``` @@ -179,24 +189,24 @@ Represents a WebSocket connection to an Agent. ```ts // WebSocket connection interface interface Connection { - // Unique ID for this connection - id: string; + // Unique ID for this connection + id: string; - // Client-specific state attached to this connection - state: State; + // Client-specific state attached to this connection + state: State; - // Update the connection's state - setState(state: State): void; + // Update the connection's state + setState(state: State): void; - // Accept an incoming WebSocket connection - accept(): void; + // Accept an incoming WebSocket connection + accept(): void; - // Close the WebSocket connection with optional code and reason - close(code?: number, reason?: string): void; + // Close the WebSocket connection with optional code and reason + close(code?: number, reason?: string): void; - // Send a message to the client - // Can be string, ArrayBuffer, or ArrayBufferView - send(message: string | ArrayBuffer | ArrayBufferView): void; + // Send a message to the client + // Can be string, ArrayBuffer, or ArrayBufferView + send(message: string | ArrayBuffer | ArrayBufferView): void; } ``` @@ -205,33 +215,35 @@ interface Connection { ```ts // Example of handling WebSocket messages export class YourAgent extends Agent { - async onMessage(connection: Connection, message: WSMessage) { - if (typeof message === 'string') { - try { - // Parse JSON message - const data = JSON.parse(message); - - if (data.type === 'update') { - // Update connection-specific state - connection.setState({ ...connection.state, lastActive: Date.now() }); - - // Update global Agent state - this.setState({ - ...this.state, - connections: this.state.connections + 1 - }); - - // Send response back to this client only - connection.send(JSON.stringify({ - type: 'updated', - status: 'success' - })); - } - } catch (e) { - connection.send(JSON.stringify({ error: 'Invalid message format' })); - } - } - } + async onMessage(connection: Connection, message: WSMessage) { + if (typeof message === "string") { + try { + // Parse JSON message + const data = JSON.parse(message); + + if (data.type === "update") { + // Update connection-specific state + connection.setState({ ...connection.state, lastActive: Date.now() }); + + // Update global Agent state + this.setState({ + ...this.state, + connections: this.state.connections + 1, + }); + + // Send response back to this client only + connection.send( + JSON.stringify({ + type: "updated", + status: "success", + }), + ); + } + } catch (e) { + connection.send(JSON.stringify({ error: "Invalid message format" })); + } + } + } } ``` @@ -253,8 +265,8 @@ Context information for a WebSocket connection. ```ts // Context available during WebSocket connection interface ConnectionContext { - // The original HTTP request that initiated the WebSocket connection - request: Request; + // The original HTTP request that initiated the WebSocket connection + request: Request; } ``` @@ -273,19 +285,19 @@ Methods and types for managing Agent state. ```ts // State management in the Agent class class Agent { - // Initial state that will be set if no state exists yet - initialState: State = {} as unknown as State; + // Initial state that will be set if no state exists yet + initialState: State = {} as unknown as State; - // Current state of the Agent, persisted across restarts - get state(): State; + // Current state of the Agent, persisted across restarts + get state(): State; - // Update the Agent's state - // Persists to storage and notifies all connected clients - setState(state: State): void; + // Update the Agent's state + // Persists to storage and notifies all connected clients + setState(state: State): void; - // Called when state is updated from any source - // Override to react to state changes - onStateUpdate(state: State, source: "server" | Connection): void; + // Called when state is updated from any source + // Override to react to state changes + onStateUpdate(state: State, source: "server" | Connection): void; } ``` @@ -294,12 +306,12 @@ class Agent { ```ts // Example of state management in an Agent interface ChatState { - messages: Array<{ sender: string; text: string; timestamp: number }>; - participants: string[]; - settings: { - allowAnonymous: boolean; - maxHistoryLength: number; - }; + messages: Array<{ sender: string; text: string; timestamp: number }>; + participants: string[]; + settings: { + allowAnonymous: boolean; + maxHistoryLength: number; + }; } interface Env { @@ -308,32 +320,34 @@ interface Env { // Inside your Agent class export class YourAgent extends Agent { - async addMessage(sender: string, text: string) { - // Update state with new message - this.setState({ - ...this.state, - messages: [ - ...this.state.messages, - { sender, text, timestamp: Date.now() } - ].slice(-this.state.settings.maxHistoryLength) // Maintain max history - }); - - // The onStateUpdate method will automatically be called - // and all connected clients will receive the update - } - - // Override onStateUpdate to add custom behavior when state changes - onStateUpdate(state: ChatState, source: "server" | Connection) { - console.log(`State updated by ${source === "server" ? "server" : "client"}`); - - // You could trigger additional actions based on state changes - if (state.messages.length > 0) { - const lastMessage = state.messages[state.messages.length - 1]; - if (lastMessage.text.includes('@everyone')) { - this.notifyAllParticipants(lastMessage); - } - } - } + async addMessage(sender: string, text: string) { + // Update state with new message + this.setState({ + ...this.state, + messages: [ + ...this.state.messages, + { sender, text, timestamp: Date.now() }, + ].slice(-this.state.settings.maxHistoryLength), // Maintain max history + }); + + // The onStateUpdate method will automatically be called + // and all connected clients will receive the update + } + + // Override onStateUpdate to add custom behavior when state changes + onStateUpdate(state: ChatState, source: "server" | Connection) { + console.log( + `State updated by ${source === "server" ? "server" : "client"}`, + ); + + // You could trigger additional actions based on state changes + if (state.messages.length > 0) { + const lastMessage = state.messages[state.messages.length - 1]; + if (lastMessage.text.includes("@everyone")) { + this.notifyAllParticipants(lastMessage); + } + } + } } ``` @@ -348,33 +362,33 @@ Schedule tasks to run at a specified time in the future. ```ts // Scheduling API for running tasks in the future class Agent { - // Schedule a task to run in the future - // when: seconds from now, specific Date, or cron expression - // callback: method name on the Agent to call - // payload: data to pass to the callback - // Returns a Schedule object with the task ID - async schedule( - when: Date | string | number, - callback: keyof this, - payload?: T - ): Promise>; - - // Get a scheduled task by ID - // Returns undefined if the task doesn't exist - async getSchedule(id: string): Promise | undefined>; - - // Get all scheduled tasks matching the criteria - // Returns an array of Schedule objects - getSchedules(criteria?: { - description?: string; - id?: string; - type?: "scheduled" | "delayed" | "cron"; - timeRange?: { start?: Date; end?: Date }; - }): Schedule[]; - - // Cancel a scheduled task by ID - // Returns true if the task was cancelled, false otherwise - async cancelSchedule(id: string): Promise; + // Schedule a task to run in the future + // when: seconds from now, specific Date, or cron expression + // callback: method name on the Agent to call + // payload: data to pass to the callback + // Returns a Schedule object with the task ID + async schedule( + when: Date | string | number, + callback: keyof this, + payload?: T, + ): Promise>; + + // Get a scheduled task by ID + // Returns undefined if the task doesn't exist + async getSchedule(id: string): Promise | undefined>; + + // Get all scheduled tasks matching the criteria + // Returns an array of Schedule objects + getSchedules(criteria?: { + description?: string; + id?: string; + type?: "scheduled" | "delayed" | "cron"; + timeRange?: { start?: Date; end?: Date }; + }): Schedule[]; + + // Cancel a scheduled task by ID + // Returns true if the task was cancelled, false otherwise + async cancelSchedule(id: string): Promise; } ``` @@ -383,44 +397,44 @@ class Agent { ```ts // Example of scheduling in an Agent interface ReminderData { - userId: string; - message: string; - channel: string; + userId: string; + message: string; + channel: string; } export class YourAgent extends Agent { - // Schedule a one-time reminder in 2 hours - async scheduleReminder(userId: string, message: string) { - const twoHoursFromNow = new Date(Date.now() + 2 * 60 * 60 * 1000); - - const schedule = await this.schedule( - twoHoursFromNow, - 'sendReminder', - { userId, message, channel: 'email' } - ); - - console.log(`Scheduled reminder with ID: ${schedule.id}`); - return schedule.id; - } - - // Schedule a recurring daily task using cron - async scheduleDailyReport() { - // Run at 08:00 AM every day - const schedule = await this.schedule( - '0 8 * * *', // Cron expression: minute hour day month weekday - 'generateDailyReport', - { reportType: 'daily-summary' } - ); - - console.log(`Scheduled daily report with ID: ${schedule.id}`); - return schedule.id; - } - - // Method that will be called when the scheduled task runs - async sendReminder(data: ReminderData) { - console.log(`Sending reminder to ${data.userId}: ${data.message}`); - // Add code to send the actual notification - } + // Schedule a one-time reminder in 2 hours + async scheduleReminder(userId: string, message: string) { + const twoHoursFromNow = new Date(Date.now() + 2 * 60 * 60 * 1000); + + const schedule = await this.schedule( + twoHoursFromNow, + "sendReminder", + { userId, message, channel: "email" }, + ); + + console.log(`Scheduled reminder with ID: ${schedule.id}`); + return schedule.id; + } + + // Schedule a recurring daily task using cron + async scheduleDailyReport() { + // Run at 08:00 AM every day + const schedule = await this.schedule( + "0 8 * * *", // Cron expression: minute hour day month weekday + "generateDailyReport", + { reportType: "daily-summary" }, + ); + + console.log(`Scheduled daily report with ID: ${schedule.id}`); + return schedule.id; + } + + // Method that will be called when the scheduled task runs + async sendReminder(data: ReminderData) { + console.log(`Sending reminder to ${data.userId}: ${data.message}`); + // Add code to send the actual notification + } } ``` @@ -433,35 +447,35 @@ Represents a scheduled task. ```ts // Represents a scheduled task type Schedule = { - // Unique identifier for the schedule - id: string; - // Name of the method to be called - callback: string; - // Data to be passed to the callback - payload: T; + // Unique identifier for the schedule + id: string; + // Name of the method to be called + callback: string; + // Data to be passed to the callback + payload: T; } & ( - | { - // One-time execution at a specific time - type: "scheduled"; - // Timestamp when the task should execute - time: number; - } - | { - // Delayed execution after a certain time - type: "delayed"; - // Timestamp when the task should execute - time: number; - // Number of seconds to delay execution - delayInSeconds: number; - } - | { - // Recurring execution based on cron expression - type: "cron"; - // Timestamp for the next execution - time: number; - // Cron expression defining the schedule - cron: string; - } + | { + // One-time execution at a specific time + type: "scheduled"; + // Timestamp when the task should execute + time: number; + } + | { + // Delayed execution after a certain time + type: "delayed"; + // Timestamp when the task should execute + time: number; + // Number of seconds to delay execution + delayInSeconds: number; + } + | { + // Recurring execution based on cron expression + type: "cron"; + // Timestamp for the next execution + time: number; + // Cron expression defining the schedule + cron: string; + } ); ``` @@ -469,32 +483,34 @@ type Schedule = { ```ts export class YourAgent extends Agent { - // Example of managing scheduled tasks - async viewAndManageSchedules() { - // Get all scheduled tasks - const allSchedules = this.getSchedules(); - console.log(`Total scheduled tasks: ${allSchedules.length}`); - - // Get tasks scheduled for a specific time range - const upcomingSchedules = this.getSchedules({ - timeRange: { - start: new Date(), - end: new Date(Date.now() + 24 * 60 * 60 * 1000) // Next 24 hours - } - }); - - // Get a specific task by ID - const taskId = "task-123"; - const specificTask = await this.getSchedule(taskId); - - if (specificTask) { - console.log(`Found task: ${specificTask.callback} at ${new Date(specificTask.time)}`); - - // Cancel a scheduled task - const cancelled = await this.cancelSchedule(taskId); - console.log(`Task cancelled: ${cancelled}`); - } - } + // Example of managing scheduled tasks + async viewAndManageSchedules() { + // Get all scheduled tasks + const allSchedules = this.getSchedules(); + console.log(`Total scheduled tasks: ${allSchedules.length}`); + + // Get tasks scheduled for a specific time range + const upcomingSchedules = this.getSchedules({ + timeRange: { + start: new Date(), + end: new Date(Date.now() + 24 * 60 * 60 * 1000), // Next 24 hours + }, + }); + + // Get a specific task by ID + const taskId = "task-123"; + const specificTask = await this.getSchedule(taskId); + + if (specificTask) { + console.log( + `Found task: ${specificTask.callback} at ${new Date(specificTask.time)}`, + ); + + // Cancel a scheduled task + const cancelled = await this.cancelSchedule(taskId); + console.log(`Task cancelled: ${cancelled}`); + } + } } ``` @@ -511,12 +527,12 @@ Execute SQL queries against the Agent's built-in SQLite database using the `this ```ts // SQL query API for the Agent's embedded database class Agent { - // Execute a SQL query with tagged template literals - // Returns an array of rows matching the query - sql>( - strings: TemplateStringsArray, - ...values: (string | number | boolean | null)[] - ): T[]; + // Execute a SQL query with tagged template literals + // Returns an array of rows matching the query + sql>( + strings: TemplateStringsArray, + ...values: (string | number | boolean | null)[] + ): T[]; } ``` @@ -525,16 +541,16 @@ class Agent { ```ts // Example of using SQL in an Agent interface User { - id: string; - name: string; - email: string; - created_at: number; + id: string; + name: string; + email: string; + created_at: number; } export class YourAgent extends Agent { - async setupDatabase() { - // Create a table if it doesn't exist - this.sql` + async setupDatabase() { + // Create a table if it doesn't exist + this.sql` CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, name TEXT NOT NULL, @@ -542,33 +558,33 @@ export class YourAgent extends Agent { created_at INTEGER ) `; - } + } - async createUser(id: string, name: string, email: string) { - // Insert a new user - this.sql` + async createUser(id: string, name: string, email: string) { + // Insert a new user + this.sql` INSERT INTO users (id, name, email, created_at) VALUES (${id}, ${name}, ${email}, ${Date.now()}) `; - } + } - async getUserById(id: string): Promise { - // Query a user by ID - const users = this.sql` + async getUserById(id: string): Promise { + // Query a user by ID + const users = this.sql` SELECT * FROM users WHERE id = ${id} `; - return users.length ? users[0] : null; - } + return users.length ? users[0] : null; + } - async searchUsers(term: string): Promise { - // Search users with a wildcard - return this.sql` + async searchUsers(term: string): Promise { + // Search users with a wildcard + return this.sql` SELECT * FROM users - WHERE name LIKE ${'%' + term + '%'} OR email LIKE ${'%' + term + '%'} + WHERE name LIKE ${"%" + term + "%"} OR email LIKE ${"%" + term + "%"} ORDER BY created_at DESC `; - } + } } ``` @@ -588,20 +604,20 @@ The Agents SDK allows your Agent to act as an MCP (Model Context Protocol) clien ```ts class Agent { - // Connect to an MCP server - async addMcpServer( - serverName: string, - url: string, - callbackHost?: string, - agentsPrefix?: string, - options?: MCPClientOptions - ): Promise<{ id: string; authUrl: string | undefined }>; - - // Disconnect from an MCP server - async removeMcpServer(id: string): Promise; - - // Get state of all connected MCP servers - getMcpServers(): MCPServersState; + // Connect to an MCP server + async addMcpServer( + serverName: string, + url: string, + callbackHost?: string, + agentsPrefix?: string, + options?: MCPClientOptions, + ): Promise<{ id: string; authUrl: string | undefined }>; + + // Disconnect from an MCP server + async removeMcpServer(id: string): Promise; + + // Get state of all connected MCP servers + getMcpServers(): MCPServersState; } ``` @@ -619,9 +635,9 @@ For complete MCP client API documentation, including OAuth configuration and adv The Agents SDK provides a set of client APIs for interacting with Agents from client-side JavaScript code, including: -* React hooks, including `useAgent` and `useAgentChat`, for connecting to Agents from client applications. -* Client-side [state syncing](/agents/api-reference/store-and-sync-state/) that allows you to subscribe to state updates between the Agent and any connected client(s) when calling `this.setState` within your Agent's code. -* The ability to call remote methods (Remote Procedure Calls; RPC) on the Agent from client-side JavaScript code using the `@callable` method decorator. +- React hooks, including `useAgent` and `useAgentChat`, for connecting to Agents from client applications. +- Client-side [state syncing](/agents/api-reference/store-and-sync-state/) that allows you to subscribe to state updates between the Agent and any connected client(s) when calling `this.setState` within your Agent's code. +- The ability to call remote methods (Remote Procedure Calls; RPC) on the Agent from client-side JavaScript code using the `@callable` method decorator. #### AgentClient @@ -632,17 +648,17 @@ import { AgentClient } from "agents/client"; // Options for creating an AgentClient type AgentClientOptions = Omit & { - // Name of the agent to connect to (class name in kebab-case) - agent: string; - // Name of the specific Agent instance (optional, defaults to "default") - name?: string; - // Other WebSocket options like host, protocol, etc. + // Name of the agent to connect to (class name in kebab-case) + agent: string; + // Name of the specific Agent instance (optional, defaults to "default") + name?: string; + // Other WebSocket options like host, protocol, etc. }; // WebSocket client for connecting to an Agent class AgentClient extends PartySocket { - static fetch(opts: PartyFetchOptions): Promise; - constructor(opts: AgentClientOptions); + static fetch(opts: PartyFetchOptions): Promise; + constructor(opts: AgentClientOptions); } ``` @@ -654,15 +670,15 @@ import { AgentClient } from "agents/client"; // Connect to an Agent instance const client = new AgentClient({ - agent: "chat-agent", // Name of your Agent class in kebab-case - name: "support-room-123", // Specific instance name - host: window.location.host, // Using same host + agent: "chat-agent", // Name of your Agent class in kebab-case + name: "support-room-123", // Specific instance name + host: window.location.host, // Using same host }); client.onopen = () => { - console.log("Connected to agent"); - // Send an initial message - client.send(JSON.stringify({ type: "join", user: "user123" })); + console.log("Connected to agent"); + // Send an initial message + client.send(JSON.stringify({ type: "join", user: "user123" })); }; client.onmessage = (event) => { @@ -680,11 +696,13 @@ client.onclose = () => console.log("Disconnected from agent"); // Send messages to the Agent function sendMessage(text) { - client.send(JSON.stringify({ - type: "message", - text, - timestamp: Date.now() - })); + client.send( + JSON.stringify({ + type: "message", + text, + timestamp: Date.now(), + }), + ); } ``` @@ -699,16 +717,16 @@ import { agentFetch } from "agents/client"; // Options for the agentFetch function type AgentClientFetchOptions = Omit & { - // Name of the agent to connect to - agent: string; - // Name of the specific Agent instance (optional) - name?: string; + // Name of the agent to connect to + agent: string; + // Name of the specific Agent instance (optional) + name?: string; }; // Make an HTTP request to an Agent function agentFetch( - opts: AgentClientFetchOptions, - init?: RequestInit + opts: AgentClientFetchOptions, + init?: RequestInit, ): Promise; ``` @@ -720,29 +738,29 @@ import { agentFetch } from "agents/client"; // Function to get data from an Agent async function fetchAgentData() { - try { - const response = await agentFetch( - { - agent: "task-manager", - name: "user-123-tasks" - }, - { - method: "GET", - headers: { - "Authorization": `Bearer ${userToken}` - } - } - ); - - if (!response.ok) { - throw new Error(`Error: ${response.status}`); - } - - const data = await response.json(); - return data; - } catch (error) { - console.error("Failed to fetch from agent:", error); - } + try { + const response = await agentFetch( + { + agent: "task-manager", + name: "user-123-tasks", + }, + { + method: "GET", + headers: { + Authorization: `Bearer ${userToken}`, + }, + }, + ); + + if (!response.ok) { + throw new Error(`Error: ${response.status}`); + } + + const data = await response.json(); + return data; + } catch (error) { + console.error("Failed to fetch from agent:", error); + } } ``` @@ -761,24 +779,24 @@ import { useAgent } from "agents/react"; // Options for the useAgent hook type UseAgentOptions = Omit< - Parameters[0], - "party" | "room" + Parameters[0], + "party" | "room" > & { - // Name of the agent to connect to - agent: string; - // Name of the specific Agent instance (optional) - name?: string; - // Called when the Agent's state is updated - onStateUpdate?: (state: State, source: "server" | "client") => void; + // Name of the agent to connect to + agent: string; + // Name of the specific Agent instance (optional) + name?: string; + // Called when the Agent's state is updated + onStateUpdate?: (state: State, source: "server" | "client") => void; }; // React hook for connecting to an Agent // Returns a WebSocket connection with setState method function useAgent( - options: UseAgentOptions + options: UseAgentOptions, ): PartySocket & { - // Update the Agent's state - setState: (state: State) => void + // Update the Agent's state + setState: (state: State) => void; }; ``` @@ -798,17 +816,17 @@ import { Message, StreamTextOnFinishCallback, ToolSet } from "ai"; // Base class for chat-specific agents class AIChatAgent extends Agent { - // Array of chat messages for the current conversation - messages: Message[]; + // Array of chat messages for the current conversation + messages: Message[]; - // Handle incoming chat messages and generate a response - // onFinish is called when the response is complete - async onChatMessage( - onFinish: StreamTextOnFinishCallback - ): Promise; + // Handle incoming chat messages and generate a response + // onFinish is called when the response is complete + async onChatMessage( + onFinish: StreamTextOnFinishCallback, + ): Promise; - // Persist messages within the Agent's local storage. - async saveMessages(messages: Message[]): Promise; + // Persist messages within the Agent's local storage. + async saveMessages(messages: Message[]): Promise; } ``` @@ -820,51 +838,99 @@ import { AIChatAgent } from "agents/ai-chat-agent"; import { Message } from "ai"; interface Env { - AI: any; // Your AI binding + AI: any; // Your AI binding } class CustomerSupportAgent extends AIChatAgent { - // Override the onChatMessage method to customize behavior - async onChatMessage(onFinish) { - // Access the AI models using environment bindings - const { openai } = this.env.AI; - - // Get the current conversation history - const chatHistory = this.messages; - - // Generate a system prompt based on knowledge base - const systemPrompt = await this.generateSystemPrompt(); - - // Generate a response stream - const stream = await openai.chat({ - model: "gpt-4o", - messages: [ - { role: "system", content: systemPrompt }, - ...chatHistory - ], - stream: true - }); - - // Return the streaming response - return new Response(stream, { - headers: { "Content-Type": "text/event-stream" } - }); - } - - // Helper method to generate a system prompt - async generateSystemPrompt() { - // Query knowledge base or use static prompt - return `You are a helpful customer support agent. + // Override the onChatMessage method to customize behavior + async onChatMessage(onFinish) { + // Access the AI models using environment bindings + const { openai } = this.env.AI; + + // Get the current conversation history + const chatHistory = this.messages; + + // Generate a system prompt based on knowledge base + const systemPrompt = await this.generateSystemPrompt(); + + // Generate a response stream + const stream = await openai.chat({ + model: "gpt-4o", + messages: [{ role: "system", content: systemPrompt }, ...chatHistory], + stream: true, + }); + + // Return the streaming response + return new Response(stream, { + headers: { "Content-Type": "text/event-stream" }, + }); + } + + // Helper method to generate a system prompt + async generateSystemPrompt() { + // Query knowledge base or use static prompt + return `You are a helpful customer support agent. Respond to customer inquiries based on the following guidelines: - Be friendly and professional - If you don't know an answer, say so - Current company policies: ...`; - } + } +} +``` + + + +#### Resumable streaming + +The `AIChatAgent` class provides **automatic resumable streaming** out of the box. When a client disconnects and reconnects during an active stream, the response automatically resumes from where it left off. This works across browser tabs and devices. + +##### How it works + +When you use `AIChatAgent` with `useAgentChat`: + +1. **During streaming**: All chunks are automatically persisted to SQLite +2. **On disconnect**: The stream continues server-side, buffering chunks +3. **On reconnect**: Client receives all buffered chunks and continues streaming + +##### Server + + + +```ts +import { AIChatAgent } from "agents/ai-chat-agent"; +import { streamText } from "ai"; +import { openai } from "@ai-sdk/openai"; + +export class ChatAgent extends AIChatAgent { + async onChatMessage() { + const result = streamText({ + model: openai("gpt-4o"), + messages: this.messages, + }); + + // Automatic resumable streaming - no extra code needed! + return result.toUIMessageStreamResponse(); + } } ``` +##### Disabling resume + +If you don't want automatic resume (for example, for short responses), disable it: + + + +```tsx +const { messages } = useAgentChat({ + agent, + resume: false, // Disable automatic stream resumption +}); +``` + + + ### Chat Agent React API #### useAgentChat @@ -878,51 +944,67 @@ import type { Message } from "ai"; // Options for the useAgentChat hook type UseAgentChatOptions = Omit< - Parameters[0] & { - // Agent connection from useAgent - agent: ReturnType; - }, - "fetch" + Parameters[0] & { + // Agent connection from useAgent + agent: ReturnType; + }, + "fetch" >; // React hook for building AI chat interfaces using an Agent function useAgentChat(options: UseAgentChatOptions): { - // Current chat messages - messages: Message[]; - // Set messages and synchronize with the Agent - setMessages: (messages: Message[]) => void; - // Clear chat history on both client and Agent - clearHistory: () => void; - // Append a new message to the conversation - append: (message: Message, chatRequestOptions?: any) => Promise; - // Reload the last user message - reload: (chatRequestOptions?: any) => Promise; - // Stop the AI response generation - stop: () => void; - // Current input text - input: string; - // Set the input text - setInput: React.Dispatch>; - // Handle input changes - handleInputChange: (e: React.ChangeEvent) => void; - // Submit the current input - handleSubmit: (event?: { preventDefault?: () => void }, chatRequestOptions?: any) => void; - // Additional metadata - metadata?: Object; - // Whether a response is currently being generated - isLoading: boolean; - // Current status of the chat - status: "submitted" | "streaming" | "ready" | "error"; - // Tool data from the AI response - data?: any[]; - // Set tool data - setData: (data: any[] | undefined | ((data: any[] | undefined) => any[] | undefined)) => void; - // Unique ID for the chat - id: string; - // Add a tool result for a specific tool call - addToolResult: ({ toolCallId, result }: { toolCallId: string; result: any }) => void; - // Current error if any - error: Error | undefined; + // Current chat messages + messages: Message[]; + // Set messages and synchronize with the Agent + setMessages: (messages: Message[]) => void; + // Clear chat history on both client and Agent + clearHistory: () => void; + // Append a new message to the conversation + append: ( + message: Message, + chatRequestOptions?: any, + ) => Promise; + // Reload the last user message + reload: (chatRequestOptions?: any) => Promise; + // Stop the AI response generation + stop: () => void; + // Current input text + input: string; + // Set the input text + setInput: React.Dispatch>; + // Handle input changes + handleInputChange: ( + e: React.ChangeEvent, + ) => void; + // Submit the current input + handleSubmit: ( + event?: { preventDefault?: () => void }, + chatRequestOptions?: any, + ) => void; + // Additional metadata + metadata?: Object; + // Whether a response is currently being generated + isLoading: boolean; + // Current status of the chat + status: "submitted" | "streaming" | "ready" | "error"; + // Tool data from the AI response + data?: any[]; + // Set tool data + setData: ( + data: any[] | undefined | ((data: any[] | undefined) => any[] | undefined), + ) => void; + // Unique ID for the chat + id: string; + // Add a tool result for a specific tool call + addToolResult: ({ + toolCallId, + result, + }: { + toolCallId: string; + result: any; + }) => void; + // Current error if any + error: Error | undefined; }; ``` @@ -935,66 +1017,65 @@ import { useAgent } from "agents/react"; import { useState } from "react"; function ChatInterface() { - // Connect to the chat agent - const agentConnection = useAgent({ - agent: "customer-support", - name: "session-12345" - }); - - // Use the useAgentChat hook with the agent connection - const { - messages, - input, - handleInputChange, - handleSubmit, - isLoading, - error, - clearHistory - } = useAgentChat({ - agent: agentConnection, - initialMessages: [ - { role: "system", content: "You're chatting with our AI assistant." }, - { role: "assistant", content: "Hello! How can I help you today?" } - ] - }); - - return ( -
-
- {messages.map((message, i) => ( -
- {message.role === 'user' ? '👤' : '🤖'} {message.content} -
- ))} - - {isLoading &&
AI is typing...
} - {error &&
Error: {error.message}
} -
- -
- - - -
-
- ); + // Connect to the chat agent + const agentConnection = useAgent({ + agent: "customer-support", + name: "session-12345", + }); + + // Use the useAgentChat hook with the agent connection + const { + messages, + input, + handleInputChange, + handleSubmit, + isLoading, + error, + clearHistory, + } = useAgentChat({ + agent: agentConnection, + initialMessages: [ + { role: "system", content: "You're chatting with our AI assistant." }, + { role: "assistant", content: "Hello! How can I help you today?" }, + ], + }); + + return ( +
+
+ {messages.map((message, i) => ( +
+ {message.role === "user" ? "👤" : "🤖"} {message.content} +
+ ))} + + {isLoading &&
AI is typing...
} + {error &&
Error: {error.message}
} +
+ +
+ + + +
+
+ ); } ``` - ### Next steps -* [Build a chat Agent](/agents/getting-started/build-a-chat-agent/) using the Agents SDK and deploy it to Workers. -* Learn more [using WebSockets](/agents/api-reference/websockets/) to build interactive Agents and stream data back from your Agent. -* [Orchestrate asynchronous workflows](/agents/api-reference/run-workflows) from your Agent by combining the Agents SDK and [Workflows](/workflows). +- [Build a chat Agent](/agents/getting-started/build-a-chat-agent/) using the Agents SDK and deploy it to Workers. +- Learn more [using WebSockets](/agents/api-reference/websockets/) to build interactive Agents and stream data back from your Agent. +- [Orchestrate asynchronous workflows](/agents/api-reference/run-workflows) from your Agent by combining the Agents SDK and [Workflows](/workflows). diff --git a/src/content/docs/agents/guides/resumable-streaming.mdx b/src/content/docs/agents/guides/resumable-streaming.mdx deleted file mode 100644 index c99a781d05d..00000000000 --- a/src/content/docs/agents/guides/resumable-streaming.mdx +++ /dev/null @@ -1,102 +0,0 @@ ---- -pcx_content_type: concept -title: Resumable Streaming -sidebar: - order: 10 ---- - -import { TypeScriptExample, WranglerConfig } from "~/components"; - -The `AIChatAgent` class provides **automatic resumable streaming** out of the box. When a client disconnects and reconnects during an active stream, the response automatically resumes from where it left off. - -## How it works - -When you use `AIChatAgent` with `useAgentChat`: - -1. **During streaming**: All chunks are automatically persisted to SQLite -2. **On disconnect**: The stream continues server-side, buffering chunks -3. **On reconnect**: Client receives all buffered chunks and continues streaming - -It just works! - -## Example - -### Server - - -```ts -import { AIChatAgent } from "agents/ai-chat-agent"; -import { streamText } from "ai"; -import { openai } from "@ai-sdk/openai"; - -export class ChatAgent extends AIChatAgent { - async onChatMessage() { - const result = streamText({ - model: openai("gpt-4o"), - messages: this.messages - }); - - // Automatic resumable streaming - no extra code needed! - return result.toUIMessageStreamResponse(); - } -} -``` - - -### Client - - -```tsx -import { useAgent } from "agents/react"; -import { useAgentChat } from "agents/ai-react"; - -function Chat() { - const agent = useAgent({ - agent: "ChatAgent", - name: "my-chat" - }); - - const { messages, input, handleInputChange, handleSubmit, status } = - useAgentChat({ - agent - // resume: true is the default - streams automatically resume on reconnect - }); - - // ... render your chat UI -} -``` - - -## Under the hood - -### Server-side (AIChatAgent) - -- Creates SQLite tables for stream chunks and metadata on startup -- Each stream gets a unique ID and tracks chunk indices -- Chunks are buffered and flushed to SQLite every 100ms for performance -- On client connect, checks for active streams and sends `CF_AGENT_STREAM_RESUMING` -- Old completed streams are cleaned up after 24 hours - -### Client-side (useAgentChat) - -- Listens for `CF_AGENT_STREAM_RESUMING` notification -- Sends `CF_AGENT_STREAM_RESUME_ACK` when ready -- Receives all buffered chunks and reconstructs the message -- Continues receiving live chunks as they arrive - -## Disabling resume - -If you do not want automatic resume (for example, for short responses), disable it: - - -```tsx -const { messages } = useAgentChat({ - agent, - resume: false // Disable automatic stream resumption -}); -``` - - -## Try it - -Refer to the [resumable-stream-chat example](https://github.com/cloudflare/agents/tree/main/examples/resumable-stream-chat) for a complete working example. Start a long response, refresh the page mid-stream, and watch it resume automatically.