WebSockets

WebSockets from agents/websockets is a capability that makes WebSocket connections to a Durable Object simple. It accepts upgrades, lets idle connections hibernate, tracks and tags connections, and speaks the Agents protocol. That lets a plain Durable Object work with useAgent and AgentClient from the client SDK.

Install the capability

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { routeAgentRequest } from "agents/routing";
import { WebSockets } from "agents/websockets";

export class RoomAgent extends DurableObject {
	webSockets = new WebSockets({
		handlers: {
			onConnect: (connection) => {
				connection.send(JSON.stringify({ type: "welcome", id: connection.id }));
			},
			onMessage: (connection, message) => {
				for (const other of this.webSockets.getConnections()) {
					if (other.id !== connection.id) other.send(message);
				}
			},
		},
	});
	lifecycle = Lifecycle.install(this).use(this.webSockets);
}

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

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { routeAgentRequest } from "agents/routing";
import { WebSockets } from "agents/websockets";

export class RoomAgent extends DurableObject<Env> {
	readonly webSockets: WebSockets = new WebSockets({
		handlers: {
			onConnect: (connection) => {
				connection.send(JSON.stringify({ type: "welcome", id: connection.id }));
			},
			onMessage: (connection, message) => {
				for (const other of this.webSockets.getConnections()) {
					if (other.id !== connection.id) other.send(message);
				}
			},
		},
	});
	readonly lifecycle = Lifecycle.install(this).use(this.webSockets);
}

export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		return (
			(await routeAgentRequest(request, env)) ??
			new Response("Not found", { status: 404 })
		);
	},
} satisfies ExportedHandler<Env>;

routeAgentRequest() sends /agents/room-agent/{name} to the RoomAgent object called {name}. Authenticate the request before you route it. For more information, refer to Route requests.

WebSockets claims every upgrade, even without handlers. It is the only catch-all capability, so the Lifecycle offers each upgrade to the other capabilities first. Without WebSockets, the Lifecycle declines upgrades. Plain HTTP requests still reach your onRequest().

Options

Option	Description
`handlers`	Connection handlers: `onConnect`, `onMessage`, `onClose`, and `onError`
`callables`	An `RpcTarget` whose methods clients call. For more information, refer to Callables
`state`	A `State` capability to sync with clients
`getConnectionTags`	`(connection, ctx) => string[]`. Labels each connection when it is accepted
`readonly`	`(connection, ctx) => boolean`. Decides whether a connection may change the synced state
`protocol`	`true` (default), `false`, or `(connection, ctx) => boolean`. Decides whether a connection gets Agents protocol frames

In every function, ctx.request is the upgrade request.

Handle connections

The handlers run inside your object's invocation context, so getCurrentAgent() returns your object and the connection:

Handler	When it runs
`onConnect(connection, ctx)`	After the upgrade is accepted. `ctx.request` is the upgrade request
`onMessage(connection, message)`	For each message that is not an Agents protocol frame
`onClose(connection, code, reason, wasClean)`	When the connection closes
`onError(connection, error)`	When the connection fails mid-session

A message wakes a hibernated object. The constructor and Lifecycle startup run again before onMessage. Keep anything you need after a wake in storage or on the connection.

Connections

A Connection is a WebSocket with these extra fields:

Field	Description
`id`	The connection ID
`uri`	The URL of the upgrade request
`tags`	The connection's tags. The ID is always the first tag
`state`	A small value stored on this connection. It survives hibernation
`setState(value)`	Replaces `state`. Pass a function to update it from the previous value, or `null` to clear it

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { WebSockets } from "agents/websockets";

