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

基于 Flask 的「企业微信智能机器人（AI Bot）」回调服务 Python 框架。

如果项目对你有帮助，欢迎点 Star ⭐

---

与「群机器人」版本的区别

维度	群机器人 wecom-bot-svr	智能机器人（本项目）
消息/回复格式	XML	JSON
ReceiveId	corp_id	空字符串 ""
主动推送	Webhook Key	response_url（每条消息携带的临时 URL）
回复类型	text / markdown / file / image / news	text / stream / template_card / stream_with_card
流式消息	不支持	✅ 核心特性（AI 打字机效果）
支持场景	群聊	单聊 + 群聊（@ 机器人）
模板卡片	不支持	✅ 支持
用户反馈	不支持	✅ 点赞/踩事件

---

快速开始

安装

pip install wecom-new-bot-svr

最小示例

from wecom_new_bot_svr import WecomNewBotServer, TextReqMsg, RspStreamMsg, RspEmptyMsg

def msg_handler(msg):
    if isinstance(msg, TextReqMsg):
        return RspStreamMsg(
            stream_id="reply-001",
            content=f"你说了：{msg.content}",
            finish=True,
        )
    return RspEmptyMsg()

def event_handler(msg):
    return RspEmptyMsg()

server = WecomNewBotServer(
    name="my_bot",
    host="0.0.0.0",
    port=5002,
    path="/wecom_bot",
    token="YOUR_TOKEN",
    aes_key="YOUR_AES_KEY",
)
server.set_message_handler(msg_handler)
server.set_event_handler(event_handler)
server.run()

---

配置说明

构造参数

参数	说明	环境变量备选
token	回调 Token	WECOM_BOT_TOKEN
aes_key	回调 AESKey（43 字符）	WECOM_BOT_AES_KEY

推荐使用环境变量注入，不在代码中硬编码：

server = WecomNewBotServer("my_bot", "0.0.0.0", 5002, "/wecom_bot")
# 从环境变量 WECOM_BOT_TOKEN / WECOM_BOT_AES_KEY 自动读取

企业微信后台配置

在企业微信管理后台创建智能机器人后，填入：

• 回调 URL：https://your-domain.com/wecom_bot
• Token：与代码保持一致
• EncodingAESKey：与代码保持一致

---

核心概念

消息类型（接收）

类	msgtype	场景
TextReqMsg	text	单聊 + 群聊
ImageReqMsg	image	单聊
MixedReqMsg	mixed	单聊 + 群聊
VoiceReqMsg	voice	单聊（已转文本）
FileReqMsg	file	单聊
VideoReqMsg	video	单聊
StreamReqMsg	stream	单聊 + 群聊（流式刷新）
EventReqMsg	event	单聊 + 群聊

回复类型（发送）

类	说明
RspTextMsg	纯文本（仅限 enter_chat 事件回复）
RspStreamMsg	流式消息，AI 打字机效果，支持 Markdown
RspTemplateCardMsg	模板卡片
RspStreamWithCardMsg	流式 + 卡片组合
RspUpdateTemplateCardMsg	更新已发出的模板卡片
RspEmptyMsg	空回包（不回复）

流式消息详解

# 单条流式回复（最常用）
return RspStreamMsg(
    stream_id="unique-id-001",   # 首次回复必填，自定义唯一 ID
    content="**AI 回答内容**\n支持 Markdown",
    finish=True,                  # True 表示内容已完整输出
)

# 多段流式推送（generator 形式）
def stream_handler(msg):
    sid = "stream-001"
    yield RspStreamMsg(stream_id=sid, content="正在思考", finish=False)
    time.sleep(1)
    yield RspStreamMsg(stream_id=sid, content="正在思考...", finish=False)
    time.sleep(1)
    yield RspStreamMsg(stream_id=sid, content="最终答案：42", finish=True)

流式刷新（StreamReqMsg）：企业微信收到流式回复后会持续推送刷新事件， 业务层需根据 stream_id 识别并返回对应的下一段内容，直到 finish=True。

---

延迟回复与生成器

框架内置 4 秒超时机制（FIRST_HANDLER_RSP_TIMEOUT_SEC）：

• handler 在 4 秒内返回 → 同步加密回包
• handler 超时 → 先回空包，结果改为通过 response_url 主动推送
• handler 返回 generator → 首条同步回包，后续条主动推送

def slow_handler(msg):
    time.sleep(10)  # 超时，框架自动改为主动推送
    return RspStreamMsg(stream_id="slow-001", content="慢速回复", finish=True)

---

事件处理

from wecom_new_bot_svr import EventReqMsg, RspTextMsg, RspEmptyMsg

def event_handler(msg: EventReqMsg):
    if msg.eventtype == "enter_chat":
        # 用户当天首次进入单聊，可回复文本或流式消息
        return RspTextMsg(content="欢迎使用智能机器人！")

    if msg.eventtype == "template_card_event":
        tce = msg.template_card_event
        print(f"卡片事件: {tce.card_type}, key={tce.event_key}")
        return RspEmptyMsg()

    if msg.eventtype == "feedback_event":
        fe = msg.feedback_event
        print(f"用户反馈: type={fe.type}")
        return RspEmptyMsg()  # 反馈事件只能回复空包

    return RspEmptyMsg()

---

Demo 运行

cd demo
pip install wecom-new-bot-svr
export WECOM_BOT_TOKEN=your_token
export WECOM_BOT_AES_KEY=your_aes_key
python demo.py

Demo 支持的指令（群聊 @ 机器人或单聊发送）：

指令	说明
help	显示帮助信息
sleep N	延迟 N 秒后回复，演示超时主动推送
stream	单条流式消息演示
stream_multi	多段流式推送演示（generator）
其他	回显消息内容

---

相关文档

• 接收消息
• 接收事件
• 被动回复消息
• 回调和回复的加解密方案

Star History