企业微信智能机器人 WebSocket 服务框架

这是 wecom-bot-svr 的长连接版本，面向企业微信「智能机器人」API 模式的 WebSocket 接入。

它和旧的群机器人回调服务不同：不需要公网回调地址、Token、EncodingAESKey，也不需要 Flask。服务进程会主动连接企业微信 WebSocket 网关 wss://openws.work.weixin.qq.com，使用 Bot ID 和 Secret 认证后接收消息。

参考文档：

• 企业微信智能机器人接入文档：https://developer.work.weixin.qq.com/document/path/101463
• 底层 SDK：wecom-aibot-sdk-python

特性

• WebSocket 长连接：主动连接企业微信网关，不需要公网回调地址。
• 自动认证：连接建立后使用 Bot ID 和 Secret 订阅机器人消息。
• 心跳保活：认证成功后按固定间隔发送心跳，检测连接状态。
• 自动重连：连接异常断开后自动指数退避重连，默认无限重试。
• 本地可调试：本地机器能访问 openws.work.weixin.qq.com 即可启动调试。
• 消息封装：封装文本、图片、语音、视频、文件、图文混排、事件等输入消息。
• 媒体上传：图片、语音、视频、文件回复会先通过 WebSocket 上传，再回复 media_id。
• 流式回复：文本和 Markdown 默认通过企微 AI Bot 的 stream 消息完成回复。

为什么本地启动也可以使用

WebSocket 长连接模式下，本项目进程扮演的是客户端角色，而不是公网 HTTP 服务端：

本地机器 / 内网机器 -> wss://openws.work.weixin.qq.com

程序启动后会主动连到企业微信 WebSocket 网关，并通过 Bot ID 和 Secret 订阅机器人消息。企业微信后续把消息沿这条长连接推送给本地进程，本地进程再通过同一条连接回复消息。

因此开发调试时可以直接在本地启动，不需要：

• 公网 IP
• 域名和 HTTPS 证书
• 回调 URL
• Token / EncodingAESKey
• 开放本机端口

只要本地机器能访问 wss://openws.work.weixin.qq.com，并且 Bot ID / Secret 配置正确，就可以收到消息并回复。

需要注意：

• 本地进程必须持续运行，终端关闭、电脑休眠或网络断开都会导致机器人离线。
• 同一个 Bot 通常只应保持一个长连接实例，多处同时启动可能互相踢下线。
• 如果公司网络拦截 WebSocket 或外网 TLS，需要放行 openws.work.weixin.qq.com。
• 长期给团队使用时，建议部署到常驻机器、CVM、办公室内网主机或后台进程中。

安装

pip install -e .

配置

export WECOM_BOT_ID="你的 Bot ID"
export WECOM_BOT_SECRET="你的 Secret"
export WECOM_BOT_NAME="机器人名称"
# 可选：默认 wss://openws.work.weixin.qq.com
export WECOM_BOT_WS_URL="wss://openws.work.weixin.qq.com"

连接参数

WecomAIBotWsServer 暴露了底层长连接相关参数：

server = WecomAIBotWsServer(
    reconnect_interval=1000,
    max_reconnect_attempts=-1,
    heartbeat_interval=30000,
    request_timeout=10000,
)

参数说明：

• reconnect_interval：初始重连间隔，单位毫秒，默认 1000。
• max_reconnect_attempts：最大重连次数，默认 -1，表示无限重连。
• heartbeat_interval：心跳间隔，单位毫秒，默认 30000。
• request_timeout：请求超时，单位毫秒，默认 10000。

默认情况下，只要不是手动退出进程、凭据错误，或同一个 Bot 被其他长连接实例踢下线，框架会持续保活并在异常断开后自动重连。

使用

from wecom_ai_bot_ws_svr import RspTextMsg, WecomAIBotWsServer


def msg_handler(req_msg, server):
    if req_msg.msg_type == "text":
        return RspTextMsg(f"你说：{req_msg.content}")
    return RspTextMsg(f"msg_type: {req_msg.msg_type}")


server = WecomAIBotWsServer(name="my-bot")
server.set_message_handler(msg_handler)
server.run()

也可以直接运行 demo：

python demo/demo.py

Handler 返回值

msg_handler(req_msg, server) 可以返回：

• RspTextMsg
• RspMarkdownMsg
• RspStreamMsg
• RspImageMsg(file_path="...")
• RspVoiceMsg(file_path="...")
• RspVideoMsg(file_path="...")
• RspFileMsg(file_path="...")
• RspTemplateCardMsg
• dict，直接作为企微回复 body
• generator，连续 yield 多条响应

文本和 Markdown 默认通过企微 AI Bot 的 stream 消息完成回复；图片、语音、视频、文件会先走 WebSocket 媒体上传，再回复 media_id。

Handler 输入消息

msg_handler 收到的 req_msg 会按 msgtype 封装成具体类型：

• TextReqMsg：文本消息，读取 req_msg.content
• ImageReqMsg：图片消息，读取 req_msg.image_url / req_msg.aes_key
• VoiceReqMsg：语音消息，读取 req_msg.content，如果有文件地址则读取 req_msg.voice_url / req_msg.aes_key
• VideoReqMsg：视频消息，读取 req_msg.video_url / req_msg.aes_key / req_msg.name / req_msg.size
• FileReqMsg：文件消息，读取 req_msg.file_url / req_msg.aes_key / req_msg.name / req_msg.size
• MixedMessageReqMsg：混合消息，读取 req_msg.msg_items；子项支持 MixedTextItem、MixedImageItem、MixedVoiceItem、MixedVideoItem、MixedFileItem

图片、文件、视频等带 url 和 aes_key 的消息，可以通过 await server.download_file(url, aes_key) 下载并解密。

主动发送

在异步上下文里可以调用：

await server.send_markdown(chat_id, "### 主动消息")
await server.send_image(chat_id, "/path/to/image.png")
await server.send_voice(chat_id, "/path/to/voice.amr")
await server.send_video(chat_id, "/path/to/video.mp4")
await server.send_file(chat_id, "/path/to/file.pdf")

chat_id 在收到消息的 req_msg.chat_id 中可取到。单聊场景通常是用户 ID，群聊场景是群 ID。