别再手动@人了！用钉钉机器人自动发送监控告警到群聊（附Python/Node.js代码）

钉钉机器人自动化告警：解放双手的智能运维实践

凌晨三点，服务器突然宕机，值班工程师强撑着睡意打开电脑，在群里@相关同事——这样的场景在技术团队中屡见不鲜。其实，通过钉钉群机器人，我们可以让告警信息像流水一样自动推送到指定群聊，不仅省去了人工干预的麻烦，还能确保关键信息不被遗漏。本文将带你从零开始，实现一套高效的自动化告警系统。

1. 为什么选择群机器人而非工作通知？

在钉钉生态中，消息推送主要有两种方式：工作通知和群机器人。对于监控告警场景，群机器人具有明显优势：

• 受众管理更灵活：工作通知需要维护接收人列表，而群机器人只需将相关人员加入群聊即可
• 历史记录可追溯：群聊消息天然形成对话上下文，便于后续问题排查
• 交互性更强：支持@特定成员、快捷回复等交互方式
• 配置更简单：无需处理复杂的用户权限体系

下表对比了两种方式的特性差异：

2. 快速搭建告警机器人

2.1 创建群聊与添加机器人

首先，我们需要创建一个专门用于接收告警的群聊：

1. 在钉钉PC端，点击"+"号选择"发起群聊"
2. 选择"项目群"类型，命名为"[项目名]监控告警中心"
3. 邀请相关成员加入群聊

添加机器人步骤：

1. 在群设置中选择"智能群助手"
2. 点击"添加机器人"，选择"自定义"类型
3. 设置机器人名称（如"监控小助手"）和头像
4. 记录生成的Webhook地址（包含access_token参数）

注意：Webhook地址是敏感信息，应当妥善保管，避免泄露

2.2 安全配置建议

为保障系统安全，建议采取以下措施：

• 为Webhook地址设置IP白名单
• 定期轮换access_token
• 限制机器人的消息发送频率
• 为不同环境（开发/测试/生产）创建独立的机器人

3. 代码实战：发送告警消息

3.1 Python实现方案

以下是使用Python发送告警消息的完整示例：

import requests import json from datetime import datetime class DingTalkAlert: def __init__(self, webhook_url): self.webhook_url = webhook_url def send_text_alert(self, content, at_mobiles=None, is_at_all=False): """ 发送文本告警 :param content: 告警内容 :param at_mobiles: 需要@的手机号列表 :param is_at_all: 是否@所有人 """ headers = {'Content-Type': 'application/json'} payload = { "msgtype": "text", "text": { "content": content }, "at": { "atMobiles": at_mobiles or [], "isAtAll": is_at_all } } response = requests.post( self.webhook_url, headers=headers, data=json.dumps(payload) ) return response.json() def send_markdown_alert(self, title, text, at_mobiles=None): """ 发送Markdown格式告警 :param title: 消息标题 :param text: Markdown内容 :param at_mobiles: 需要@的手机号列表 """ headers = {'Content-Type': 'application/json'} payload = { "msgtype": "markdown", "markdown": { "title": title, "text": text }, "at": { "atMobiles": at_mobiles or [] } } response = requests.post( self.webhook_url, headers=headers, data=json.dumps(payload) ) return response.json() # 使用示例 if __name__ == "__main__": webhook = "https://oapi.dingtalk.com/robot/send?access_token=your_token" alert = DingTalkAlert(webhook) # 发送文本告警 alert.send_text_alert("数据库连接数超过阈值！", at_mobiles=["13800138000"]) # 发送Markdown告警 markdown_content = """## 服务器告警 **时间**: {time} **主机**: 10.0.0.1 **问题**: CPU使用率95% **建议**: 立即检查运行进程 """.format(time=datetime.now().strftime("%Y-%m-%d %H:%M:%S")) alert.send_markdown_alert("服务器告警", markdown_content)

3.2 Node.js实现方案

对于Node.js技术栈的团队，可以使用以下实现：

const axios = require('axios'); const moment = require('moment'); class DingTalkBot { constructor(webhookUrl) { this.webhookUrl = webhookUrl; } async sendText(content, atMobiles = [], isAtAll = false) { const payload = { msgtype: 'text', text: { content }, at: { atMobiles, isAtAll } }; try { const response = await axios.post(this.webhookUrl, payload); return response.data; } catch (error) { console.error('发送钉钉消息失败:', error.message); throw error; } } async sendMarkdown(title, text, atMobiles = []) { const payload = { msgtype: 'markdown', markdown: { title, text }, at: { atMobiles } }; try { const response = await axios.post(this.webhookUrl, payload); return response.data; } catch (error) { console.error('发送钉钉Markdown消息失败:', error.message); throw error; } } } // 使用示例 (async () => { const bot = new DingTalkBot('https://oapi.dingtalk.com/robot/send?access_token=your_token'); // 发送文本消息 await bot.sendText('服务异常：订单服务响应超时', ['13800138000']); // 发送Markdown消息 const markdownContent = `## 服务监控告警 **服务名称**: 订单服务 **异常时间**: ${moment().format('YYYY-MM-DD HH:mm:ss')} **错误信息**: 接口响应时间超过2000ms **相关链接**: [查看详情](http://monitor.example.com)`; await bot.sendMarkdown('订单服务告警', markdownContent); })();

