1. 为什么我放着现成的通知方案不用非要自己写一个2024 年下半年开始我大量时间都花在跑 AI Agent 上。多轮推理任务动辄二十分钟起步代码生成、测试、修复、再生成的循环要转好几圈。问题很快就来了任务跑完了我不知道人又不能在终端前面干等。最开始我用的是最土的办法——Agent 跑完后把结果写进日志文件我每隔几分钟去翻一眼。试过的朋友应该懂这个“每隔几分钟”非常反人类你去看的时候任务还没跑完你刚离开座位它就结束了然后再等下一轮。一天下来时间全碎在看日志上了。后来我试过让 Agent“跑完继续跟我聊”也就是在 Prompt 里写“完成后把结果告诉我”。结果更尴尬我人没在电脑前跟谁聊而且长会话一开后面所有的任务上下文都会被之前的对话污染Token 消耗也明显往上飙。还有同事推荐用 Server酱、PushPlus 这类微信推送服务把结果当成消息推到微信里。这确实是可行的方案我也短暂用过。但用了一段时间后发现几个很现实的问题第三方平台的调用限额对高频任务不太友好免费档每天只有几次配额真正跑起来根本不够用消息内容格式受限只能发纯文本Agent 生成的结构化 JSON 日志、错误堆栈、任务统计数据塞进去就是一团乱麻所有请求都走第三方中转企业内部用的时候数据痕迹全部落在别人的服务器上这个心理门槛其实挺高的。琢磨了一周之后我决定写一个自己的微信推送服务。需求非常明确Agent 跑完任务后通过 HTTP 接口给我推一条消息消息以微信通知的形式出现手机上能实时看到。一句话总结就是——自己搭一个通知通道然后把“通知”这件事变成 Agent 的一个工具。2. 微信推送服务第0版企业微信 Webhook 接收代码我第一个想到的就是企业微信群机器人。原因不复杂个人微信的接口是封死的想给个人微信发消息只能依赖第三方服务而企业微信群机器人是官方开放的 Webhook 接口往群里发消息不需要审核拿到 Webhook 地址就能调。但有一个前提需要说清楚用企业微信群机器人接收方必须有一个企业微信账号并且被拉进机器人所在的那个群。我是自己注册了一个企业微信建了一个只有自己的群然后把机器人拉进去这样消息就相当于推给了自己。把这个思路讲给同事之后几个人的小团队也复制了同一套方案。服务端的实现其实很简单。我用的 FastAPI本质就是一个接收 POST 请求、转发到企业微信 Webhook 的中间层。核心代码长这样import httpx import uvicorn from fastapi import FastAPI, HTTPException app FastAPI() # 企微机器人 Webhook建议用环境变量 WEBHOOK_URL https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key_here app.post(/notify) async def notify_wechat(union_id: str, content: str): if not union_id or not content: raise HTTPException(status_code400, detail缺少 union_id 或 content) payload { msgtype: text, text: { content: content, mentioned_list: [all] # 默认 所有人实际看需不需要 } } async with httpx.AsyncClient(timeout10) as client: resp await client.post(WEBHOOK_URL, jsonpayload) if resp.status_code ! 200: raise HTTPException(status_code502, detailf企微推送失败: {resp.text}) return {status: ok, union_id: union_id, sent_at: datetime.now().isoformat()}接口设计成了两个参数union_id和content。union_id是用来区分任务来源的后面 Agent 并发跑多个任务的时候就知道消息归属谁content就是推给微信的实际文本内容。启动方式也没什么玄学一个uvicorn命令的事uvicorn main:app --host 0.0.0.0 --port 9100为什么选 9100 这个端口纯粹是个人习惯避开常规的 8000、8080省得跟同机部署的其他开发服务撞车。跑起来之后用curl验证一下curl -X POST http://localhost:9100/notify?union_idtestcontenthello手机企业微信里收到“hello”的那一刻第0版就算跑通了。这里有一个细节值得多说一句为什么不直接用脚本调企业微信 Webhook非要套一个中间服务原因有三个。第一Webhook 地址如果散落在各个 Agent 配置里换一次 key 就要全改一遍中间服务只暴露自己的接口Webhook 地址是唯一的。第二企微机器人有频率限制中间层可以在转发前做兼容处理限流、重试、合并消息这些逻辑加在脚本里会污染 Agent 的代码。第三中间服务可以做权限校验后续接多个人一起用的时候只需要管理自己服务的用户维度而不是把同一个 Webhook 到处分。3. Agent 侧接入让“通知”变成 Agent 的一个工具服务端好了接下来就是重头戏怎么让 AI Agent 在任务结束时主动调这个通知接口。最直接的办法是改 Prompt——在系统提示词里写“任务完成后调用 http://xxx/notify 发送结果”。技术上可行但实际跑下来效果不稳定。Agent 有时候会把参数拼错有时候在任务中间就提前触发了通知还有时候完全忘了这回事。大模型的“自觉”是靠不住的你得给它提供一个结构化的调用入口。正确做法是把通知封装成一个工具Tool/Function让 Agent 框架在需要时以函数调用的方式触发。以 Claude Agent SDK 为例def send_wechat_notification(union_id: str, content: str) - str: 向微信发送任务完成通知。 Args: union_id: 任务标识用来区分不同任务的消息 content: 要推送的内容可以是结果摘要、错误信息等 import httpx resp httpx.post( http://localhost:9100/notify, params{union_id: union_id, content: content} ) return f通知已发送状态码 {resp.status_code}然后在 Agent 的工具注册表里加上这个函数tools [ { name: send_wechat_notification, description: 任务完成后向用户微信发送通知包含任务结果摘要。仅在任务全部结束时调用一次。, input_schema: { type: object, properties: { union_id: { type: string, description: 任务ID或标识, }, content: { type: string, description: 要推送的消息内容, } }, required: [union_id, content] } } ]注意description里我特意写了“仅在任务全部结束时调用一次”。这句话对 Agent 的行为约束非常关键不加这句话它可能在每个中间步骤都推一条消息你的手机会被通知轰炸。如果你用的是 LangGraph思路也完全一样把通知函数挂到某个节点之后自动执行from langgraph.graph import StateGraph, END def execute_task(state): # 任务执行逻辑 result run_agent_task(state[task]) # 节点结束前推送通知 send_wechat_notification( union_idstate[task_id], contentf任务完成结果: {result.get(summary, 无)} ) return {result: result} graph.add_node(execute_task, execute_task) graph.add_edge(execute_task, END)对接 Continue 这类开源 AI 编程工具的时候也一样在自定义函数里把通知逻辑包好就行机制是通用的。还有一点值得提很多 Agent 跑在 Docker 容器或内网机器上这些环境没有公网 IPAgent 自身无法访问外网服务。所以通知服务必须和 Agent 部署在同一个内网网络里或者通过内网穿透暴露出来。我本地的做法是直接用内网地址Agent 容器用 Docker 的host.docker.internal来访问宿主机服务。如果你的是生产环境、多个 Agent 分布在不同的机器上我建议把这个服务部署在公司的公网网关上或者用已有的内网穿透通道。4. 多任务并发时的通知治理任务 ID 与消息聚合第 0 版跑通之后我很快遇到了第二个问题Agent 并发跑多个任务的时候微信会被推吐。你想象一下这个场景一个批量脚本同时启动 5 个 Agent每个 Agent 按时完成任务并各自发通知。如果你不幸把 20 多个文件都丢给 Agent 批处理那手机上的通知能直接刷屏。而且同一个任务可能有中途更新、最终结果、失败告警等多条状态通通推过来就是灾难。解决思路是给任务打标签并在服务端做消息聚合。我在通知接口里加了一个字段task_id让 Agent 每次带上它所属的任务标识。服务端不直接转发每条消息而是按task_id做一个简单的聚合窗口短时间内的消息合并成一条推送from collections import defaultdict import asyncio message_buffer defaultdict(list) buffer_lock asyncio.Lock() AGGREGATE_WINDOW 5 # 5秒内的消息合并推送 async def append_and_flush(union_id, task_id, content): key f{union_id}:{task_id} async with buffer_lock: message_buffer[key].append(content) if len(message_buffer[key]) 1: # 第一次收到该 task 的消息创建一个延迟冲洗任务 asyncio.create_task(flush_later(key)) async def flush_later(key): await asyncio.sleep(AGGREGATE_WINDOW) async with buffer_lock: contents message_buffer.pop(key, []) if contents: merged \n.join(contents) await forward_to_wecom(merged)如果你的 Agent 任务都是长流程中间状态并不多这个聚合窗口可以设小一点如果任务非常碎片化可以加大窗口减少推送频率。5 秒是一个比较能兼顾实时性和通知压力的平衡点。配合任务 ID 的还有一个习惯Agent 通知内容里第一行必须包含任务 ID 和最终状态。我会约定格式为[任务ID] 状态: ✅完成 | ⏳进行中 | ❌失败这样在手机端扫一眼就能判断哪些任务需要人工介入哪些可以继续放任跑。另外还要提一个容易被忽略的点企微机器人消息内容有长度限制大约不超过 4096 字节Agent 生成的内容一长就容易被截断。所以最好在通知前对content做裁剪只保留关键信息MAX_CONTENT_BYTES 3800 # 留出余量 def trim_content(content: str) - str: bytes_content content.encode(utf-8) if len(bytes_content) MAX_CONTENT_BYTES: return content return bytes_content[:MAX_CONTENT_BYTES].decode(utf-8, errorsignore) ……(已截断)多任务通知的另一个实操经验是设置“安静时段”。白天人可能在开会、写代码、跟人对需求每一条通知都即时到达但晚上人睡觉了Agent 批量苟活跑任务通知发过来也没人看。我在配置里加了时间判断晚上 22 点到早上 8 点之间的消息只记录不推送早上统一发一条汇总。5. 踩坑实录企业微信 Webhook 的六个真实教训这个服务前后跑了两个多月我踩过的坑至少有六个值得单独拿出来说每一个都花了我不少时间排查。5.1 端口和防火墙本机通了服务器上不通本机curl测试一切正常部署到服务器上就是超时。排查了半天才发现服务器安全组默认只开放了 80、443 等少数端口9100 没有被放行。这不是代码问题是网络策略问题。解决方式很简单控制台里把 9100 加到安全组入站规则或者更省心一点用 Nginx 做反向代理把/notify路径代理到内网端口。5.2 微信侧请求用的是长连接别频繁断连企业微信 Webhook 接口本身不要求长连接但你用httpx.AsyncClient的时候很容易为每次请求建一个新的 client。我之前就是每次调用都async with httpx.AsyncClient()后来发现偶尔会出现Connection reset by peer的报错。原因是企微侧对同一客户端 IP 的频繁新建连接有限流和冷启动保护。解决办法是复用httpx.AsyncClient实例或者直接用requests的单例 Session。5.3 重试风暴服务端挂了一个Agent 一起重试Webhook 直接被限流企业微信 Webhook 官方的频率限制是 20 条/分钟不同套餐有差异。我当时在通知服务里加了重试机制每 2 秒重试一次。结果某个中午企业微信那边接口抖动我的服务不断重试瞬间把额度打满后面正常消息全被丢弃群里一片寂静。解决方式是加退避策略指数退避起步 1 秒最大间隔 30 秒最多重试 3 次如果还是失败把消息落盘到本地等网络恢复后再补推。5.4 消息类型只用 textmarkdown 消息其实是企微暗坑企微机器人支持text、markdown、news、template_card等消息类型。我一开始为了排版好看直接用了markdown类型结果 Agent 返回的内容里带着各种 Markdown 特殊字符渲染出来特别难看有的甚至直接解析失败被拒收。结论通知类消息就用纯文本text。文本里不要放|符号企微对某些特殊字符有转义处理会导致内容被截断或替换。5.5 Agent “幻觉”调用通知工具参数是瞎传的这个问题最有意思。有一次 Agent 跑完任务通知里的内容完全不是我想要的结果而是一段它“编造”的任务总结。排查之后发现原因Agent 在执行中途生成了send_wechat_notification的调用但传给content的不是真实的执行结果而是它认为的“预期结果”。这就是大模型推理的不确定性。对策有两个一是工具的描述里强调“内容必须是实际执行结果的真实摘要禁止臆造”二是在 Agent 代码里对content的长度做约束过短或过长的内容都不推。但说实在的最有用的还是把通知触发权放到代码节点让 Agent 只生成结果数据由外部代码负责调用通知工具。5.6 Server酱 与自建服务的选择焦虑最后说一个选型上的体会。Server酱 这类服务更适合个人开发者做少量测试依赖它的免费配额接口也很稳。但一旦你要给团队里的多个 Agent 跑生产任务、要对接内部系统自建一个 Webhook 中转服务的收益就很明显了消息格式、推送频率、内容裁剪、时间策略全部在自己手里不受第三方限制。如果你们公司内部已经有钉钉、飞书那其实不需要重复造轮子直接用它们自带的群机器人接口做同样的中转服务就行原理一模一样。6. 进阶玩法从文本通知到可以回看的通知中心跑通基础版之后我逐渐意识到一个更深层的需求通知只是入口我更想要的是通知的可追溯性。微信消息刷过去了就找不回来了尤其 Agent 任务跑了半小时结束后我打开手机看到一条“任务完成”但具体的完整日志还是得回到终端里翻。如果人能通过通知里的链接直接跳到对应的日志页面甚至在线看板整个体验就闭环了。所以我给通知服务加了一个存储模块每次推送的消息都存一份到 SQLite 里按task_id索引。通知内容里的摘要后面附一个链接形如http://your-internal-server:9100/tasks/{task_id}点开就是在线的任务状态页里面能看到完整的执行日志、开始/结束时间、产出文件列表。到这里这个微信推送服务已经不只是“跑步机上的铃铛”而是一个轻量级的任务通知中心。Agent 跑完任务微信响一声我点开链接就能拿到全部上下文。就算不在电脑前也能在手机上把所有关键状态看个明明白白。如果你的需求只是“结果能到手机上就行”那第 2 节的代码完全够用如果你也像我一样有任务治理、多 Agent 并发、历史追溯的需求我建议直接按第 4、第 5 节的思路把服务做得稍微完整一点。做一次所有 Agent 项目都能接着用边际收益还是很高的。现在这个服务已经跑了两个多月我手头的 Agent 任务基本都接进来了。每天的推送量在 50~100 条左右群机器人稳稳扛住。我自己感受最深的一句话就是工具这个东西不是越复杂越好而是“刚好能用”最好。微信推送服务听起来很小但它解决的是我日常工作中最高频的等待焦虑。你如果也在做 Agent 开发或者正在为“任务跑完不知道怎么知道”发愁完全可以照这个思路搭一个。