export class LobbyAgent extends DurableObject {
	webSockets = new WebSockets({
		handlers: {
			onConnect: (connection) => {
				connection.setState({ joinedAt: Date.now() });
			},
			onClose: (connection) => {
				const visitor = connection.state;
				console.log(
					`${connection.id} stayed ${Date.now() - (visitor?.joinedAt ?? 0)} ms`,
				);
			},
		},
	});
	lifecycle = Lifecycle.install(this).use(this.webSockets);
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { WebSockets } from "agents/websockets";

type Visitor = { joinedAt: number };

export class LobbyAgent extends DurableObject<Env> {
	readonly webSockets = new WebSockets({
		handlers: {
			onConnect: (connection) => {
				connection.setState({ joinedAt: Date.now() });
			},
			onClose: (connection) => {
				const visitor = connection.state as Visitor | null;
				console.log(
					`${connection.id} stayed ${Date.now() - (visitor?.joinedAt ?? 0)} ms`,
				);
			},
		},
	});
	readonly lifecycle = Lifecycle.install(this).use(this.webSockets);
}

connection.setState() is not synced to other clients. To share one value with every client, use the State capability.

Find connections

Method	Description
`getConnections(tag?)`	Iterates over every open connection, or those with one tag
`getConnection(id)`	Returns one connection by ID, or `undefined`

Tag connections

Pass getConnectionTags to label each connection when it is accepted. Then send to one group with getConnections(tag):

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { WebSockets } from "agents/websockets";

export class ChatRoom extends DurableObject {
	webSockets = new WebSockets({
		getConnectionTags: (_connection, ctx) => {
			const room = new URL(ctx.request.url).searchParams.get("room");
			return room ? [`room:${room}`] : [];
		},
		handlers: {
			onMessage: (_connection, message) => {
				for (const peer of this.webSockets.getConnections("room:general")) {
					peer.send(message);
				}
			},
		},
	});
	lifecycle = Lifecycle.install(this).use(this.webSockets);
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { WebSockets } from "agents/websockets";

export class ChatRoom extends DurableObject<Env> {
	readonly webSockets: WebSockets = new WebSockets({
		getConnectionTags: (_connection, ctx) => {
			const room = new URL(ctx.request.url).searchParams.get("room");
			return room ? [`room:${room}`] : [];
		},
		handlers: {
			onMessage: (_connection, message) => {
				for (const peer of this.webSockets.getConnections("room:general")) {
					peer.send(message);
				}
			},
		},
	});
	readonly lifecycle = Lifecycle.install(this).use(this.webSockets);
}

use(handlers) adds another set of handlers to the same connections, for example for a protocol of your own. Added handlers run before the configured ones. An added onMessage can return true to claim a message, so later handlers do not see it:

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { WebSockets } from "agents/websockets";

export class EchoAgent extends DurableObject {
	webSockets = new WebSockets({
		handlers: {
			onMessage: (connection, message) => connection.send(message),
		},
	});
	lifecycle = Lifecycle.install(this).use(this.webSockets);