4. 高级功能与优化实践

4.1 消息卡片优化

纯文本告警可能信息量有限，我们可以使用Markdown或ActionCard格式创建更丰富的告警卡片：

def send_action_card_alert(self, title, text, single_title, single_url): """ 发送动作卡片告警 :param title: 卡片标题 :param text: 卡片内容 :param single_title: 按钮文字 :param single_url: 点击跳转链接 """ payload = { "msgtype": "actionCard", "actionCard": { "title": title, "text": text, "singleTitle": single_title, "singleURL": single_url, "btnOrientation": "0" } } response = requests.post( self.webhook_url, headers={'Content-Type': 'application/json'}, data=json.dumps(payload) ) return response.json()

优秀告警卡片应包含：

• 明确的告警级别（如紧急、警告、提示）
• 关键指标数值与阈值
• 问题发生时间
• 相关资源链接
• 建议处理措施

4.2 频率限制与错误处理

钉钉机器人有发送频率限制（默认20条/分钟），超出限制会导致消息发送失败。建议：

• 实现消息队列缓冲，避免突发流量
• 对非紧急告警进行聚合发送
• 添加重试机制处理网络波动

from ratelimit import limits, sleep_and_retry class RateLimitedAlert(DingTalkAlert): @sleep_and_retry @limits(calls=15, period=60) # 保守控制在15条/分钟 def send_text_alert(self, content, at_mobiles=None, is_at_all=False): return super().send_text_alert(content, at_mobiles, is_at_all)

4.3 告警分级与路由

不同级别的告警应区别处理：

• 紧急告警（P0）：立即@相关人员，使用电话提醒
• 重要告警（P1）：@相关团队，要求30分钟内响应
• 一般告警（P2）：记录但不立即通知
• 提示信息：每日汇总报告

实现示例：

def send_alert_by_level(self, level, content, related_team): at_mobiles = { 'P0': ['13800138000', '13900139000'], # 运维负责人 'P1': ['13500135000', '13600136000'], # 开发团队 'P2': [], 'INFO': [] }.get(level, []) prefix = { 'P0': '【紧急】', 'P1': '【重要】', 'P2': '【警告】', 'INFO': '【提示】' }.get(level, '') full_content = f"{prefix}{content}\n相关团队：{related_team}" return self.send_text_alert(full_content, at_mobiles, level == 'P0')

5. 监控系统集成实践

将钉钉机器人接入常见监控系统：

5.1 Prometheus + Alertmanager

在Alertmanager配置中添加钉钉webhook：

receivers: - name: 'dingtalk' webhook_configs: - url: 'http://your-alert-proxy/dingtalk?token=your_token' send_resolved: true

5.2 Zabbix集成

创建告警媒介类型：

1. 在Zabbix控制台进入"管理" → "告警媒介类型"
2. 创建新类型，名称"钉钉机器人"
3. 脚本内容示例：

#!/bin/bash # 参数：$1=接收人 $2=主题 $3=内容 WEBHOOK="https://oapi.dingtalk.com/robot/send?access_token=your_token" CONTENT=$(echo "$3" | jq -Rs '.') PAYLOAD="{\"msgtype\":\"text\",\"text\":{\"content\":\"$2\n$CONTENT\"}}" curl -s -X POST -H "Content-Type: application/json" -d "$PAYLOAD" "$WEBHOOK"

5.3 自定义监控脚本

对于自建监控系统，可以直接调用前面实现的Python/Node.js类。关键点：

• 添加适当的异常处理
• 记录消息发送日志
• 实现消息模板系统
• 考虑性能影响（异步发送）

import asyncio from concurrent.futures import ThreadPoolExecutor class AsyncDingTalkAlert(DingTalkAlert): def __init__(self, webhook_url, max_workers=5): super().__init__(webhook_url) self.executor = ThreadPoolExecutor(max_workers=max_workers) async def async_send_text_alert(self, content, at_mobiles=None, is_at_all=False): loop = asyncio.get_event_loop() return await loop.run_in_executor( self.executor, self.send_text_alert, content, at_mobiles, is_at_all )

在实际项目中，我们通常会将这些功能封装成独立的服务，提供REST API供各个系统调用，实现告警的统一管理和分发。