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

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

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

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

快速开始

安装

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()

配置说明

构造参数

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

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：与代码保持一致

核心概念

消息类型（接收）

回复类型（发送）

流式消息详解

# 单条流式回复（最常用）
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 支持的指令（群聊 @ 机器人或单聊发送）：

相关文档

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

Star History