将 AI 小助手接入企业微信：基于回调接口的群聊机器人实现

可以简单理解为：

企业微信：提供 UI 和消息推送 你的 AI 小助手：负责'想答案' 中间这个回调接口：负责「翻译 + 转发 + 回填」

二、企业微信端：你必须准备的 3 个关键参数

在写代码前，先在企业微信后台把几个关键信息准备好。

2.1 创建自建应用 / 机器人

1. 管理员登录企业微信后台；
2. 进入【应用管理】 →【自建】→【创建应用】；
3. 按提示填写名称、图标即可。

创建完成后，记下：

• CorpID（企业 ID）
• AgentID（应用 ID）
• Secret（应用密钥）

其中 CorpID 稍后会和 Token、EncodingAESKey 一起用于解密； Secret 一般用于你「主动调用企业微信接口」，本篇只做回调，不用到。

2.2 配置消息回调

在刚刚创建的自建应用里，找到：

• 「接收消息」或「回调配置」模块

配置三项：

1. URL：你后端可公网访问的地址，比如 https://your-domain.com/wechat/callback
2. Token：自己随便写一个，比如 my_wechat_token_2024
3. EncodingAESKey：企业微信自动生成的一串 43 位字符串

这三项记下来，对应到代码里的环境变量：

三、项目结构：只新增一个 wechat_adapter.py

假设你当前项目结构：

project/
├── app/
│   ├── server.py # 已有：包含 /v1/chat
│   └── ...
├── docker-compose.yml
└── Dockerfile

现在只需要：

• 在 app/ 下加一个 wechat_adapter.py，做企业微信适配；
• 在 server.py 里把这个 router 挂起来。

四、后端实现：FastAPI + wechatpy

这里用 FastAPI 举例（Flask 思路完全一样），并使用 wechatpy 来处理消息加解密，避免自己手写 AES。

4.1 安装依赖

pip install fastapi uvicorn wechatpy httpx

（如果你已经有 FastAPI，只需加 wechatpy 和 httpx 即可。）

4.2 编写 wechat_adapter.py

# app/wechat_adapter.py
import os
from fastapi import APIRouter, Request
from fastapi.responses import PlainTextResponse
from wechatpy.enterprise.crypto import WeChatCrypto
from wechatpy.enterprise import parse_message, create_reply
import httpx

# 从环境变量读取配置
WECHAT_TOKEN = os.getenv("WECHAT_TOKEN")
WECHAT_ENCODING_AES_KEY = os.getenv("WECHAT_ENCODING_AES_KEY")
WECHAT_CORP_ID = os.getenv("WECHAT_CORP_ID")

# 你自己的 AI 服务接口
AGENT_BACKEND_URL = os.getenv("AGENT_BACKEND_URL", "http://localhost:8000/v1/chat")

router = APIRouter()

@router.get("/callback")
async def wechat_verify(
    msg_signature: str,
    timestamp: str,
    nonce: str,
    echostr: str
):
    """
    企业微信在配置回调 URL 时，会发一个 GET 请求校验。
    要做的就是：用 WeChatCrypto 解密 echostr，然后原样返回。
    """
    crypto = WeChatCrypto(WECHAT_TOKEN, WECHAT_ENCODING_AES_KEY, WECHAT_CORP_ID)
    try:
        echostr_decoded = crypto.decrypt_message(echostr, msg_signature, timestamp, nonce)
        return PlainTextResponse(content=echostr_decoded)
    except Exception:
        # 验签失败就返回空，企业微信会认为不通过
        return PlainTextResponse(content="")

@router.post("/callback")
async def wechat_callback(request: Request):
    """
    真正收消息的入口。
    1. 解密 XML
    2. 解析文本内容
    3. 调用内部 AI 服务
    4. 构造并加密回复
    """
    query = request.query_params
    msg_signature = query.get("msg_signature")
    timestamp = query.get("timestamp")
    nonce = query.get("nonce")
    raw_xml = await request.body()

    crypto = WeChatCrypto(WECHAT_TOKEN, WECHAT_ENCODING_AES_KEY, WECHAT_CORP_ID)
    try:
        msg_xml = crypto.decrypt_message(raw_xml, msg_signature, timestamp, nonce)
    except Exception:
        return PlainTextResponse(content="")

    # 解析企业微信 XML
    msg = parse_message(msg_xml)

    # 只处理文本消息
    if msg.type != "text":
        reply = create_reply("暂时只支持文本消息哦～", msg)
        encrypted = crypto.encrypt_message(reply.render(), nonce, timestamp)
        return PlainTextResponse(content=encrypted)

    # 用户 ID（企业微信里的 userid）
    user_id = msg.source
    content = msg.content.strip()

    # [可选] 去掉前面的 @机器人 文本，只保留真正问题
    # 例如："@AI 小助手 帮我写日报" → "帮我写日报"
    if content.startswith("@"):
        idx = content.find(" ")
        if idx != -1:
            content = content[idx + 1 :].strip()

    # 调用你自己的 AI 接口
    async with httpx.AsyncClient(timeout=30) as client:
        try:
            resp = await client.post(
                AGENT_BACKEND_URL,
                json={
                    "question": content,
                    "session_id": f"wx_{user_id}",
                    "user_id": f"wx_{user_id}",
                    "user_level": "normal",
                }
            )
            data = resp.json()
            answer = data.get("answer", "后端没有返回 answer 字段，请检查。")
        except Exception as e:
            answer = f"后端服务异常：{e}"

    # 构造回复 XML 并加密
    reply = create_reply(answer, msg)
    encrypted = crypto.encrypt_message(reply.render(), nonce, timestamp)
    return PlainTextResponse(content=encrypted)

