エージェントは HTTP リクエストを処理し、Server-Sent Events (SSE) で応答をストリーミングできます。このページでは onRequest メソッドと SSE のパターンを説明します。
onRequest メソッドを定義すると、エージェントへの HTTP リクエストを処理できます。
import { Agent } from "agents";
export class APIAgent extends Agent {
async onRequest(request) {
const url = new URL(request.url);
// Route based on path
if (url.pathname.endsWith("/status")) {
return Response.json({ status: "ok", state: this.state });
}
if (url.pathname.endsWith("/action")) {
if (request.method !== "POST") {
return new Response("Method not allowed", { status: 405 });
}
const data = await request.json();
await this.processAction(data.action);
return Response.json({ success: true });
}
return new Response("Not found", { status: 404 });
}
async processAction(action) {
// Handle the action
}
}import { Agent } from "agents";
export class APIAgent extends Agent {
async onRequest(request: Request): Promise<Response> {
const url = new URL(request.url);
// Route based on path
if (url.pathname.endsWith("/status")) {
return Response.json({ status: "ok", state: this.state });
}
if (url.pathname.endsWith("/action")) {
if (request.method !== "POST") {
return new Response("Method not allowed", { status: 405 });
}
const data = await request.json<{ action: string }>();
await this.processAction(data.action);
return Response.json({ success: true });
}
return new Response("Not found", { status: 404 });
}
async processAction(action: string) {
// Handle the action
}
}SSE を使うと、長時間続く HTTP 接続上でクライアントへデータをストリーミングできます。トークンを段階的に生成する AI モデルの応答に向いています。
ReadableStream で SSE ストリームを手動作成します。
export class StreamAgent extends Agent {
async onRequest(request) {
const encoder = new TextEncoder();
const stream = new ReadableStream({
async start(controller) {
// Send events
controller.enqueue(encoder.encode("data: Starting...\n\n"));
for (let i = 1; i <= 5; i++) {
await new Promise((r) => setTimeout(r, 500));
controller.enqueue(encoder.encode(`data: Step ${i} complete\n\n`));
}
controller.enqueue(encoder.encode("data: Done!\n\n"));
controller.close();
},
});
return new Response(stream, {
headers: {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",
},
});
}
}export class StreamAgent extends Agent {
async onRequest(request: Request): Promise<Response> {
const encoder = new TextEncoder();
const stream = new ReadableStream({
async start(controller) {
// Send events
controller.enqueue(encoder.encode("data: Starting...\n\n"));
for (let i = 1; i <= 5; i++) {
await new Promise((r) => setTimeout(r, 500));
controller.enqueue(encoder.encode(`data: Step ${i} complete\n\n`));
}
controller.enqueue(encoder.encode("data: Done!\n\n"));
controller.close();
},
});
return new Response(stream, {
headers: {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",
},
});
}
}SSE メッセージは次の形式に従います。
data: your message here\n\nイベント種別と ID も含められます。
event: update\n
id: 123\n
data: {"count": 42}\n\nAI SDK ↗ は SSE ストリーミングを組み込みで提供します。
import { Agent } from "agents";
import { streamText } from "ai";
import { createWorkersAI } from "workers-ai-provider";
export class ChatAgent extends Agent {
async onRequest(request) {
const { prompt } = await request.json();
const workersai = createWorkersAI({ binding: this.env.AI });
const result = streamText({
model: workersai("@cf/zai-org/glm-4.7-flash"),
prompt: prompt,
});
return result.toTextStreamResponse();
}
}import { Agent } from "agents";
import { streamText } from "ai";
import { createWorkersAI } from "workers-ai-provider";
interface Env {
AI: Ai;
}
export class ChatAgent extends Agent<Env> {
async onRequest(request: Request): Promise<Response> {
const { prompt } = await request.json<{ prompt: string }>();
const workersai = createWorkersAI({ binding: this.env.AI });
const result = streamText({
model: workersai("@cf/zai-org/glm-4.7-flash"),
prompt: prompt,
});
return result.toTextStreamResponse();
}
}SSE 接続は長時間続くことがあります。クライアント切断は次のように扱います。
- 進捗を永続化する — エージェント状態 に書き込み、クライアントが再開できるようにします
- エージェントルーティングを使う — クライアントはセッションストアなしで 同じエージェントインスタンスへ再接続 できます
- タイムアウト制限はない — Cloudflare Workers では、SSE 応答の継続時間に実効的な上限はありません
export class ResumeAgent extends Agent {
async onRequest(request) {
const url = new URL(request.url);
const lastEventId = request.headers.get("Last-Event-ID");
if (lastEventId) {
// Client is resuming - send events after lastEventId
return this.resumeStream(lastEventId);
}
return this.startStream();
}
async startStream() {
// Start new stream, saving progress to this.state
}
async resumeStream(fromId) {
// Resume from saved state
}
}export class ResumeAgent extends Agent {
async onRequest(request: Request): Promise<Response> {
const url = new URL(request.url);
const lastEventId = request.headers.get("Last-Event-ID");
if (lastEventId) {
// Client is resuming - send events after lastEventId
return this.resumeStream(lastEventId);
}
return this.startStream();
}
async startStream(): Promise<Response> {
// Start new stream, saving progress to this.state
}
async resumeStream(fromId: string): Promise<Response> {
// Resume from saved state
}
}| 機能 | WebSockets | SSE |
|---|---|---|
| 方向 | 双方向 | サーバー → クライアントのみ |
| プロトコル | ws:// / wss:// |
HTTP |
| バイナリデータ | はい | いいえ(テキストのみ) |
| 再接続 | 手動 | 自動(ブラウザー) |
| 向いている用途 | 対話型アプリ、チャット | ストリーミング応答、通知 |
推奨: 対話型アプリケーションには WebSockets を使います。AI 応答のストリーミングやサーバープッシュ通知には SSE を使います。
WebSocket のドキュメントは WebSockets を参照してください。
WebSockets
双方向のリアルタイム通信です。
状態管理
ストリームの進捗とエージェントの状態を永続化します。
チャットエージェントを構築する
AI チャットで応答をストリーミングします。