企业官网建设流程全解析

Cloudflare Agents SDK 实战模式：从 AI 聊天到实时协作的六大用例与最佳实践

【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills

导读

本文以 Cloudflare Agents SDK 官方文档中的 Patterns & Use Cases 为骨架，系统讲解如何在 Cloudflare 边缘构建有状态、全局分布的 AI Agent：从带工具调用的 AI 聊天、人工介入（Human-in-the-Loop）、任务队列与定时调度，到手动 WebSocket 聊天、AI 邮件处理与实时协作应用。你将掌握AIChatAgent与Agent两大类的核心编程模型，并结合本仓库的 agents-sdk 参考文档（API、配置、坑点）得到可直接落地的完整示例与生产级注意事项。该参考位于cloudflare-deploySkill 中，配套的 SKILL 决策树将其定位为"构建有状态 AI Agent"的入口（见 SKILL.md）。

前置知识：两个核心类与三类入口

Cloudflare Agents SDK 构建在 Durable Objects 之上，提供持久状态、WebSocket、SQL、调度与 AI 集成能力。使用前需要理解两个类：

• AIChatAgent：面向 AI 聊天场景，自带自动流式输出（auto-streaming）、消息历史管理、工具调用与可恢复流（resumable streaming）。
• Agent（基类）：提供完整控制力，用于自定义逻辑、WebSocket、邮件与 SQL，也是实时协作、任务队列、邮件处理等模式的基础。

SDK 提供了三个包入口（见 README.md）：

模式一：带工具调用的 AI 聊天（AI Chat with Tools）

这是最典型的入门模式：Agent 端调用streamText()返回流式响应，客户端用 React Hooks 渲染聊天界面。

服务端：定义模型与工具

在onChatMessage(onFinish)中调用this.streamText()，通过tools字段注册可执行工具：

