要是你手上刚好跑着一个自托管的 OpenClaw 智能体服务想让它在微信里随叫随到这篇就是给你准备的。前前后后折腾了三个晚上从拿到公众号测试号、配置回调 URL、处理验签和 XML 消息到把不同用户的会话对上号、路由给不同智能体最终把整条链路彻底跑通微信里发一句话OpenClaw 干活结果再原路返回。整个过程不算复杂但坑是真的不少。这篇文章记录的是完整实战包括每一步的取舍逻辑和排查过程适合手上已经跑通 OpenClaw、想加一个 IM 入口的朋友也适合从来没碰过公众号回调、第一次做这类对接的人照着走一遍。先交代一下背景。OpenClaw 在我这儿是一个自托管的智能体运行时可以配置多模型、多智能体平时我习惯通过网页或者命令行跟它对话干活没问题但人不在电脑前的时候就等于失联。把微信接进来本质上是给它开一扇“随时随地能喊它干活”的门。这个需求听起来简单真正落地时涉及的不只是“转发一条消息”而是消息怎么验真、怎么解析、怎么和会话对上号、怎么在不同智能体之间分发以及微信那 5 秒被动回复时限怎么绕过去。下面按我实际趟过的顺序来写。1. 为什么非要把微信接进来以及入口选型的思路1.1 先想清楚你要的是“入口”还是“平台”做这件事之前我先逼自己明确了一个原则我要的不是“把微信整个搬过来”而是给 OpenClaw 加一个消息入口。换句话说微信只是信道真正的业务逻辑、对话管理、工具调用全部留在 OpenClaw 里。这个定位决定了后面所有设计——网关层越薄越好逻辑尽可能往里收。选择微信作为入口原因很直接它是我日常使用频率最高的聊天软件。Agent 接进微信意味着我在手机上不用额外装任何东西不用打开网页后台随手就能抛一个任务过去。相比之下自建 App 成本太高网页端使用成本也高接别的 IM 当然也可以但论“随手可用”微信对我来说还是最顺的那个。这里有个方向性的坑必须先说清楚。微信生态里做“机器人”民间常有两种做法。第一种是模拟个人号收发消息我一开始心动过但很快放弃理由有三方案不稳定客户端接口说变就变封号风险一直悬在头上辛辛苦苦搭的服务可能一夜归零没有正规回调协议消息同步、多端登录这些都要自己拿逆向工程去补。真要这么干等于整条链路都建在沙地上不符合“自托管 Agent”要稳定长期跑的前提。第二种是走公众号开发者模式也就是微信官方提供的服务器回调能力。官方协议、公开文档、稳定回调订阅号和个人订阅号都支持开启开发者模式还有专门的测试号可供开发调试。我当时的选择标准就一条这个入口是否建立在官方协议之上。只要协议是官方公开的剩下的事都是工程问题可以慢慢磨协议是灰色的话技术再漂亮也是给自己埋雷。最后我选了公众号而且是先用测试号联调全通了才考虑正式号。1.2 整体架构网关、会话、路由三层接入方案定下来之后我在纸上画了一下完整的消息流发现核心其实就三块网关层、会话层、路由层。网关层负责接收微信的 HTTP 回调完成验签、报文解析和被动回复会话层负责维护“微信用户 ↔ OpenClaw 会话”的映射管好身份和上下文隔离路由层根据消息内容决定交给哪个智能体再把结果送回链路。三层各干各的尽量不互相依赖。网关层只认 HTTP 和 XML不关心下游是不是 OpenClaw会话层只存映射不做业务判断路由层才真正碰业务规则。这个拆法带来的好处我在联调阶段体会特别深。出了问题时先定位是网关没收到、验签挂了、会话丢了还是 Agent 本身卡了排查范围一下子就缩得很小。我实际部署里网关是一个独立的 FastAPI 服务单独跑一个进程。OpenClaw 作为一套自托管的智能体运行时我通过它提供的 HTTP 接口来创建对话和发送消息。不管当前版本有没有现成的微信通道把“发消息”这个动作封装成一个 client路由层就不用关心底层细节。提示如果 OpenClaw 当前版本没有现成的微信通道不要急着往框架里写通道插件。先把网关独立出来用最普通的 HTTP 调用把链路打通确认端到端能跑通之后再回头考虑“要不要集成得更优雅”。我一开始就试图把通道做成框架插件被框架内部那一堆抽象绕了很久最后退回独立网关半小时就通了。先跑起来再谈架构这个是接入类项目最实用的顺序。2. 回调解析微信消息是怎么进来的2.1 服务器配置与签名校验的那些细节微信公众平台的接入逻辑核心是“服务器配置”里的三个值URL、Token、EncodingAESKey。URL 是接收回调的地址Token 是你自己定的一串校验字符串EncodingAESKey 用来做消息加解密。后台提交配置时微信服务器会向你的 URL 发一个 GET 请求带四个参数signature、timestamp、nonce、echostr。你必须把 Token、timestamp、nonce 三个字符串先按字典序排序、再拼接成一个字符串做一次 SHA1 哈希和 signature 比对。一致就把 echostr 原样返回服务器配置就算验证通过不一致微信后台直接提示“验证失败”。逻辑不长坑却不少。第一个坑是排序拼接必须严格按字典序不是按参数传入顺序或者你想当然的顺序。微信的算法是sort([token, timestamp, nonce])之后再拼我一开始偷懒直接字符串相加结果签名死活对不上排查了二十分钟才发现是顺序问题。import hashlib def check_signature(token: str, timestamp: str, nonce: str, signature: str) - bool: tmp .join(sorted([token, timestamp, nonce])) return hashlib.sha1(tmp.encode(utf-8)).hexdigest() signature第二个坑是 GET 和 POST 共用同一个 URL但 GET 只负责验证。验证通过之后微信推送用户消息全部走 POST。如果代码里只处理了 POST后台提交配置时会直接失败反过来如果只写了 GET消息进来没人接。两者都要有且 GET 的处理要极简只做验签和回显。第三个坑是 EncodingAESKey 相关的。公众号后台有两种加解密模式明文模式和安全模式。开发初期强烈建议先选明文模式把整条链路跑通再考虑加密。安全模式用的是 AES-256-CBC涉及 PKCS7 填充、密钥处理、签名和时间戳校验一圈流程走下来步骤很多。一旦解密出错你根本分不清是消息本身的问题还是解密的问题调试难度直接翻倍。我是明文全通之后才切到安全模式对照文档把解密一步步补上。把“协议验证”和“业务逻辑”分开调试这是双向协议接入最省力的思路。注意签名校验要写成纯函数只依赖 token、timestamp、nonce、signature 四个入参。GET 验证和 POST 消息处理共用同一个校验函数免得两处逻辑漂移。2.2 消息体解析XML、消息类型和必要的防御POST 回调的消息体是 XML。一个典型的文本消息长这样xml ToUserName![CDATA[gh_xxxx]]/ToUserName FromUserName![CDATA[oXXXX_用户openid]]/FromUserName CreateTime1234567890/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[帮我总结一下这篇文章]]/Content MsgId1234567890123456/MsgId /xml解析用 Python 标准库的xml.etree.ElementTree就完全够用不需要上重型 XML 框架。核心要提取的字段FromUserName是用户的 openid这是后面会话映射的钥匙Content是消息文字MsgType是消息类型MsgId是消息的唯一 ID。import xml.etree.ElementTree as ET def parse_message(xml_body: str) - dict: root ET.fromstring(xml_body) return {child.tag: child.text for child in root}这里有几个容易被忽略的点逐个说。首先是 CDATA。微信返回的 XML 里所有字符串字段都用 CDATA 包着解析库会直接给出里面的值不用你手动拆标记。但自己拼回复 XML 时必须把用户内容放进![CDATA[...]]里否则内容里出现、、这类符号会直接破坏 XML 结构微信就丢掉这条回复。其次是消息类型。除了 text你还会收到 event 类型典型的就是用户关注公众号时推来的 subscribe 事件。如果解析层不区分类型直接把 event 当文本消息Content 字段就是空的路由层拿到空串然后干出各种莫名其妙的事。所以解析层必须先看 MsgTypeevent 单独走事件处理text 走对话流程image 等其他类型先统一返回“暂不支持该类型消息”。第三是重复消息。微信在请求超时或网络异常时会重试推送同一条消息重试时 MsgId 相同。不做去重的话用户的同一句话会被 OpenClaw 处理两遍浪费 token 不说还可能重复触发工具调用。我用了一个简单的 LRU 集合记录最近 1000 条 MsgId重复的直接丢弃。这个防御看着小后面救了我好几次尤其是接异步任务之后微信重试频率明显上来了。3. 会话路由多用户、多会话怎么对上号3.1 会话映射设计openid 是唯一的钥匙微信回调给到你手里的用户身份就是 openid相当于用户在公众号体系里的身份证号。不同用户 openid 不同同一用户每次进来 openid 稳定不变。所以会话映射表的设计就围绕 openid 展开。我的方案是维护一张映射表键是 openid值是一个会话 ID。这个会话 ID 是网关自行生成的 UUID代表 OpenClaw 侧的一个独立对话实例。用户第一次发消息就为这个 openid 生成一个新会话 ID后续消息全部复用同一个直到用户主动重置或会话过期。import uuid class SessionStore: def __init__(self): self._map {} def get_or_create(self, openid: str) - str: if openid not in self._map: self._map[openid] str(uuid.uuid4()) return self._map[openid] def reset(self, openid: str): self._map.pop(openid, None)这个映射本身简单但有个隐患进程重启后内存表就没了。用户隔天继续对话openid 没变映射却清了系统会重新生成会话 ID等于强制重置上下文。如果接受“重启即失忆”内存表就够如果希望长期保持上下文就得落到 SQLite 或 Redis 里。我选了 SQLite一张表三列够用还好排查。注意我见过有人图省事直接把 openid 当会话 ID 传给 Agent觉得“反正一个用户一个会话”。单机单用户调试时没问题多用户一上来就串了——两个用户的消息会进入同一个对话历史。我在设计上强制区分“身份 ID”和“会话 ID”进框架的消息一律只带会话 ID身份 ID 只存在于网关层映射表。这个隔离是会话路由不出错的前提。3.2 路由分发单 Agent 还是多 Agent链路通了之后紧接着的问题就是OpenClaw 里可能配了多个智能体有的擅长写代码有的擅长查资料有的就是闲聊。微信进来的一句话交给谁我的路由规则特别直白以消息前缀为准没前缀就走默认。比如/code 帮我写个脚本里的前缀/code路由到编程专用智能体/search 查一下xxx路由到搜索智能体其余消息全部进默认智能体。规则直观、可控排查时看日志里的前缀就知道走的是哪条链路。ROUTE_MAP { /code: agent_coder, /search: agent_search, /write: agent_writer, } def route_message(openid: str, content: str): session_id session_store.get_or_create(openid) content content.strip() if content.startswith(/): cmd, _, rest content.partition( ) if cmd in ROUTE_MAP: return session_id, ROUTE_MAP[cmd], rest.strip() return session_id, DEFAULT_AGENT, content这里有个微妙的设计取舍**切换智能体后会话上下文要不要跟着切**我的做法是多一层映射openid 不直接映射到单个会话 ID而是映射到“各智能体各自的会话 ID”。也就是说同一个用户在/code智能体下的对话历史和默认智能体下的历史互不干扰。用户切回/code继续聊昨天的代码上下文还在。成本只是多维护一个字典套字典收益却是每条业务线的上下文都独立不会因为用户从闲聊切到编程就把代码上下文冲掉。会话过期策略也得定。OpenClaw 的上下文有长度限制聊得久了历史会越来越长。我的策略是每个会话记录最后一次活跃时间超过 24 小时没活跃就重置调用 Agent 时如果发现上下文偏长就主动让框架裁剪历史。裁剪阈值取决于模型我跑了几天后取了一个相对稳的值保留最近 20 轮对话。设太小浪费上下文设太大容易超限这个值可以根据自己用的模型和任务类型再调。4. 实操过程从零到跑通4.1 依赖和骨架实际跑通的服务端结构很简单不复杂wechat-gateway/ ├── main.py # FastAPI 应用回调入口 ├── crypto.py # 验签与解密安全模式下用 ├── session.py # 会话映射存储 ├── router.py # 路由规则 ├── openclaw_client.py # 封装对 OpenClaw 的调用 └── config.yaml # Token、URL、密钥等配置依赖就四样fastapi、uvicorn、requests或 httpx、pyyaml。OpenClaw 侧我走的是它提供的 HTTP 接口网关里封装成一个 client。封装的时候我刻意不引入框架 SDK只用普通 HTTP POST好处是框架升级换接口时网关只需要改一个文件里的一个函数不用跟着框架的版本走。4.2 核心代码验签、解析、回复FastAPI 应用主体大概长这样逻辑不算多但每个环节都不能缺import time from fastapi import FastAPI, Request, Response app FastAPI() XML_REPLY_TMPL xml ToUserName![CDATA[{to_user}]]/ToUserName FromUserName![CDATA[{from_user}]]/FromUserName CreateTime{create_time}/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[{content}]]/Content /xml def build_text_reply(from_user: str, to_user: str, content: str) - str: return XML_REPLY_TMPL.format( to_userfrom_user, from_userto_user, create_timeint(time.time()), contentcontent, ) app.get(/wechat/callback) async def handle_get(signature: str, timestamp: str, nonce: str, echostr: str): if check_signature(TOKEN, timestamp, nonce, signature): return Response(contentechostr, media_typetext/plain) raise HTTPException(status_code403, detailsignature mismatch) app.post(/wechat/callback) async def handle_post(request: Request): raw await request.body() # 安全模式下这里要先把 raw 解密再解析 msg parse_message(raw.decode(utf-8)) if msg.get(MsgType) event: return Response(contenthandle_event(msg), media_typeapplication/xml) openid msg.get(FromUserName) if not dedup(msg.get(MsgId)): return Response(status_code200) content msg.get(Content, ).strip() if not content: return Response(contentbuild_text_reply(openid, ACCOUNT_ID, 空消息忽略), media_typeapplication/xml) session_id, agent_name, payload route_message(openid, content) reply openclaw_client.send_message(session_id, agent_name, payload) return Response(contentbuild_text_reply(openid, ACCOUNT_ID, reply), media_typeapplication/xml)关于被动回复有一个躲不开的硬约束微信要求你的服务器在5 秒内响应这个 POST 请求。如果 OpenClaw 处理得很快5 秒内能返回直接同步回填 XML 就行。但如果智能体要跑工具、搜资料耗时很容易超过 5 秒。这时候有三个办法先回“收到正在处理”再用客服消息接口主动推送结果。客服消息对刚刚互动过的用户有 48 小时窗口完全够用。异步任务化。POST 进来先把消息落库立刻返 200 空串后台线程跑任务跑完用客服消息推给用户。同步死等。如果绝大多数消息本身就在 2 秒内完成同步处理也没问题超时的个别情况再单独处理。我这边的情况是大约八成的普通消息能在 3 秒内返回所以默认走同步但路由到搜索这类慢任务时切到异步通道。两套模式共存由路由层决定走哪条而不是所有消息一刀切。提示微信被动回复里Content 长度不要超过 2048 字节否则可能被截断甚至失败。OpenClaw 的输出经常一大段一大段我在 client 层做了切分逻辑超过 2000 字节就把文本按段落切块通过客服消息分多条发送被动回复只放第一段。这个细节实测救了我很多次长回复十有八九会栽在这。4.3 联调和上线要点本地开发时微信服务器访问不到你的 localhost所以需要一个公网可达的地址。我这边是用内网穿透工具把本地端口映射到一个临时域名后台填这个域名就能联调。这个阶段最实用的技巧是先用命令行工具模拟微信的回调把 GET 验证和 POST 消息分开测试确认服务端逻辑没问题之后再在公众号后台真实发消息。否则你一边调后台一边调代码出错了根本分不清是配置问题还是代码问题。联调时我把日志级别拉满每个关键节点都打日志收到回调、验签结果、解析后的消息字段、路由结果、OpenClaw 返回耗时、回复内容长度。这套日志后来帮我定位过几次“消息发出去了但没回复”的问题基本都是回复格式不对导致微信不展示。上线之后又加了请求耗时统计专门盯 5 秒超时率这个指标。5. 常见问题与避坑实录5.1 签名校验不过签名校验不过是我踩的第一个坑。除了前面说的拼接顺序问题常见的还有三个原因后台配置的 Token 和代码里的 Token 不一致多一个空格都算错服务器系统时间不准导致 timestamp 和微信服务器偏差过大GET 和 POST 的校验逻辑不一致GET 多了 echostr 参数导致处理分支分叉。建议把签名校验写成纯函数只依赖四个参数GET 和 POST 统一调用能少掉一半这种破事。5.2 被动回复超时触发微信重试微信的重试机制是把双刃剑。好处是消息不容易丢坏处是处理时间超过 5 秒后微信会重发同一条消息于是同一个任务可能被跑两遍。我的处理思路是两条并行一是 5 秒内无法完成的动作立刻异步化并先给用户一个即时反馈二是所有消息入口都过 MsgId 去重内存里保留最近 1000 条重复消息直接丢弃。这套组合上线之后重复处理基本绝迹。5.3 用户会话串号、上下文错乱会话串号的表现是A 用户问了个代码问题B 用户发消息却收到了 A 的上下文答复。排查时先看日志里两条消息的 session_id 是否一致。如果不一致但回复内容一样说明路由层没问题是 Agent 侧接口对 session_id 不敏感用了同一个全局会话。如果映射表本身串了把“身份 ID 与会话 ID”的隔离做好就能解决。压测时我特意用两个测试账号交替发消息专门验证这个点确保每个用户各说各话。5.4 长回复被截断、XML 转义出错长回复和 XML 转义是两回事但经常一起出现。转义出错的表现是用户消息里带或符号返回的 XML 非法微信直接丢弃回复。最终我统一用 CDATA 包裹所有用户生成内容CDATA 内部只排除]]这个结束序列其余符号不需要转义简单可靠。长回复靠 2048 字节切分兜底。这两条加进代码之后格式类问题再没出现过。5.5 非文本消息怎么处理非文本消息一定会在 MsgType 上体现出来所以解析层必须先按类型分流。image、voice、video 这类消息里带的只是媒体 ID不是实际文件要变成可用输入还得额外调用素材下载接口。我这个版本没做多模态输入统一返回“暂时只支持文字消息”等以后接入视觉模型再扩展。“先兜底、后扩展”这个思路比一开始就把所有消息类型全接上要稳得多。为了方便后面维护我把这些坑整理成一张速查表贴在网关项目的 README 里现象根因处理办法提交配置验证失败Token 不一致、拼接顺序错误、服务器时间偏差核对三项参数按字典序排序拼接后 SHA1消息发了没回复被动回复超时、XML 格式非法、消息被去重误吞看网关日志确认 200 与响应体检查去重逻辑回复内容错乱会话映射串号、Agent 接口忽略 session_id日志对比 openid 与 session_id核对调用长回复被截断Content 超过 2048 字节切分多条用客服消息推送这次把微信接进 OpenClaw最大的收获其实不是“跑通了”这个结果而是对这类协议对接的路径有了更清楚的认识。难的地方从来不在框架本身而在协议对齐——微信的协议是文档驱动、回调驱动的摸清它的脾气之后剩下的事就是把自己的网关和会话层实现清楚。真正省时间的经验是把网关做得越“笨”越好只负责验签、解析、转发把会话层做得越“简单”越好只存一张映射表不掺任何业务业务判断全部放到路由层。这样不管微信协议怎么调整、OpenClaw 怎么升级你改动的范围始终被压缩在很小的文件里。最后再分享一个小建议如果条件允许先用公众号测试号把链路跑通再申请正式的订阅号或服务号。测试号后台随便折腾不会影响真实用户配置项又和正式号基本一致迁移成本很低。我就是靠着测试号把验签、会话、路由、异步回复全走了一遍切到正式号时只改了一个账号 ID半小时上线。