介绍如何在保留现有 AI 服务核心逻辑的前提下，通过在企业微信后台配置消息回调 URL，并编写一个适配层（使用 FastAPI 和 wechatpy），将 AI 小助手变成企业微信群聊中的@机器人。文章涵盖了后端代码实现、Nginx 配置、Docker Compose 环境变量设置以及故障排查指南，实现了低成本、易上手的群聊机器人集成方案。
你也许已经有了一个「看起来还挺像样」的 AI 小助手服务，比如：
/v1/chat；但现实是：同事们每天真正打开的是企业微信，很少会专门去打开一个新网页跟机器人聊天。
这篇文章就做一件很实用的小事：
在不动你现有 AI 服务核心逻辑的前提下，
用一个企业微信'回调接口'，
把它变成「群聊里的 @ 机器人」。
假设你现在的 AI 服务长这样：
POST /v1/chat返回：
{
"answer": "上午开会，下午写代码……",
"skill": "daily_plan",
"duration_ms": 1234
}
请求体：
{
"question": "今天帮我规划下工作",
"session_id": "xxx",
"user_id": "u_123",
"user_level": "normal"
}
我们要做的只是：
/wechat/callback）；question，连同 user_id 一起转发给 /v1/chat；answer，再按照企业微信要求的格式加密返回。可以简单理解为：
企业微信：提供 UI 和消息推送
你的 AI 小助手：负责'想答案'
中间这个回调接口：负责「翻译 + 转发 + 回填」
在写代码前，先在企业微信后台把几个关键信息准备好。
创建完成后，记下：
其中 CorpID 稍后会和 Token、EncodingAESKey 一起用于解密；
Secret 一般用于你「主动调用企业微信接口」，本篇只做回调，不用到。
在刚刚创建的自建应用里，找到：
配置三项：
https://your-domain.com/wechat/callbackmy_wechat_token_2024这三项记下来，对应到代码里的环境变量：
| 环境变量名 | 对应企业微信配置 |
|---|---|
WECHAT_TOKEN | Token |
WECHAT_ENCODING_AES_KEY | EncodingAESKey |
WECHAT_CORP_ID | 企业 CorpID |
wechat_adapter.py假设你当前项目结构：
project/
├── app/
│ ├── server.py # 已有：包含 /v1/chat
│ └── ...
├── docker-compose.yml
└── Dockerfile
现在只需要：
app/ 下加一个 wechat_adapter.py，做企业微信适配；server.py 里把这个 router 挂起来。这里用 FastAPI 举例（Flask 思路完全一样），并使用 wechatpy 来处理消息加解密，避免自己手写 AES。
pip install fastapi uvicorn wechatpy httpx
（如果你已经有 FastAPI，只需加 wechatpy 和 httpx 即可。）
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)\n```
#### 4.3 在 `server.py` 中挂载 router
```python
# 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：用于企业微信验证 URLPOST /wechat/callback：用于接收真实消息并回复企业微信需要访问你服务器的公网地址，所以一般会放在 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 启动 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
app 服务已启动；/v1/chat 的实现）；在群里直接 @ 它，例如：
张三：@AI 小助手 帮我写一份今天的工作计划：
上午要开会，下午写代码，晚上看两篇技术文章。
AI 小助手：
好的，这是你今天的简要工作计划：
1. 上午：准备和参加会议，记录行动项……
2. 下午：根据会议结论编写代码……
3. 晚上：选择两篇和当前项目相关的技术文章……
重点检查：
WECHAT_TOKEN、EncodingAESKey、CorpID 是否和后台一致；/wechat/callback 前缀）；app:8000/wechat/。可以在回调接口里先打印日志，确认 GET 请求有没有到后端。
可能原因：
/wechat/callback 配置的是内网地址，企业微信访问不到；/v1/chat 报错，导致 answer 为空或超时；排查方法：
msg.content，确保消息已经到达你服务；curl 或 Postman 测试 /v1/chat 是否工作正常。说明你发送的是图片、文件等非文本类型，而代码里只接受 msg.type == "text"。可以根据实际需要扩展对其他类型的支持。
当基础版本跑通后，你可以逐步加一些'工程化'功能：
user_id = msg.source 传给 /v1/chat；/v1/chat 不变；dingtalk_adapter.py、feishu_adapter.py，用同样的'回调 → 转发'模式接入。/v1/chat。wechat_adapter.py + 一点 Nginx / Docker 配置，就能把你的 AI 小助手真正「塞进」日常企业工作流。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online