你是不是也遇到过这种场景白天把一个 AI Agent 的任务挂上去让它在后台跑数据清洗、批量生成文案或者处理一整套自动化流程然后你有事出门了回来打开电脑一看任务早就跑完了但你完全没有第一时间拿到结果白白浪费了中间那几个小时。这个情况我碰到过太多次所以后来就花了一下午写了这个微信推送服务专门解决“AI Agent 跑完任务怎么通知你”这个问题。我做的事情其实很简单让我的 AI Agent 在任务开始、结束、异常这三个关键时刻自动往微信上推送一条消息。不管我是在开会、在吃饭还是在路上只要手机震一下我就知道任务的当前状态了。如果你的工作流里也经常跑耗时的 Agent 任务这篇文章应该能帮你省掉不少盯着进度条的功夫。我会把完整的实现思路、代码和踩坑记录都分享出来新人照着做也能在半小时内跑通。1. 为什么 AI Agent 任务离不开“主动通知”先聊聊我的真实痛点1.1 从盯屏幕到“被通知”我经历过哪些场景先说一个最典型的场景。我有一批历史数据要做清洗和特征提取原本是写脚本跑后来发现这一类事情用 AI Agent 来做更省心——它能根据数据情况自己调整处理步骤。但问题来了一份 50 万行的数据Agent 要在中间调用好几轮大模型接口做字段判断一轮调用就是几十秒整条流水线跑下来动辄一个多小时。这一个多小时里我总不能一直盯着终端吧屏幕保护都能出来好几轮。中间我可能去开个会、下楼拿个快递、做顿饭结果就是任务其实提前 20 分钟就跑完了但我到点才回来看到时间被白白浪费了。类似的情况还有 Agent 批量生成短视频文案、批量爬取网页内容、跑回归测试等等都是耗时型任务。后来我也试过在本地终端播放提示音但只要我离开工位就听不见。试过用邮件通知但邮件客户端不常开的话一样是延迟。真正让我下定决心做微信推送的是有一次我在外面任务跑完了我却等了整整两个多小时才看到结果。从那天起我就觉得AI Agent 跑任务这件事缺的不是计算能力而是一个可靠的“完成了”的信号。1.2 通知方案横向对比为什么我最终选了微信这条通道市面上常见的通知方案其实不少我大概列了一圈各有各的问题。邮件通知通用性没问题但没有谁会一直盯着邮箱而且邮件很容易被丢进垃圾箱。我试过一次任务跑完了邮件躺在回收站里白等了半天。钉钉/企业微信的群机器人 Webhook这个方案很轻往群聊里发消息就行但问题在于“群”。个人项目没有必要专门拉一个群而且机器人消息容易被群消息刷掉。Telegram Bot接口好用、实时性也强但在国内使用链路不够顺畅而且身边大部分同事朋友都用不上。Server酱、PushPlus 之类的第三方推送平台胜在接入快代码就几行但免费额度有限而且数据要经过第三方服务我这种喜欢自托管的人不太放心。自己对接企业微信应用消息推送这也是我最终选的方案。企业微信里创建一个自建应用给它发消息消息会直接通过微信上绑定的“企业微信”会话送达不需要对方也是企业成员一个人自用绰绰有余。关键是官方 API 稳定、免费配额足够个人使用完全够还不用经过第三方。选型企业微信还有一个原因是它打通了“微信”这个人人都在用的高频 App。通知的本质是触达触达率最高的入口就在微信里。我不用安装任何新的软件也不用担心手机通知权限被关。1.3 整体方案架构Agent、推送服务、企业微信三者如何协作整个架构其实非常简单一句话就能说清AI Agent 在任务的几个关键节点调用一个“推送模块”推送模块再通过企业微信的 API 把消息发到我的微信上。最上层是 AI Agent不管你是用现成的框架还是像我一样写了几行调度逻辑的任务脚本都可以接入。它只需要在“开始时”“成功时”“失败时”三个位置调用推送函数。中间是推送模块我把它单独封装成了一个脚本文件里面就两个函数一个负责获取 access_token一个负责发送文本消息。Agent 那边只管调用不用关心微信接口的细节。最底层是企业微信服务端它提供两个接口一个是获取 token 的一个是发送消息的。消息从企业微信的服务器推到微信客户端链路是官方维护的稳定性和速度都有保障。这样拆的好处是解耦。以后我想把推送从企业微信换成其他的只需要改推送模块Agent 那边的代码一行都不用动。如果你已经在用 n8n、Dify、Coze 这类平台也可以在它们的 Webhook 节点里面直接调用这个服务同样能实现任务通知。2. 微信推送服务的核心原理企业微信应用消息是怎么跑通的2.1 四个关键概念CorpID、Secret、AgentId、access_token在做代码之前先得把这几个概念搞明白。我第一次接触企业微信 API 的时候也被这一堆 ID 绕晕过其实理清之后很简单。CorpID企业 ID这是你的企业微信账号的唯一编号相当于“你是谁”。创建企业微信后就有在管理后台的“我的企业”页面能看到。Secret应用密钥每个自建应用都有一个独立的密钥相当于“应用密码”。它决定了这个应用能不能调用 API所以不能泄露。AgentId应用 ID你在企业微信里创建的自建应用会有一个编号用来区分消息是从哪个应用发出来的。Agent 消息推送时这个参数是必填的。access_token访问令牌这是调用企业微信接口的“临时通行证”。每次发送消息之前都要用 CorpID 和 Secret 去换一个 tokentoken 有效期通常是 7200 秒过期了需要重新获取。很多教程里会把 CorpID 和 AgentId 搞混其实可以这样记CorpID 是整个企业的身份证AgentId 是你创建的那个应用的门牌号Secret 是开门用的钥匙。2.2 消息发送的完整请求链路两个 API 搞定推送整个推送过程就两步。第一步拿到 access_tokenGET https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpidCORPIDcorpsecretSECRET返回结果大概是这样的{ errcode: 0, errmsg: ok, access_token: xxxxx, expires_in: 7200 }第二步拿着 token 去发消息POST https://qyapi.weixin.qq.com/cgi-bin/message/send?access_tokenACCESS_TOKEN请求体是 JSON{ touser: all, msgtype: text, agentid: 1000002, text: { content: AI Agent 任务已完成 } }touser 支持指定成员但我自己一个人用直接填“all”最简单。如果以后要分发给不同成员就把成员的 UserID 用竖线拼起来比如“zhangsan|lisi”。这两个接口都是标准的 HTTPS 请求不复杂但有一个细节值得注意access_token 要尽量复用。如果你每一个任务每一次推送都先去获取一次 token企业微信 API 对获取 token 的接口有频控限制短时间调用太频繁会被临时封禁。所以我在推送模块里做了一个缓存token 没过期就一直用旧的。2.3 个人微信为什么不能直接推聊聊自建应用这条合规通道的妙处聊到这里可能有朋友会问能不能直接给个人微信发消息我最初也想过用 itchat 那种个人号机器人后来专门研究了一下放弃了。个人微信私聊接口属于非官方能力账号很容易被限制登录消息能不能发出去全靠运气。作为生产环境里的通知服务这种不确定性是不可接受的。企业微信之所以能成为合规的替代方案是因为它有一个天然优势企业微信消息可以联动到微信。你在微信 App 里可以收到企业微信应用的推送不需要额外打开企业微信客户端。也就是说我通知最终落在的还是微信这个超级入口上但消息通道是企业微信官方 API稳定性有保障。我举个例子你就明白了。企业微信的应用消息推送相当于“官方快递”走的是企业微信服务器的正规链路消息体量再大也不会封号。而个人号机器人相当于让一个普通人帮你跑腿送文件运气好能送到运气不好连人带文件都搭进去。我选择前者图的就是踏实。3. 实操过程从零搭建一个可用的微信推送服务3.1 第一步注册企业微信并创建自建应用获取关键参数如果你目前没有企业微信先去官网注册一个。个人也可以注册不需要营业执照。注册过程中会让你填写企业名称和行业类型随便填一个靠谱的就行后面不会审核太严。注册完进入企业微信管理后台按下面几步操作在左侧菜单找到“应用管理”进入“应用”页面。拉到最底部找到“自建”板块点击“创建应用”。填写应用名称我起的是“Task 通知”再上传一个应用 Logo选一个可见范围。这里要看情况选择如果只有你自己使用选择“全部成员”或指定一个人都可以建议选指定成员避免不必要的打扰。创建成功后在应用详情页就能看到AgentId和Secret。Secret 默认隐藏点击“查看”并发送到你的企业微信验证一下或者用管理员微信扫码就能看到。再到“我的企业”页面最底下有个“企业 ID”这就是CorpID。在同一个应用详情页里你还会看到“企业可信 IP”这个配置项。这是企业微信的一个安全机制只有配置过的 IP 才能调用这个应用的 API。如果你和我一样Agent 跑在云服务器上就填服务器的公网 IP如果跑在本地就填家里宽带的出口 IP。这一步踩坑概率最高。我第一次配置的时候忘了填 IP哪怕代码写得完全正确接口也永远返回“not allow to access from your ip”。所以只要消息发不出去第一件事就去查这个 IP 白名单。3.2 第二步写一个轻量 Python 推送模块封装 token 和发送逻辑参数齐了代码就很简单。我用 Python 的 requests 库写了这个推送模块整个模块就 50 行左右。import json import time import requests class WeChatPusher: def __init__(self, corp_id, secret, agent_id): self.corp_id corp_id self.secret secret self.agent_id agent_id self.token None self.token_expire_at 0 def _get_token(self): if self.token and time.time() self.token_expire_at: return self.token url https://qyapi.weixin.qq.com/cgi-bin/gettoken params { corpid: self.corp_id, corpsecret: self.secret, } resp requests.get(url, paramsparams, timeout10).json() if resp.get(errcode) ! 0: raise RuntimeError(f获取 token 失败: {resp}) self.token resp[access_token] self.token_expire_at time.time() resp[expires_in] - 200 return self.token def send_text(self, content): token self._get_token() url fhttps://qyapi.weixin.qq.com/cgi-bin/message/send?access_token{token} payload { touser: all, msgtype: text, agentid: self.agent_id, text: {content: content}, safe: 0, } headers {Content-Type: application/json} resp requests.post(url, jsonpayload, headersheaders, timeout10).json() if resp.get(errcode) ! 0: raise RuntimeError(f发送消息失败: {resp}) return resp这里有两个小细节值得说明。一个是 token 的过期时间我减了 200 秒预留缓冲避免在临界点时出现高并发请求导致个别 token 失效另一个是每次请求都设置了 10 秒超时避免网络异常时程序卡死。如果你不想自己维护这个类也可以在代码里去掉类写成两个普通函数。但强烈建议保留 token 缓存的逻辑我在后面会专门讲一下为什么。3.3 第三步在你的 AI Agent 代码中接入三个关键推送时机推送模块写好了接下来就是接入 AI Agent 的代码。我举一个比较有代表性的例子假设你写了一个 Agent让它批量给客户生成营销文案每生成一批就调用一次大模型接口整个流程大概需要跑十几分钟。在 Agent 的主流程里我在三个地方加了推送pusher WeChatPusher( corp_id你的CorpID, secret你的Secret, agent_id你的AgentId ) def run_agent(task_list): # 1. 任务开始 pusher.send_text(f[任务开始] 本轮共 {len(task_list)} 条文案待生成预计需要 15 分钟) results [] for idx, task in enumerate(task_list): try: # 这里是你原本的 Agent 执行逻辑 result agent.generate_copywriting(task) results.append(result) except Exception as e: # 2. 任务异常立即通知 pusher.send_text(f[任务失败] 第 {idx1} 条生成失败原因{str(e)}) raise # 每完成 5 条可以发一次进度 if (idx 1) % 5 0: pusher.send_text(f[进度] 已完成 {idx1}/{len(task_list)} 条) # 3. 任务成功 summary f[任务完成] 共生成 {len(results)} 条文案耗时 12 分钟 pusher.send_text(summary)我在这个示例里加了两个细节进度推送和异常推送。进度推送看起来简单实际上对长时间任务很重要——它让你知道 Agent 没有卡死还在正常跑。异常推送更是必不可少很多时候 Agent 跑着跑着因为 API 超时挂了你要是没看到失败通知还以为它在正常运行结果白白等了一个小时。有朋友可能会问任务里用到的 API Key 硬编码在代码里会不会不安全我自己的做法是放在环境变量里不提交到 Git 仓库。这个推送模块虽然说风险不大但 Secret 泄露了对方就能用你的应用发消息也麻烦。3.4 第四步消息模板与多任务并发别让你的通知变成一堆乱码接好之后你会发现一个问题一旦任务多了通知消息就会变得混乱。今天跑了一个数据任务明天跑了三个文案任务每条消息都在说“任务完成”但你根本分不清是哪一个完成了。所以我后来做了一套简单的消息模板。我把每条推送消息设计成了这个格式【任务类型】文案生成 【任务ID】task-20240615-001 【当前状态】已完成 【结果摘要】共 20 条成功 19 条失败 1 条 【耗时】12 分 30 秒 【执行时间】2024-06-15 18:30:00如果任务失败就替换“当前状态”为“失败”然后在“结果摘要”里附上错误堆栈的前几行。这样收到任何一条消息你都能在 3 秒内判断要不要去处理。再来说多任务并发的问题。我一开始是直接在每个子任务里都调用 pusher.send_text结果发现的问题是发送频率太高企业微信单应用每分钟有频控限制。后来我引入了一个简单的消息缓冲机制多个任务共享同一个 pusher 实例同一个时间窗口内的消息合并成一条用列表拼起来再发送。这个做法不复杂但在任务多了以后效果非常明显既能避免频控也不会刷屏。如果你用线程池或者 asyncio 跑多个 Agent 子任务建议在推送模块里加一个线程锁避免多个线程同时刷新 token 造成冲突。用 Python 的threading.Lock包住_get_token方法就够用了。4. 常见问题与排查技巧实录我替你先踩过的那些坑4.1 消息没到先用这张速查表定位问题我在搭建过程中遇到过不少问题也帮朋友排查过好几次把最高频的问题整理成了一张表。现象常见原因排查方法与解决接口返回not allow to access from your ip服务器 IP 没有配置到应用白名单检查应用详情页的“企业可信 IP”确认是否已添加当前出口 IPinvalid credentialCorpID 或 Secret 填错重新复制参数注意 Secret 是否多复制了空格invalid agentidAgentId 填写成了应用名称或别的 IDAgentId 是一个纯数字 ID去自建应用详情页看消息收到但在微信里没有提醒微信里的企业微信通知被折叠或关闭在微信的“企业微信联系人”中打开“接收消息通知”并关闭勿扰模式消息发送成功但没有内容content 参数传了空字符串或纯符号检查 content 是否为空建议至少写一个字符请求超时服务器到企业微信 API 网络不稳定设置稍长的超时时间观察一段时间如果频繁超时考虑更换网络出口上面这根表格基本能覆盖 90% 的问题。如果你是按我的代码来写的遇到错误时直接把接口返回的 JSON 打出来看errmsg字段会把原因说得很清楚只是很多人懒得看一眼就到处问人。记住这句话企业微信的报错信息是这个世界少有的“说明白了就是明白了”的报错。4.2 我踩过的几个大坑Token 并发刷新、长文本截断、特殊字符转义聊几个我觉得最有价值的坑。第一个是Token 并发刷新。多线程任务里我第一次没有加锁结果好几个线程同时发现 token 过期并同时去刷新。企业微信对 gettoken 接口同一秒的并发调用有限制结果一个线程刷新成功了其他线程拿到新的 token 还没用上接口就被临时限流了。后来我在_get_token里加了锁还要判断“当前 token 是否已经快过期如果是只让一个线程去刷新其他线程等一会儿再用旧的”。第二个是长文本截断。有一次把 Agent 的完整执行日志直接丢到 content 里发企业微信直接返回了“invalid message content length”查了文档才知道文本消息的长度上限是 2048 字节。中文字符一个占 3 个字节也就是说纯中文最多发 682 个字符左右。这个长度平时够用但如果想发日志就不够了。我的对策是推送消息只放摘要完整日志散落到文件里如果确实需要看全文就在消息里附一个日志文件的路径。第三个是特殊字符转义。这个更隐蔽。有一次我在内容里放了一个换行符和一个制表符结果消息发出来了但在微信上显示成了奇怪的空白。后来我仔细对比才发现是 JSON 里的\n被解释成了真正的换行而我又没注意到。解决办法是内容里如果有用户输入或者从文件里读取的文本一定要先做转义或者替换把控制字符清理掉再发送。这个细节不处理偶发性问题排查起来老费劲了。4.3 稳定性优化重试机制、消息持久化和后续扩展想法推送服务的稳定性直接决定了可靠性。如果 Agent 跑了半小时结果通知消息因为网络抖动没发出去那整个通知链条就是失败的。我对推送模块做了三个层次的加固。第一层是发送重试。send_text里捕获requests.RequestException和接口错误码比如 45009 这种触发频控的情况退避 3 秒后重试最多重试 3 次。实践下来因为网络抖动导致的偶发失败基本都能被重试消掉。第二层是消息持久化。我在本地维护了一个push_log.jsonl文件每次推送把时间、内容、发送结果追加进去。这样做的最直接收益是即使消息真的没发出去我事后也能从日志里看到“这个时间点有一条消息应该发出去”不至于整个过程中处于信息真空的状态。第三层是扩展方向。现在的通知只有文本但其实企业微信 API 还支持 markdown 消息、图片消息、文件消息。我曾经想升级成发图片Agent 跑完生成图表之后直接截图推送到微信。后来代码也写好了展示效果确实比纯文本好很多至少不用打开电脑看图。如果你跑的 Agent 会输出图表、PDF 或 Excel 报告建议往这个方向扩展体验是质的飞跃。最后分享一个小细节这个服务我大概用了三个月最直观的变化是我不再焦虑了。以前挂任务每隔几分钟就想切回终端看一眼出去吃个饭都心里不踏实。现在任务一挂手机放口袋里该干嘛干嘛消息来了再开电脑处理。最后再分享一个我在使用中摸索出来的小技巧不要把“任务完成”和“任务失败”的消息都用同一个声音。企业微信应用消息本身没法区分但你可以根据微信通知栏的内容文本把成功消息统一写成“完成”失败消息统一写成“失败”。时间长了你会有条件反射的看到“失败”两个字心里咯噔一下马上跑去开电脑看到“完成”两个字就从容地把手里的事情忙完。这样看通知的效率是真的高也让我对每个任务的状态一目了然。