WebSockets

来自 agents/websockets 的 WebSockets 是一项功能，可简化 Durable Object 的 WebSocket 连接。它接受升级请求、允许空闲连接休眠、跟踪并标记连接，并使用 Agents 协议。这样，普通 Durable Object 也能与客户端 SDK中的 useAgent 和 AgentClient 配合使用。

安装 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() 会将 /agents/room-agent/{name} 路由到名为 {name} 的 RoomAgent 对象。路由请求之前请先验证身份。更多信息请参阅路由请求。

WebSockets 会接管每个升级请求，即使没有处理程序也是如此。它是唯一的全捕获功能，因此 Lifecycle 会先将每个升级请求交给其他功能处理。没有 WebSockets 时，Lifecycle 会拒绝升级请求。普通 HTTP 请求仍会到达你的 onRequest()。

选项

选项	说明
`handlers`	连接处理程序：`onConnect`、`onMessage`、`onClose` 和 `onError`
`callables`	客户端可调用方法的 `RpcTarget`。更多信息请参阅可调用方法
`state`	用于与客户端同步的 `State` 功能
`getConnectionTags`	`(connection, ctx) => string[]`。连接接受时为其添加标签
`readonly`	`(connection, ctx) => boolean`。决定连接是否可以更改同步状态
`protocol`	`true`（默认值）、`false` 或 `(connection, ctx) => boolean`。决定连接是否接收 Agents 协议帧

在每个函数中，ctx.request 都是升级请求。

处理连接

处理程序在对象的调用上下文中运行，因此 getCurrentAgent() 会返回你的对象和连接：

处理程序	触发时机
`onConnect(connection, ctx)`	升级请求被接受后。`ctx.request` 是升级请求
`onMessage(connection, message)`	对于每条非 Agents 协议帧消息
`onClose(connection, code, reason, wasClean)`	连接关闭时
`onError(connection, error)`	连接在会话中途失败时

消息会唤醒休眠的对象。在执行 onMessage 之前，构造函数和 Lifecycle 启动流程会再次运行。需要在唤醒后保留的数据，请存入存储或连接中。

连接

Connection 是带有以下额外字段的 WebSocket：

字段	说明
`id`	连接 ID
`uri`	升级请求的 URL
`tags`	连接的标签。ID 始终是第一个标签
`state`	存储在此连接上的小型值，可在休眠期间保留
`setState(value)`	替换 `state`。传入函数可根据先前值更新；传入 `null` 可清除它

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() 不会与其他客户端同步。若要与所有客户端共享同一个值，请使用 State 功能。

查找连接

方法	说明
`getConnections(tag?)`	遍历每个开放连接，或遍历具有某个标签的连接
`getConnection(id)`	按 ID 返回一个连接；找不到时返回 `undefined`

为连接添加标签

传入 getConnectionTags，在连接被接受时为其添加标签。然后使用 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) 会为同一批连接添加另一组处理程序，例如用于你自己的协议。新增的处理程序会先于已配置的处理程序运行。新增的 onMessage 可以返回 true 来接管消息，这样后续处理程序就不会收到该消息：

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

可调用方法

使用可调用方法，让应用界面能够在代理上执行操作。例如，停止按钮可以调用服务器方法来取消运行。客户端通过现有连接发送参数，并接收方法的结果或错误。

在 RpcTarget 上声明这些操作，并将其作为 callables 传递给 WebSockets。客户端可通过 useAgent 和 AgentClient 中的 agent.stub 或 agent.call() 调用它们。

选择可调用方法

对于由应用客户端发起的操作（例如停止运行或更改代理设置），请使用可调用方法。
对于模型可以选择调用的操作，请使用工具。
当另一个 Worker 直接调用你的对象时，请使用 Durable Object RPC。

声明可调用方法

此示例为 Pi 代理添加 stopRun() 可调用方法。请使用该指南中的软件包安装步骤和绑定。请将身份验证放在路由 Worker 中，确保每位用户只能连接自己的代理。此示例对象中的所有会话都属于该用户。

stopRun() 是应用自定义的方法。它会验证会话 ID，然后请求 Pi 中止该会话正在运行的任务并撤回排队的输入。它不会撤销已完成的操作。

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

该目标就是完整的远程接口。客户端只能调用其类中声明的方法，不能调用 Durable Object 自身的方法，因此你可以精确决定对外开放哪些内容：

只能调用类原型上的方法。实例属性（包括赋给字段的箭头函数）不可调用。
继承自 Object 或 RpcTarget 的构造函数和方法永远不可调用。
如果类声明了名为 then 的方法，WebSockets 会抛出错误。

任何已连接的客户端都可以调用这些方法。请将所有参数视为不可信输入，并在请求到达对象之前检查调用者身份。更多信息请参阅身份验证。

从客户端调用

使用 agents/client 中的 AgentStub 和你的可调用方法类来指定 agent.stub 类型。每个方法都会变成一个异步函数，参数和返回类型保持一致：

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

将经过授权的代理名称和 Pi 返回的会话 ID 传递给 StopRunButton。从服务器文件导出 RunCallables，以便客户端导入其类型。import type 只引入类型，不会将服务器代码打包进客户端。

