乐于分享
好东西不私藏

详解微信龙虾插件openclaw-weixin源码,并实现一个Python版本的简易sdk

详解微信龙虾插件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)通信,主要接口包括:

接口
功能
getUpdates
长轮询获取新消息
sendMessage
发送文本/图片/视频/文件
getUploadUrl
获取CDN上传预签名
sendTyping
发送输入状态提示
getConfig
获取账号配置(typing ticket 等)

3. 消息处理流水线

  • inbound.ts:接收并解析消息
  • send.ts / send-media.ts:发送文本和媒体文件
  • process-message.ts:消息处理的核心逻辑

4. CDN 与媒体处理

支持图片上传、视频转发、文件传输,并包含图片解密和音频转码功能。

接口详情

所有接口都是POST请求,返回JSON数据。通用请求头有几个固定的:

Header
说明
Content-Type application/json
AuthorizationType
固定值 ilink_bot_token
Authorization Bearer <token>

(登录后获取)
X-WECHAT-UIN
随机 uint32 的 base64 编码

主要接口有5个:

接口
路径
说明
getUpdates
getupdates
长轮询获取新消息
sendMessage
sendmessage
发送消息(文本/图片/视频/文件)
getUploadUrl
getuploadurl
获取 CDN 上传预签名 URL
getConfig
getconfig
获取账号配置(typing ticket 等)
sendTyping
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。

流程是:

  1. 调用 getUploadUrl 获取预签名URL
  2. 用AES-128-ECB加密文件
  3. 上传到CDN
  4. 获得 filekey 和 downloadEncryptedQueryParam
  5. 用这些参数发送消息

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_rawsizethumb_rawfilemd5thumb_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?
消息唯一 ID
from_user_id string?
发送者 ID
to_user_id string?
接收者 ID
create_time_ms number?
创建时间戳(ms)
session_id string?
会话 ID
message_type number? 1

 = USER, 2 = BOT
message_state number? 0

 = NEW, 1 = GENERATING, 2 = FINISH
item_list MessageItem[]?
消息内容列表
context_token string?
会话上下文令牌,回复时需回传

MessageItem 是消息内容:

字段
类型
说明
type number 1

 TEXT, 2 IMAGE, 3 VOICE, 4 FILE, 5 VIDEO
text_item { text: string }?
文本内容
image_item ImageItem?
图片(含 CDN 引用和 AES 密钥)
voice_item VoiceItem?
语音(SILK 编码)
file_item FileItem?
文件附件
video_item VideoItem?
视频
ref_msg RefMessage?
引用消息

所有媒体类型(图片/语音/文件/视频)通过CDN传输,使用AES-128-ECB加密。CDNMedia 包含两个字段:

字段
类型
说明
encrypt_query_param string?
CDN 下载/上传的加密参数
aes_key string?
base64 编码的 AES-128 密钥

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}")    # 保存Token    auth.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_id    context = message.context_token    if 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) == 0or (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()

功能对比

两个版本的功能对比如下:

功能
Node.js版本
Python版本
二维码登录
文本消息
图片消息
⏳ 开发中
视频消息
⏳ 开发中
文件消息
⏳ 开发中
CDN上传
⏳ 开发中
输入状态
用户授权
多账号
上下文隔离
日志上传
监控调试

Python版本目前不支持的功能主要是CDN上传和用户授权管理。

CDN上传需要实现AES-128-ECB加密,参考Node.js插件的实现就行。用户授权管理可以自己实现简单的白名单机制。

总结

openclaw-weixin = 微信开放了「底层基础设施」:官方消息通道 + 本地 AI 接入标准 。你完全可以抛开 OpenClaw,直接用这套接口做任何你想做的微信 AI / 机器人 / 服务
我整理的Python SDK已经放在项目里了,有需要可以直接用。如果遇到问题,欢迎交流。

https://github.com/peter123023/weixin_bot

本站文章均为手工撰写未经允许谢绝转载:夜雨聆风 » 详解微信龙虾插件openclaw-weixin源码,并实现一个Python版本的简易sdk

猜你喜欢

  • 暂无文章