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.mdImport用途agents服务端 Agent 类与生命周期agents/reactuseAgent()Hook建立 WebSocket 连接与 RPCagents/ai-reactuseAgentChat()Hook构建 AI 聊天 UI模式一带工具调用的 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 AIChatAgentEnv { 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 AIChatAgentEnv { 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 ProcessingAgent基类内置任务队列queue/dequeue与定时调度schedule能力适合视频处理、日志清理、定期统计等后台任务。export class TaskAgent extends AgentEnv { 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.sqlDELETE 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 AgentEnv { 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区分AgentEnv, State, ConnState的第三个类型参数可为conn.state提供类型安全api.md。广播通过this.connections.forEach(...)向所有活跃连接推送消息。生命周期钩子onRequest处理 HTTP、onEmail处理邮件、onConnect/onMessage处理 WebSocket同一 Agent 可同时承载多种入口api.md 的 Lifecycle Hooks 章节。模式五AI 邮件处理Email Processing with AIAgent 可以通过onEmail()接收路由到 Worker 的邮件将邮件入库、用 LLM 生成摘要并实时推送给连接中的客户端还能按需自动回复。export class EmailAgent extends AgentEnv { async onEmail(email: AgentEmail) { const [text, from, subject] [await email.text(), email.from, email.headers.get(subject) || ]; this.sqlINSERT 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 AgentEnv { 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.mdimport { Agent, callable } from agents; export class MyAgent extends AgentEnv { 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 服务器的工具注入到streamTextawait 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。高频坑点速查症状原因解法setState()不同步直接修改了this.state始终用setState({...this.state, ...})不可变更新消息历史无限膨胀AIChatAgent的this.messages持续累积周期裁剪如this.messages this.messages.slice(-50)SQL 注入字符串插值拼接使用参数化模板sqlWHERE id ${id}而非${id}WebSocket 超时onConnect中未调用conn.accept()连接后立即conn.accept()超出调度上限每个 Agent 超 1000 个定时任务用getSchedules()监控、清理已完成任务可恢复流不生效流 ID 不固定直接用AIChatAgent其自动处理Agent not foundDO 绑定缺失或类名不匹配核对wrangler.jsonc的 binding 与 class_name关键限额参考CPU 每请求30s标准/ 300s最大每实例内存128MB与 WebSocket 共享每 Agent 存储10GBSQLite定时任务每 Agent 1000 个SQL 每表列数100行大小上限 2MBWebSocket 单条消息32MiB单 DO 实例请求速率约 1000 req/s高流量需自行限流完整限额表见 gotchas.md。总结如何选择你的模式场景推荐模式核心类AI 聊天 工具模式一AIChatAgent需要人类确认的工具模式二execute: clientAIChatAgent React后台批量任务/定时处理模式三queue scheduleAgent自定义实时协议模式四手动 WSAgent邮件摘要/自动回复模式五onEmail AIAgent多人协作/游戏模式六广播 状态Agent这六大模式覆盖了 Cloudflare Agents SDK 绝大多数生产场景且可以自由组合例如AI 聊天 任务队列 定时清理就是一个完整的产品。进一步学习可参考仓库中的 api.md类与生命周期、configuration.mdWrangler 与路由与 gotchas.md坑点与限额它们与本文共同构成完整的 Agents SDK 参考手册。【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考