import { AIChatAgent } from "agents"; import { openai } from "@ai-sdk/openai"; import { tool } from "ai"; import { z } from "zod"; export class ChatAgent extends AIChatAgent<Env> { async onChatMessage(onFinish) { return this.streamText({ model: openai("gpt-4"), messages: this.messages, // Auto-managed 自动维护的消息历史 tools: { getWeather: tool({ description: "Get current weather", parameters: z.object({ city: z.string() }), execute: async ({ city }) => `Weather in ${city}: Sunny, 72°F` }), searchDocs: tool({ description: "Search documentation", parameters: z.object({ query: z.string() }), execute: async ({ query }) => JSON.stringify( this.sql<{title, content}>`SELECT title, content FROM docs WHERE content LIKE ${'%' + query + '%'}` ) }) }, onFinish, }); } }

关键点说明：

• this.messages：由框架自动维护的对话历史，无需自行持久化；传入streamText()即可让模型感知上下文。
• 工具定义：每个工具包含description（帮助模型判断何时调用）、parameters（用 zod 声明参数结构，保证类型安全）与execute（实际执行逻辑）。
• 工具内可直接使用 SQL：如searchDocs在工具中直接查询 SQLite，体现了 Agent 状态、数据库与 AI 能力在同一对象上的融合。SQL 采用参数化模板字符串（sql`...LIKE ${'%' + query + '%'}`），可有效防止注入（详见后文坑点章节）。
• onFinish：用于将本轮响应写回this.messages，实现历史闭环。

客户端：React 聊天 UI

客户端通过useAgent()建立与 Agent 的连接，再用useAgentChat()获取聊天状态与操作函数：

import { useAgent } from "agents/react"; import { useAgentChat } from "agents/ai-react"; function ChatUI() { const agent = useAgent({ agent: "ChatAgent" }); const { messages, input, handleInputChange, handleSubmit, isLoading } = useAgentChat({ agent }); return ( <div> {messages.map(m => <div key={m.id}>{m.role}: {m.content}</div>)} <form onSubmit={handleSubmit}> <input value={input} onChange={handleInputChange} disabled={isLoading} /> <button disabled={isLoading}>Send</button> </form> </div> ); }

useAgentChat还暴露stop（停止生成）与clearHistory（清空历史），并支持maxSteps（最大工具迭代次数）与resume（断线自动恢复）等选项（见 api.md 的 Client Hooks 章节）。其status取值为"ready" | "submitted" | "streaming" | "error"，可用于渲染加载态。

模式二：人工介入（Human-in-the-Loop，客户端工具）

当工具执行需要人类确认（如付款、删除数据）时，可以让服务端只声明工具、由客户端负责执行：在工具定义中将execute设为字符串"client"，表示由客户端完成执行。

// Server export class ChatAgent extends AIChatAgent<Env> { async onChatMessage(onFinish) { return this.streamText({ model: openai("gpt-4"), messages: this.messages, tools: { confirmAction: tool({ description: "Ask user to confirm", parameters: z.object({ action: z.string() }), execute: "client", // Client-side execution }) }, onFinish, }); } }

客户端在useAgentChat中通过onToolCall拦截工具调用并返回结果：

// Client const { messages } = useAgentChat({ agent, onToolCall: async (toolCall) => { if (toolCall.toolName === "confirmAction") { return { confirmed: window.confirm(`Confirm: ${toolCall.args.action}?`) }; } } });

该模式的价值在于：敏感操作永远不离开用户控制。模型只负责决定"是否需要确认"，真正执行确认的是浏览器端，天然适合审批流、支付确认、高风险变更等场景。从 API 设计看，useAgentChat的onToolCall回调（api.md）与maxSteps配合，可以控制一次对话中工具迭代的轮数上限，防止无限循环。

模式三：任务队列与定时调度（Task Queue & Scheduled Processing）

Agent基类内置任务队列（queue/dequeue）与定时调度（schedule）能力，适合视频处理、日志清理、定期统计等后台任务。

export class TaskAgent extends Agent<Env> { onStart() { this.schedule("*/5 * * * *", "processQueue", {}); // Every 5 min 每 5 分钟 this.schedule("0 0 * * *", "dailyCleanup", {}); // Daily 每日 } async onRequest(req: Request) { await this.queue("processVideo", { videoId: (await req.json()).videoId }); return Response.json({ queued: true }); } async processQueue() { const tasks = await this.dequeue(10); for (const task of tasks) { if (task.name === "processVideo") await this.processVideo(task.data.videoId); } } async dailyCleanup() { this.sql`DELETE FROM logs WHERE created_at < ${Date.now() - 86400000}`; } }

要点拆解：

• onStart()：Agent 初始化/重启时执行，适合注册定时任务或建表。官方建议在onStart()而非onRequest()中创建 SQL 表（见 gotchas.md 的 Best Practices）。
• schedule()三种形式（见 api.md 的 Scheduling 章节）：
  • 指定日期：await this.schedule(new Date("2026-12-25"), "sendGreeting", {msg:"Hi"})
  • 延迟秒数：await this.schedule(60, "checkStatus", {})
  • Cron 表达式：await this.schedule("0 0 * * *", "dailyCleanup", {})
  • 取消调度：await this.cancelSchedule(scheduleId)
• 任务队列：queue(name, data)入队，dequeue(n)批量取出最多 n 个任务，按task.name分发处理。这是将慢操作（视频转码、批量导入）与请求路径解耦的标准手段。
• 配额警示：每个 Agent 最多 1000 个定时任务，建议用getSchedules()监控数量，接近上限时清理已完成任务（见 gotchas.md 的 Rate Limits 章节）。

模式四：手动 WebSocket 聊天（非 AI 自定义协议）

当业务协议不是"AI 对话"而是自定义实时协议时，直接使用Agent基类的onConnect/onMessage手动管理连接、状态与广播：

export class ChatAgent extends Agent<Env> { async onConnect(conn: Connection, ctx: ConnectionContext) { conn.accept(); conn.setState({userId: ctx.request.headers.get("X-User-ID") || "anon"}); conn.send(JSON.stringify({type: "history", messages: this.state.messages})); } async onMessage(conn: Connection, msg: WSMessage) { const newMsg = {userId: conn.state.userId, text: JSON.parse(msg as string).text, timestamp: Date.now()}; this.setState({messages: [...this.state.messages, newMsg]}); this.connections.forEach(c => c.send(JSON.stringify(newMsg))); } }

实现要点：

• conn.accept()必须调用：未调用会导致 WebSocket 连接超时（见 gotchas.md 的对应坑点）。
• 连接状态：conn.setState({...})存储单个连接的状态（如userId），与 Agent 全局状态this.state区分；Agent<Env, State, ConnState>的第三个类型参数可为conn.state提供类型安全（api.md）。
• 广播：通过this.connections.forEach(...)向所有活跃连接推送消息。
• 生命周期钩子：onRequest处理 HTTP、onEmail处理邮件、onConnect/onMessage处理 WebSocket，同一 Agent 可同时承载多种入口（api.md 的 Lifecycle Hooks 章节）。

模式五：AI 邮件处理（Email Processing with AI）

Agent 可以通过onEmail()接收路由到 Worker 的邮件，将邮件入库、用 LLM 生成摘要并实时推送给连接中的客户端，还能按需自动回复。

export class EmailAgent extends Agent<Env> { async onEmail(email: AgentEmail) { const [text, from, subject] = [await email.text(), email.from, email.headers.get("subject") || ""]; this.sql`INSERT INTO emails (from_addr, subject, body) VALUES (${from}, ${subject}, ${text})`; const { text: summary } = await generateText({ model: openai("gpt-4o-mini"), prompt: `Summarize: ${subject}\n\n${text}` }); this.connections.forEach(c => c.send(JSON.stringify({type: "new_email", from, summary}))); if (summary.includes("urgent")) await this.schedule(0, "sendAutoReply", { to: from }); } }

配套配置：要让邮件触达 Agent，需要开启邮件路由。代码侧在 Worker 入口导出email处理器并调用routeAgentEmail()：

import { routeAgentEmail } from "agents"; export default { fetch: (req: Request, env: Env) => routeAgent(req, env), email: (message: ForwardableEmailMessage, env: Env) => { return routeAgentEmail(message, env); } }

同时在 Cloudflare Dashboard 中将邮箱路由目标设置为 "Workers with Durable Objects" 并绑定对应 Worker（详见 configuration.md 的 Email Routing 章节）。本例还展示了调度与 AI 的组合：检测到 "urgent" 关键词后通过schedule(0, ...)立即触发自动回复任务。

模式六：实时协作（Real-time Collaboration）

利用 Durable Object 的强一致状态与 WebSocket 广播，可以构建多人在线协作应用。下面的GameAgent展示了玩家加入、计分与开局控制的完整流程：

export class GameAgent extends Agent<Env> { initialState = { players: [], gameStarted: false }; async onConnect(conn: Connection, ctx: ConnectionContext) { conn.accept(); const playerId = ctx.request.headers.get("X-Player-ID") || crypto.randomUUID(); conn.setState({ playerId }); const newPlayer = { id: playerId, score: 0 }; this.setState({...this.state, players: [...this.state.players, newPlayer]}); this.connections.forEach(c => c.send(JSON.stringify({type: "player_joined", player: newPlayer}))); } async onMessage(conn: Connection, msg: WSMessage) { const m = JSON.parse(msg as string); if (m.type === "move") { this.setState({ ...this.state, players: this.state.players.map(p => p.id === conn.state.playerId ? {...p, score: p.score + m.points} : p) }); this.connections.forEach(c => c.send(JSON.stringify({type: "player_moved", playerId: conn.state.playerId}))); } if (m.type === "start" && this.state.players.length >= 2) { this.setState({...this.state, gameStarted: true}); this.connections.forEach(c => c.send(JSON.stringify({type: "game_started"}))); } } }

该模式的关键设计：

• initialState：声明式定义初始状态，Agent 重启后从持久存储恢复。
• 不可变更新：所有状态变更都通过setState({...this.state, ...})生成新对象——这是 SDK 的硬性要求，直接修改this.state不会同步（见 gotchas.md）。
• 单写者语义：所有连接的消息汇聚到同一个 DO 实例，天然避免并发写入冲突，状态强一致。

部署与路由：让 Agent 对外可访问

上述所有 Agent 都需要在 Worker 入口路由并写入 wrangler 配置才能运行。

Wrangler 配置

在wrangler.jsonc中注册 Durable Object 绑定与迁移：

{ "name": "my-agents-app", "durable_objects": { "bindings": [ {"name": "MyAgent", "class_name": "MyAgent"} ] }, "migrations": [ {"tag": "v1", "new_sqlite_classes": ["MyAgent"]} ], "ai": { "binding": "AI" } }

migrations中的new_sqlite_classes声明 Agent 使用 SQLite 存储（这是 Agent 状态与this.sql的基础）；ai绑定使env.AI可用。完整配置与多 Agent 场景参见 configuration.md。

路由

推荐使用routeAgent帮助函数自动按 URL 模式路由：

import { routeAgent } from "agents"; export default { fetch(request: Request, env: Env) { return routeAgent(request, env); } }

多 Agent 可按路径前缀分发（如/chat→ChatAgent、/task→TaskAgent），高级场景也可用idFromName/idFromString手动构造确定性或随机 ID 再get(id).fetch(request)（configuration.md 的 Agent Routing 章节）。本地开发与上线分别使用npx wrangler dev与npx wrangler deploy，密钥通过npx wrangler secret put OPENAI_API_KEY注入。

生产级进阶：RPC、MCP 与常见坑点

用@callable()暴露类型安全的 RPC

除 WebSocket 外，Agent 还可以通过@callable()装饰器暴露方法，供客户端agent.method()直接调用（api.md）：

import { Agent, callable } from "agents"; export class MyAgent extends Agent<Env> { @callable() async processTask(input: {text: string}): Promise<{result: string}> { return { result: await this.env.AI.run("@cf/meta/llama-3.1-8b-instruct", {prompt: input.text}) }; } } // Client: const result = await agent.processTask({ text: "Hello" });

注意：@callable方法必须返回 JSON 可序列化值（普通对象/数组/原始类型），返回Date实例等对象会失败（gotchas.md）。

通过 MCP 接入第三方工具

SDK 支持 Model Context Protocol，可将外部 MCP 服务器的工具注入到streamText：

await this.mcp.registerServer("github", { url: env.MCP_SERVER_URL, auth: { type: "oauth", clientId: env.GITHUB_CLIENT_ID, clientSecret: env.GITHUB_CLIENT_SECRET } }); const tools = await this.mcp.getAITools(["github"]); return this.streamText({ model: openai("gpt-4"), messages: this.messages, tools, onFinish });

OAuth 凭据通过npx wrangler secret put GITHUB_CLIENT_ID等命令配置（configuration.md 的 MCP 章节）。MCP 连接不随 DO 休眠保留，需要在onStart()中重新注册（gotchas.md）。

高频坑点速查

关键限额参考

• CPU 每请求：30s（标准）/ 300s（最大）
• 每实例内存：128MB（与 WebSocket 共享）
• 每 Agent 存储：10GB（SQLite）
• 定时任务：每 Agent 1000 个
• SQL 每表列数：100，行大小上限 2MB
• WebSocket 单条消息：32MiB
• 单 DO 实例请求速率：约 1000 req/s，高流量需自行限流

完整限额表见 gotchas.md。

总结：如何选择你的模式

这六大模式覆盖了 Cloudflare Agents SDK 绝大多数生产场景，且可以自由组合（例如"AI 聊天 + 任务队列 + 定时清理"就是一个完整的产品）。进一步学习可参考仓库中的 api.md（类与生命周期）、configuration.md（Wrangler 与路由）与 gotchas.md（坑点与限额），它们与本文共同构成完整的 Agents SDK 参考手册。

【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills