エージェントは、リアルタイムの双方向通信向けに WebSocket 接続をサポートします。このページではサーバー側の WebSocket 処理を説明します。クライアント側の接続は Client SDK を参照してください。
エージェントには、タイミングごとに発火するライフサイクルフックがあります。
| フック | 呼び出されるタイミング |
|---|---|
onStart(props?) |
エージェントの初回起動時に 1 度(接続の前) |
onRequest(request) |
HTTP リクエストを受信したとき(WebSocket 以外) |
onConnect(connection, ctx) |
新しい WebSocket 接続が確立したとき |
onMessage(connection, message) |
WebSocket メッセージを受信したとき |
onClose(connection, code, reason, wasClean) |
WebSocket 接続が閉じたとき |
onError(connection, error) |
接続上で WebSocket エラーが発生したとき |
onError(error) |
サーバーレベルのエラーが発生したとき(特定の接続に紐づかない) |
shouldSendProtocolMessages(connection, ctx) |
この接続にプロトコルメッセージ(identity、state、MCP)を送るかどうか。デフォルト: true |
onStart() は、エージェントの初回起動時に、接続が確立する前に 1 度呼び出されます。
export class MyAgent extends Agent {
async onStart() {
// Initialize resources
console.log(`Agent ${this.name} starting...`);
// Load data from storage
const savedData = this.sql`SELECT * FROM cache`;
for (const row of savedData) {
// Rebuild in-memory state from persistent storage
}
}
onConnect(connection) {
// By the time connections arrive, onStart has completed
}
}export class MyAgent extends Agent {
async onStart() {
// Initialize resources
console.log(`Agent ${this.name} starting...`);
// Load data from storage
const savedData = this.sql`SELECT * FROM cache`;
for (const row of savedData) {
// Rebuild in-memory state from persistent storage
}
}
onConnect(connection: Connection) {
// By the time connections arrive, onStart has completed
}
}WebSocket 接続を受け付けるには、Agent に onConnect と onMessage メソッドを定義します。
import { Agent, Connection, ConnectionContext, WSMessage } from "agents";
export class ChatAgent extends Agent {
async onConnect(connection, ctx) {
// Connections are automatically accepted
// Access the original request for auth, headers, cookies
const url = new URL(ctx.request.url);
const token = url.searchParams.get("token");
if (!token) {
connection.close(4001, "Unauthorized");
return;
}
// Store user info on this connection
connection.setState({ authenticated: true });
}
async onMessage(connection, message) {
if (typeof message === "string") {
// Handle text message
const data = JSON.parse(message);
connection.send(JSON.stringify({ received: data }));
}
}
}import { Agent, Connection, ConnectionContext, WSMessage } from "agents";
export class ChatAgent extends Agent {
async onConnect(connection: Connection, ctx: ConnectionContext) {
// Connections are automatically accepted
// Access the original request for auth, headers, cookies
const url = new URL(ctx.request.url);
const token = url.searchParams.get("token");
if (!token) {
connection.close(4001, "Unauthorized");
return;
}
// Store user info on this connection
connection.setState({ authenticated: true });
}
async onMessage(connection: Connection, message: WSMessage) {
if (typeof message === "string") {
// Handle text message
const data = JSON.parse(message);
connection.send(JSON.stringify({ received: data }));
}
}
}接続中の各クライアントには、一意の Connection オブジェクトがあります。
| プロパティ / メソッド | 型 | 説明 |
|---|---|---|
id |
string |
この接続の一意な識別子 |
uri |
string | null |
元の WebSocket アップグレードリクエストの URL。ハイバネーションをまたいで保持されます |
state |
State |
接続ごとの状態オブジェクト |
setState(state) |
void |
接続の状態を更新します |
send(message) |
void |
このクライアントへメッセージを送信します |
close(code?, reason?) |
void |
接続を閉じます |
tags |
readonly string[] |
getConnectionTags で割り当てたタグ。先頭のタグは常に接続 ID です |
server |
string |
エージェントインスタンス名(Agent 上の this.name と同じ) |
各接続固有のデータ(ユーザー情報、設定など)を保存します。
export class ChatAgent extends Agent {
async onConnect(connection, ctx) {
const userId = new URL(ctx.request.url).searchParams.get("userId");
connection.setState({
userId: userId || "anonymous",
role: "user",
joinedAt: Date.now(),
});
}
async onMessage(connection, message) {
// Access connection-specific state
console.log(`Message from ${connection.state.userId}`);
}
}interface ConnectionState {
userId: string;
role: "admin" | "user";
joinedAt: number;
}
export class ChatAgent extends Agent {
async onConnect(
connection: Connection<ConnectionState>,
ctx: ConnectionContext,
) {
const userId = new URL(ctx.request.url).searchParams.get("userId");
connection.setState({
userId: userId || "anonymous",
role: "user",
joinedAt: Date.now(),
});
}
async onMessage(connection: Connection<ConnectionState>, message: WSMessage) {
// Access connection-specific state
console.log(`Message from ${connection.state.userId}`);
}
}接続中の全クライアントへメッセージを送るには this.broadcast() を使います。
export class ChatAgent extends Agent {
async onMessage(connection, message) {
// Broadcast to all connected clients
this.broadcast(
JSON.stringify({
from: connection.id,
message: message,
timestamp: Date.now(),
}),
);
}
// Broadcast from any method
async notifyAll(event, data) {
this.broadcast(JSON.stringify({ event, data }));
}
}export class ChatAgent extends Agent {
async onMessage(connection: Connection, message: WSMessage) {
// Broadcast to all connected clients
this.broadcast(
JSON.stringify({
from: connection.id,
message: message,
timestamp: Date.now(),
}),
);
}
// Broadcast from any method
async notifyAll(event: string, data: unknown) {
this.broadcast(JSON.stringify({ event, data }));
}
}ブロードキャストから除外する接続 ID の配列を渡します。
// Broadcast to everyone except the sender
this.broadcast(
JSON.stringify({ type: "user-typing", userId: "123" }),
[connection.id], // Do not send to the originator
);// Broadcast to everyone except the sender
this.broadcast(
JSON.stringify({ type: "user-typing", userId: "123" }),
[connection.id], // Do not send to the originator
);接続にタグを付けて、フィルタしやすくします。接続確立時にタグを割り当てるには getConnectionTags() をオーバーライドします。
export class ChatAgent extends Agent {
getConnectionTags(connection, ctx) {
const url = new URL(ctx.request.url);
const role = url.searchParams.get("role");
const tags = [];
if (role === "admin") tags.push("admin");
if (role === "moderator") tags.push("moderator");
return tags; // Up to 9 tags, max 256 chars each
}
// Later, broadcast only to admins
notifyAdmins(message) {
for (const conn of this.getConnections("admin")) {
conn.send(message);
}
}
}export class ChatAgent extends Agent {
getConnectionTags(connection: Connection, ctx: ConnectionContext): string[] {
const url = new URL(ctx.request.url);
const role = url.searchParams.get("role");
const tags: string[] = [];
if (role === "admin") tags.push("admin");
if (role === "moderator") tags.push("moderator");
return tags; // Up to 9 tags, max 256 chars each
}
// Later, broadcast only to admins
notifyAdmins(message: string) {
for (const conn of this.getConnections("admin")) {
conn.send(message);
}
}
}| メソッド | シグネチャ | 説明 |
|---|---|---|
getConnections |
(tag?: string) => Iterable<Connection> |
すべての接続を取得します。タグで絞り込むこともできます |
getConnection |
(id: string) => Connection | undefined |
ID で接続を取得します |
getConnectionTags |
(connection, ctx) => string[] |
オーバーライドして接続にタグを付けます |
broadcast |
(message, without?: string[]) => void |
すべての接続へ送信します |
isConnectionReadonly |
(connection) => boolean |
接続が readonly かどうかを確認します |
isConnectionProtocolEnabled |
(connection) => boolean |
この接続でプロトコルメッセージが有効かどうかを確認します |
メッセージは文字列、またはバイナリ(ArrayBuffer / ArrayBufferView)です。
export class FileAgent extends Agent {
async onMessage(connection, message) {
if (message instanceof ArrayBuffer) {
// Handle binary upload
const bytes = new Uint8Array(message);
await this.processFile(bytes);
connection.send(
JSON.stringify({ status: "received", size: bytes.length }),
);
} else if (typeof message === "string") {
// Handle text command
const command = JSON.parse(message);
// ...
}
}
}export class FileAgent extends Agent {
async onMessage(connection: Connection, message: WSMessage) {
if (message instanceof ArrayBuffer) {
// Handle binary upload
const bytes = new Uint8Array(message);
await this.processFile(bytes);
connection.send(
JSON.stringify({ status: "received", size: bytes.length }),
);
} else if (typeof message === "string") {
// Handle text command
const command = JSON.parse(message);
// ...
}
}
}接続エラーと切断を処理します。onError メソッドには 2 つのオーバーロードがあります。WebSocket 接続エラー用と、サーバーレベルエラー用です。
export class ChatAgent extends Agent {
// WebSocket connection error
// Server-level error (not tied to a specific connection)
onError(connectionOrError, error) {
if (error) {
console.error(`Connection ${connectionOrError.id} error:`, error);
} else {
console.error("Server error:", connectionOrError);
}
}
async onClose(connection, code, reason, wasClean) {
console.log(`Connection ${connection.id} closed: ${code} ${reason}`);
this.broadcast(
JSON.stringify({
event: "user-left",
userId: connection.state?.userId,
}),
);
}
}export class ChatAgent extends Agent {
// WebSocket connection error
onError(connection: Connection, error: unknown): void;
// Server-level error (not tied to a specific connection)
onError(error: unknown): void;
onError(connectionOrError: Connection | unknown, error?: unknown) {
if (error) {
console.error(
`Connection ${(connectionOrError as Connection).id} error:`,
error,
);
} else {
console.error("Server error:", connectionOrError);
}
}
async onClose(
connection: Connection,
code: number,
reason: string,
wasClean: boolean,
) {
console.log(`Connection ${connection.id} closed: ${code} ${reason}`);
this.broadcast(
JSON.stringify({
event: "user-left",
userId: connection.state?.userId,
}),
);
}
}デフォルトの onError 実装はエラーをログに出して再スローします。カスタムのエラー処理、レポート、回復ロジックを追加するにはオーバーライドします。
| 種類 | 説明 |
|---|---|
string |
テキストメッセージ(通常は JSON) |
ArrayBuffer |
バイナリデータ |
ArrayBufferView |
バイナリデータの型付き配列ビュー |
エージェントはハイバネーションに対応します。非アクティブ時にスリープし、必要になったら起動します。WebSocket 接続を維持したままリソースを節約できます。
ハイバネーションはデフォルトで有効です。無効にするには次のようにします。
export class AlwaysOnAgent extends Agent {
static options = { hibernate: false };
}export class AlwaysOnAgent extends Agent {
static options = { hibernate: false };
}- エージェントはアクティブで、接続を処理しています
- メッセージのない非アクティブ期間のあと、エージェントはハイバネーション(スリープ)します
- WebSocket 接続は開いたままです(Cloudflare が処理します)
- メッセージが届くと、エージェントが起動します
- 通常どおり
onMessageが呼び出されます
| 保持される | 保持されない |
|---|---|
this.state(エージェントの状態) |
メモリ上の変数 |
connection.state |
タイマー / インターバル |
SQLite データ(this.sql) |
進行中の Promise |
| 接続のメタデータ | ローカルキャッシュ |
重要なデータはクラスプロパティではなく、this.state または SQLite に保存します。
export class MyAgent extends Agent {
initialState = { counter: 0 };
// Do not do this - lost on hibernation
localCounter = 0;
onMessage(connection, message) {
// Persists across hibernation
this.setState({ counter: this.state.counter + 1 });
// Lost after hibernation
this.localCounter++;
}
}export class MyAgent extends Agent<Env, { counter: number }> {
initialState = { counter: 0 };
// Do not do this - lost on hibernation
private localCounter = 0;
onMessage(connection: Connection, message: WSMessage) {
// Persists across hibernation
this.setState({ counter: this.state.counter + 1 });
// Lost after hibernation
this.localCounter++;
}
}接続ごとの状態で、オンラインのユーザーを追跡します。ユーザーが切断すると、接続状態は自動でクリーンアップされます。
export class PresenceAgent extends Agent {
onConnect(connection, ctx) {
const url = new URL(ctx.request.url);
const name = url.searchParams.get("name") || "Anonymous";
connection.setState({
name,
joinedAt: Date.now(),
lastSeen: Date.now(),
});
// Send current presence to new user
connection.send(
JSON.stringify({
type: "presence",
users: this.getPresence(),
}),
);
// Notify others that someone joined
this.broadcastPresence();
}
onClose(connection) {
// No manual cleanup needed - connection state is automatically gone
this.broadcastPresence();
}
onMessage(connection, message) {
if (message === "ping") {
connection.setState((prev) => ({
...prev,
lastSeen: Date.now(),
}));
connection.send("pong");
}
}
getPresence() {
const users = {};
for (const conn of this.getConnections()) {
if (conn.state) {
users[conn.id] = {
name: conn.state.name,
lastSeen: conn.state.lastSeen,
};
}
}
return users;
}
broadcastPresence() {
this.broadcast(
JSON.stringify({
type: "presence",
users: this.getPresence(),
}),
);
}
}type UserState = {
name: string;
joinedAt: number;
lastSeen: number;
};
export class PresenceAgent extends Agent {
onConnect(connection: Connection<UserState>, ctx: ConnectionContext) {
const url = new URL(ctx.request.url);
const name = url.searchParams.get("name") || "Anonymous";
connection.setState({
name,
joinedAt: Date.now(),
lastSeen: Date.now(),
});
// Send current presence to new user
connection.send(
JSON.stringify({
type: "presence",
users: this.getPresence(),
}),
);
// Notify others that someone joined
this.broadcastPresence();
}
onClose(connection: Connection) {
// No manual cleanup needed - connection state is automatically gone
this.broadcastPresence();
}
onMessage(connection: Connection<UserState>, message: WSMessage) {
if (message === "ping") {
connection.setState((prev) => ({
...prev!,
lastSeen: Date.now(),
}));
connection.send("pong");
}
}
private getPresence() {
const users: Record<string, { name: string; lastSeen: number }> = {};
for (const conn of this.getConnections<UserState>()) {
if (conn.state) {
users[conn.id] = {
name: conn.state.name,
lastSeen: conn.state.lastSeen,
};
}
}
return users;
}
private broadcastPresence() {
this.broadcast(
JSON.stringify({
type: "presence",
users: this.getPresence(),
}),
);
}
}export class ChatRoom extends Agent {
onConnect(connection, ctx) {
const url = new URL(ctx.request.url);
const username = url.searchParams.get("username") || "Anonymous";
connection.setState({ username });
// Notify others
this.broadcast(
JSON.stringify({
type: "join",
user: username,
timestamp: Date.now(),
}),
[connection.id], // Do not send to the joining user
);
}
onMessage(connection, message) {
if (typeof message !== "string") return;
const { username } = connection.state;
this.broadcast(
JSON.stringify({
type: "message",
user: username,
text: message,
timestamp: Date.now(),
}),
);
}
onClose(connection) {
const { username } = connection.state || {};
if (username) {
this.broadcast(
JSON.stringify({
type: "leave",
user: username,
timestamp: Date.now(),
}),
);
}
}
}type Message = {
type: "message" | "join" | "leave";
user: string;
text?: string;
timestamp: number;
};
export class ChatRoom extends Agent {
onConnect(connection: Connection, ctx: ConnectionContext) {
const url = new URL(ctx.request.url);
const username = url.searchParams.get("username") || "Anonymous";
connection.setState({ username });
// Notify others
this.broadcast(
JSON.stringify({
type: "join",
user: username,
timestamp: Date.now(),
} satisfies Message),
[connection.id], // Do not send to the joining user
);
}
onMessage(connection: Connection, message: WSMessage) {
if (typeof message !== "string") return;
const { username } = connection.state as { username: string };
this.broadcast(
JSON.stringify({
type: "message",
user: username,
text: message,
timestamp: Date.now(),
} satisfies Message),
);
}
onClose(connection: Connection) {
const { username } = (connection.state as { username: string }) || {};
if (username) {
this.broadcast(
JSON.stringify({
type: "leave",
user: username,
timestamp: Date.now(),
} satisfies Message),
);
}
}
}デフォルトでは、エージェントはすべての接続へ JSON テキストフレーム(identity、状態同期、MCP サーバー一覧)を送ります。特定の接続で抑制するには shouldSendProtocolMessages をオーバーライドします。JSON テキストフレームを扱えないバイナリ専用クライアントなどが対象です。
export class IoTAgent extends Agent {
shouldSendProtocolMessages(connection, ctx) {
const url = new URL(ctx.request.url);
return url.searchParams.get("protocol") !== "binary";
}
}export class IoTAgent extends Agent {
shouldSendProtocolMessages(
connection: Connection,
ctx: ConnectionContext,
): boolean {
const url = new URL(ctx.request.url);
return url.searchParams.get("protocol") !== "binary";
}
}false を返すと、接続時にもブロードキャスト経由でも、identity、state、MCP サーバー一覧のフレームは届きません。通常のメッセージの送受信、RPC、プロトコル以外の通信には参加できます。
実行時に接続の状態を確認するには isConnectionProtocolEnabled(connection) を使います。
次のプロパティは、任意の Agent メソッド内の this で使えます。
| プロパティ | 型 | 説明 |
|---|---|---|
this.name |
string |
このエージェントのインスタンス名 |
this.state |
State |
現在のエージェント状態(SQLite から遅延読み込み) |
this.env |
Env |
Worker の環境バインディング |
this.ctx |
DurableObjectState |
Durable Object のコンテキスト(ストレージ、アラームなど) |
this.sql |
template tag | エージェントの SQLite ストレージに対してクエリを実行する SQL テンプレートタグ |
this.mcp |
MCPClientManager |
外部 MCP サーバーへ接続するための MCP クライアントマネージャー |
ブラウザーからの接続には、Agents のクライアント SDK を使います。
- Vanilla JS:
agents/clientのAgentClient - React:
agents/reactのuseAgentフック
詳細は Client SDK を参照してください。