详解微信龙虾插件openclaw-weixin源码,并实现一个Python版本的简易sdk
这两天微信接入龙虾openclaw很火爆,这里对插件openclaw-weixin源码进行一些分析和讲解,并实现一个Python版本的简易sdk。
为什么要接入微信Bot
微信Bot这个需求挺常见的。企业想做智能客服,个人想做AI助手,甚至有人想搞群管理机器人。但微信官方的接口限制很多,不是随便就能接入的。
微信这两天终于发布了插件 @tencent-weixin/openclaw-weixin,专门用于接入openclaw等ai agent。这个项目是Nodejs实现,因为openclaw是用Nodejs实现的。
插件是针对openclaw开放的接口,也意味着微信开放了个人号官方机器人接口,本质是:微信官方开放了:私聊消息通道 + 鉴权体系,你完全可以抛开 OpenClaw,直接用这套接口做任何你想做的微信 AI / 机器人 / 服务。
不需要像以前那样,通过破解的方式,还要时刻担心被微信封号了。我也用Python实现了一套简易的SDK,包括登录认证、消息收发、状态管理等功能,并且做了一个Web控制台做例子。
整个过程不算复杂,下面我把代码再过一下,方便大家学习。
Node.js版本:官方插件,功能最完整
先说Node.js版本,这是腾讯官方提供的实现,功能最全,专用于openclaw使用接入。
安装很简单,一行命令搞定:
npx -y @tencent-weixin/openclaw-weixin-cli install
如果报错也可以手动安装:
openclaw plugins install "@tencent-weixin/openclaw-weixin"openclaw config set plugins.entries.openclaw-weixin.enabled true
登录也方便,执行命令后终端会显示二维码,手机扫码确认就行:
openclaw channels login --channel openclaw-weixin
支持的功能包括:
-
二维码登录认证 -
长轮询消息接收 -
文本/图片/视频/文件消息发送 -
CDN上传(AES-128-ECB加密) -
输入状态指示 -
用户授权管理(allowFrom白名单) -
多账号支持 -
上下文隔离
代码结构也很清晰,分成了几个核心模块:
核心模块介绍
1. 认证模块
负责微信登录和账号管理。扫码登录后,令牌会自动保存到本地,无需额外操作。
2. API 通信层
通过 HTTP JSON API 与微信后端网关(https://ilinkai.weixin.qq.com)通信,主要接口包括:
|
|
|
|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
3. 消息处理流水线
-
inbound.ts:接收并解析消息 -
send.ts / send-media.ts:发送文本和媒体文件 -
process-message.ts:消息处理的核心逻辑
4. CDN 与媒体处理
支持图片上传、视频转发、文件传输,并包含图片解密和音频转码功能。
所有接口都是POST请求,返回JSON数据。通用请求头有几个固定的:
|
|
|
|---|---|
Content-Type |
application/json |
AuthorizationType |
ilink_bot_token |
Authorization |
Bearer <token>
|
X-WECHAT-UIN |
|
主要接口有5个:
|
|
|
|
|---|---|---|
|
|
getupdates |
|
|
|
sendmessage |
|
|
|
getuploadurl |
|
|
|
getconfig |
|
|
|
sendtyping |
|
长轮询获取消息
getUpdates 是长轮询接口,服务端在有新消息或超时后返回。
请求体很简单:
{"get_updates_buf": ""}
get_updates_buf 是同步游标,首次请求传空字符串,后续用上次返回的值。
响应体:
{"ret": 0,"msgs": [...],"get_updates_buf": "<新游标>","longpolling_timeout_ms": 35000}
msgs 是消息列表,get_updates_buf 是新的同步游标,下次请求时回传。longpolling_timeout_ms 是服务端建议的下次长轮询超时时间。
发送消息
sendMessage 发送消息给用户。
请求体:
{"msg": {"to_user_id": "<目标用户 ID>","context_token": "<会话上下文令牌>","item_list": [{"type": 1,"text_item": { "text": "你好" }}]}}
type 字段:1=TEXT,2=IMAGE,3=VOICE,4=FILE,5=VIDEO。
这里有个坑:sendmessage 接口成功时返回空对象 {},失败时才返回 {"ret": xxx, "msg": "xxx"}。我一开始只检查 ret=0,导致空对象被当作失败处理,消息发成功了却报错。
CDN上传
如果要发送图片、视频、文件,必须先上传到微信CDN。
流程是:
-
调用 getUploadUrl获取预签名URL -
用AES-128-ECB加密文件 -
上传到CDN -
获得 filekey和downloadEncryptedQueryParam -
用这些参数发送消息
getUploadUrl 请求体:
{"filekey": "<文件标识>","media_type": 1,"to_user_id": "<目标用户 ID>","rawsize": 12345,"rawfilemd5": "<明文 MD5>","filesize": 12352,"thumb_rawsize": 1024,"thumb_rawfilemd5": "<缩略图明文 MD5>","thumb_filesize": 1040}
media_type:1=IMAGE,2=VIDEO,3=FILE。
rawsize 是原文件明文大小,rawfilemd5 是原文件明文MD5,filesize 是AES-128-ECB加密后的密文大小。
图片和视频还需要缩略图,所以有 thumb_rawsize、thumb_rawfilemd5、thumb_filesize 三个字段。
响应体:
{"upload_param": "<原图上传加密参数>","thumb_upload_param": "<缩略图上传加密参数>"}
拿到这些参数后,用AES-128-ECB加密文件,PUT上传到CDN URL,然后用返回的 encrypt_query_param 构造 CDNMedia 引用,放入 MessageItem 发送。
目前Python SDK只实现了文本消息,CDN上传功能还在开发中。如果只需要发文本,现有的SDK已经够用了。
消息结构
微信的消息结构设计得挺合理的。
WeixinMessage 是顶层消息对象:
|
|
|
|
|---|---|---|
seq |
number? |
|
message_id |
number? |
|
from_user_id |
string? |
|
to_user_id |
string? |
|
create_time_ms |
number? |
|
session_id |
string? |
|
message_type |
number? |
1
2 = BOT |
message_state |
number? |
0
1 = GENERATING, 2 = FINISH |
item_list |
MessageItem[]? |
|
context_token |
string? |
|
MessageItem 是消息内容:
|
|
|
|
|---|---|---|
type |
number |
1
2 IMAGE, 3 VOICE, 4 FILE, 5 VIDEO |
text_item |
{ text: string }? |
|
image_item |
ImageItem? |
|
voice_item |
VoiceItem? |
|
file_item |
FileItem? |
|
video_item |
VideoItem? |
|
ref_msg |
RefMessage? |
|
所有媒体类型(图片/语音/文件/视频)通过CDN传输,使用AES-128-ECB加密。CDNMedia 包含两个字段:
|
|
|
|
|---|---|---|
encrypt_query_param |
string? |
|
aes_key |
string? |
|
Python版本Sdk
利用AI编程辅助,实现Python版的sdk,方便Python版本机器人的接入。
代码:https://github.com/peter123023/weixin_bot
本地安装也很简单:
git clone https://github.com/peter123023/weixin-bot-sdk.gitcd weixin-bot-sdkpip install -e .
1. 登录获取Token
from weixin_bot_sdk.auth import AuthManager# 创建认证管理器auth = AuthManager()# 执行登录流程result = auth.login(timeout=300,on_status_change=lambda msg: print(f"状态: {msg}"),auto_display_qr=True)if result.success:print(f"登录成功!Token: {result.bot_token}")# 保存Tokenauth.save_token(result, "weixin_token.txt")
2. 创建机器人
from weixin_bot_sdk import WeixinClient, WeixinConfigfrom weixin_bot_sdk.models import Message# 加载配置config = WeixinConfig.from_token_file("weixin_token.txt")# 创建客户端client = WeixinClient(config)# 定义消息处理器def handle_message(message: Message):text = message.get_text()from_user = message.from_user_idcontext = message.context_tokenif text.lower() == "hello":# 发送输入状态client.send_typing(from_user, True)# 发送回复client.send_text_message(from_user,"你好!我是微信机器人 🤖",context)# 停止输入状态client.send_typing(from_user, False)# 注册处理器client.on_message(handle_message)# 开始接收消息client.start_polling()
example里面有个一个Web控制台,提供图形化demo例子:web_chat。
启动后访问 http://localhost:5000 就能用,功能包括:
-
二维码登录 -
实时消息显示 -
发送消息 -
机器人状态管理 -
消息历史记录 -
重新登录/退出功能


实战经验:踩过的坑
开发过程中踩了一些坑,这里总结一下。
1. 二维码内容是URL不是base64
登录时获取二维码,返回的 qrcode_img_content 是URL,不是base64图片数据。
我一开始以为是图片数。正确的做法是直接用这个URL显示二维码:
# 错误做法img_data = base64.b64decode(qrcode_img_content)# 正确做法qr_code_url = qrcode_img_content
2. sendmessage成功时返回空对象
前面说过,sendmessage 接口成功时返回空对象 {},失败时才返回错误信息。
正确的判断逻辑是:
# 错误做法if result.get("ret") == 0:return True# 正确做法is_success = (len(result) == 0) or (result.get("ret") == 0)if is_success:return True
3. 长轮询超时
频繁出现超时错误,可以增加超时时间:
config = WeixinConfig( long_poll_timeout_ms=60000# 60秒)
4. Token过期
重启后提示token过期,保存token到文件,重启时重新加载:
# 保存tokenwith open("token.txt", "w") as f: f.write(token)# 加载tokenwith open("token.txt", "r") as f: token = f.read()
功能对比
两个版本的功能对比如下:
|
|
|
|
|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Python版本目前不支持的功能主要是CDN上传和用户授权管理。
CDN上传需要实现AES-128-ECB加密,参考Node.js插件的实现就行。用户授权管理可以自己实现简单的白名单机制。
总结
openclaw-weixin = 微信开放了「底层基础设施」:官方消息通道 + 本地 AI 接入标准 。你完全可以抛开 OpenClaw,直接用这套接口做任何你想做的微信 AI / 机器人 / 服务。我整理的Python SDK已经放在项目里了,有需要可以直接用。如果遇到问题,欢迎交流。
https://github.com/peter123023/weixin_bot

夜雨聆风