前几天组里一个小伙伴跑过来问我,能不能把 Workbuddy 接到企业微信里,让大家直接在企微里发一句“创建任务 明天下午三点 给客户回个电话”,Workbuddy 就能自动把任务建好、到点再提醒。这个需求我太熟了。以前我们内部工具只有 Web 后台,大家要切过去看板,消息发出去没人及时响应,接入企微之后,相当于把“人找系统”变成了“系统找人”。我把完整接入过程整理了一遍,从企微后台配置、回调服务、消息解密到主动推送,基本是一条主线走完,踩过的坑也标出来,给后来想接企微又不想从零看文档的朋友省点时间。
1. 动手之前,先确认接入方案和消息链路
1.1 Workbuddy 到底是什么,接入企微能带来什么
Workbuddy 在团队里定位是一个内部任务智能助手,能解析自然语言指令、管理任务、到点提醒、生成简单汇报。它原本跑在内部服务器上,有 Web 页面,也有命令行入口。但实际用下来发现,大家不会每天都打开后台,反而企业微信天天开着。所以接入企微的核心目的,就是把人最常待的地方变成 Workbuddy 的前端:用户在企微里发一条消息,Workbuddy 能听懂并响应,主动提醒也能直接推送到企微的消息列表里。
接入不是简单加一个 webhook,而是要打通一条双向链路。上行链路是用户在企微应用里发消息,企微把消息推送到回调服务,Workbuddy 解析后执行命令;下行链路是 Workbuddy 产生结果或定时提醒,通过企微 API 主动发送消息给指定用户。如果你在规划时只想着“能发消息”,后面很容易漏掉回调这一半,导致用户根本没法在企微里给 Workbuddy 下指令。
1.2 三种接入方式对比:群机器人、自建应用、第三方平台
我一开始也想图省事,直接用群机器人 webhook。它确实最快,只要拿一个 webhook 地址,往群里发文本或 markdown,十分钟就能跑通。但它只能单向推送,不能接收用户消息。Workbuddy 是需要双向对话的,用户要发“创建任务”进来,群机器人完全做不到。
第二种是企微自建应用,这也是我最终采用的方案。它能收消息、发消息,还能给指定用户推提醒,消息列表里会显示 Workbuddy 的应用名,体验很原生。代价是需要企业管理员权限,要配置回调服务,开发量稍微多一些。
第三种是接入第三方开放平台,面向开发者做对外分发,需要走应用审核流程,适合做公开的 SaaS 产品,不适合咱们内部快速自用。对比下来,自建应用是 Workbuddy 这种内部助手的最优解。
| 接入方式 | 消息方向 | 开发成本 | 适合场景 |
|---|---|---|---|
| 群机器人 webhook | 只能单向发送 | 极低 | 告警通知、定时推送 |
| 自建应用 | 双向收发,可指定成员 | 中等 | 内部智能助手、对话式工具 |
| 第三方开放平台 | 双向收发 | 高,需审核 | 对外分发、SaaS 产品 |
1.3 消息链路设计上的两个关键点
第一,自建应用的消息路径依赖回调推送。用户在企微里对你的应用发消息,企微会向你在后台配置的“接收消息服务器 URL”发起 POST 请求,请求体是加密的 XML。服务需要解密、解析消息内容,再返回响应。这里要注意,企微不希望回调服务阻塞太长时间,如果处理逻辑比较重,应该先返回空串,再用异步任务处理,避免企微反复重试。
第二,主动推送不能直接查一下用户就发。企微对外发送消息走的是应用消息接口,需要先获取 access_token,再带 agentid 调用。很多人在这里把“被动回复”和“主动推送”搞混:被动回复通常在回调接口里返回加密 XML,主动推送则是一个独立的 HTTP API。把这两个概念分开,后面才不会乱。
2. 前置准备:企业微信后台配置和参数获取
2.1 创建自建应用,拿到三个核心参数
接入企业微信的第一步,是让管理员在管理后台创建一个自建应用。路径一般是:企业微信管理后台 -> 应用管理 -> 自建 -> 创建应用。需要填写应用名称、Logo,选择可见范围。创建完成后,你会得到三个最关键的参数:Corp ID、AgentId、Secret。
| 参数 | 含义 | 使用位置 |
|---|---|---|
| Corp ID | 企业 ID,唯一标识你的企业 | 获取 access_token、回调验证签名 |
| AgentId | 应用 ID,标识你的应用 | 发消息时必须带上 |
| Secret | 应用密钥,用来获取 access_token | 服务端保存,不能泄露 |
Corp ID 在企业微信管理后台“我的企业”里可以看到,AgentId 和 Secret 在自建应用的详情页里。特别提醒,Secret 一般只在创建或重置时完整显示一次,建议复制到配置中心或环境变量里,不要直接硬编码在代码中,更不要提交到 Git。
2.2 配置接收消息服务器 URL
在自建应用详情页里,找到“接收消息”配置。企业微信要求填三个东西:URL、Token、EncodingAESKey。URL 必须是公网可访问的 HTTPS 地址,比如https://wb.example.com/qw/callback,Token 是你自己定义的任意字符串,EncodingAESKey 可以随机生成,也可以自己填一组 43 位字符。
配置后企微会发起一次验证请求,这时服务需要响应一个 GET 请求。我看到不少人在这里卡住,因为企微要求你正确解密echostr并原样返回。如果返回的不是明文 echostr,后台会一直提示“验证失败”。这个环节只要通了,后续的 POST 消息回调才能进得来。
2.3 配置可信 IP 和通讯录权限
调用企微 API 时,有个常见错误码 60020,表示当前访问 IP 不在白名单。所以在后台的“企业可信 IP”里,要把部署服务的服务器公网 IP 加进去。如果是本地联调,不要把笔记本 IP 写进去,而是直接把回调服务放在测试服务器上,用测试机的出口 IP 绑定白名单,这样后续部署不会因为换 IP 而踩坑。
另外,应用的可见范围决定了哪些用户能使用这个应用,同时也会限制调用通讯录 API 时可读取的成员范围。如果你希望按部门推送消息,要保证应用可见范围覆盖对应部门。这个权限是跟着应用一起走的,别光配了接口权限,忘了可见范围,否则发消息时会报“用户不在可见范围”。
3. 本地先跑通回调验证和消息解密
3.1 为什么企微的消息体要做加密
企业微信回调消息采用 AES 加密,并且带一个 msg_signature 签名。加密的原因很简单:企微推送的是员工在聊天框里的内容,属于敏感信息,不能以明文形式裸奔到公网。所以它在推送时用 AES-CBC 加密,加密用的 key 由 EncodingAESKey 派生出来,签名用的 token 则是你自己配的 Token。
这个设计意味着,回调服务必须实现“验证签名 -> 解密 -> 解析明文”三步。好消息是企微官方提供了各语言的 SDK,比如 Python 版本里的 WXBizMsgCrypt 类,建议直接用,不要自己重写加解密。自己实现一个看似不难,但一旦字符编码、Base64 填充出问题,排查成本极高,而且企微文档里对细节的说明粒度有限,踩坑容易上头。
3.2 最小回调服务长什么样
这里给一个 Flask 实现的骨架,核心就是两条路由:GET 处理验证,POST 处理消息。代码里的decrypt_message和encrypt_message是对官方 WXBizMsgCrypt 的封装,实际对接时直接替换成官方 SDK 对应方法即可。
from flask import Flask, request from workbuddy_qw import decrypt_message import xml.etree.ElementTree as ET app = Flask(__name__) TOKEN = "your_token" ENCODING_AES_KEY = "your_encoding_aes_key" CORP_ID = "your_corp_id" @app.route("/qw/callback", methods=["GET", "POST"]) def qw_callback(): if request.method == "GET": # 验证回调 URL msg_signature = request.args.get("msg_signature") timestamp = request.args.get("timestamp") nonce = request.args.get("nonce") echostr = request.args.get("echostr") ret, reply = decrypt_echostr(msg_signature, timestamp, nonce, echostr, TOKEN, ENCODING_AES_KEY, CORP_ID) if ret == 0: return reply return "verify failed", 403 # POST 消息 msg_signature = request.args.get("msg_signature") timestamp = request.args.get("timestamp") nonce = request.args.get("nonce") body = request.get_data(as_text=True) ret, xml_content = decrypt_message(msg_signature, timestamp, nonce, body, TOKEN, ENCODING_AES_KEY, CORP_ID) if ret != 0: return "decrypt failed", 403 root = ET.fromstring(xml_content) msg_type = root.findtext("MsgType") content = root.findtext("Content") from_user = root.findtext("FromUserName") msg_id = root.findtext("MsgId") # 这里把消息交给 Workbuddy 的核心处理模块 handle_workbuddy_command(from_user, content, msg_id) # 企业微信要求快速返回,先返回空串 return ""本地调试时,先用 GET 验证把 URL 配通,再用 POST 打一条测试消息。你可以在handle_workbuddy_command里先只打印from_user和content,确认能拿到明文后,再往下接真正的业务逻辑。
3.3 三个最常导致解密失败的坑
第一,EncodingAESKey 不对。填的时候少位、多空格都会导致解密失败,后台不会告诉你具体原因,只会看到AES decrypt fail。第二,Token 与回调配置不一致,签名验证不过时,后台会提示“token 无效”。第三,把 GET 验证和 POST 消息处理混在一起。GET 验证走的是 echostr 解密,POST 走的是消息解密,两种处理逻辑不同,一旦共用一段代码,大概率会返回乱码或验签失败。
还有一个容易被忽略的点:在企微管理后台配置完回调后,最好让一个在可见范围内的同事给应用发一条消息,然后在服务端打印解密后的 XML。看到 XML 里有MsgType和Content字段,说明链路已经通了。
4. 把企微消息变成 Workbuddy 能听懂的指令
4.1 消息明文 XML 里到底有哪些字段
当你成功解密一条文本消息后,拿到的明文 XML 大概是这样:
<xml> <ToUserName><![CDATA[ww1234567890]]></ToUserName> <FromUserName><![CDATA[ZhangSan]]></FromUserName> <CreateTime>1710000000</CreateTime> <MsgType><![CDATA[text]]></MsgType> <Content><![CDATA[创建任务 明天下午三点 给客户回电话]]></Content> <MsgId>1234567890</MsgId> <AgentID>1000002</AgentID> </xml>FromUserName是企业微信中的用户 ID,不是姓名,也不是手机号。Content是用户发的文本内容。MsgId是消息唯一 ID。这里要特别注意MsgId,因为企业微信对未确认的消息会重试推送,如果你不做去重,同一个指令可能被执行两次,创建出重复任务。我在接入早期就因为这个吃过亏,用户连续收到两条“已创建任务”的回复。
4.2 指令格式设计:先固定规则,再上智能解析
Workbuddy 的智能体现在两个地方:一是对任务的时间表达做解析,比如“明天下午三点”“半小时后”;二是对动作词的识别。但初始接入阶段,我不建议直接上大模型做任意语义解析,因为用户输入千奇百怪,没有约束时你会被各种漏识别拖垮。
我采用的方式是设计一组前缀指令,让模型在指令框架内做参数填充:
| 指令前缀 | 用法示例 | 对应动作 |
|---|---|---|
| 创建任务 | 创建任务 明天下午3点 给客户回电话 | 创建一条带提醒的任务 |
| 查询任务 | 查询任务 今天 | 返回今天的任务列表 |
| 完成任务 | 完成 123 | 把任务ID为123的任务标记完成 |
| 提醒我 | 提醒我 30分钟后 喝水 | 创建一条一次性提醒 |
解析逻辑可以先正则,比如把“明天下午三点”切成“明天”“下午”“三点”,再映射成具体时间。实测下来用户配合度比想象中高,因为入口是企微聊天框,他们天然会把它当成真人对话助手,讲话会更清晰。等积累一批有问题的输入后,再考虑引入大模型做兜底解析,这样既保证稳定,又不至于被自由输入淹没。
4.3 用 MsgId 做消息去重
前面提到企业微信可能重试推送。一个最简单可靠的做法是用 Redis 的SETNX把MsgId作为 key,存 60 秒,处理前先判断是否处理过:
import redis r = redis.Redis(host="127.0.0.1", port=6379, db=0) def is_duplicate(msg_id): return not r.set(msg_id, "1", nx=True, ex=60)这样即使企微重试同一事件,第二次也能被拦截掉。key 的过期时间不要太短,至少大于你的异步任务处理耗时。如果业务处理超过 60 秒,就适当调大到 300 秒。这个去重逻辑虽然简单,但能省掉后续很多数据脏问题。
5. 主动推送:把 Workbuddy 的提醒和结果发回企微
5.1 获取 access_token 的正确姿势
企业微信接口调用前都要先获取 access_token。接口是:
GET https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=CORP_ID&corpsecret=SECRET返回结果里有access_token,有效期 7200 秒。网上很多教程直接每次请求都重新获取,但这样会在高并发时触发限流,而且没必要。我习惯用 Redis 缓存 token,过期前自动刷新:
import requests, redis r = redis.Redis(host="127.0.0.1", port=6379, db=0) TOKEN_KEY = "qw_access_token" def get_access_token(corp_id, secret): token = r.get(TOKEN_KEY) if token: return token.decode() resp = requests.get("https://qyapi.weixin.qq.com/cgi-bin/gettoken", params={"corpid": corp_id, "corpsecret": secret}, timeout=5).json() if resp.get("errcode") == 0: # 提前 300 秒过期,防止边界时间失效 r.set(TOKEN_KEY, resp["access_token"], ex=resp["expires_in"] - 300) return resp["access_token"] raise Exception(f"get token failed: {resp}")这里有个细节:expires_in和实际有效期会有偏差,最好把过期时间缩短一点,避免刚好在临界点拿到旧 token 而报 40014 或 42001。我一开始没做提前过期,结果每周都会碰到一两次 token 刚过期导致的发送失败,后来改成提前 300 秒刷新就再没出现过。
5.2 发送文本消息的完整实现
获取 token 后,就可以调用消息推送接口。最常见的是发给单个用户:
def send_text_message(user_id, content, agent_id, access_token): url = "https://qyapi.weixin.qq.com/cgi-bin/message/send" data = { "touser": user_id, "msgtype": "text", "agentid": agent_id, "text": {"content": content}, "safe": 0, "enable_id_trans": 0 } resp = requests.post(url, params={"access_token": access_token}, json=data, timeout=5).json() return resptouser可以是单个成员 ID,也可以用|分隔多个用户;如果想发给整个部门,用toparty。agentid必须填,否则企微不知道这条消息是哪个应用发的,接收方消息列表里显示的应用名也会不对。
调用成功后,用户在企业微信中会收到来自该应用的一条消息。这里有一个比较隐蔽的限制:应用消息连续发多条短文本时会被折叠,体验并不好。所以我在 Workbuddy 里会把同类提醒合并成一条列表消息,而不是每个任务单独发一条。
5.3 除了文本,还能发卡片消息
企业微信应用消息还支持textcard,这是一种带标题、描述、链接的消息卡片,实测下来比纯文本更清晰。例如给用户推送一条任务卡片,点击卡片可以跳转到 Workbuddy 的详情页:
{ "touser": "ZhangSan", "msgtype": "textcard", "agentid": 1000002, "textcard": { "title": "任务提醒:给客户回电话", "description": "<div class=\"gray\">时间:明天下午3点</div>\n<div class=\"normal\">状态:待处理</div>", "url": "https://wb.example.com/task/123", "btntxt": "查看详情" } }这种卡片比纯文本更不容易被淹没在聊天流里,也方便后续扩展审批跳转。不过要记住,url 需要是企业微信配置的“可信域名”,否则用户点开会被拦截。如果还没配置可信域名,可以先继续用纯文本,等域名验证通过后再切换成卡片。
5.4 定时任务推送时的节奏控制
Workbuddy 的提醒往往集中在早晚高峰,比如 9 点一批、14 点一批。如果没有节奏控制,很容易触发企微的接口频率上限。我建议把批处理任务拆成小批次,每批最多 20 条,批次之间 sleep 1 秒。另外,可以在任务表里加一个sent_status字段,只推送给未发送的记录,推送成功后更新状态,这样即使服务重启也不会重复推送。
6. 生产部署和排障实录
6.1 从本地到服务器的部署衔接
本地跑通只是第一步,真正要让团队成员用起来,回调服务必须部署到一台公网可达的服务器上。域名和 HTTPS 证书是硬要求,企业微信回调 URL 不认 IP 裸奔和 HTTP。部署时我用 Nginx 做反向代理,把/qw/callback转发到 Flask 服务,同时开启 HTTPS。
Nginx 配置里还有一个容易忽略的点:企业微信回调可能来自多个出口 IP,如果你依赖 IP 做接口访问控制,这里就要小心别把回调出口挡在防火墙外面。正确的做法是只限制主动调企微 API 的服务出口 IP,不要限制企微进入回调的 IP。
6.2 高频错误码对照
| 错误码 | 含义 | 处理建议 |
|---|---|---|
| 40014 | access_token 不合法或已过期 | 清缓存重新获取 token |
| 42001 | access_token 过期 | 同上,并检查 Redis 过期时间 |
| 60011 | 用户不在应用的可见范围 | 调整应用可见范围 |
| 60020 | 访问 IP 不在白名单 | 把服务器公网 IP 加入可信 IP |
| 45009 | 接口调用频率超限 | 降低发送频次,分批推送 |
| 40091 | 语音识别接口错误 | 检查是否误把语音当文本处理 |
对照这个表格,大部分发送失败的问题都能快速定位。我的习惯是把这些错误码和解决建议直接写到运维文档里,团队其他人遇到同类问题也能自己查,不用每次都来问我。
6.3 我踩过的三个坑
第一个坑是回调验证成功,但收不到消息。排查后发现是应用没有开启完整的接收消息权限,回调里拿不到Content字段。企微后台对自建应用的接口权限控制比较细,建议在自建应用详情页里检查“接收消息”的配置模式,确认已经设置为兼容模式或密文模式,并且在应用权限里允许读取消息内容。
第二个坑是主动发送消息时,touser传了中文姓名而不是成员 ID。企业微信 API 只认成员 ID,不认姓名。要先通过通讯录接口查询成员 ID,或者在后端把 ID 和姓名对应关系表维护好。我之前在测试环境里直接传姓名,返回的却是成功状态,后来才发现是同名的另一个人收到了消息,非常尴尬。
第三个坑是消息处理耗时太长。Workbuddy 解析任务时调用了一个比较慢的模型接口,导致回调响应超过企微的等待时间,企微就不断重试,消息被重复处理。后来我在回调入口先回一个空串,再用消息队列异步处理业务逻辑,问题就解决了。
6.4 日志和监控不能省
接入企微后,消息链路会经过企微服务器、Nginx、Flask、Workbuddy 核心,任何一环出问题都容易让人抓狂。所以我给每条消息打了一个trace_id,日志里记录user_id、msg_id、msg_type、耗时和结果。在 Nginx 层面也记录回调请求的状态码和时间。这样用户反馈“没收到提醒”时,我可以很快定位是企微没推过来,还是 Workbuddy 没发出去。
这次接入做完后,最大的体会是,很多人以为接入企业微信就是配一个 webhook,实际上它有一套完整的事件回调机制、加密协议和 API 权限模型。把双向链路跑通,后面接审批、日程、智能问答就都是同一套骨架。团队现在把 Workbuddy 的入口藏进了企微,大家不用再记网址,直接发消息就能用,日常使用率翻了不止一倍。如果你也在做类似接入,建议先从最小闭环开始:验证回调 -> 收到消息 -> 解析一条指令 -> 主动回一条消息,跑通后再逐步加功能,千万别一上来就把所有业务逻辑塞进去。