HTTP 与 Server-Sent Events

Agent 可处理 HTTP 请求并使用 Server-Sent Events (SSE) 流式返回响应。本页涵盖 onRequest 方法与 SSE 模式。

处理 HTTP 请求

定义 onRequest 方法处理发往 agent 的 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
	}
}

Server-Sent Events (SSE)

SSE 允许经长连接 HTTP 向 client 流式发送数据。适合逐 token 生成的 AI model 响应。

手动 SSE

使用 ReadableStream 手动创建 SSE stream：

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 消息格式

SSE 消息遵循特定格式：

data: your message here\n\n

也可包含 event 类型与 ID：

event: update\n
id: 123\n
data: {"count": 42}\n\n

配合 AI SDK

AI 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 连接可能长寿命。优雅处理 client 断开：

持久化进度 — 写入 agent 状态以便 client resume
使用 agent 路由 — Client 可重连同一 agent 实例而无需 session store
无 timeout 限制 — 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
	}
}

WebSocket 与 SSE 对比

特性	WebSockets	SSE
方向	双向	仅 Server → Client
协议	`ws://` / `wss://`	HTTP
二进制数据	是	否（仅文本）
重连	手动	自动（browser）
适用场景	交互应用、聊天	流式响应、通知

建议： 交互应用用 WebSocket。流式 AI 响应或 server-push 通知用 SSE。

WebSocket 文档请参阅 WebSockets。

后续步骤

WebSockets

双向实时通信。

状态管理

持久化流进度与 agent 状态。

构建 chat agent

AI 聊天的流式响应。