你也许已经有了一个「看起来还挺像样」的 AI 小助手服务,比如:
- 有 HTTP 接口
/v1/chat; - 能识别不同 Skill(待办、日报、FAQ 等);
- 甚至已经有网页版前端。
但现实是:同事们每天真正打开的是企业微信,很少会专门去打开一个新网页跟机器人聊天。
这篇文章就做一件很实用的小事:
介绍如何在现有 AI 服务基础上,通过添加一层回调适配层,将其接入企业微信群聊。方案无需重写核心逻辑,仅需配置企业微信回调 URL,使用 FastAPI 和 wechatpy 处理消息加解密与转发。步骤包括准备 CorpID 等参数、编写适配器代码、挂载路由、配置 Nginx 及 Docker 环境变量。最终实现在企业微信群中通过@提及触发 AI 回复,支持文本交互及后续扩展如配额控制与多平台对接。
你也许已经有了一个「看起来还挺像样」的 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)
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:用于企业微信验证 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