1. 项目概述当OpenClaw遇见Telegram最近在折腾一个挺有意思的事儿把OpenClaw这个智能体框架接入了Telegram。简单来说就是让你在Telegram里能像跟一个真人助手聊天一样指挥OpenClaw去完成各种任务比如查资料、写代码、管理服务器甚至控制智能家居。这听起来像是给OpenClaw装上了翅膀让它从一个运行在命令行或Web界面的工具变成了一个随时待命的“随身助理”。为什么是Telegram它的Bot API设计得非常清晰、稳定而且用户基数庞大几乎成了各种自动化工具和机器人服务的首选对接平台。BotFather是Telegram官方提供的“机器人制造机”所有Telegram机器人的创建、配置、管理都从这里开始。所以这个项目的核心就是通过BotFather创建一个机器人拿到唯一的身份凭证Bot Token然后让OpenClaw能够接收和处理这个机器人收到的所有消息并做出智能响应。整个过程就像是在OpenClaw和Telegram之间架起一座双向通信的桥梁。我踩过不少坑从BotToken的获取与保管到Webhook的设置与调试再到OpenClaw侧消息适配器的编写每一步都有需要注意的细节。这篇文章我就把这次实战的完整过程、核心原理和避坑经验梳理出来目标是让你看完就能自己动手搭一个并且知道每一步为什么要这么做。2. 核心思路与架构设计2.1 为什么选择Webhook模式Telegram Bot提供了两种消息获取方式长轮询Long Polling和Webhook。长轮询就是你的服务端程序不停地、主动地向Telegram服务器询问“有没有新消息给我”。这种方式实现简单适合开发和测试初期因为不需要公网IP。但它有个致命缺点延迟和资源消耗。你的服务器需要维持一个持续的连接去“问”如果问得太频繁可能会被限流问得不频繁消息就有延迟。而Webhook则是一种“推送”模式。你告诉Telegram服务器一个公开的URL地址你的服务端接口。一旦有用户给你的机器人发送消息Telegram服务器会主动将这个消息打包成一个HTTP POST请求发送到你指定的这个URL上。你的服务端只需要监听这个接口收到请求后解析、处理、再回复即可。对于OpenClaw这种需要实时交互、且可能部署在自有服务器拥有公网IP或通过内网穿透暴露服务的场景Webhook是生产环境的不二之选。它实时性高Telegram服务器一收到消息就立刻推过来资源消耗低你的服务端是被动接收无需维持主动查询连接也更可靠减少了因网络波动导致轮询失败的消息丢失风险。因此我们的架构就基于Webhook来构建。2.2 整体数据流与组件职责整个系统的数据流可以清晰地分为几个阶段用户触发用户在Telegram中向你的Bot发送一条消息或命令。Telegram推送Telegram服务器将这条消息封装成一个结构化的JSON数据包通过HTTPS POST请求发送到你预先设置好的Webhook URL。接收与路由你的服务端承载OpenClaw有一个HTTP服务比如用FastAPI、Flask搭建在监听Webhook URL。它接收到请求后首先进行安全验证验证请求是否真的来自Telegram然后解析JSON提取出关键信息用户ID、聊天ID、消息内容等。OpenClaw处理服务端将提取出的消息内容按照OpenClaw能理解的格式进行封装然后调用OpenClaw的核心处理引擎。OpenClaw根据其内部配置的技能Skills和记忆Memory来理解用户意图执行相应操作可能是调用一个API、运行一段脚本、查询数据库等并生成一个文本或包含图片、文件的响应。响应回传服务端拿到OpenClaw的响应后将其再次封装成Telegram Bot API要求的格式通过调用Telegram的sendMessage等接口将消息发送回对应的聊天窗口。用户接收用户在Telegram中看到机器人的回复。在这个流程中我们的核心开发工作集中在第3、4、5步即构建一个“适配器”它既能听懂Telegram的话解析Webhook又能让OpenClaw听懂转换消息格式最后还要把OpenClaw的“话”翻译成Telegram能发送的格式。注意Webhook URL必须是HTTPS的。这意味着你的服务端必须配置SSL证书。在开发测试阶段你可以使用内网穿透工具如ngrok、localtunnel来获得一个临时的HTTPS地址或者在自己的服务器上配置好SSL。Telegram不会向HTTP地址发送Webhook。3. 实战第一步从BotFather获取通行证3.1 创建你的第一个Telegram Bot一切始于BotFather。它本身就是一个Telegram Bot你需要先在Telegram应用中搜索并打开与它的对话。启动对话在Telegram中搜索BotFather点击打开聊天窗口。创建新Bot向它发送命令/newbot。设置名称BotFather会问你给机器人起什么名字。这个名字是用户在聊天列表中看到的名字比如“我的智能助手OpenClaw”。可以包含空格和特殊字符。设置用户名接下来需要设置一个唯一的用户名Username必须以bot结尾例如my_openclaw_assistant_bot。这个用户名是唯一的用于在Telegram中通过username找到你的Bot也是API调用的一部分。获取Token创建成功后BotFather会发给你一串重要的信息——HTTP API Token。它长得像这样1234567890:ABCdefGhIJKlmNoPQRsTUVwxyZ-abc123def456。这串Token就是你的Bot的“密码”和“身份证”。任何人拿到这串Token就完全控制了你的Bot。所以请立即将其妥善保存并绝对不要提交到公开的代码仓库如GitHub。最佳实践是将其设置为环境变量。3.2 Token的安全管理与环境变量配置在服务器上我们可以这样操作# 将Token写入一个只有当前用户可读的环境变量文件 echo export TELEGRAM_BOT_TOKEN你的_Actual_Token_Here ~/.bashrc # 或者为了更安全使用系统级服务环境文件或像下面这样临时设置 export TELEGRAM_BOT_TOKEN你的_Actual_Token_Here # 然后让配置生效 source ~/.bashrc在你的OpenClaw服务端代码中通过os.environ.get(TELEGRAM_BOT_TOKEN)来读取它。这样Token就不会硬编码在代码里。3.3 初步验证与基础信息设置拿到Token后你可以立即用它来测试Bot的基础API是否通畅并设置一些基本信息。# 使用curl测试getMe方法验证Token有效 curl -X GET https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/getMe如果返回包含ok: true和你的Bot信息说明Token有效。你还可以通过BotFather设置Bot的头像/setuserpic、描述/setdescription、关于信息/setabouttext以及指令菜单/setcommands。指令菜单非常有用你可以设置像这样的指令start - 开始使用 help - 获取帮助 status - 查看系统状态 task - 创建新任务这样用户在聊天中输入/时就会弹出这些提示提升用户体验。4. 构建Webhook接收器连接Telegram与OpenClaw4.1 选择Web框架与搭建基础服务我们需要一个轻量级的Web框架来快速搭建接收Webhook的HTTP服务。Python生态里FastAPI是当前的首选因为它异步性能好、自动生成API文档、类型提示完善。当然使用熟悉的Flask或Django也可以。首先安装依赖pip install fastapi uvicorn python-telegram-bot这里我们安装了python-telegram-bot这个官方推荐的、功能强大的库它封装了Telegram Bot API能极大简化我们的工作。接下来创建一个主文件比如main.pyfrom fastapi import FastAPI, Request, HTTPException from telegram import Update from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext import os import logging import asyncio # 配置日志 logging.basicConfig(format%(asctime)s - %(name)s - %(levelname)s - %(message)s, levellogging.INFO) logger logging.getLogger(__name__) # 从环境变量获取Token BOT_TOKEN os.environ.get(TELEGRAM_BOT_TOKEN) if not BOT_TOKEN: raise ValueError(请设置 TELEGRAM_BOT_TOKEN 环境变量) # 初始化FastAPI应用和Telegram Bot应用 app FastAPI() # 注意python-telegram-bot v20.x 使用 Application telegram_app Application.builder().token(BOT_TOKEN).build() # 这里暂时先定义几个空的处理器后面会填充 async def start(update: Update, context: CallbackContext) - None: await update.message.reply_text(你好我是由OpenClaw驱动的助手。) async def echo(update: Update, context: CallbackContext) - None: # 这是最简单的回声测试后续会替换为OpenClaw调用 await update.message.reply_text(f你说了: {update.message.text}) # 将处理器添加到Dispatcher telegram_app.add_handler(CommandHandler(start, start)) telegram_app.add_handler(MessageHandler(filters.TEXT ~filters.COMMAND, echo)) # FastAPI的Webhook接收端点 app.post(/webhook) async def webhook(request: Request): 接收Telegram发送的Webhook更新。 try: # 1. 获取原始JSON数据 json_data await request.json() # 2. 将数据转换为Telegram的Update对象 update Update.de_json(json_data, telegram_app.bot) # 3. 将Update对象放入Application的更新队列中进行处理 await telegram_app.update_queue.put(update) # 4. 立即返回200 OK给Telegram避免超时重试 return {status: ok} except Exception as e: logger.error(f处理Webhook时出错: {e}) # 返回错误状态码Telegram可能会重试 raise HTTPException(status_code500, detailstr(e)) app.get(/) async def root(): return {message: Telegram Bot Webhook Server is running.} # 启动函数可选用于本地测试轮询模式 async def main() - None: # 注意在生产Webhook模式下我们不需要调用 run_polling # 这里只是示例结构 pass if __name__ __main__: # 本地开发时可以用uvicorn直接运行 import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这段代码搭建了一个最基础的架子。它创建了一个FastAPI应用定义了一个/webhook端点来接收POST请求。当请求到来时它把JSON数据转换成Telegram库能理解的Update对象然后放入处理队列。start和echo函数是消息处理器分别对应/start命令和普通文本消息。4.2 设置与验证Webhook服务跑起来后假设运行在https://your-server.com我们需要告诉Telegram把消息发到哪里。# 使用curl设置Webhook curl -F urlhttps://your-server.com/webhook https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/setWebhook如果返回{ok:true,result:true,description:Webhook was set}说明设置成功。你可以通过以下命令检查当前Webhook信息或删除Webhook# 获取Webhook信息 curl https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/getWebhookInfo # 删除Webhook切换回轮询模式 curl https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/deleteWebhook一个关键细节Telegram发送的Webhook请求其IP地址是动态的。官方提供了一个IP地址列表。在生产环境中如果你的服务有防火墙可能需要允许这些IP段。不过更常见的做法是在Webhook处理函数中验证一个名为X-Telegram-Bot-Api-Secret-Token的自定义请求头。你可以在设置Webhook时指定一个密钥然后在你的服务端验证它这是一个重要的安全措施。# 设置带密钥的Webhook curl -F urlhttps://your-server.com/webhook -F secret_tokenYourSuperSecretTokenHere https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/setWebhook然后在FastAPI的/webhook端点中添加验证SECRET_TOKEN os.environ.get(WEBHOOK_SECRET_TOKEN) app.post(/webhook) async def webhook(request: Request): if request.headers.get(X-Telegram-Bot-Api-Secret-Token) ! SECRET_TOKEN: raise HTTPException(status_code403, detailForbidden) # ... 其余处理逻辑5. 核心集成让OpenClaw处理消息5.1 设计消息适配层现在Webhook能收消息了echo函数也能回消息了。但我们的目标是让OpenClaw来处理消息。这就需要一个适配层将Telegram的Update对象转换成OpenClaw能处理的输入并将OpenClaw的输出转换回Telegram消息。OpenClaw通常通过其API或SDK被调用。假设OpenClaw提供了一个简单的HTTP API端点http://localhost:8001/v1/chat/completions类似OpenAI API接收JSON格式的请求。首先我们改造echo函数或者创建一个新的消息处理器import aiohttp import json async def handle_message(update: Update, context: CallbackContext) - None: 处理用户发送的文本消息转发给OpenClaw并回复结果。 user_message update.message.text chat_id update.message.chat_id # 构造OpenClaw API请求 openclaw_request { model: 你的OpenClaw模型名称, # 例如 claude-3-haiku messages: [ {role: system, content: 你是一个有帮助的助手通过Telegram与用户交流。}, {role: user, content: user_message} ], stream: False # 为了简单先使用非流式 } try: async with aiohttp.ClientSession() as session: async with session.post( http://localhost:8001/v1/chat/completions, # OpenClaw API地址 jsonopenclaw_request, timeoutaiohttp.ClientTimeout(total30) # 设置超时 ) as resp: if resp.status 200: result await resp.json() # 解析OpenClaw的回复。具体结构取决于OpenClaw API的实现 # 假设返回格式类似OpenAI: {“choices”:[{“message”:{“content”: “...”}}]} reply_text result.get(choices, [{}])[0].get(message, {}).get(content, 抱歉我没有收到回复。) else: reply_text f调用OpenClaw API失败状态码{resp.status} except asyncio.TimeoutError: reply_text 请求超时请稍后再试。 except Exception as e: logger.error(f处理消息时发生异常: {e}) reply_text 处理你的请求时出了点问题。 # 将回复发送回Telegram await update.message.reply_text(reply_text) # 更新消息处理器用handle_message替换echo telegram_app.add_handler(MessageHandler(filters.TEXT ~filters.COMMAND, handle_message))5.2 处理复杂交互上下文与状态管理简单的问答做到了但真实的助手需要上下文。用户可能会问“上一句提到的那个项目是什么”。OpenClaw自身可能有记忆机制但我们需要确保对话的上下文被正确传递。方案一依赖OpenClaw的会话记忆。如果OpenClaw的API支持传递conversation_id或能自动维护会话状态例如每次调用都传递完整的历史消息那么适配层就相对简单。我们只需要在本地缓存每个chat_id对应的对话历史并在每次请求时将整个历史记录或最近N条发送给OpenClaw。from collections import defaultdict from typing import Deque from collections import deque # 简单的内存缓存记录每个聊天最近10轮对话 conversation_history defaultdict(lambda: deque(maxlen20)) # 保存10轮对话userassistant async def handle_message_with_context(update: Update, context: CallbackContext) - None: user_message update.message.text chat_id update.message.chat_id # 获取该聊天的历史 history conversation_history[chat_id] # 将用户新消息加入历史 history.append({role: user, content: user_message}) # 构造包含完整历史的请求 openclaw_request { model: 你的模型, messages: [ {role: system, content: 系统提示词...}, *list(history) # 将deque展开为列表 ], stream: False } # ... 调用OpenClaw API ... if resp.status 200: result await resp.json() assistant_reply result[choices][0][message][content] # 将助手的回复也加入历史 history.append({role: assistant, content: assistant_reply}) await update.message.reply_text(assistant_reply)方案二利用Telegram Bot的Context和持久化。python-telegram-bot库的Context对象CallbackContext可以存储聊天特定的数据。对于更复杂的状态比如多轮表单填写、任务创建流程可以使用ConversationHandler。对于生产环境历史记录应该持久化到数据库如SQLite、Redis中而不是内存。5.3 支持多媒体与指令扩展Telegram消息不只有文本还有图片、文档、位置等。OpenClaw也可能具备多模态能力。我们需要扩展处理器来支持这些类型。from telegram.ext import filters # 处理图片消息 async def handle_photo(update: Update, context: CallbackContext): photo update.message.photo[-1] # 获取最高分辨率的图片 file_id photo.file_id # 可以通过 bot.get_file(file_id) 下载文件到本地或获取URL # 然后将图片路径或URL描述给OpenClaw如果其API支持 # 例如将图片上传到图床将URL作为文本的一部分发送“[图片]描述...” await update.message.reply_text(f收到图片文件ID: {file_id}。已记录。) # 处理文档消息 async def handle_document(update: Update, context: CallbackContext): document update.message.document file_name document.file_name # 类似地下载并处理文档 await update.message.reply_text(f收到文档: {file_name}) # 注册处理器 telegram_app.add_handler(MessageHandler(filters.PHOTO, handle_photo)) telegram_app.add_handler(MessageHandler(filters.DOCUMENT, handle_document))对于指令除了/start我们可以定义更多直接映射到OpenClaw的特定技能或功能。async def cmd_status(update: Update, context: CallbackContext): # 调用OpenClaw的“检查系统状态”技能或者直接查询本地服务状态 status_info 系统运行正常。\n当前负载低。\n今日请求数42。 await update.message.reply_text(status_info) async def cmd_task(update: Update, context: CallbackContext): # 可能触发一个创建任务的对话流程 await update.message.reply_text(请告诉我任务的具体内容。) # 这里可以返回一个状态码配合ConversationHandler进入多轮对话 telegram_app.add_handler(CommandHandler(status, cmd_status)) telegram_app.add_handler(CommandHandler(task, cmd_task))6. 部署、优化与问题排查6.1 生产环境部署考量本地开发测试完成后需要部署到稳定的服务器。以下是关键步骤服务器与域名准备一台拥有公网IP的云服务器VPS并配置一个指向该IP的域名例如bot.yourdomain.com。SSL证书为域名申请SSL证书可以使用Let‘s Encrypt免费证书。这是Telegram Webhook的强制要求。反向代理使用Nginx或Caddy作为反向代理处理SSL终止并将HTTPS请求转发到内部运行的FastAPI服务例如运行在127.0.0.1:8000。进程管理使用系统服务管理器如systemd或进程守护工具如supervisor来管理你的Python应用进程确保其崩溃后能自动重启。更新Webhook URL将Webhook设置为你的生产域名例如https://bot.yourdomain.com/webhook并记得加上安全令牌。一个简单的Nginx配置示例/etc/nginx/sites-available/your_botserver { listen 443 ssl http2; server_name bot.yourdomain.com; ssl_certificate /path/to/fullchain.pem; ssl_certificate_key /path/to/privkey.pem; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; # 传递Telegram的Secret Token proxy_set_header X-Telegram-Bot-Api-Secret-Token $http_x_telegram_bot_api_secret_token; } }6.2 性能与可靠性优化异步处理确保你的代码是异步的使用async/await避免阻塞主线程。FastAPI和python-telegram-bot都支持异步。消息队列如果消息处理逻辑非常耗时比如调用一个很慢的模型可以考虑引入消息队列如Redis RQ或Celery。Webhook端点快速接收消息后将任务放入队列立即返回200给Telegram然后由后台工作进程异步处理并调用Telegram API发送回复。这能有效防止Webhook超时Telegram默认超时时间较短。错误处理与重试对OpenClaw API的调用要有完善的错误处理和重试机制。网络波动、模型服务暂时不可用等情况都需要考虑。日志记录记录所有入站和出站消息、API调用状态和错误便于后期排查问题。可以将日志结构化后输出到文件或日志收集系统。6.3 常见问题排查实录在实际操作中你几乎一定会遇到下面这些问题问题1Webhook设置失败返回400 Bad Request。可能原因URL格式错误、不是HTTPS、SSL证书无效或不被信任、服务器防火墙/安全组未开放443端口。排查用浏览器或curl -I https://your-server.com/webhook检查你的URL是否能正常访问并返回200。检查SSL证书是否有效且域名匹配。可以用openssl s_client -connect your-server.com:443检查。确保服务器安全组和防火墙允许入站443端口连接。问题2Webhook设置成功但收不到消息。可能原因你的服务端代码没有正确返回200 OKWebhook处理函数有未捕获的异常导致进程崩溃Nginx等反向代理配置错误。排查查看服务端日志确认/webhook端点是否被调用。在Webhook处理函数开头和结尾添加详细日志打印接收到的数据。使用curl -X POST -H Content-Type: application/json -d {test: data} https://your-server.com/webhook手动模拟Telegram的请求看你的服务如何响应。检查getWebhookInfo看是否有 pending update count待处理更新数这表示Telegram尝试发送但你的服务没有正确处理。问题3机器人可以收到消息但回复很慢或超时。可能原因OpenClaw API响应慢网络延迟你的处理函数是同步阻塞的。解决优化OpenClaw服务性能或为其设置更合理的超时时间。确保你的所有IO操作网络请求、数据库查询都是异步的。如前所述引入消息队列将耗时操作异步化。问题4OpenClaw返回了内容但Telegram消息发送失败。可能原因回复内容过长Telegram文本消息有长度限制包含了Telegram不支持的格式调用TelegramsendMessageAPI时出错。解决对长文本进行分割。Telegram单条消息限约4096个字符需要实现分片发送逻辑。检查返回内容中是否有特殊字符导致API调用失败。在调用sendMessage的代码周围添加try-except并记录Telegram API返回的具体错误信息。问题5openclaw llamap svr operator(): got exception: { error: { code: 400, ...分析这个错误提示来源于OpenClaw服务内部表明它在处理请求时遇到了问题返回了400错误。这通常与发送给OpenClaw API的请求格式不正确有关。排查核对API格式仔细检查你构造的openclaw_requestJSON对象其结构必须与OpenClaw服务期望的API文档完全一致。检查字段名是model还是engine、消息数组的格式、role的值是user/assistant还是human/ai。检查模型名称确认model字段的值是OpenClaw服务中已加载且可用的正确模型名称。查看OpenClaw服务日志这个错误信息很可能在OpenClaw服务的日志中有更详细的记录去查看那里能获得根本原因。简化测试先用一个最简单的请求只包含一条用户消息测试排除是上下文历史构造复杂导致的错误。7. 进阶玩法与扩展思路基础功能跑通后这个集成的想象力空间还很大1. 技能Skill的热管理与调用OpenClaw的核心是技能。可以通过特定的Telegram命令来动态管理技能。例如/skills列出当前所有可用技能。/enable_skill skill_name启用某个技能。/disable_skill skill_name禁用某个技能。 这需要在适配层维护一个技能状态表并在调用OpenClaw时通过系统提示词或元数据告知其当前可用的技能列表。2. 权限管理与用户隔离不是所有用户都能使用所有功能。可以在代码中维护一个授权用户ID列表ALLOWED_USER_IDS在Webhook处理最开始进行校验。还可以结合OpenClaw的用户系统实现更精细的权限控制。3. 流式响应Streaming如果OpenClaw API支持流式输出类似OpenAI的stream我们可以实现打字机效果在OpenClaw生成内容的同时就一段段地发送到Telegram体验更好。这需要处理Telegram的编辑消息API。4. 内联查询Inline Query实现内联模式用户在其他聊天中输入你的Bot用户名 关键词就能直接看到并发送Bot提供的快捷结果。这需要处理InlineQuery类型的更新。5. 与CI/CD流水线集成这也是搜索热词中提到的场景gitlab → webhook → jenkins。你可以创建一个特殊的Telegram命令授权给运维人员用来触发部署、查看构建状态、重启服务等。Bot接收到命令后通过OpenClaw解析意图并调用相应的Jenkins API或执行Shell脚本再将结果返回。这相当于一个安全的、对话式的运维入口。整个集成过程从BotFather那里拿到钥匙到架起Webhook的桥梁再到精心雕琢消息适配器每一步都需要耐心调试。最深的体会是日志是你的眼睛在各个环节打好日志能帮你快速定位问题是出在Telegram推送、网络接收、消息转换还是OpenClaw处理阶段。另外异步编程思维很重要尤其是在处理网络IO时用对了async/await能避免很多性能瓶颈。最后从简单的回声测试开始逐步增加功能每完成一小步就验证一下这种渐进的方式能让复杂的集成过程变得清晰可控。 实战指南：将OpenClaw智能体接入Telegram，打造对话式AI助手
1. 项目概述当OpenClaw遇见Telegram最近在折腾一个挺有意思的事儿把OpenClaw这个智能体框架接入了Telegram。简单来说就是让你在Telegram里能像跟一个真人助手聊天一样指挥OpenClaw去完成各种任务比如查资料、写代码、管理服务器甚至控制智能家居。这听起来像是给OpenClaw装上了翅膀让它从一个运行在命令行或Web界面的工具变成了一个随时待命的“随身助理”。为什么是Telegram它的Bot API设计得非常清晰、稳定而且用户基数庞大几乎成了各种自动化工具和机器人服务的首选对接平台。BotFather是Telegram官方提供的“机器人制造机”所有Telegram机器人的创建、配置、管理都从这里开始。所以这个项目的核心就是通过BotFather创建一个机器人拿到唯一的身份凭证Bot Token然后让OpenClaw能够接收和处理这个机器人收到的所有消息并做出智能响应。整个过程就像是在OpenClaw和Telegram之间架起一座双向通信的桥梁。我踩过不少坑从BotToken的获取与保管到Webhook的设置与调试再到OpenClaw侧消息适配器的编写每一步都有需要注意的细节。这篇文章我就把这次实战的完整过程、核心原理和避坑经验梳理出来目标是让你看完就能自己动手搭一个并且知道每一步为什么要这么做。2. 核心思路与架构设计2.1 为什么选择Webhook模式Telegram Bot提供了两种消息获取方式长轮询Long Polling和Webhook。长轮询就是你的服务端程序不停地、主动地向Telegram服务器询问“有没有新消息给我”。这种方式实现简单适合开发和测试初期因为不需要公网IP。但它有个致命缺点延迟和资源消耗。你的服务器需要维持一个持续的连接去“问”如果问得太频繁可能会被限流问得不频繁消息就有延迟。而Webhook则是一种“推送”模式。你告诉Telegram服务器一个公开的URL地址你的服务端接口。一旦有用户给你的机器人发送消息Telegram服务器会主动将这个消息打包成一个HTTP POST请求发送到你指定的这个URL上。你的服务端只需要监听这个接口收到请求后解析、处理、再回复即可。对于OpenClaw这种需要实时交互、且可能部署在自有服务器拥有公网IP或通过内网穿透暴露服务的场景Webhook是生产环境的不二之选。它实时性高Telegram服务器一收到消息就立刻推过来资源消耗低你的服务端是被动接收无需维持主动查询连接也更可靠减少了因网络波动导致轮询失败的消息丢失风险。因此我们的架构就基于Webhook来构建。2.2 整体数据流与组件职责整个系统的数据流可以清晰地分为几个阶段用户触发用户在Telegram中向你的Bot发送一条消息或命令。Telegram推送Telegram服务器将这条消息封装成一个结构化的JSON数据包通过HTTPS POST请求发送到你预先设置好的Webhook URL。接收与路由你的服务端承载OpenClaw有一个HTTP服务比如用FastAPI、Flask搭建在监听Webhook URL。它接收到请求后首先进行安全验证验证请求是否真的来自Telegram然后解析JSON提取出关键信息用户ID、聊天ID、消息内容等。OpenClaw处理服务端将提取出的消息内容按照OpenClaw能理解的格式进行封装然后调用OpenClaw的核心处理引擎。OpenClaw根据其内部配置的技能Skills和记忆Memory来理解用户意图执行相应操作可能是调用一个API、运行一段脚本、查询数据库等并生成一个文本或包含图片、文件的响应。响应回传服务端拿到OpenClaw的响应后将其再次封装成Telegram Bot API要求的格式通过调用Telegram的sendMessage等接口将消息发送回对应的聊天窗口。用户接收用户在Telegram中看到机器人的回复。在这个流程中我们的核心开发工作集中在第3、4、5步即构建一个“适配器”它既能听懂Telegram的话解析Webhook又能让OpenClaw听懂转换消息格式最后还要把OpenClaw的“话”翻译成Telegram能发送的格式。注意Webhook URL必须是HTTPS的。这意味着你的服务端必须配置SSL证书。在开发测试阶段你可以使用内网穿透工具如ngrok、localtunnel来获得一个临时的HTTPS地址或者在自己的服务器上配置好SSL。Telegram不会向HTTP地址发送Webhook。3. 实战第一步从BotFather获取通行证3.1 创建你的第一个Telegram Bot一切始于BotFather。它本身就是一个Telegram Bot你需要先在Telegram应用中搜索并打开与它的对话。启动对话在Telegram中搜索BotFather点击打开聊天窗口。创建新Bot向它发送命令/newbot。设置名称BotFather会问你给机器人起什么名字。这个名字是用户在聊天列表中看到的名字比如“我的智能助手OpenClaw”。可以包含空格和特殊字符。设置用户名接下来需要设置一个唯一的用户名Username必须以bot结尾例如my_openclaw_assistant_bot。这个用户名是唯一的用于在Telegram中通过username找到你的Bot也是API调用的一部分。获取Token创建成功后BotFather会发给你一串重要的信息——HTTP API Token。它长得像这样1234567890:ABCdefGhIJKlmNoPQRsTUVwxyZ-abc123def456。这串Token就是你的Bot的“密码”和“身份证”。任何人拿到这串Token就完全控制了你的Bot。所以请立即将其妥善保存并绝对不要提交到公开的代码仓库如GitHub。最佳实践是将其设置为环境变量。3.2 Token的安全管理与环境变量配置在服务器上我们可以这样操作# 将Token写入一个只有当前用户可读的环境变量文件 echo export TELEGRAM_BOT_TOKEN你的_Actual_Token_Here ~/.bashrc # 或者为了更安全使用系统级服务环境文件或像下面这样临时设置 export TELEGRAM_BOT_TOKEN你的_Actual_Token_Here # 然后让配置生效 source ~/.bashrc在你的OpenClaw服务端代码中通过os.environ.get(TELEGRAM_BOT_TOKEN)来读取它。这样Token就不会硬编码在代码里。3.3 初步验证与基础信息设置拿到Token后你可以立即用它来测试Bot的基础API是否通畅并设置一些基本信息。# 使用curl测试getMe方法验证Token有效 curl -X GET https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/getMe如果返回包含ok: true和你的Bot信息说明Token有效。你还可以通过BotFather设置Bot的头像/setuserpic、描述/setdescription、关于信息/setabouttext以及指令菜单/setcommands。指令菜单非常有用你可以设置像这样的指令start - 开始使用 help - 获取帮助 status - 查看系统状态 task - 创建新任务这样用户在聊天中输入/时就会弹出这些提示提升用户体验。4. 构建Webhook接收器连接Telegram与OpenClaw4.1 选择Web框架与搭建基础服务我们需要一个轻量级的Web框架来快速搭建接收Webhook的HTTP服务。Python生态里FastAPI是当前的首选因为它异步性能好、自动生成API文档、类型提示完善。当然使用熟悉的Flask或Django也可以。首先安装依赖pip install fastapi uvicorn python-telegram-bot这里我们安装了python-telegram-bot这个官方推荐的、功能强大的库它封装了Telegram Bot API能极大简化我们的工作。接下来创建一个主文件比如main.pyfrom fastapi import FastAPI, Request, HTTPException from telegram import Update from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext import os import logging import asyncio # 配置日志 logging.basicConfig(format%(asctime)s - %(name)s - %(levelname)s - %(message)s, levellogging.INFO) logger logging.getLogger(__name__) # 从环境变量获取Token BOT_TOKEN os.environ.get(TELEGRAM_BOT_TOKEN) if not BOT_TOKEN: raise ValueError(请设置 TELEGRAM_BOT_TOKEN 环境变量) # 初始化FastAPI应用和Telegram Bot应用 app FastAPI() # 注意python-telegram-bot v20.x 使用 Application telegram_app Application.builder().token(BOT_TOKEN).build() # 这里暂时先定义几个空的处理器后面会填充 async def start(update: Update, context: CallbackContext) - None: await update.message.reply_text(你好我是由OpenClaw驱动的助手。) async def echo(update: Update, context: CallbackContext) - None: # 这是最简单的回声测试后续会替换为OpenClaw调用 await update.message.reply_text(f你说了: {update.message.text}) # 将处理器添加到Dispatcher telegram_app.add_handler(CommandHandler(start, start)) telegram_app.add_handler(MessageHandler(filters.TEXT ~filters.COMMAND, echo)) # FastAPI的Webhook接收端点 app.post(/webhook) async def webhook(request: Request): 接收Telegram发送的Webhook更新。 try: # 1. 获取原始JSON数据 json_data await request.json() # 2. 将数据转换为Telegram的Update对象 update Update.de_json(json_data, telegram_app.bot) # 3. 将Update对象放入Application的更新队列中进行处理 await telegram_app.update_queue.put(update) # 4. 立即返回200 OK给Telegram避免超时重试 return {status: ok} except Exception as e: logger.error(f处理Webhook时出错: {e}) # 返回错误状态码Telegram可能会重试 raise HTTPException(status_code500, detailstr(e)) app.get(/) async def root(): return {message: Telegram Bot Webhook Server is running.} # 启动函数可选用于本地测试轮询模式 async def main() - None: # 注意在生产Webhook模式下我们不需要调用 run_polling # 这里只是示例结构 pass if __name__ __main__: # 本地开发时可以用uvicorn直接运行 import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这段代码搭建了一个最基础的架子。它创建了一个FastAPI应用定义了一个/webhook端点来接收POST请求。当请求到来时它把JSON数据转换成Telegram库能理解的Update对象然后放入处理队列。start和echo函数是消息处理器分别对应/start命令和普通文本消息。4.2 设置与验证Webhook服务跑起来后假设运行在https://your-server.com我们需要告诉Telegram把消息发到哪里。# 使用curl设置Webhook curl -F urlhttps://your-server.com/webhook https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/setWebhook如果返回{ok:true,result:true,description:Webhook was set}说明设置成功。你可以通过以下命令检查当前Webhook信息或删除Webhook# 获取Webhook信息 curl https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/getWebhookInfo # 删除Webhook切换回轮询模式 curl https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/deleteWebhook一个关键细节Telegram发送的Webhook请求其IP地址是动态的。官方提供了一个IP地址列表。在生产环境中如果你的服务有防火墙可能需要允许这些IP段。不过更常见的做法是在Webhook处理函数中验证一个名为X-Telegram-Bot-Api-Secret-Token的自定义请求头。你可以在设置Webhook时指定一个密钥然后在你的服务端验证它这是一个重要的安全措施。# 设置带密钥的Webhook curl -F urlhttps://your-server.com/webhook -F secret_tokenYourSuperSecretTokenHere https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/setWebhook然后在FastAPI的/webhook端点中添加验证SECRET_TOKEN os.environ.get(WEBHOOK_SECRET_TOKEN) app.post(/webhook) async def webhook(request: Request): if request.headers.get(X-Telegram-Bot-Api-Secret-Token) ! SECRET_TOKEN: raise HTTPException(status_code403, detailForbidden) # ... 其余处理逻辑5. 核心集成让OpenClaw处理消息5.1 设计消息适配层现在Webhook能收消息了echo函数也能回消息了。但我们的目标是让OpenClaw来处理消息。这就需要一个适配层将Telegram的Update对象转换成OpenClaw能处理的输入并将OpenClaw的输出转换回Telegram消息。OpenClaw通常通过其API或SDK被调用。假设OpenClaw提供了一个简单的HTTP API端点http://localhost:8001/v1/chat/completions类似OpenAI API接收JSON格式的请求。首先我们改造echo函数或者创建一个新的消息处理器import aiohttp import json async def handle_message(update: Update, context: CallbackContext) - None: 处理用户发送的文本消息转发给OpenClaw并回复结果。 user_message update.message.text chat_id update.message.chat_id # 构造OpenClaw API请求 openclaw_request { model: 你的OpenClaw模型名称, # 例如 claude-3-haiku messages: [ {role: system, content: 你是一个有帮助的助手通过Telegram与用户交流。}, {role: user, content: user_message} ], stream: False # 为了简单先使用非流式 } try: async with aiohttp.ClientSession() as session: async with session.post( http://localhost:8001/v1/chat/completions, # OpenClaw API地址 jsonopenclaw_request, timeoutaiohttp.ClientTimeout(total30) # 设置超时 ) as resp: if resp.status 200: result await resp.json() # 解析OpenClaw的回复。具体结构取决于OpenClaw API的实现 # 假设返回格式类似OpenAI: {“choices”:[{“message”:{“content”: “...”}}]} reply_text result.get(choices, [{}])[0].get(message, {}).get(content, 抱歉我没有收到回复。) else: reply_text f调用OpenClaw API失败状态码{resp.status} except asyncio.TimeoutError: reply_text 请求超时请稍后再试。 except Exception as e: logger.error(f处理消息时发生异常: {e}) reply_text 处理你的请求时出了点问题。 # 将回复发送回Telegram await update.message.reply_text(reply_text) # 更新消息处理器用handle_message替换echo telegram_app.add_handler(MessageHandler(filters.TEXT ~filters.COMMAND, handle_message))5.2 处理复杂交互上下文与状态管理简单的问答做到了但真实的助手需要上下文。用户可能会问“上一句提到的那个项目是什么”。OpenClaw自身可能有记忆机制但我们需要确保对话的上下文被正确传递。方案一依赖OpenClaw的会话记忆。如果OpenClaw的API支持传递conversation_id或能自动维护会话状态例如每次调用都传递完整的历史消息那么适配层就相对简单。我们只需要在本地缓存每个chat_id对应的对话历史并在每次请求时将整个历史记录或最近N条发送给OpenClaw。from collections import defaultdict from typing import Deque from collections import deque # 简单的内存缓存记录每个聊天最近10轮对话 conversation_history defaultdict(lambda: deque(maxlen20)) # 保存10轮对话userassistant async def handle_message_with_context(update: Update, context: CallbackContext) - None: user_message update.message.text chat_id update.message.chat_id # 获取该聊天的历史 history conversation_history[chat_id] # 将用户新消息加入历史 history.append({role: user, content: user_message}) # 构造包含完整历史的请求 openclaw_request { model: 你的模型, messages: [ {role: system, content: 系统提示词...}, *list(history) # 将deque展开为列表 ], stream: False } # ... 调用OpenClaw API ... if resp.status 200: result await resp.json() assistant_reply result[choices][0][message][content] # 将助手的回复也加入历史 history.append({role: assistant, content: assistant_reply}) await update.message.reply_text(assistant_reply)方案二利用Telegram Bot的Context和持久化。python-telegram-bot库的Context对象CallbackContext可以存储聊天特定的数据。对于更复杂的状态比如多轮表单填写、任务创建流程可以使用ConversationHandler。对于生产环境历史记录应该持久化到数据库如SQLite、Redis中而不是内存。5.3 支持多媒体与指令扩展Telegram消息不只有文本还有图片、文档、位置等。OpenClaw也可能具备多模态能力。我们需要扩展处理器来支持这些类型。from telegram.ext import filters # 处理图片消息 async def handle_photo(update: Update, context: CallbackContext): photo update.message.photo[-1] # 获取最高分辨率的图片 file_id photo.file_id # 可以通过 bot.get_file(file_id) 下载文件到本地或获取URL # 然后将图片路径或URL描述给OpenClaw如果其API支持 # 例如将图片上传到图床将URL作为文本的一部分发送“[图片]描述...” await update.message.reply_text(f收到图片文件ID: {file_id}。已记录。) # 处理文档消息 async def handle_document(update: Update, context: CallbackContext): document update.message.document file_name document.file_name # 类似地下载并处理文档 await update.message.reply_text(f收到文档: {file_name}) # 注册处理器 telegram_app.add_handler(MessageHandler(filters.PHOTO, handle_photo)) telegram_app.add_handler(MessageHandler(filters.DOCUMENT, handle_document))对于指令除了/start我们可以定义更多直接映射到OpenClaw的特定技能或功能。async def cmd_status(update: Update, context: CallbackContext): # 调用OpenClaw的“检查系统状态”技能或者直接查询本地服务状态 status_info 系统运行正常。\n当前负载低。\n今日请求数42。 await update.message.reply_text(status_info) async def cmd_task(update: Update, context: CallbackContext): # 可能触发一个创建任务的对话流程 await update.message.reply_text(请告诉我任务的具体内容。) # 这里可以返回一个状态码配合ConversationHandler进入多轮对话 telegram_app.add_handler(CommandHandler(status, cmd_status)) telegram_app.add_handler(CommandHandler(task, cmd_task))6. 部署、优化与问题排查6.1 生产环境部署考量本地开发测试完成后需要部署到稳定的服务器。以下是关键步骤服务器与域名准备一台拥有公网IP的云服务器VPS并配置一个指向该IP的域名例如bot.yourdomain.com。SSL证书为域名申请SSL证书可以使用Let‘s Encrypt免费证书。这是Telegram Webhook的强制要求。反向代理使用Nginx或Caddy作为反向代理处理SSL终止并将HTTPS请求转发到内部运行的FastAPI服务例如运行在127.0.0.1:8000。进程管理使用系统服务管理器如systemd或进程守护工具如supervisor来管理你的Python应用进程确保其崩溃后能自动重启。更新Webhook URL将Webhook设置为你的生产域名例如https://bot.yourdomain.com/webhook并记得加上安全令牌。一个简单的Nginx配置示例/etc/nginx/sites-available/your_botserver { listen 443 ssl http2; server_name bot.yourdomain.com; ssl_certificate /path/to/fullchain.pem; ssl_certificate_key /path/to/privkey.pem; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; # 传递Telegram的Secret Token proxy_set_header X-Telegram-Bot-Api-Secret-Token $http_x_telegram_bot_api_secret_token; } }6.2 性能与可靠性优化异步处理确保你的代码是异步的使用async/await避免阻塞主线程。FastAPI和python-telegram-bot都支持异步。消息队列如果消息处理逻辑非常耗时比如调用一个很慢的模型可以考虑引入消息队列如Redis RQ或Celery。Webhook端点快速接收消息后将任务放入队列立即返回200给Telegram然后由后台工作进程异步处理并调用Telegram API发送回复。这能有效防止Webhook超时Telegram默认超时时间较短。错误处理与重试对OpenClaw API的调用要有完善的错误处理和重试机制。网络波动、模型服务暂时不可用等情况都需要考虑。日志记录记录所有入站和出站消息、API调用状态和错误便于后期排查问题。可以将日志结构化后输出到文件或日志收集系统。6.3 常见问题排查实录在实际操作中你几乎一定会遇到下面这些问题问题1Webhook设置失败返回400 Bad Request。可能原因URL格式错误、不是HTTPS、SSL证书无效或不被信任、服务器防火墙/安全组未开放443端口。排查用浏览器或curl -I https://your-server.com/webhook检查你的URL是否能正常访问并返回200。检查SSL证书是否有效且域名匹配。可以用openssl s_client -connect your-server.com:443检查。确保服务器安全组和防火墙允许入站443端口连接。问题2Webhook设置成功但收不到消息。可能原因你的服务端代码没有正确返回200 OKWebhook处理函数有未捕获的异常导致进程崩溃Nginx等反向代理配置错误。排查查看服务端日志确认/webhook端点是否被调用。在Webhook处理函数开头和结尾添加详细日志打印接收到的数据。使用curl -X POST -H Content-Type: application/json -d {test: data} https://your-server.com/webhook手动模拟Telegram的请求看你的服务如何响应。检查getWebhookInfo看是否有 pending update count待处理更新数这表示Telegram尝试发送但你的服务没有正确处理。问题3机器人可以收到消息但回复很慢或超时。可能原因OpenClaw API响应慢网络延迟你的处理函数是同步阻塞的。解决优化OpenClaw服务性能或为其设置更合理的超时时间。确保你的所有IO操作网络请求、数据库查询都是异步的。如前所述引入消息队列将耗时操作异步化。问题4OpenClaw返回了内容但Telegram消息发送失败。可能原因回复内容过长Telegram文本消息有长度限制包含了Telegram不支持的格式调用TelegramsendMessageAPI时出错。解决对长文本进行分割。Telegram单条消息限约4096个字符需要实现分片发送逻辑。检查返回内容中是否有特殊字符导致API调用失败。在调用sendMessage的代码周围添加try-except并记录Telegram API返回的具体错误信息。问题5openclaw llamap svr operator(): got exception: { error: { code: 400, ...分析这个错误提示来源于OpenClaw服务内部表明它在处理请求时遇到了问题返回了400错误。这通常与发送给OpenClaw API的请求格式不正确有关。排查核对API格式仔细检查你构造的openclaw_requestJSON对象其结构必须与OpenClaw服务期望的API文档完全一致。检查字段名是model还是engine、消息数组的格式、role的值是user/assistant还是human/ai。检查模型名称确认model字段的值是OpenClaw服务中已加载且可用的正确模型名称。查看OpenClaw服务日志这个错误信息很可能在OpenClaw服务的日志中有更详细的记录去查看那里能获得根本原因。简化测试先用一个最简单的请求只包含一条用户消息测试排除是上下文历史构造复杂导致的错误。7. 进阶玩法与扩展思路基础功能跑通后这个集成的想象力空间还很大1. 技能Skill的热管理与调用OpenClaw的核心是技能。可以通过特定的Telegram命令来动态管理技能。例如/skills列出当前所有可用技能。/enable_skill skill_name启用某个技能。/disable_skill skill_name禁用某个技能。 这需要在适配层维护一个技能状态表并在调用OpenClaw时通过系统提示词或元数据告知其当前可用的技能列表。2. 权限管理与用户隔离不是所有用户都能使用所有功能。可以在代码中维护一个授权用户ID列表ALLOWED_USER_IDS在Webhook处理最开始进行校验。还可以结合OpenClaw的用户系统实现更精细的权限控制。3. 流式响应Streaming如果OpenClaw API支持流式输出类似OpenAI的stream我们可以实现打字机效果在OpenClaw生成内容的同时就一段段地发送到Telegram体验更好。这需要处理Telegram的编辑消息API。4. 内联查询Inline Query实现内联模式用户在其他聊天中输入你的Bot用户名 关键词就能直接看到并发送Bot提供的快捷结果。这需要处理InlineQuery类型的更新。5. 与CI/CD流水线集成这也是搜索热词中提到的场景gitlab → webhook → jenkins。你可以创建一个特殊的Telegram命令授权给运维人员用来触发部署、查看构建状态、重启服务等。Bot接收到命令后通过OpenClaw解析意图并调用相应的Jenkins API或执行Shell脚本再将结果返回。这相当于一个安全的、对话式的运维入口。整个集成过程从BotFather那里拿到钥匙到架起Webhook的桥梁再到精心雕琢消息适配器每一步都需要耐心调试。最深的体会是日志是你的眼睛在各个环节打好日志能帮你快速定位问题是出在Telegram推送、网络接收、消息转换还是OpenClaw处理阶段。另外异步编程思维很重要尤其是在处理网络IO时用对了async/await能避免很多性能瓶颈。最后从简单的回声测试开始逐步增加功能每完成一小步就验证一下这种渐进的方式能让复杂的集成过程变得清晰可控。