Chat SDK Messenger 示例用 Cloudflare Agents 构建 Telegram AI 机器人的完整实战指南【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agents导读本文基于当前仓库中的 chat-sdk-messenger 示例系统讲解如何在 Cloudflare Workers 上将一个Chat SDK messenger 运行时嵌入 Agents SDK 的Agent中用两个子 AgentSubagent分别承载 Chat SDK 状态存储与 Think 驱动的 AI 回复。示例以 Telegram 为具体适配器但整体架构与具体 Messenger 无关。读完本文你将掌握Chat SDK 与 Agents SDK 的集成边界、Webhook 入口与状态/AI 逻辑的分离方式、Think 托管 Fiber 的幂等回复与恢复策略以及如何将这套架构迁移到 Slack、Discord 等其他聊天平台。这个示例展示了什么Chat SDK是 Cloudflare 提供的一套多平台聊天运行时Telegram 只是本文示例选择的具体适配器。示例的核心价值在于证明同一个Chat()实例可以同时挂载多个 Messenger 适配器因此「入口 / 状态 / AI」三件套可以在不修改底层的前提下被复用到 Slack、Discord、Teams、Google Chat 等平台。示例中一共涉及三个 Agent 角色见 src/index.tsChatIngressAgent顶层入口 Agent拥有 Chat SDK 运行时与 Webhook 入口负责把 Telegram 事件规范化成 Chat SDK 的Thread与Message对象ThinkMessengerStateAgent状态子 Agent以 Agents SDK 子 Agent 的形式承载 Chat SDK 的订阅subscriptions、锁locks、队列queues、缓存cache与列表lists等基础设施状态ConversationAgent extends Think对话子 Agent按 Chat SDK 的thread.id为粒度持有 AI 消息历史并负责模型调用。关键设计原则是把 provider 相关的边界压到最窄Telegram 的配置与渲染只存在于入口层状态与 AI 行为保持跨平台可复用。目录结构速览示例代码严格按照上述边界拆分与 README 中的说明一一对应examples/chat-sdk-messenger/src/ admin/ # Admin 目录与回复任务展示辅助函数 demos/ # Chat SDK 演示菜单卡片、文件、Markdown 等 intelligence/ # Think 对话、消息转换、回复策略 provider/telegram.ts # Telegram webhook 设置 state # 由 cloudflare/think/messengers 提供后端 index.ts # 入口编排与 Chat SDK 事件接线 client.tsx # Admin 前端React menu.tsx # 菜单 / 按钮动作工程脚本定义在 package.jsonnpm start启动本地 Vite 开发服务器npm run deploy执行构建并部署npm test运行 Vitest 测试套件配置见 src/tests/vitest.config.ts。本地运行1. 安装依赖在仓库根目录执行npm install2. 创建 Telegram Bot 并配置环境变量通过 BotFather 创建 Telegram 机器人后复制环境变量模板cp examples/chat-sdk-messenger/.env.example examples/chat-sdk-messenger/.env填写以下三项对应 .env.exampleTELEGRAM_BOT_TOKENyour-bot-token-from-botfather TELEGRAM_WEBHOOK_SECRET_TOKENgenerate-a-random-secret TELEGRAM_BOT_USERNAMEyour_bot_username其中TELEGRAM_WEBHOOK_SECRET_TOKEN是必填项Telegram 会用它来为 Webhook 请求签名x-telegram-bot-api-secret-token请求头比对逻辑可在 packages/think/src/messengers/telegram.ts 的telegramSecretTokenVerifier中看到。在 src/index.ts 中这两个 token 缺一不可否则createBot()会直接抛错。3. 启动本地开发服务器npm startVite 插件默认会开启一个Quick Tunnel让 Telegram 能够访问你本地的 Webhook见 vite.config.ts 中的cloudflare({ tunnel: { autoStart: true } })。启动后 Vite 插件会打印一个公网trycloudflare.com域名。在浏览器中打开该 HTTPS 地址点击设置面板中的Set webhook here按钮即可完成 Webhook 指向。这个按钮在http://localhost下是禁用状态因为 Telegram 要求 Webhook 必须是 HTTPS URL。它会先调用 Telegram 的getWebhookInfo检查当前 Webhook 地址仅在需要时才执行setWebhook幂等逻辑见 src/provider/telegram.ts。也可以手动设置 Webhookcurl -X POST https://api.telegram.org/bot$TELEGRAM_BOT_TOKEN/setWebhook \ -H Content-Type: application/json \ -d { url: https://your-tunnel.example.com/webhooks/telegram, secret_token: $TELEGRAM_WEBHOOK_SECRET_TOKEN }Webhook 路径常量WEBHOOK_PATH /webhooks/telegram定义在 src/provider/telegram.ts。4. 与机器人互动私聊DM直接发消息即可获得 AI 回复发送/menu打开演示菜单发送/reset清空该线程的 AI 历史。群聊Group首次 机器人会让它订阅当前线程之后再次 机器人、或发送/ask ...即可触发 AI 回复/menu打开演示菜单/reset清空当前线程的 AI 历史。路由判定逻辑集中在 src/intelligence/messages.ts 的shouldRouteToAi()DM 一律走 AI群聊只有被 或以/ask开头才走 AI/menu与/reset优先拦截。5. 关于 Workers AI本示例使用 Workers AI。在 wrangler.jsonc 中ai: { binding: AI, remote: true }remote: true表示本地开发时的 AI 调用也会打到你的 Cloudflare 账户远程执行。模型在ConversationAgent中指定见下文的「Think 驱动的 AI 回复」。部署到生产环境1. 存储 Secretswrangler secret put TELEGRAM_BOT_TOKEN wrangler secret put TELEGRAM_WEBHOOK_SECRET_TOKEN wrangler secret put TELEGRAM_BOT_USERNAME2. 部署npm run deploy部署脚本是vite build wrangler deploy。打开部署后的 Worker 根路径可以查看管理面板点击Set webhook here即可把 Telegram Webhook 指向部署环境下的/webhooks/telegram路由。3. Worker 配置要点examples/chat-sdk-messenger/wrangler.jsonc 中的关键配置{ ai: { binding: AI, remote: true }, name: chat-sdk-messenger, main: src/index.ts, compatibility_date: 2026-06-11, compatibility_flags: [nodejs_compat], assets: { not_found_handling: single-page-application, run_worker_first: [/agents/*, /setup/*, /webhooks/*] }, durable_objects: { bindings: [ { name: ChatIngressAgent, class_name: ChatIngressAgent } ] }, migrations: [ { tag: v1, new_sqlite_classes: [ChatIngressAgent] } ], observability: { logs: { enabled: true } } }注意run_worker_first/webhooks/*Telegram Webhook、/setup/*Webhook 设置接口和/agents/*Agent RPC 路由必须优先交给 Worker 处理而不能回退到静态资源 SPA。架构深入一个入口 Agent 两个子 Agent整体数据流摘自 README 架构图Worker 只绑定顶层ChatIngressAgent一个 Durable Object其余全部通过子 Agent 路由到达ChatIngressAgent Chat({ adapters: { telegram } }) ThinkMessengerStateAgent # Chat SDK 基础设施状态 ConversationAgent # 每个 thread 的 Think 消息与模型调用子 Agent 的导出非常关键ConversationAgent与ThinkMessengerStateAgent都必须从 Worker 入口导出见 src/index.ts子 Agent 路由才能正确解析。ChatIngressAgent的运行时创建ChatIngressAgent在onStart()阶段创建一个 Chat SDK 运行时见 src/index.tsexport { ThinkMessengerStateAgent } from cloudflare/think/messengers; export class ChatIngressAgent extends Agent { onStart() { this.ensureAdminSchema(); try { this.bot this.createBot(); } catch (error) { this.botStartupError toError(error); } } private createBot() { return new Chat({ userName, adapters: { telegram }, state: createChatSdkState({ agent: ThinkMessengerStateAgent, keyShard: (key) shardTelegramStateKey(key, this.shardThread), shardKey: this.shardThread }), concurrency: { strategy: burst, debounceMs: 600 } }); } }值得注意的实现细节concurrency: { strategy: burst, debounceMs: 600 }Chat SDK 对每个线程的可见回复做 600ms 去抖的 burst 调度保证用户可见的回复按线程串行输出shardThreadsrc/index.ts把threadId按冒号拆分成provider:chatId前缀作为状态分片键shardTelegramStateKey只对dedupe:telegram:前缀的缓存键做分片见 packages/think/src/messengers/telegram.tsbot.registerSingleton()把 Chat SDK 运行时注册为单例供后续 Webhook 请求复用。ChatIngressAgent通过onRequest接收 Webhook 请求并转交给 Chat SDKsrc/index.tsasync onRequest(request: Request): PromiseResponse { const url new URL(request.url); if (request.method ! POST || url.pathname ! WEBHOOK_PATH) { return new Response(Not found, { status: 404 }); } const bot this.getBot(); if (bot instanceof Error) { return setupErrorResponse(bot); } return bot.webhooks.telegram(request, { waitUntil: (task: Promiseunknown) this.ctx.waitUntil(task) }); }事件接线方面示例注册了四类事件处理器src/index.tsonNewMention首次 即订阅线程、onDirectMessage、onSubscribedMessage与onAction处理菜单按钮与审批动作。适配其他 Messengerprovider 相关代码被刻意压缩得很小。要把这个示例移植到其他平台只需四步导入或创建另一个 Chat SDK 适配器把它加进createBot()的adapters对象把该 provider 的 Webhook 路径路由到同一个ChatIngressAgent仅在 provider 的 UX 有差异时调整菜单/动作。多 provider 入口可以共享完全相同的状态与 AI 子 Agentconst bot new Chat({ userName, adapters: { telegram, slack, discord }, state: createChatSdkState() });核心边界在于provider 适配器统一产出 Chat SDK 的Thread和Message对象此后ThinkMessengerStateAgent与ConversationAgent无需关心消息来自哪个平台。状态子 AgentChat SDK 状态的后端实现Chat SDK 的状态适配器由agents/chat-sdk包提供packages/agents/src/chat-sdk/index.ts 中的createChatSdkState()。本示例复用了 Think 的 messenger state agent 别名使其与一等公民的 Think messenger API 保持一致import { ThinkMessengerStateAgent } from cloudflare/think/messengers; import { createChatSdkState } from agents/chat-sdk;ThinkMessengerStateAgent定义在 packages/think/src/messengers/chat-sdk.ts它直接继承自agents/chat-sdk的ChatSdkStateAgentpackages/agents/src/chat-sdk/agent.ts。状态 Agent 承载的数据ChatSdkStateAgent是一个纯基础设施 Agent它只负责存储 Chat SDK 的订阅、锁、队列、缓存与列表全部落在 Durable Object 的 SQLite 中不拥有任何渠道人格、工具或推理逻辑。其migrate()方法创建五张表packages/agents/src/chat-sdk/agent.ts表用途关键方法chat_sdk_state_subscriptions线程订阅关系subscribe/unsubscribe/isSubscribedchat_sdk_state_locks每线程互斥锁含过期acquireLock/releaseLock/extendLock/forceReleaseLockchat_sdk_state_queue每线程消息队列enqueue/popQueue/queueDepthchat_sdk_state_cache通用 KV 缓存含 TTLcacheGet/cacheSet/cacheSetIfNotExists/cacheDeletechat_sdk_state_lists有序列表历史消息等listAppend/listGet所有带 TTL 的数据都会通过schedule()定时清理过期项cleanupExpired相关索引也在迁移时一并创建。与社区方案的对比社区包chat-state-cloudflare-do覆盖了通用 Workers 方案自行带一个 Durable Object 绑定作为 Chat SDK 状态适配器。本示例展示的是Agents SDK 版本如果你的应用里已经有一个 Agent那么 Chat SDK 状态完全可以放进子 Agent 中而无需单独的顶层绑定。这带来的直接好处是状态生命周期与父 Agent 的生命周期保持一致并天然获得 Agents SDK 的子 Agent 路由、SQLite 与调度能力。Think 驱动的 AI 回复AI 路径被刻意做得小而清晰examples/chat-sdk-messenger/src/intelligence/ conversation-agent.ts # ConversationAgent extends Think delivery.ts # 托管回复快照与失败策略 messages.ts # Chat SDK Message - AI SDK UIMessage 辅助函数RPC 安全的TextStreamCallback与 Telegram 投递辅助函数均来自cloudflare/think/messengers导出清单见 packages/think/src/messengers/index.ts。ConversationAgent每线程的 AI 会话// examples/chat-sdk-messenger/src/intelligence/conversation-agent.ts import { Think } from cloudflare/think; import type { ToolSet } from ai; export class ConversationAgent extends Think { override getModel() { return cf/moonshotai/kimi-k2.7-code; } override getSystemPrompt(): string { return [ You are a concise assistant replying inside a chat thread., Answer the users latest message directly., Use plain text or simple Markdown only., Do not expose hidden reasoning, tool calls, or internal state. ].join(\n); } override getTools(): ToolSet { return {}; } async resetConversation(): Promisevoid { await this.clearMessages(); } }ConversationAgent使用 Think 的messages/ Session 存储作为一个 Chat SDKthread.id的权威 AI 历史conversationNameForThread()直接返回thread.id。Chat SDK 自身的消息历史则被视作平台/事件历史作为后续回填的可选素材。回复路径Thinkchat()流式转 Chat SDK 流式 post回复路径使用 Think 的chat()RPC 流在一个托管 Fiber 中把文本增量转发给 Chat SDK 的流式 post API。入口处的enqueueConversationReplysrc/index.tsawait this.startFiber( AI_REPLY_FIBER_NAME, async (fiber) { fiber.stash(aiReplySnapshot(accepted, thread, message)); await this.answerWithConversationAgent(thread, message, fiber); }, { idempotencyKey: ai-reply:${thread.id}:${message.id}, metadata: { provider: telegram, threadId: thread.id, messageId: message.id }, waitForCompletion: true } );随后answerWithConversationAgent完成以下序列src/index.ts创建TextStreamCallback设置visibleSoftLimit: TELEGRAM_STREAM_SOFT_LIMIT先发起一个有界的 Chat SDK 流式 post作为第一条可见消息调用 Think 的agent.chat(toThinkUserMessage(message), callback)StreamCallback持续收集完整文本模型回合完成后把剩余文本通过splitTelegramMessageText切成多个 provider 安全的后续消息逐个 post所有可见投递工作完成后把 fiber 状态 stash 为completed。TextStreamCallback与 Telegram 软上限TextStreamCallback定义于 packages/think/src/messengers/delivery.ts它实现 AI SDK 的StreamCallback接口用TextSegmentJoiner汇总流式文本段并维护可见文本visibleText、软上限visibleSoftLimit与请求 idonStart事件中拿到 Think 的requestId等状态。Telegram 相关常量packages/think/src/messengers/telegram.tsexport const TELEGRAM_STREAM_SOFT_LIMIT 3_400; export const TELEGRAM_FOLLOWUP_CHUNK_LIMIT 3_500;软上限 3400 字符第一条可见消息流式发送到约 3400 字符即停止后续分块上限 3500 字符模型完成后剩余长文本用splitTelegramMessageTextpackages/think/src/messengers/telegram.ts优先按段落\n\n、其次换行、再其次空格的位置切块逐条 post。这样设计的原因是 Telegram 有单条消息长度上限短回复保持「直播」观感长回复避免触发message is not modified的 final-edit no-op 错误或单条消息截断。isExpectedTelegramFinalEditNooppackages/think/src/messengers/telegram.ts专门识别「已经达到软上限后再编辑产生 VALIDATION_ERROR / message is not modified」这类预期内的投递完成而不是模型取消。失败与恢复策略恢复逻辑有一套显式的可见策略见 src/index.ts 的onFiberRecovered与 src/intelligence/delivery.ts 的模式判定函数Fiber 阶段恢复动作accepted流式尚未开始用快照还原 Chat SDKThread/Message重放 AI 回复返回{ status: completed }结算托管 Fiberstreaming流式已开始发送一句简洁的「回复被中断」致歉消息同样将 Fiber 结算为 completedcompleted重复 Webhook 直接忽略快照由aiReplySnapshot()生成src/intelligence/delivery.ts序列化 Chat SDK 的thread.toJSON()与message.toJSON()配合bot.reviver()反序列化还原回复目标。重复 Webhook 通过幂等键ai-reply:${thread.id}:${message.id}复用已保留的 Fiber而不是启动第二次可见回复。失败模式的判定aiReplyFailureModesrc/intelligence/delivery.ts遵循以下规则预期内的 final-edit no-op → 视为投递完成返回null模型回合已完成但投递失败 → 标记为error终态不自动重试模型回合未完成但有部分可见文本 → 发apologize消息其余 →error。模型完成后completedModelTurn true的溢出分块失败仍是投递失败除非所有预期的可见块都已成功 post否则 Fiber 不能被视为完成。waitForCompletion: true的另一个作用是保持 Chat SDK handler 挂起直到托管 Fiber 达到终态从而每个 Chat SDK 线程同一时刻只有一条可见 AI 回复——既避免了持久化的 Webhook 接受绕过 Chat SDK 的 burst/debounce UX也避免了 Telegram 占位消息或流式消息的重叠。从流式错误中取消模型调用如果第一条可见流式 post 失败且不是预期 no-opanswerWithConversationAgent会从TextStreamCallback.requestId()拿到 Think 的请求 id然后调用agent.cancelChat(requestId, errorMessage)取消对话子 Agent 上的模型调用src/index.ts再让 callback 以错误结束。这保证了「可见侧投递失败」与「模型侧继续烧钱生成」之间的一致性。Admin 管理面板Worker 根路径会提供一个轻量管理面板前端见 src/client.tsx通过useAgent()连接父级ChatIngressAgent。它展示Telegram 配置状态与当前 Webhook 命令getSetupInfo()已经过 AI 路径的 Chat SDK 会话列表listConversations()SQLite 表chat_admin_conversations在onStart()时创建每个 Chat SDK 线程对应的ConversationAgent名称conversationName即thread.id选中会话的最近托管 AI 回复任务listReplyJobs()来自listFibers({ name: AI_REPLY_FIBER_NAME })选中ConversationAgent的 Think 聊天面板useAgentChat。Think 面板的设计约束Think 面板刻意设计为仅内部使用从浏览器发出的消息只进入 Think session用于检查、调试或引导不会 post 回 Telegram。把消息写回 Messenger 应始终是显式的渠道动作避免操作员消息、机器人消息与合成用户消息混在一起。面板还会显示紧凑的消息诊断信息role、短消息 id 与文本长度。这些字段便于判断一条意外的 assistant 消息究竟来自真实二次回合、内部管理 prompt还是重放/恢复的 messenger 回合。浏览器子 Agent 访问门禁浏览器访问ConversationAgent子 Agent 由ChatIngressAgent.onBeforeSubAgent()把关src/index.ts只允许 Admin 目录中已记录的会话名被访问其它一律 404。该目录由父 Agent 持有的 SQLite 表维护通过recordConversation()在每次入站消息时 upsert。生产环境行为要点README 的生产行为章节 把重试与恢复策略显式化便于移植到其他 provider幂等键为ai-reply:${thread.id}:${message.id}同一 Chat SDK 消息的 provider 重试复用已保留的 FiberwaitForCompletion: true保持 Chat SDK handler 挂起直到可见回复达到托管 Fiber 终态保留 per-thread burst/debounce 行为长模型回合可能超出 provider Webhook 超时若 Telegram 在原始回复仍运行于同一 isolate 时重试重复投递会加入活跃的托管 Fiber重启后重复投递观察保留状态要么直接返回要么执行恢复completed的重复投递被忽略可见回复已完成溢出分块失败仍是投递失败Fiber 只有在第一条可见流与所有后续分块都成功后才会完成interrupted的重复投递会恢复序列化的快照、执行同一恢复策略并在应用级恢复成功后调用resolveFiber()error与aborted的 Fiber 是终态本示例不自动重试生产机器人可以增加操作员命令、重试按钮或人工对账流程。未来的迭代可以在有「模型驱动写入」的审批 UX 之后接入 Chat SDK 的createChatTools。Telegram 行为细节Telegram 是内置适配器示例处理器围绕 Telegram 机器人 UX 编写私聊消息直接获得 AI 响应/menu与/reset除外群聊中首次 订阅线程已订阅的群聊线程中机器人只响应后续 或/ask开头的消息/menu打开 Chat SDK 演示菜单/reset清空当前 Chat SDK 线程的 Think 会话长 AI 回复拆分多条 Telegram 消息第一条流式发送至软上限剩余部分在模型回合完成后追加发送。命令识别用正则实现src/intelligence/messages.ts均支持bot用户名后缀如/askMyBot。扩展规模按租户路由父 Agent当前getIngressAgentName()辅助函数固定返回defaultsrc/index.ts。更大的应用可以在 Worker 边界验证 Webhook、解析 update 后按租户 / bot / chat 路由到不同的父 Agent 名称ChatIngressAgent:tenant-a ThinkMessengerStateAgent:telegram:-100123 ThinkMessengerStateAgent:slack:T123 ChatIngressAgent:tenant-b ThinkMessengerStateAgent:discord:987路由逻辑位于 Worker 顶层 fetch 中getAgentByName(env.ChatIngressAgent, getIngressAgentName(request))src/index.ts。已知限制与注意事项Telegram Webhook URL 必须公网可达。Quick Tunnel URL 是临时的tunnel 地址变化后需要重新点击Set webhook hereTELEGRAM_WEBHOOK_SECRET_TOKEN必填否则无法校验 Webhook 请求签名Telegram 回调数据callback data上限64 字节按钮 action id 应保持简短Telegram 机器人无法拉取完整历史聊天记录适配器历史仅限于机器人可见/可缓存的部分AI 路径只把 assistant 文本渲染进 Messenger推理过程、工具调用、工具结果与其他未知消息部分只在 Admin Think 面板中可见不会 post 进 Telegramdefault父 Agent 有意保持简单高流量机器人应考虑路由到更具体的父 Agent 名称Admin 面板属于开发/控制平面生产环境暴露前必须加真实认证。相关源码与文档索引示例说明文档examples/chat-sdk-messenger/README.md入口编排examples/chat-sdk-messenger/src/index.tsAI 会话 Agentexamples/chat-sdk-messenger/src/intelligence/conversation-agent.ts回复快照与失败/恢复策略examples/chat-sdk-messenger/src/intelligence/delivery.ts路由与消息转换examples/chat-sdk-messenger/src/intelligence/messages.tsTelegram Webhook 设置examples/chat-sdk-messenger/src/provider/telegram.tsChat SDK 状态适配器入口packages/agents/src/chat-sdk/index.tsChat SDK 状态 Agent 实现packages/agents/src/chat-sdk/agent.tsThink messenger 导出与别名packages/think/src/messengers/index.ts、packages/think/src/messengers/chat-sdk.tsTelegram 投递辅助软上限/分块/no-op 判定packages/think/src/messengers/telegram.tsTextStreamCallback实现packages/think/src/messengers/delivery.ts路由与快照策略测试examples/chat-sdk-messenger/src/tests/intelligence.test.tsWorker 配置examples/chat-sdk-messenger/wrangler.jsonc仓库为只读状态以上所有文件均可在当前仓库中直接查看运行与部署请遵循本文「本地运行」与「部署」两节的命令将本示例作为一个可独立部署的 Telegram AI 机器人起步模板。【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考