	constructor(ctx, env) {
		super(ctx, env);
		// Runs before the configured handlers. Return true to claim a message.
		this.webSockets.use({
			onMessage: (connection, message) => {
				if (message !== "ping") return false;
				connection.send("pong");
				return true;
			},
		});
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { WebSockets } from "agents/websockets";

export class EchoAgent extends DurableObject<Env> {
	readonly webSockets = new WebSockets({
		handlers: {
			onMessage: (connection, message) => connection.send(message),
		},
	});
	readonly lifecycle = Lifecycle.install(this).use(this.webSockets);

	constructor(ctx: DurableObjectState, env: Env) {
		super(ctx, env);
		// Runs before the configured handlers. Return true to claim a message.
		this.webSockets.use({
			onMessage: (connection, message) => {
				if (message !== "ping") return false;
				connection.send("pong");
				return true;
			},
		});
	}
}

Callables

Use callables to let your application's interface invoke actions on an agent. For example, a Stop button can call a server method that cancels a run. The client sends arguments over its existing connection and receives the method's result or error.

Declare these actions on an RpcTarget and pass it to WebSockets as callables. Clients invoke them through agent.stub or agent.call() from useAgent and AgentClient.

Choose a callable

Use a callable for an action initiated by an application client, such as stopping a run or changing an agent's settings.
Use a tool for an action the model can choose to invoke.
Use Durable Object RPC when another Worker calls your object directly.

Declare callables

This example adds a stopRun() callable to the Pi agent. Use that guide's package installation and bindings. Keep authentication in the routing Worker, so each user can connect only to their own agent. All sessions in this example's object belong to that user.

stopRun() is an application-defined method. It validates the session ID, then asks Pi to abort that session's running work and withdraw queued inputs. It does not undo completed actions.

src/server.jsjs

import { DurableObject, RpcTarget } from "cloudflare:workers";
import { createModels } from "@earendil-works/pi-ai/models";
import { createRegistry, Harness } from "@earendil-works/pi-durable";
import { PiHarness } from "agents/harness/pi";
import { Lifecycle } from "agents/lifecycle";
import { createAI } from "agents/models/pi-ai";
import { WebSockets } from "agents/websockets";

export class RunCallables extends RpcTarget {
	#harness;

	constructor(harness) {
		super();
		this.#harness = harness;
	}

	async stopRun(sessionId) {
		if (
			typeof sessionId !== "string" ||
			!/^[1-9]\d{0,15}$/.test(sessionId) ||
			!Number.isSafeInteger(Number(sessionId))
		) {
			throw new Error("sessionId must be a positive integer string");
		}
		await this.#harness.session(sessionId).abort();
	}
}

export class Assistant extends DurableObject {
	ai = createAI({ binding: this.env.AI });
	registry = createRegistry();
	harness = new PiHarness({
		harness: ({ storage, context, onWake }) => {
			const models = createModels();
			models.setProvider(this.ai.provider);
			return Harness.open(
				storage,
				{ models, registry: this.registry, onWake },
				context,
			);
		},
		defaults: { model: this.ai("@cf/zai-org/glm-5.3-flash") },
	});
	webSockets = new WebSockets({
		callables: new RunCallables(this.harness),
	});
	lifecycle = Lifecycle.install(this).use(this.harness).use(this.webSockets);

	async ask(prompt) {
		const { text } = await this.harness.prompt(prompt);
		return text;
	}
}

src/server.tsts

import { DurableObject, RpcTarget } from "cloudflare:workers";
import { createModels } from "@earendil-works/pi-ai/models";
import { createRegistry, Harness } from "@earendil-works/pi-durable";
import { PiHarness } from "agents/harness/pi";
import { Lifecycle } from "agents/lifecycle";
import { createAI } from "agents/models/pi-ai";
import { WebSockets } from "agents/websockets";

export class RunCallables extends RpcTarget {
	readonly #harness: PiHarness;

	constructor(harness: PiHarness) {
		super();
		this.#harness = harness;
	}

	async stopRun(sessionId: string): Promise<void> {
		if (
			typeof sessionId !== "string" ||
			!/^[1-9]\d{0,15}$/.test(sessionId) ||
			!Number.isSafeInteger(Number(sessionId))
		) {
			throw new Error("sessionId must be a positive integer string");
		}
		await this.#harness.session(sessionId).abort();
	}
}

export class Assistant extends DurableObject<Env> {
	readonly ai = createAI({ binding: this.env.AI });
	readonly registry = createRegistry();
	readonly harness = new PiHarness({
		harness: ({ storage, context, onWake }) => {
			const models = createModels();
			models.setProvider(this.ai.provider);
			return Harness.open(
				storage,
				{ models, registry: this.registry, onWake },
				context,
			);
		},
		defaults: { model: this.ai("@cf/zai-org/glm-5.3-flash") },
	});
	readonly webSockets = new WebSockets({
		callables: new RunCallables(this.harness),
	});
	readonly lifecycle = Lifecycle.install(this)
		.use(this.harness)
		.use(this.webSockets);

	async ask(prompt: string) {
		const { text } = await this.harness.prompt(prompt);
		return text;
	}
}

The target is the complete remote interface. Clients can only call methods declared on its class, not the Durable Object's own methods, so you decide exactly what is exposed:

Only methods on the class prototype are callable. Instance properties, including arrow functions assigned to fields, are not.
The constructor and methods inherited from Object or RpcTarget are never callable.
WebSockets throws when the class declares a method named then.

Any connected client can call any of these methods. Treat every argument as untrusted input, and check who is calling before the request reaches the object. For more information, refer to Authentication.

Call from a client

Type agent.stub with AgentStub from agents/client and your callables class. Each method becomes an async function with the same arguments and return type:

import { useState } from "react";
import type { AgentStub } from "agents/client";
import { useAgent } from "agents/react";
import type { RunCallables } from "./server";

type StopRunButtonProps = { agentName: string; sessionId: string };

export function StopRunButton({ agentName, sessionId }: StopRunButtonProps) {
	const agent = useAgent({ agent: "Assistant", name: agentName });
	const runs = agent.stub as AgentStub<RunCallables>;
	const [pending, setPending] = useState(false);
	const [status, setStatus] = useState("");

	const stop = async () => {
		setPending(true);
		setStatus("");
		try {
			await agent.ready;
			await runs.stopRun(sessionId);
			setStatus("Stop requested.");
		} catch (error) {
			setStatus(
				error instanceof Error ? error.message : "Could not stop the run.",
			);
		} finally {
			setPending(false);
		}
	};

	return (
		<>
			<button onClick={stop} disabled={pending}>
				{pending ? "Stopping…" : "Stop"}
			</button>
			<p role="status">{status}</p>
		</>
	);
}

Pass the authorized agent name and the session ID returned by Pi to StopRunButton. Export RunCallables from your server file so the client can import its type. import type includes only the type, keeping server code out of the client bundle.

AgentStub<RunCallables> describes the remote interface to TypeScript. Runtime argument validation still happens in stopRun().

agent.call() takes the method name and an array of arguments, without types:

await agent.call("stopRun", [sessionId]);

AgentClient from agents/client has the same call() and stub. Non-streaming calls time out after 30 seconds by default. Pass { timeout } as the third argument to call() to change it for one call, or set defaultCallTimeout on the client. A timeout of 0 turns it off.

When a method throws, the client's promise rejects with the error message. When a method does not exist, it rejects with Method <name> does not exist.

Stream results

Return a ReadableStream to send results as they are produced. Each chunk reaches the client as soon as it is read:

import { RpcTarget } from "cloudflare:workers";

class CountdownCallables extends RpcTarget {
	countdown(from) {
		const start = Math.min(Math.max(Math.trunc(from), 1), 10);
		return new ReadableStream({
			start(controller) {
				for (let n = start; n >= 0; n--) controller.enqueue(n);
				controller.close();
			},
		});
	}
}

import { RpcTarget } from "cloudflare:workers";

class CountdownCallables extends RpcTarget {
	countdown(from: number): ReadableStream<number> {
		const start = Math.min(Math.max(Math.trunc(from), 1), 10);
		return new ReadableStream<number>({
			start(controller) {
				for (let n = start; n >= 0; n--) controller.enqueue(n);
				controller.close();
			},
		});
	}
}

Pass stream callbacks to call() to read the chunks:

await agent.call("countdown", [3], {
	stream: {
		onChunk: (chunk) => console.log(chunk),
		onDone: () => console.log("done"),
		onError: (error) => console.error(error),
	},
});

Streaming calls have no default timeout.

Context

Each call runs inside your object's invocation context, with the calling connection in scope. getCurrentAgent() returns your object and the connection, so a method can reply to the caller or broadcast to everyone:

import { DurableObject, RpcTarget } from "cloudflare:workers";
import { getCurrentAgent, Lifecycle } from "agents/lifecycle";
import { WebSockets } from "agents/websockets";

class RoomCallables extends RpcTarget {
	#room;

	constructor(room) {
		super();
		this.#room = room;
	}

	say(text) {
		const { connection } = getCurrentAgent();
		for (const peer of this.#room.webSockets.getConnections()) {
			if (peer.id !== connection?.id) peer.send(text);
		}
	}
}

export class Room extends DurableObject {
	webSockets = new WebSockets({
		callables: new RoomCallables(this),
	});
	lifecycle = Lifecycle.install(this).use(this.webSockets);
}

import { DurableObject, RpcTarget } from "cloudflare:workers";
import { getCurrentAgent, Lifecycle } from "agents/lifecycle";
import { WebSockets } from "agents/websockets";

class RoomCallables extends RpcTarget {
	readonly #room: Room;

	constructor(room: Room) {
		super();
		this.#room = room;
	}

	say(text: string): void {
		const { connection } = getCurrentAgent();
		for (const peer of this.#room.webSockets.getConnections()) {
			if (peer.id !== connection?.id) peer.send(text);
		}
	}
}

export class Room extends DurableObject<Env> {
	readonly webSockets: WebSockets = new WebSockets({
		callables: new RoomCallables(this),
	});
	readonly lifecycle = Lifecycle.install(this).use(this.webSockets);
}

Transports

The client chooses how calls travel with its transport option:

Transport	How calls travel	Hibernates	Results
`"cf-websocket"` (default)	JSON `rpc` frames over a hibernating WebSocket	Yes	Must be JSON-serializable. A `ReadableStream` streams. An `RpcTarget` result is an error.
`"capnweb"`	Native methods on a Cap'n Web ↗︎ session	No	An `RpcTarget` result arrives as a live stub the client can keep calling. Streams and pipelined calls work.

const agent = useAgent({
	agent: "Assistant",
	name: agentName,
	transport: "capnweb",
});

The same callables target serves both transports, and handlers such as onConnect and onMessage run for both. A Cap'n Web connection does not hibernate: it keeps the object in memory while it is open, and the client reconnects after an eviction. Use the default transport unless you need results passed by reference.

Sync state

Pass a State capability as state to sync one value with useAgent().state. On connect, each client receives its identity frame and then the current value. When a client calls setState(), the change goes through the state's validator before it is stored. For an example, refer to Sync state with clients.

WebSockets does not broadcast changes on its own. Call broadcastState(source) from the state's onChanged hook. It sends the new value to every connection except source, which already has it.

Readonly connections

A readonly connection can still receive state and call methods, but its setState() calls are refused. The client gets a cf_agent_state_error frame and its onStateUpdateError callback runs.

Decide when the connection is accepted with the readonly option, or change it later with setReadonly():

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { State } from "agents/state";
import { WebSockets } from "agents/websockets";

export class BoardAgent extends DurableObject {
	board = new State({
		initialState: { cards: [] },
		onChanged: (_state, source) => this.webSockets.broadcastState(source),
	});
	webSockets = new WebSockets({
		state: this.board,
		readonly: (_connection, ctx) =>
			new URL(ctx.request.url).searchParams.get("mode") === "view",
		protocol: (_connection, ctx) =>
			new URL(ctx.request.url).searchParams.get("client") !== "sensor",
	});
	lifecycle = Lifecycle.install(this).use(this.board).use(this.webSockets);

	// Called over RPC from the Worker after it checks the user's role.
	async promote(connectionId) {
		await this.lifecycle.start();
		const connection = this.webSockets.getConnection(connectionId);
		if (!connection) return false;
		this.webSockets.setReadonly(connection, false);
		return this.webSockets.isReadonly(connection) === false;
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { State } from "agents/state";
import { WebSockets } from "agents/websockets";

type Board = { cards: string[] };

export class BoardAgent extends DurableObject<Env> {
	readonly board: State<Board> = new State<Board>({
		initialState: { cards: [] },
		onChanged: (_state, source) => this.webSockets.broadcastState(source),
	});
	readonly webSockets: WebSockets = new WebSockets({
		state: this.board,
		readonly: (_connection, ctx) =>
			new URL(ctx.request.url).searchParams.get("mode") === "view",
		protocol: (_connection, ctx) =>
			new URL(ctx.request.url).searchParams.get("client") !== "sensor",
	});
	readonly lifecycle = Lifecycle.install(this)
		.use(this.board)
		.use(this.webSockets);

	// Called over RPC from the Worker after it checks the user's role.
	async promote(connectionId: string): Promise<boolean> {
		await this.lifecycle.start();
		const connection = this.webSockets.getConnection(connectionId);
		if (!connection) return false;
		this.webSockets.setReadonly(connection, false);
		return this.webSockets.isReadonly(connection) === false;
	}
}

Do not trust a query parameter on its own to decide who can write. Verify the user in your Worker before you route the request.

Control the protocol per connection

By default, every connection gets the Agents protocol frames: an identity frame on connect, then the current state when state is set, then state broadcasts. Clients such as useAgent wait for the identity frame before ready resolves.

Set protocol to a function to turn the frames off for some connections, such as binary-only clients. A connection without the protocol still sends and receives your own messages and can call methods. In the previous example, connections with ?client=sensor get no protocol frames.

Set protocol: false to drive the connect sequence yourself. Then send the frames with sendConnectFrames(), and apply state frames from clients with applyStateFrame():

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { State } from "agents/state";
import { WebSockets } from "agents/websockets";

export class PresenceAgent extends DurableObject {
	presence = new State({
		initialState: { online: 0 },
		onChanged: (_state, source) => this.webSockets.broadcastState(source),
	});
	webSockets = new WebSockets({
		state: this.presence,
		protocol: false,
		handlers: {
			onConnect: (connection) => {
				// Send identity and the current state when you choose.
				this.webSockets.sendConnectFrames(connection);
			},
			onMessage: (connection, message) => {
				if (typeof message !== "string") return;
				const frame = JSON.parse(message);
				if (this.webSockets.applyStateFrame(connection, frame)) return;
				connection.send(message);
			},
		},
	});
	lifecycle = Lifecycle.install(this).use(this.presence).use(this.webSockets);
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { State } from "agents/state";
import { WebSockets } from "agents/websockets";

type Presence = { online: number };

export class PresenceAgent extends DurableObject<Env> {
	readonly presence: State<Presence> = new State<Presence>({
		initialState: { online: 0 },
		onChanged: (_state, source) => this.webSockets.broadcastState(source),
	});
	readonly webSockets: WebSockets = new WebSockets({
		state: this.presence,
		protocol: false,
		handlers: {
			onConnect: (connection) => {
				// Send identity and the current state when you choose.
				this.webSockets.sendConnectFrames(connection);
			},
			onMessage: (connection, message) => {
				if (typeof message !== "string") return;
				const frame: unknown = JSON.parse(message);
				if (this.webSockets.applyStateFrame(connection, frame)) return;
				connection.send(message);
			},
		},
	});
	readonly lifecycle = Lifecycle.install(this)
		.use(this.presence)
		.use(this.webSockets);
}

Methods

Method	Description
`getConnections(tag?)`	Iterates over open connections, optionally by tag
`getConnection(id)`	Returns one open connection, or `undefined`
`use(handlers)`	Adds a handler set that runs before the configured handlers
`broadcastState(except?)`	Sends the current state to every protocol-enabled connection except `except`
`sendConnectFrames(connection, identity?)`	Sends the identity frame, then the current state when one is stored
`sendIdentity(connection, identity?)`	Sends the identity frame alone. `identity` defaults to the object's name and its class name in kebab case
`sendState(connection)`	Sends the current state to one connection
`applyStateFrame(connection, frame)`	Applies a client's state frame, refusing readonly connections. Returns `true` when `frame` was a state frame
`isReadonly(connection)`	Returns whether the connection is readonly
`setReadonly(connection, readonly?)`	Marks a connection readonly, or writable again with `false`
`isProtocolEnabled(connection)`	Returns whether protocol frames reach the connection
`setProtocolEnabled(connection, enabled)`	Turns protocol frames on or off for one connection

None of the send methods send anything to a connection without the protocol. When the object is destroyed, WebSockets closes every open connection with code 1001.

On the Agent class

The Agent class installs WebSockets and calls its own onConnect(), onMessage(), onClose(), and onError() methods. Its @callable() methods are its client interface, instead of a callables target. For more information, refer to WebSockets, Callable methods, and Readonly connections.