4.3 在 server.py 中挂载 router

# app/server.py
from fastapi import FastAPI
from app.wechat_adapter import router as wechat_router

app = FastAPI(title="My AI Agent Service")

# 原有 /v1/chat 等接口
# @app.post("/v1/chat")
# async def chat(...):
#     ...

# 新增企业微信路由
app.include_router(wechat_router, prefix="/wechat")

本地启动（开发阶段）：

uvicorn app.server:app --host 0.0.0.0 --port 8000 --reload

现在，你有了如下几个接口：

• POST /v1/chat：原来的 AI 接口
• GET /wechat/callback：用于企业微信验证 URL
• POST /wechat/callback：用于接收真实消息并回复

五、让企业微信找到你的服务：Nginx / 域名配置

企业微信需要访问你服务器的公网地址，所以一般会放在 Nginx 之后。

假设你有一个域名 https://bot.example.com，Nginx 配置大致如下（核心是 /wechat/ 这段）：

server {
    listen 80;
    server_name bot.example.com;

    location / {
        proxy_pass http://app:8000/;
    }

    # 专门给企业微信回调用
    location /wechat/ {
        proxy_pass http://app:8000/wechat/;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

重载 Nginx：

nginx -s reload

然后在企业微信后台把回调 URL 填写为：

https://bot.example.com/wechat/callback

保存时，企业微信会向这个 URL 发一个 GET 请求进行签名验证。只要你代码和配置写对，后台就会提示「验证通过」。

六、Docker Compose 中增加环境变量（如在容器里跑）

如果你是用 docker-compose 启动 FastAPI 应用，在 docker-compose.yml 里为 app 服务增加环境变量：

services:
  app:
    build: .
    environment:
      - OPENAI_API_KEY=sk-xxxx
      - WECHAT_TOKEN=my_wechat_token_2024
      - WECHAT_ENCODING_AES_KEY=你的 43 位 AESKey
      - WECHAT_CORP_ID=你的企业 CorpID
      - AGENT_BACKEND_URL=http://app:8000/v1/chat
    ports:
      - "8000:8000"

重新构建并启动：

docker-compose up -d --build

七、在企业微信群里真实体验一把

1. 确保：
   • 你的 app 服务已启动；
   • Nginx 反向代理配置正确；
   • 企业微信后台的回调验证已经通过。
2. 在企业微信中：
   • 找到你刚刚创建的自建应用 / 机器人；
   • 将它添加到某个测试群。
3. 对同事来说，这就是一个普通的群聊机器人：
   • 不需要任何技术知识；
   • 会记住对话上下文（取决于你 /v1/chat 的实现）；
   • 会调用你现有的多技能 / RAG / 缓存逻辑。

在群里直接 @ 它，例如：

张三：@AI 小助手 帮我写一份今天的工作计划：
上午要开会，下午写代码，晚上看两篇技术文章。
AI 小助手：好的，这是你今天的简要工作计划：
1. 上午：准备和参加会议，记录行动项……
2. 下午：根据会议结论编写代码……
3. 晚上：选择两篇和当前项目相关的技术文章……

八、简单故障排查 Checklist

1. 企业微信后台提示「URL 验证失败」

重点检查：

• WECHAT_TOKEN、EncodingAESKey、CorpID 是否和后台一致；
• URL 是否和你代码里的路由匹配（是否带 /wechat/callback 前缀）；
• Nginx 是否正确转发到你的 app:8000/wechat/。

可以在回调接口里先打印日志，确认 GET 请求有没有到后端。

2. 群里 @ 机器人没反应

可能原因：

1. /wechat/callback 配置的是内网地址，企业微信访问不到；
2. 你的 AI 服务 /v1/chat 报错，导致 answer 为空或超时；
3. 机器人没有被拉进当前群。

排查方法：

• 先在后端打印收到的 msg.content，确保消息已经到达你服务；
• 再单独用 curl 或 Postman 测试 /v1/chat 是否工作正常。

3. 机器人一直回复「只支持文本」

说明你发送的是图片、文件等非文本类型，而代码里只接受 msg.type == "text"。可以根据实际需要扩展对其他类型的支持。

九、进阶扩展思路（可选）

当基础版本跑通后，你可以逐步加一些'工程化'功能：

1. 按用户配额控制用量
   • 把 user_id = msg.source 传给 /v1/chat；
   • 在 AI 服务侧按人统计 Token 消耗，设置每日上限。
2. 按部门开放不同 Skill
   • 若能从企业微信接口拿到部门信息，可在内部配置文件中，为不同部门开放不同功能。
3. 对接钉钉 / 飞书
   • 保持 /v1/chat 不变；
   • 新增 dingtalk_adapter.py、feishu_adapter.py，用同样的'回调 → 转发'模式接入。

十、总结

• 不动核心，只加一层适配： 企业微信只是换了一个入口，你的 AI 逻辑仍只暴露 /v1/chat。
• 成本低，上手快： 一份 wechat_adapter.py + 一点 Nginx / Docker 配置，就能把你的 AI 小助手真正「塞进」日常企业工作流。
• 具备可扩展空间： 后续可以按配额、权限、统计、管理后台等方向继续进化，而这一层群聊回调接口几乎无须改变。