最近在折腾 AI Agent 的时候我碰到一个特别现实的问题Agent 跑一个长任务比如数据清洗、批量文件处理、爬虫抓取、代码仓库扫描跑起来之后我总不能一直盯着终端看吧。有时候任务跑完了我在忙别的根本不知道有时候任务中途挂了我在摸鱼也没人告诉我。等到我想起来去检查进度发现 Agent 早就跑完半小时了甚至已经因为异常退出了。后来我干脆写了个微信推送服务把 Agent 的“任务完成”“任务失败”“关键步骤进度”全部推到微信上。手机一震我就知道任务跑完了点开就能看结果摘要。这个服务我用了两三个月稳定可靠现在分享出来希望能帮到同样被“Agent 跑完没人通知”困扰的朋友。这个内容适合谁适合正在做 Agent 开发、自动化流程搭建、长期值守任务、定时批处理脚本的朋友。不管你是用 Python 写 Agent、用 LangGraph 编排复杂流程、用 n8n 拉工作流还是用 Claude/OpenAI API 调模型干活这套微信推送的思路都可以直接套进去。1. 为什么 Agent 需要一套专属的通知机制1.1 Agent 任务的“长周期”特性很多人对 AI Agent 的印象还停留在“一问一答”的对话框里但实际上真正的 Agent 应用场景往往是多步骤、长耗时的。我举个自己的例子我写过一个小型 Agent它会从某内部系统拉取数据、做格式校验、调用大模型生成摘要、再把结果写入数据库。整个流程跑一次需要 5 到 20 分钟如果是批量处理可能要跑几个小时。在这种场景下人和 Agent 的关系是典型的“异步协作”。Agent 在后台运行人不可能一直守着。这时候如果 Agent 没有一个主动触达人的渠道任务跑完或者跑挂了信息就滞留在服务器日志里等于白跑。你不可能因为一个号跑完了就专门登服务器看一眼日志。1.2 通知回路要解决的核心问题我在设计这个推送服务之前先梳理了三个必须解决的问题第一任务结束要有感知。不管成功还是失败都要第一时间让人知道这样才能决定下一步动作。比如看到“任务失败错误日志摘要”我可以马上修正参数重新跑不用等下次检查才发现。第二关键节点要有反馈。长任务的中间过程也要有轻量级的进度通知。我一般在任务开始和每个大步骤完成时各推一条不用太频繁否则微信会被刷屏。第三多个任务要能区分。我经常同时跑两三个 Agent 任务推送消息必须带上任务名称和类型不然微信里堆了一堆消息根本分不清哪个是哪个。1.3 为什么“终端打印”和“看日志”不够用我知道有人会说终端里 print 一下不就行了或者把日志写到文件里跑完 tail 一下也可以。说实话这些方案在小规模场景下确实够用我自己早期就是这么干的。但当 Agent 跑在远程服务器上或者跑在 Docker 容器里你不可能随时开着一个 SSH 窗口盯着输出。日志文件的痛点也很明显它是被动等待的你要主动去查才会看到。更重要的是服务器上下文中如果有多个 Agent 在运行日志会交错在一起可读性很差。所以我当时定了一个设计目标通知必须主动、实时、跨平台。手机是最好的载体因为手机是随身携带的而微信是国内的国民应用推送到达率高接收体验也最好。2. 方案选型为什么我盯上了微信推送2.1 常见通知方案横向对比在敲定微信推送之前我把市面上常用的通知渠道挨个过了一遍。邮件最传统但我的实际体验是邮件很容易被归入垃圾箱或者被折叠到“促销”分类里通知的实时性大打折扣。几个小时后才看到邮件那和没有通知差不多。钉钉和飞书都有自己的机器人配置不算复杂在团队协作里用很合适。但如果只是个人项目、个人使用拉一个钉钉群或者飞书群来收通知显得有点小题大做而且这些 App 的重度用户群没有微信那么广。Telegram Bot 在技术圈很受欢迎API 也简单海外 VPS 上跑的话体验很好。但对国内用户来说日常使用频率有限很多人不会特意装和登录 Telegram。最后剩下两个方向一类是 Server酱、PushPlus 这类专门做微信推送的第三方服务另一类是企业微信应用消息。我重点测了这两类后面单独展开讲。2.2 微信推送的两条主流实现路径先说 Server酱和 PushPlus 这类第三方服务。原理不复杂你在它的网站上用微信扫码绑定拿到一个专属的 token然后你调用它提供的 HTTP 接口把消息内容和 token 一起 POST 过去它就把消息推到你的微信上。整个过程几分钟就能搞定对只想知道“Agent 跑完没”的人来说这可能是最省事的方案。Server酱的主要限制是每天免费的条数有限对于高频推送场景不太够用。PushPlus 也有类似的限制免费用户有发送条数上限且单条消息有长度限制。如果你的 Agent 一天要跑十几个任务每个任务推好几条消息免费额度可能就见底了。再说企业微信应用消息。这个方法稍微绕一点但完全不依赖第三方服务可控性和稳定性会好很多。思路是注册一个企业微信个人也能注册在里面创建一个自建应用拿到企业 ID、应用 AgentId 和 Secret然后调用企业微信的接口发送应用消息。消息会直接推到你企业微信的“应用消息”里你可以在微信里绑定企业微信这样实际上也是推到微信上。从长期主义的角度看企业微信这条链路更“干净”没有每日条数限制也没有对第三方服务可用性的依赖适合正式项目的需求。我在最终方案里是以企业微信为主链路同时保留了 Server酱 作为备用通道一旦某一边出问题随时切换。2.3 我最终选择的推送链路最终方案如下Agent 任务进程在任意阶段产生通知事件 - 调用统一推送模块 - 模块组装消息 - 通过企业微信 API 发送应用消息 - 用户在企业微信/微信中实时收到通知。整个过程是异步的推送接口的失败不应该影响主业务流程。我专门设计了“推送失败静默降级”的机制推送 API 报错就打印一条日志不影响 Agent 主体任务继续跑。3. 微信推送服务的核心实现3.1 用企业微信创建应用并获取密钥第一步是企业微信注册。进入企业微信管理后台选择“应用管理”然后点“创建应用”。应用名称随便填比如“Agent通知中心”。创建完成后你会看到几个关键参数AgentId、Secret以及企业 IDCorpId。这三个参数建议放到环境变量里不要写死在代码中因为后期在不同机器上部署的时候只改环境变量就行了不用动代码。有一点要特别注意企业微信要求你先加入企业才能注册应用但个人用户也可以创建企业微信团队不需要真的拉一个公司。我当初注册的时候就选了“个人创建”流程整个过程大概十分钟没有任何成本。3.2 获取 access_token企业微信的 API 和其他微信 API 类似发送消息前先要用 CorpId 和 Secret 换取一个 access_token。接口很简单GET https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpidYOUR_CORPIDcorpsecretYOUR_SECRET返回结果是一个 JSON里面有 access_token 和过期时间。access_token 的有效期是 7200 秒也就是 2 小时但建议不要每次发消息都去请求一次可以缓存到本地内存里过期再刷新。官方没有强制限制获取频率但频繁请求会显得很浪费而且很容易触达频率限额。我把获取 token 的逻辑做成了一个单例并记录过期时间在过期前 5 分钟自动刷新。这样无论 Agent 一天推多少条消息token 都只会在过期时更新一次。3.3 发送文本消息拿到 access_token 之后发送应用消息的接口是POST https://qyapi.weixin.qq.com/cgi-bin/message/send?access_tokenACCESS_TOKEN请求体大致长这样{ touser: all, msgtype: text, agentid: 1000002, text: { content: Agent任务完成数据清洗任务已跑完共处理 1283 条记录耗时 3分42秒。 }, safe: 0 }touser 可以填具体的用户 ID 或部门 ID个人使用的话直接填 “all” 就行。agentid 就是创建应用时拿到的 AgentId。text.content 里放你要推送的消息内容。需要注意发送消息这个接口返回的 JSON 里有一个 errcode 字段必须为 0 才算发送成功。我之前踩过一次坑发送接口返回了错误码但是我没做 errcode 判断导致实际没推送到微信我还以为推送正常。后来在推送模块里严格判断 errcode非零则记录日志并尝试备用通道。3.4 封装成通用推送模块在实际项目里我不会在 Agent 的业务代码里直接裸调 HTTP 接口而是封装成一个 PushService 模块。这样 Agent 代码里的调用只涉及一行代码可读性和可维护性都提高不少。下面是一段 Python 示例我把企业微信推送的常用逻辑封装成了一个类。这类代码我公开过好几次大家的反馈都比较实用你可以直接复制改改就能用import time import json import requests class WeChatPusher: def __init__(self, corpid, corpsecret, agentid): self.corpid corpid self.corpsecret corpsecret self.agentid agentid self._token None self._token_expires 0 def _get_token(self): if self._token and time.time() self._token_expires - 300: return self._token url https://qyapi.weixin.qq.com/cgi-bin/gettoken params {corpid: self.corpid, corpsecret: self.corpsecret} resp requests.get(url, paramsparams, timeout5).json() if resp.get(errcode) ! 0: raise RuntimeError(f获取token失败: {resp}) self._token resp[access_token] self._token_expires time.time() resp[expires_in] return self._token def send_text(self, content, touserall): token self._get_token() url https://qyapi.weixin.qq.com/cgi-bin/message/send payload { touser: touser, msgtype: text, agentid: self.agentid, text: {content: content}, safe: 0 } headers {Content-Type: application/json} resp requests.post( url, params{access_token: token}, headersheaders, datajson.dumps(payload, ensure_asciiFalse).encode(utf-8), timeout5 ).json() if resp.get(errcode) ! 0: raise RuntimeError(f消息发送失败: {resp}) return resp这个类只实现了文本消息如果你想推 Markdown 或图片可以通过 msgtype 来扩展。我一般通知内容都用文本因为 Agent 完成任务后的关键信息就那么几行。3.5 用 Server酱做备用通道企业微信方案虽然稳定但毕竟依赖企业微信这个中间层万一哪天对方接口调整或者网络问题推送就断了。我后来加了一个备用通道用的是 Server酱。Server酱的调用方式更简单只要向它的接口 GET 或 POST 一个标题和正文它就会推到微信。我当时在推送模块里加了一个 fallback 逻辑企业微信发送抛异常或者返回非零 errcode 时自动切成 Server酱 发送。这块代码不复杂就不展开写了。4. Agent 与推送服务的对接实践4.1 在 Agent 任务结束点调用推送有了推送模块剩下的核心问题只有一个在 Agent 代码的哪里调用推送最简单的做法是在主流程的最后加一行代码同步调用推送接口。但这有一个隐患如果推送服务本身出问题卡住了会影响 Agent 主流程的退出。更合理的做法是使用异步或线程池的方式把推送任务丢到后台线程去执行主流程立刻结束互不影响。我试过直接在线程里跑推送后来发现偶尔会丢消息因为 Python 主进程在脚本退出时不等待子线程执行完成。更稳妥的做法是使用线程池加一点缓冲机制或者干脆在推送模块内部用独立的 daemon 线程并维护一个简单队列。对于个人项目来说用 requests 同步发送并设置超时也已经够用只要把超时时间设置短一点比如 3 秒主流程最多只会被拖 3 秒问题不大。4.2 用装饰器统一处理 Agent 结果我后来把“开始任务”“任务成功”“任务失败”三种通知做了一个统一的装饰器写在 Agent 的主执行函数上。这样代码调用处不需要到处塞推送逻辑只要在函数定义时加一个notify_on_finish(任务名称)的装饰器就能自动在开始、结束、异常时都推送消息。这段装饰器的实现思路是包装函数在调用前推送一条“开始执行”在函数正常返回后推送结果摘要在抛出异常后推送错误堆栈。为了不让推送消息过长我在装饰器里对结果做了截断处理只保留前 500 个字符。这样既能看到结果概貌又不会因为消息体太长导致推送失败。这样一个装饰器解决了我 90% 的推送需求。剩下 10% 更细的推送比如任务内某个关键步骤完成我就在对应位置手动调用一次推送服务。4.3 在 n8n 和 LangGraph 里的接入思路不光是自研 Python Agent现在很多人用 n8n 搭自动化流程或者用 LangGraph 编排多 Agent 协作。这些平台也同样需要通知能力。在 n8n 里你可以用 HTTP Request 节点直接调用企业微信的发送消息接口或者调用你自己部署的推送服务。我见过有朋友在 n8n 里创建了一个公共的“微信通知”子流程每个分支任务完成时都调用这个子流程传一个标题和内容进去就行。这样不同节点的编排看起来清爽很多。LangGraph 的思路也类似在状态图的结束节点或者在某个关键节点的 after 回调里触发推送函数。我习惯把推送逻辑放到一个单独的工具模块里在 LangGraph 的 tool 节点里引用这样核心图逻辑不会被推送代码污染。4.4 多 Agent 并行时的消息聚合与去重如果同时跑多个 Agent微信收到的消息会非常多。我建议在推送模块里加一个“聚合”逻辑同一任务名在短时间内比如 30 秒内触发多条推送时只发第一条和最后一条中间的那些合并成一条“省略了 N 条中间过程”的消息。我遇到过更麻烦的问题两个 Agent 同时调用推送模块企业微信的 token 刷新出现竞态导致其中一个的 access_token 过期消息推送失败。后来我在 token 刷新逻辑里加了线程锁确保同一时刻只有一个线程在刷新 token这个问题就彻底消失了。这个细节看似不起眼但在多 Agent 场景下非常值得注意。如果你也在做多 Agent 项目代码里只要涉及共享可变状态都要考虑并发安全性。5. 实操中的问题与避坑指南5.1 推送频率限制企业微信对应用消息的发送是有频率限制的虽然不像一些免费第三方服务那么严格但也不建议把 Agent 的每个日志都往外推。我踩过坑之后给自己定了一条规则正常情况下单个 Agent 任务推送不超过 5 条分别是“任务开始”“阶段完成最多3条”“任务结果”。如果 Agent 有大量进度要展示那就等任务结束汇总成一条发而不是实时刷屏。5.2 微信收不到消息的排查流程如果推送接口返回正常但微信里就是收不到消息我从实践中总结出一个排查顺序先确认企业微信里的“我的企业”是否绑定了微信。这个很关键如果不绑定消息只会出现在企业微信 App 里不会天然进微信。我早期测试时就是没绑导致我以为推送服务有问题排查了半天。再检查 AgentId 和 Secret 是否匹配。企业微信后台有好几个 Secret获取时容易拿错尤其是拿了一个别的应用的 Secret调用接口会返回“invalid secret”之类的错误。遇到问题时先重新生成 Secret 再试一遍。最后检查消息内容是否超长或者包含非法字符。企业微信对文本消息长度有一定限制超长会被拒绝。另外消息内容里如果有被拦截的敏感词也会导致发送失败。我的做法是在推送模块里加了内容长度检查和非法字符过滤。5.3 消息格式的兼容性问题企业微信支持文本、Markdown、图片、图文等消息类型。我只推荐文本消息因为它的兼容性最好不同客户端上显示都不会有偏差。Markdown 消息在手机端和企业微信客户端的渲染效果略有差异如果你追求稳定就不要在关键通知里依赖 Markdown 的排版。如果你有推送图片的需求比如推送图表、截图企业微信也支持传文件或图片需要先把文件上传拿到 media_id再作为消息发送。我做过一次模型训练完推送 loss 曲线图的场景最终是通过图片消息实现的效果不错但这块需要额外的上传逻辑按需扩展。5.4 千万不要推送敏感信息这是一个必须反复强调的原则。Agent 处理的数据很可能包含敏感信息比如数据库连接串、内部业务数据、个人隐私等。推送到微信意味着这些内容会出现在第三方服务器上虽然企业微信本身是企业级产品在传输加密上比较可靠但“可触达的终端设备”不等于安全边界。我在推送模块里做了关键信息脱敏数据库密码、API Key、token 一律用星号替换业务数据只推送统计量如“处理了 N 条、成功 M 条、失败 K 条”不推送具体明细。这样既满足通知需求又不会因为一条推送把核心机密泄露出去。5.5 保证 Agent 主流程不因推送而崩溃推送模块在高频调用时偶尔会遇到企业微信接口超时的情况。如果你在 Agent 主流程里同步调推送推送卡 30 秒Agent 也跟着卡 30 秒非常影响体验。我的做法是给所有推送请求加 5 秒超时并且把推送失败直接吞掉只记日志。有一点要想清楚通知是辅助能力Agent 的核心功能才是主体绝对不能因为推送服务的问题导致核心任务失败。这个优先级顺序一定要明确。6. 围绕微信推送的更多玩法这个推送服务搭建起来之后我不只是拿它给 Agent 做通知还扩展了几个实际特别有用的场景。第一个是定时提醒。我写了个循环脚本每天早上 9 点自动跑一次库存数据同步结束后推送一条报告到微信。这个和 Agent 本身没太大关系纯粹是把这个推送组件复用到了定时任务上。第二个是监控告警。我有一台小服务器上面跑了好几个服务我用一个轻量级 Agent 定期检查服务健康状态一旦发现服务挂了立即推送告警到微信。“服务异常”的消息我会特意加上醒目的文本前缀比如“紧急”微信公众号的推送会在锁屏界面显示出来肉眼看到就立刻处理。第三个是多 Agent 编排的结果汇合。当我在跑一个多 Agent 的场景时每个子 Agent 的结果不会分别推送而是统一收集到一个地方等主 Agent 聚合完毕后只推送一条完整的结果消息。这样微信不会刷屏信息也更聚焦。第四个我觉得特别实用的是对接手机端的“快捷确认”。企业微信应用消息支持模板卡片消息里可以带链接按钮。我在一个需要人工审核的 Agent 流程里做了个卡片点击“通过”就在手机上触发审核通过的后续流程。这个交互直接把 Agent 从一个无人值守的黑盒变成了一个会主动问人要反馈的半自动化系统。这些扩展玩法证明了一个事实推送服务虽然简单但它能极大释放 Agent 的生产力因为人可以在正确的时间点介入而不是被动地盯着进度条。7. 最后聊几点个人感受整套微信推送服务从构思到落地花了我大概一个周末的时间。写代码本身只占一小部分更多的时间花在选型、调试和踩坑上。现在回顾有几个决策我现在依然认为是对的第一选择了企业微信应用消息作为主链路。虽然第三方服务更快接入但它受制于免费额度和服务可用性长期来看不如自建链路稳定。第二推送模块独立封装不跟业务耦合。这个模块现在可以给任何 Python 项目用换个项目只要改环境变量代码一行不用动。第三严格遵守“推送失败不影响主流程”的原则。这条路让 Agent 在极端情况下还能保持核心任务稳定性而不是因为通知通道出问题导致整个任务挂掉。如果你也想给自己的 Agent 加一个微信通知功能我的建议是先小步快跑用 Server酱 或 PushPlus 体验一下通知带来的效率提升再认真评估要不要切换到企业微信链路。无论选哪条路记住一个核心原则——通知只是辅助渠道Agent 的任务可靠性和代码的健壮性永远比通知本身更重要。最后再给你一个我最近才总结出来的小技巧在 Agent 启动时先推一条“我准备干活了”的消息再在结束时推一条结果消息。你可能觉得开始阶段没必要提醒但实际体验下来一旦消息到达你心里会有一种“这家伙真的开始跑了”的确定感。特别是你要同时排队好几个 Agent 任务的时候开始提醒能帮你确认调度逻辑是否正确避免你重复提交任务。等你用顺手了你会感谢当初给自己做了这个通知服务。