把 AI 小助手接入企业微信：用一个回调接口做群聊机器人实战篇

Ne0inhk

— 9 min read

你也许已经有了一个「看起来还挺像样」的 AI 小助手服务，比如：

  • 有 HTTP 接口 /v1/chat；
  • 能识别不同 Skill（待办、日报、FAQ 等）；
  • 甚至已经有网页版前端。

但现实是：同事们每天真正打开的是企业微信，很少会专门去打开一个新网页跟机器人聊天。

这篇文章就做一件很实用的小事：

在不动你现有 AI 服务核心逻辑的前提下，
用一个企业微信“回调接口”，
把它变成「群聊里的 @ 机器人」。

一、整体思路：后端不重写，只加一层「翻译器」

假设你现在的 AI 服务长这样：

  • 接口：POST /v1/chat

返回：

{ "answer": "上午开会，下午写代码……", "skill": "daily_plan", "duration_ms": 1234 }

请求体：

{ "question": "今天帮我规划下工作", "session_id": "xxx", "user_id": "u_123", "user_level": "normal" }

我们要做的只是：

  1. 在企业微信后台配置一个消息回调 URL（比如 /wechat/callback）；
  2. 在后端写一个超薄适配层：
    • 收到企业微信推来的加密 XML；
    • 解密 → 提取出「用户 ID + 文本内容」；
    • 把「文本内容」转成 question，连同 user_id 一起转发给 /v1/chat；
    • 拿到 answer，再按照企业微信要求的格式加密返回。

可以简单理解为：

企业微信：提供 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_TOKENToken
WECHAT_ENCODING_AES_KEYEncodingAESKey
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 + 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 小助手真正「塞进」日常企业工作流。
  • 具备可扩展空间：
    后续可以按配额、权限、统计、管理后台等方向继续进化，而这一层群聊回调接口几乎无须改变。

Read more

【Python基础】（五）Python 库使用全攻略：从标准库到第三方库，让开发效率翻倍

【Python基础】（五）Python 库使用全攻略：从标准库到第三方库，让开发效率翻倍

目录 编辑 前言 一、Python 库的核心认知：什么是库？为什么要用库？ 1.1 库的本质：现成的 "代码工具箱" 1.2 库的分类：标准库 vs 第三方库 （1）标准库：Python 自带的 "基础工具箱" （2）第三方库：全球开发者共建的 "扩展工具箱" 1.3 使用库的核心优势：效率翻倍的关键 二、标准库实战：内置工具的高效用法 2.1 日期时间处理：datetime库（计算日期差、格式转换） 实战需求：计算你和心爱的人认识多少天 扩展用法：

By Ne0inhk
【c++】智能指针（auto_ptr, unique_ptr, shared_ptr, weak_ptr）

【c++】智能指针（auto_ptr, unique_ptr, shared_ptr, weak_ptr）

小编个人主页详情<—请点击 小编个人gitee代码仓库<—请点击 c++系列专栏<—请点击 倘若命中无此运，孤身亦可登昆仑，送给屏幕面前的读者朋友们和小编自己! 目录 * 前言 * 一、为什么需要智能指针 * 二、智能指针 * RAII * 像指针一样使用 * 三、auto_ptr * 概念讲解 * 模拟实现 * 测试 * 四、unique_ptr * 概念讲解 * 模拟实现 * 测试 * 五、shared_ptr * 概念讲解 * 模拟实现 * 测试 * 六、shared_ptr的循环引用 * 场景引入 * 如何解决循环引用问题 * 七、weak_ptr * 概念讲解 * 模拟实现 * 测试

By Ne0inhk

如何快速实现STL转STEP：制造业工程师的完整指南

如何快速实现STL转STEP：制造业工程师的完整指南 【免费下载链接】stltostpConvert stl files to STEP brep files 项目地址: https://gitcode.com/gh_mirrors/st/stltostp STL转STEP是制造业工程师在3D格式转换过程中经常遇到的关键需求，特别是在CAD数据交换场景下。本文将为您提供STL转STEP工具的完整使用指南，帮助您快速掌握这一实用的3D格式转换技能。🚀 📋 工具概述与核心优势 stltostp是一款专为STL到STEP格式转换设计的命令行工具，具有以下显著特点： * 零依赖：无需OpenCascade或FreeCAD等第三方库 * 智能边缘合并：基于公差值的精确顶点优化 * 双格式支持：兼容ASCII和二进制STL文件 * 跨平台运行：支持Windows和Linux系统 🔧 安装与配置步骤 环境准备 确保系统已安装CMake和C++编译器，然后执行以下命令： git clone https://gitcode.com/gh_mirrors/st/stlt

By Ne0inhk

C++ 中 operator() 重载详解

C++ 中 operator() 重载详解 1. operator() 重载基础概念 1.1 函数对象定义 * 函数对象（Functor）：重载了 operator()的类实例，可以像函数一样被调用 * 语法格式：ReturnType operator()(ParameterList) const * 灵活性：支持多种参数列表的重载版本 1.2 基础示例 classAdder{public:intoperator()(int a,int b)const{return a + b;}};// 调用示例 Adder adder;int result =adder(5,3);// 等价于 adder.operator()(5, 3)

By Ne0inhk