AgentStub<RunCallables> 会向 TypeScript 描述远程接口。运行时参数仍由 stopRun() 验证。

agent.call() 接受方法名称和参数数组，不提供类型信息：

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

agents/client 中的 AgentClient 也提供相同的 call() 和 stub。非流式调用默认在 30 秒后超时。向 call() 的第三个参数传入 { timeout }，可更改单次调用的超时时间；也可以在客户端上设置 defaultCallTimeout。将超时设为 0 可关闭超时。

方法抛出错误时，客户端 Promise 会以该错误消息拒绝。方法不存在时，Promise 会以 Method <name> does not exist 拒绝。

流式传输结果

返回 ReadableStream，即可在生成结果时逐步发送。每个分块一经读取就会到达客户端：

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();
			},
		});
	}
}

向 call() 传入流回调以读取分块：

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

流式调用没有默认超时时间。

上下文

每次调用都在对象的调用上下文中运行，并可访问调用连接。getCurrentAgent() 会返回你的对象和连接，因此方法可以回复调用方或向所有客户端广播：

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

传输方式

客户端通过 transport 选项选择调用的传输方式：

传输方式	调用的传输方式	支持休眠	结果
`"cf-websocket"`（默认）	在支持休眠的 WebSocket 上使用 JSON `rpc` 帧	是	必须可序列化为 JSON。`ReadableStream` 会流式传输。返回 `RpcTarget` 会产生错误。
`"capnweb"`	Cap'n Web ↗ 会话上的原生方法	否	返回的 `RpcTarget` 会作为实时存根传递给客户端，客户端可继续调用它。支持流和流水线调用。

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

相同的 callables 目标可服务于两种传输方式，onConnect 和 onMessage 等处理程序也会在两种方式下运行。Cap'n Web 连接不会休眠：只要连接保持打开，对象就会留在内存中；对象被逐出后，客户端会重新连接。除非需要通过引用传递结果，否则请使用默认传输方式。

同步状态

将 State 功能作为 state 传入，即可将一个值同步到 useAgent().state。连接建立时，每个客户端都会收到身份帧，然后收到当前值。客户端调用 setState() 时，变更会先经过状态验证器，再进行存储。示例请参阅与客户端同步状态。

WebSockets 不会自行广播变更。请在状态的 onChanged 钩子中调用 broadcastState(source)。它会将新值发送给除 source 之外的所有连接，因为来源连接已经拥有该值。

只读连接

只读连接仍可接收状态和调用方法，但其 setState() 调用会被拒绝。客户端会收到 cf_agent_state_error 帧，并运行其 onStateUpdateError 回调。

使用 readonly 选项决定何时接受连接，之后也可以通过 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;
	}
}

不要仅凭查询参数判断谁可以写入。路由请求之前，请在 Worker 中验证用户身份。

按连接控制协议

默认情况下，每个连接都会接收 Agents 协议帧：连接时发送身份帧，设置 state 后发送当前状态，并在状态变更时广播。useAgent 等客户端会等待身份帧，之后 ready 才会完成。

将 protocol 设为函数，即可对某些连接关闭这些帧，例如仅支持二进制数据的客户端。未启用协议的连接仍可发送和接收自定义消息，也可调用方法。在前面的示例中，带有 ?client=sensor 的连接不会收到协议帧。

设置 protocol: false 可自行控制连接序列。然后使用 sendConnectFrames() 发送帧，并通过 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);
}

方法

方法	说明
`getConnections(tag?)`	遍历开放连接；也可按标签筛选
`getConnection(id)`	返回一个开放连接；找不到时返回 `undefined`
`use(handlers)`	添加一组先于已配置处理程序运行的处理程序
`broadcastState(except?)`	将当前状态发送给所有启用协议且不等于 `except` 的连接
`sendConnectFrames(connection, identity?)`	发送身份帧；如果已存储状态，则随后发送当前状态
`sendIdentity(connection, identity?)`	仅发送身份帧。`identity` 默认为对象名称和其类名的 kebab-case 形式
`sendState(connection)`	将当前状态发送给单个连接
`applyStateFrame(connection, frame)`	应用客户端的状态帧，并拒绝只读连接。当 `frame` 是状态帧时返回 `true`
`isReadonly(connection)`	返回连接是否为只读
`setReadonly(connection, readonly?)`	将连接标记为只读；传入 `false` 可恢复写入权限
`isProtocolEnabled(connection)`	返回协议帧是否会发送到该连接
`setProtocolEnabled(connection, enabled)`	为单个连接启用或关闭协议帧

如果连接未启用协议，任何发送方法都不会向其发送内容。对象销毁时，WebSockets 会以代码 1001 关闭所有开放连接。

在 Agent 类上使用

Agent 类会安装 WebSockets，并调用自身的 onConnect()、onMessage()、onClose() 和 onError() 方法。它的 @callable() 方法就是客户端接口，无需另设 callables 目标。更多信息请参阅 WebSockets、Callable 方法和只读连接。