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

你也许已经有了一个「看起来还挺像样」的 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,只需加 wechatpyhttpx 即可。)

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_TOKENEncodingAESKeyCorpID 是否和后台一致;
  • 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.pyfeishu_adapter.py,用同样的“回调 → 转发”模式接入。

十、总结

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

Read more

黄仁勋力荐:OpenClaw不止是下一个ChatGPT,更是AI“动手时代”的破局者

黄仁勋力荐:OpenClaw不止是下一个ChatGPT,更是AI“动手时代”的破局者

在2026年GTC大会上,英伟达创始人兼CEO黄仁勋抛出了一个振聋发聩的判断:“OpenClaw绝对是下一个ChatGPT”。 这一评价并非夸大其词,而是精准点出了AI产业的核心演进方向——从“被动回答”的语言交互,转向“主动行动”的任务执行。ChatGPT开启了大语言模型(LLM)的普及时代,让AI具备了理解和生成人类语言的能力,但它始终停留在“军师”的角色,只能提供方案建议;而OpenClaw的出现,彻底打破了这一局限,将AI变成了能动手干活的“数字员工”,完成了AI从“认知”到“执行”的关键跃迁,成为连接AI能力与现实场景的核心桥梁。 下面我将从技术本质出发,拆解OpenClaw的核心架构、关键技术实现,结合代码示例、架构图与流程图,深入解析其如何实现“行动型AI”的突破,以及为何能被黄仁勋寄予厚望,成为AI产业的下一个里程碑。 一、认知跃迁:从“回答型AI”到“行动型AI”的本质区别 要理解OpenClaw的价值,首先需要明确它与ChatGPT这类“回答型AI”的核心差异。

【AI】高效交互的艺术:AI提示工程与大模型对话指南

【AI】高效交互的艺术:AI提示工程与大模型对话指南

🔥小龙报:个人主页 🎬作者简介:C++研发,嵌入式,机器人等方向学习者 ❄️个人专栏:《AI》 ✨ 永远相信美好的事情即将发生 文章目录 * 前言 * 一、ChatatGPT介绍 * 二、什么是提示工程? * 三、大语言模型的底层原理 * 四、AI的相关术语 * 五、如何与AI(以ChatatGPT为例)更好交流 * 5.1 使用AI的核心 * 5.2 提示组成结构 * 5.3 创建好的提示的策略 * 5.4 提示的类别 * 5.5 创建在和AI提示的进阶框架 * 5.6如何减少AI回答的空洞无味感 * 5.7 如何提高AI回答的可读性 * 六、使用AI的更多技巧 * 6.1 高效提示的原则 * 6.

Stable Diffusion v1.5技术解密:从创意到商业化的AI图像生成革命

Stable Diffusion v1.5技术解密:从创意到商业化的AI图像生成革命 【免费下载链接】stable_diffusion_v1_5Stable Diffusion is a latent text-to-image diffusion model capable of generating photo-realistic images given any text input. 项目地址: https://ai.gitcode.com/openMind/stable_diffusion_v1_5 还在为AI图像生成的复杂技术而头疼吗?想要快速掌握这个价值数十亿美元市场背后的核心技术吗?今天,我将带你深入Stable Diffusion v1.5的技术内核,揭秘它如何从实验室走向商业化应用的完整路径。无论你是技术开发者、内容创作者还是企业决策者,这篇文章都将成为你进入AIGC世界的通行证 🚀 开篇:AI图像生成的三大痛点与解决方案 在深入技术细节之前,

七大AIGC测试工具横向评测:赋能软件测试的AI利器

七大AIGC测试工具横向评测:赋能软件测试的AI利器

在AI技术迅猛发展的2025年,AIGC(人工智能生成内容)工具已深度融入软件测试领域,显著提升测试效率和质量。本次评测聚焦七大主流工具:CodeWhisperer、GitHub Copilot、Testim、Selenium AI、Test.ai、Mabl和Functionize。评测基于实际测试场景(如Web/API测试、移动端兼容性验证),从核心功能、优缺点、适用性及成本四维度展开。目标是为测试工程师提供数据驱动的决策参考。评测方法包括工具实测(使用Python/Java测试脚本)、用户反馈分析(来源Stack Overflow和GitHub议题)及性能基准测试(错误检测率、执行速度)。以下是详细横向比较。 一、工具核心功能与评测结果 1. Amazon CodeWhisperer * 功能亮点:基于AWS的AI代码助手,专精于测试脚本生成。支持Python、Java等语言,能自动补全测试用例(如Selenium脚本),并集成漏洞扫描。实测中,生成100行测试代码的平均时间仅5秒,错误率低于5%。 * 优点: