企业微信API实现AI Agent任务完成通知

1. 项目背景与核心需求去年在开发一个自动化数据处理AI Agent时我遇到了一个典型问题当Agent在后台完成耗时任务后如何第一时间获知结果传统方案是不断刷新日志文件或监控终端输出但这显然不够高效。于是我开始探索如何实现任务完成后的主动通知机制。微信作为国内最高频的通讯工具自然成为首选通知渠道。经过几轮迭代我开发了一套轻量级的微信推送服务核心解决以下痛点AI Agent执行时间不可预测人工轮询效率低下关键节点需要即时反馈如任务开始/失败/完成多设备环境下需要统一通知入口企业场景中需将运行结果同步给团队成员2. 技术方案选型与对比2.1 主流通知方案对比通知方式实现难度及时性跨平台性开发成本邮件通知★★☆★★☆★★★★★☆SMS短信★★★★★★★★★★★★钉钉/飞书机器人★★☆★★★★★☆★★☆微信推送★★★★★★★★★★★☆选择微信推送的核心优势用户打开率高达98%对比邮件20%支持富文本消息图文/文件/卡片无需额外安装客户端企业微信接口稳定可靠2.2 微信通知实现路径方案AServer酱优点5分钟快速接入缺点消息模板固定企业场景有频次限制方案B企业微信API优点官方支持功能完整缺点需要企业资质认证方案C微信公众测试号优点个人开发者可用缺点接口权限有限最终选择企业微信API方案因其支持markdown消息格式单日消息上限1000条可创建应用专属通知3. 详细实现步骤3.1 企业微信应用创建注册企业微信无需企业资质个人可注册在「应用管理」创建自建应用记录三个关键参数CORP_ID wwxxxxxx AGENT_ID 1000002 SECRET xxxxxxxx3.2 访问令牌获取import requests def get_access_token(): url fhttps://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid{CORP_ID}corpsecret{SECRET} resp requests.get(url).json() return resp[access_token] # 有效期2小时重要令牌需要缓存复用避免频繁请求触发限流3.3 消息推送实现支持多种消息类型模板文本消息def send_text(content, to_userall): url fhttps://qyapi.weixin.qq.com/cgi-bin/message/send?access_token{token} payload { touser: to_user, msgtype: text, agentid: AGENT_ID, text: {content: content}, safe: 0 } return requests.post(url, jsonpayload).json()Markdown消息def send_markdown(content): payload { touser: all, msgtype: markdown, agentid: AGENT_ID, markdown: {content: content} } # 同文本消息发送逻辑3.4 与AI Agent集成在Agent的关键节点添加通知钩子class MyAgent: def on_task_start(self): send_text(f 任务开始执行\nID: {self.task_id}) def on_task_fail(self, error): send_markdown(f❌ 任务失败\n\n{error}\n) def on_task_success(self, result): send_markdown(f✅ 任务完成\n耗时: {self.elapsed}\n结果摘要:\n{result[:200]}...)4. 高级功能实现4.1 消息卡片交互通过「查看详情」按钮跳转网页def send_card(title, desc, url): payload { touser: all, msgtype: textcard, agentid: AGENT_ID, textcard: { title: title, description: desc, url: url, btntxt: 查看详情 } }4.2 文件上传通知先上传文件获取media_iddef upload_file(filepath): url fhttps://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token{token}typefile with open(filepath, rb) as f: return requests.post(url, files{media: f}).json()[media_id]再发送文件消息def send_file(media_id): payload { touser: all, msgtype: file, agentid: AGENT_ID, file: {media_id: media_id} }5. 实战经验与避坑指南5.1 频率限制策略企业微信API限制规则单个应用上限500次/分钟单个用户上限30次/分钟优化建议合并通知内容如将多个步骤结果汇总发送重要消息添加重试机制def safe_send(func, *args, retry3): for i in range(retry): try: return func(*args) except Exception as e: if i retry - 1: raise time.sleep(2 ** i)5.2 消息模板设计原则优秀通知应包含状态标识✅/❌/⚠️任务唯一ID关键时间戳错误详情/结果摘要相关资源链接示例模板【数据清洗任务】ID: ETL-20240520-001 状态: ✅ 完成 开始: 2024-05-20 14:00:23 耗时: 12分35秒 记录处理: 24,582条 异常数据: 127条已归档 详情: https://internal.com/reports/ETL-20240520-0015.3 安全防护措施敏感信息脱敏处理def sanitize(text): for pattern in [password, token, key]: text re.sub(f{pattern}[^\s], f{pattern}***, text) return text接收人白名单控制ALLOWED_USERS [zhangsan, lisi] def validate_user(user): if user not in ALLOWED_USERS: raise ValueError(fUnauthorized user: {user})6. 扩展应用场景6.1 分布式任务监控在Celery等分布式系统中添加通知app.task(bindTrue) def data_processing(self, input_data): try: send_text(fCelery任务开始: {self.request.id}) result process_data(input_data) send_card(处理完成, f生成报告: {len(result)}条, report_url) return result except Exception as e: send_markdown(f\n{traceback.format_exc()}\n) raise6.2 自动化测试报告Pytest集成示例def pytest_terminal_summary(terminalreporter): passed len(terminalreporter.stats.get(passed, [])) failed len(terminalreporter.stats.get(failed, [])) send_markdown( f**测试结果**\n f- ✅ 通过: {passed}\n f- ❌ 失败: {failed}\n f- ⏱ 耗时: {time.time() - start_time:.1f}s )6.3 服务器监控告警结合Prometheus Alertmanager配置receivers: - name: wechat webhook_configs: - url: http://alert-handler:5000/wechat send_resolved: true处理服务示例app.route(/wechat, methods[POST]) def handle_alert(): alert request.json[alerts][0] send_markdown( f**[{alert[status].upper()}]** {alert[labels][alertname]}\n f\n{alert[annotations][description]}\n )