做后端开发和运维的人,大概率都有过这样的经历:监控系统报警了,邮件发出去没人及时看,值班群里的消息又被聊天记录淹没,等真正有人处理的时候,已经过去了半小时。我自己是在一次线上变更踩坑之后,才认真把钉钉机器人开发这件事从头到尾折腾了一遍。它看起来只是"往群里推一条消息"的小功能,实际牵扯到机器人选型、安全签名、消息类型选择、回调接入和限流处理,每个环节都有让人意外的地方。
这篇文章不是官方文档的复述,而是我从自定义机器人、企业内部应用机器人一直到 Stream 模式完整实践下来的记录。如果你正在考虑把告警通知、自动化任务反馈、群内命令交互接到钉钉上,可以照着这套思路做。先给个结论:不要一上来就选自定义机器人,先想清楚你只需要"单向发消息",还是需要"机器人能收到消息再回复",这两条路的工作量差了好几倍。
1. 先分清三种机器人形态,再决定从哪条路上手
很多人在第一步就走错了。看到一个 Webhook 地址就往里 POST 数据,结果后面要做消息回调时发现自定义机器人根本不支持,只能推倒重来。钉钉机器人开发里常见的路径其实是三条,先花五分钟搞清楚它们分别能干什么,比急着写代码划算得多。
1.1 自定义群机器人:五分钟跑通,适合单向通知
自定义群机器人是最轻量的一种。它不需要去开发者后台申请应用,直接在群设置里添加即可:群设置、智能群助手、添加机器人、自定义机器人,创建完成后会得到一个 Webhook 地址。只要向这个地址 POST 一段 JSON,消息就会进群。
它的特点非常鲜明:
- 只能主动推送消息,不能接收群里用户对它的 @ 和回复;
- 没有独立的开发者身份,拿不到发送者是谁、群成员有哪些;
- 安全策略依赖关键词、加签、IP 白名单中的一种或多种;
- 适合监控告警、构建结果通知、定时提醒这类"单向广播"场景。
如果你的需求就是"系统出问题时往群里面吼一声",自定义机器人是成本最低的选择。但要注意,Webhook 一旦泄露,任何人都能往群里发消息,所以安全设置不能省。
1.2 企业内部应用机器人:能收消息、能对话的"正规军"
当需求变成"用户在群里 @ 机器人,机器人能解析内容并回复"时,就要用企业内部应用机器人。它通过开发者后台创建企业内部应用,然后在应用里添加机器人能力,设置机器人名称、头像、消息接收模式,最后发布到组织内使用。
相比自定义机器人,它多了几个关键能力:
- 可以接收群内 @ 消息,具备双向交互能力;
- 可以获取发送者的用户标识,做权限判断;
- 可以结合通讯录接口,按部门或指定人发送工作通知;
- 可以申请更多开放能力,比如读取消息、读取组织架构、发送互动卡片等。
代价是开发链路变长了。你需要管理 AppKey 和 AppSecret,用它换取全局 access_token,再带着 token 调用发送和接收相关接口。权限申请、发布审核这些流程也会多出几步。但如果目标是做一个"群内运维助手"或者"自动问答机器人",这条路是正解。
1.3 Stream 模式:不需要公网回调的接入方式
Stream 模式严格来说不是第三种"机器人类型",而是企业内部应用机器人接收消息的一种接入方式。但它值得单独拿出来讲,因为太多人被回调卡住了。
传统回调流程是:用户在群里 @ 机器人,钉钉服务器把你的公网地址作为回调 URL,把事件 POST 过来。这个模式要求你有一个公网可达的 HTTPS 地址,还要有域名、证书,甚至要过备案相关流程。开发阶段尤其痛苦,本地调试时只能在临时服务器上改来改去,非常麻烦。
Stream 模式反过来了:由你的程序主动和钉钉服务端建立一条长连接,事件到达后从这条连接推送给你的程序。你不需要开放任何公网端口,也不需要有公网 IP。开发机、内网服务器、甚至一台没外网映射的笔记本都能跑。我的建议很直接:只要你有"接收消息"的需求,直接用 Stream 模式,别再去折腾 HTTP 回调。
1.4 三种路径怎么选:一张表定性
| 选型维度 | 自定义群机器人 | 企业内部应用机器人(HTTP回调) | 企业内部应用机器人(Stream模式) |
|---|---|---|---|
| 开发成本 | 极低,数分钟可跑通 | 较高,需要公网服务 | 中等,本地即可调试 |
| 能否接收消息 | 不能 | 能 | 能 |
| 能否获取发送者身份 | 不能 | 能 | 能 |
| 典型场景 | 告警推送、通知广播 | 正式生产环境的对话机器人 | 开发调试、内网部署、对话机器人 |
| 主要门槛 | Webhook 泄露风险 | 公网域名、HTTPS 证书 | 需常驻长连接进程 |
我通常会被问到:"自定义机器人能不能接收到回复?"答案是不能,这是它的架构决定的。如果你一开始就打算做"群内交互",请直接选企业内部应用机器人。如果只是推送通知,选自定义机器人。
2. 机器人创建与安全参数:Webhook、加签、关键词到底在防什么
确定方向后,下一个容易出问题的点是安全参数。很多人把 Webhook 当普通 URL 用,不设置加签,也不关心关键词匹配,结果要么消息发不出去,要么 Webhook 泄露导致群里被灌垃圾消息。这里把创建流程和安全设计一起说清楚。
2.1 创建机器人时容易忽略的配置细节
自定义机器人在群设置里创建时,会要求选择一种安全设置,常见三种:自定义关键词、加签、IP 地址段。这三种可以同时配置,但注意一个关键点:如果你加了 IP 白名单,就必须保证服务器出口 IP 在名单里,动态 IP 环境不要选这个。
企业内部应用机器人则不是用群里的 Webhook,而是走应用体系。创建时要填机器人名称、头像、简介,还要设置消息接收模式和应用发布范围。很多人在创建后直接测试,发现怎么都收不到消息,原因往往是忘了把机器人发布出去,或者没有在测试群里添加这个机器人。发布后还要注意权限申请,比如读取通讯录这类敏感权限,需要额外的授权流程。
另外,一个组织里经常有人重复创建同一个名字的机器人。同一个群内机器人数量是有限制的,如果建了又不用,记得删掉,否则后面真正要加的时候可能提示达到上限。
2.2 加签计算:HMAC-SHA256 的完整流程
加签是目前最推荐的安全方式。它的原理很简单:请求 URL 上带上 timestamp 和 sign,钉钉服务端用同样的算法验证签名。timestamp 的存在是为了防止别人拿到一个已经发出的请求后原样重放,所以每次请求都必须重新生成。
签名规则是固定的:把 timestamp 和 secret 拼成字符串timestamp\nsecret,用 HMAC-SHA256 加密,结果做 Base64 编码,再做 URL 编码,放进请求参数里。用 Python 写出来是这样:
import time import hmac import hashlib import base64 import urllib.parse def generate_sign(secret: str) -> tuple[str, str]: timestamp = str(round(time.time() * 1000)) string_to_sign = f"{timestamp}\n{secret}" hmac_code = hmac.new( secret.encode("utf-8"), string_to_sign.encode("utf-8"), digestmod=hashlib.sha256, ).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) return timestamp, sign这里有两个非常容易踩的坑。
第一个是 timestamp 必须是毫秒级字符串,不是秒。如果用time.time()直接转字符串,大概率是秒,签名验证必然失败。
第二个是 Base64 之后一定要做 URL 编码。Base64 结果里可能出现+、/、=这几个字符,直接在 URL query 里使用会被解析成空格或分割符,导致签名比对失败。用quote_plus编码后放在 query 上,这才是钉钉服务端期望的格式。
实际的发送请求最终是这样的:
import requests def send_signed_message(webhook: str, secret: str, payload: dict) -> dict: timestamp, sign = generate_sign(secret) url = f"{webhook}×tamp={timestamp}&sign={sign}" resp = requests.post(url, json=payload, timeout=5) return resp.json()2.3 关键词和 IP 白名单的实际局限
关键词模式有个天然约束:消息文本必须包含你设置的关键词,否则接口直接拒绝。比如你设置关键词为"告警",那么所有推送消息里都必须带上"告警"两个字。对监控告警来说这还能接受,但对业务通知、报表消息来说就很别扭,因为你永远要在文案里强行嵌入固定词。
IP 白名单则是从来源 IP 做限制。看起来最安全,实际在动态 IP 环境或经过代理转发的环境里经常失效。有时候本地测试正常,部署到服务端后突然发不出去,排查半天发现是出口 IP 变了,或者请求走了代理节点。
所以我的建议是:如果你能用加签,就别依赖关键词和 IP 白名单。加签不限制消息内容,也不依赖网络环境,只要 secret 不泄露,别人拿不到合法签名就发不了消息。
2.4 企业内部应用拿 access_token 的规范做法
企业内部应用机器人走的是 OpenAPI,第一步是用 AppKey 和 AppSecret 获取 access_token。这个 token 有有效期,通常是 7200 秒,必须做缓存,不能每个请求都重新获取。频繁刷新 token 既浪费网络请求,也可能触发接口频率限制。
获取 token 的接口逻辑很简单:
import requests def get_access_token(app_key: str, app_secret: str) -> str: resp = requests.get( "https://oapi.dingtalk.com/gettoken", params={"appkey": app_key, "appsecret": app_secret}, timeout=5, ) data = resp.json() if data.get("errcode") != 0: raise RuntimeError(f"gettoken failed: {data}") return data["access_token"]拿到 token 后,后续调发送消息、读取会话等接口都要在请求头或参数里带上它。一个值得记住的经验是:token 千万不要打到日志里。出问题排查时可以临时打印脱敏后的前几位,但不能整段输出,因为它在整个应用体系里等同于一串高级凭证。
3. 推送消息的代码骨架:文本、Markdown、ActionCard一次讲透
消息推送本身不复杂,真正复杂的是不同消息类型的选择和字段设计。这一节我把最常用的四种消息类型全部拆开,直接给可复用的代码和字段说明。
3.1 先封装一个最基础的发送函数
不管哪种消息类型,发送动作都是一样的:往 Webhook 或 OpenAPI 地址 POST 一段 JSON。我先封装一个基础函数,后面所有类型都复用:
import requests def push_to_dingtalk(url: str, payload: dict, timeout: int = 5) -> dict: resp = requests.post(url, json=payload, timeout=timeout) result = resp.json() if result.get("errcode") != 0: raise RuntimeError(f"push failed: {result}") return result这里有个新手很容易忽略的问题:只看 HTTP 状态码是不够的。接口可能返回 200,但业务层面的 errcode 不是 0。比如消息格式不对、关键词不匹配、签名错误,都会返回 200 加一个非零错误码。所以必须判断errcode,而不是只要没有抛异常就当发送成功。
3.2 text 和 markdown:什么时候用什么
text 是最简单的消息类型,适合纯文本告警和通知:
{ "msgtype": "text", "text": { "content": "线上服务心跳检测失败,请立刻排查。" }, "at": { "atMobiles": ["13800000000"], "isAtAll": false } }注意atMobiles里传的是手机号字符串列表。如果需要 @ 所有人,把isAtAll设为 true。但 @ 所有人不要滥用,尤其是夜间告警,频繁 @ 所有人很容易被群成员投诉。一般只有最高级别故障才用。
markdown 类型适合结构化信息,比如状态汇总、列表、链接:
{ "msgtype": "markdown", "markdown": { "title": "构建告警", "text": "#### 服务异常\n> 接口响应时间超过3秒\n\n[查看监控面板](https://example.com/monitor)" }, "at": { "atMobiles": [], "isAtAll": false } }这里有两个细节:
title字段在会话折叠区作为摘要展示,不要留空;- 钉钉的 markdown 是子集语法,不是完整标准。它支持标题、加粗、引用、列表和链接,但不支持复杂的 HTML 标签。想用 CSS 样式或者复杂排版,基本做不到。
如果在 markdown 里要 @ 某个具体成员,必须在 text 内容里显式写出@手机号,否则客户端不会显示艾特效果。这是 text 和 markdown 的一个明显差异,我踩过一次之后才记住。
3.3 ActionCard:让人在群里做决策
ActionCard 是我个人最喜欢的一种类型。它把一段文本和若干个按钮放在同一张卡片里,用户可以直接在群里点击按钮跳转到对应页面。非常适合发布确认、审批处理、故障认领这类场景。
下面是一个带两个按钮的示例:
{ "msgtype": "actionCard", "actionCard": { "title": "变更确认", "text": "检测到未完成的变更单,是否立即执行发布?", "btnOrientation": "0", "buttons": [ { "title": "确认发布", "action_url": "https://example.com/confirm?id=123" }, { "title": "取消", "action_url": "https://example.com/cancel?id=123" } ] } }btnOrientation控制按钮排布方向,"0" 表示竖向排列,"1" 表示横向排列。按钮太多时建议用竖向,不然手机上会挤成一团。如果只需要一个整体跳转链接,也可以不配 buttons,只配singleTitle和singleURL,卡片本身就能点。
ActionCard 的按钮跳转 URL 通常要能直达内部系统,这就要求你的业务系统有对应的免登录或者带 token 的链接。如果跳转过去还要再登录一次,用户很快就会嫌麻烦。
3.4 FeedCard:多条独立消息的聚合推送
FeedCard 适合一次推送多条具备独立跳转的条目,比如多条告警、多篇资讯、多个工单。它的结构是外层一个 msgtype,里面一个 links 数组,每条链接包含 title、messageURL、picURL。
{ "msgtype": "feedCard", "feedCard": { "links": [ { "title": "网关1 响应异常", "messageURL": "https://example.com/alert/1", "picURL": "https://example.com/pic/1.png" }, { "title": "网关2 响应异常", "messageURL": "https://example.com/alert/2", "picURL": "https://example.com/pic/2.png" } ] } }说实话,FeedCard 在监控告警场景里的实用性一般,因为用户需要在群里逐条点击,效率不如一条 markdown 汇总高。它更适合资讯类、推荐类场景。如果你在纠结消息格式,优先保证信息在会话流里一眼能看完,而不是让用户点很多次。
3.5 频率限制和错误重试的基本逻辑
推送接口不是无限量的。每个机器人每分钟能发的消息数有上限,官方文档会写明限流规则。超出后接口会返回频控错误,提示你稍后再试。
处理限流时最忌讳"无限重试"。我见过有脚本在发送失败后死循环重试,结果机器人的限制越来越严,本来还能发的消息彻底发不出去了。正确做法是:
- 对可重试错误(网络超时、频控)做有限次退避重试,比如最多 3 次;
- 对参数类错误(格式错误、关键词不匹配)直接记录日志,不要重试;
- 把失败消息写入本地队列,由另一个定时任务补推。
告警类场景还要防止"告警风暴":一次故障触发几十条细节告警,全部推出去,群里面瞬间刷屏。我会在推送程序里做聚合,把同一时间段、同一服务的告警合并成一条,等故障恢复后再发一条汇总。这样既不会漏消息,也不会打扰所有人。
4. 从"只会发"到"能对话":Stream 模式接入消息回调
只做推送,自定义机器人就够了。但如果你想做"群内 @ 机器人,机器人自动回复"的交互型应用,就必须进入消息回调环节。这一节重点讲 Stream 模式的接入逻辑和工程化细节。
4.1 为什么 Stream 模式让开发舒服很多
在没有 Stream 模式之前,企业内部应用机器人接收消息要靠 HTTP 回调。你需要准备好一个公网 HTTPS 地址,而且这个地址要稳定、证书要有效。每次改点代码都要重新部署到公网服务器上验证,非常影响调试效率。
Stream 模式的思路是"反向建连":程序启动时通过官方 SDK 主动连接钉钉服务端,建立一条长连接。钉钉收到新事件后,沿着这条连接把事件数据推过来。你的程序只需要保证能访问外网即可,不需要任何入站端口。
这样做有几个直接的好处:
- 本地开发环境可以直接调试,不用内网穿透工具;
- 部署环境不需要配置公网入口,减少了暴露面;
- 避免了 HTTPS 证书、反向代理这些额外组件的运维负担。
代价是程序需要保持一条常驻长连接,进程重启后要自动重连。这在生产环境里对应一个简单结论:要用进程守护工具把机器人进程管起来,或者直接跑在容器平台里并由平台负责重启。
4.2 一个最小可用的人机对话闭环
以 Stream 模式为例,程序骨架大概是这样的。伪代码展示了最重要的三个环节:建立连接、注册消息回调、返回回复。
import dingtalk_stream def on_message(msg): text = msg.get("text", {}).get("content", "").strip() sender_id = msg.get("senderStaffId", "") conversation_id = msg.get("conversationId", "") msg_id = msg.get("msgId", "") if not text: return # 去掉 @ 机器人的部分,只看真正的指令 command = extract_command(text) reply_text = handle_command(command, sender_id) reply_to_conversation(conversation_id, reply_text) def main(): client = dingtalk_stream.DingTalkStreamClient() client.register_callback("message", on_message) client.start_forever()这个回调结构在不同版本 SDK 里细节会有差异,但核心思路一致:拿到消息内容、发送者标识、会话标识,然后决定怎么回复。回复的方式有两种,一种是直接在回调返回内容,另一种是拿 conversationId 再调用一次发送接口。后者更灵活,因为你可以先做耗时的查询再回复,甚至分多次回复。
第一次调通这个闭环时,我发现一个常被人忽略的问题:消息内容里会包含机器人的名字,类似@机器人名 查订单 123。如果直接拿全文做关键词匹配,"查订单"前面那段很容易干扰解析。所以进入业务逻辑前,要把 @ 机器人的片段去掉,提取真正的指令部分。
4.3 消息去重和会话状态管理
线上运行一段时间后,你会发现同一个事件可能被推送多次。程序处理消息时如果没有幂等保护,用户发一句话,机器人可能回复两三遍。解决方法是按消息 ID 去重:用一个缓存存最近处理过的 msg_id,设置 10 分钟左右的过期时间。
seen_msg_ids = set() def is_duplicate(msg_id: str) -> bool: if msg_id in seen_msg_ids: return True seen_msg_ids.add(msg_id) return False生产环境不要用进程内 set,因为进程重启后去重记录就丢了。用 Redis 或者带持久化的存储更稳。
会话状态管理是交互型机器人另一个核心点。机器人至少要能区分当前对话来自哪个群。以conversationId作为 Key,把每个会话的上下文存在 Redis 里,比如当前用户身份、临时输入数据、上次查询条件等。不然用户在多个群里分别 @ 机器人,消息会互相串。
权限设计也要从第一天就想清楚。不是所有人都有资格让机器人执行查询命令。我会维护一个操作员手机号白名单,消息进来先判断senderStaffId是否在名单里,不在就回复"无权限",并且只把这类拒绝事件写日志,不暴露内部信息。
5. 落地部署中的高频坑位与排查思路
这部分是我最想写的内容。代码能跑通是一回事,部署到真实环境稳定运行是另一回事。下面的问题我全部遇到过,按出现频率排序。
5.1 消息发出去了但群里没人看到
表现:程序返回正常,errcode 为 0,群里却始终看不到消息。这种问题最迷惑人,因为从接口层面看是成功的。
排查思路按顺序来:
- 先确认收到了返回且 errcode 为 0,如果 errcode 非 0,直接看 errmsg;
- 检查 Webhook 对应的群是不是你当前看的那个群,把 URL 里的 access_token 和群里机器人管理页展示的地址逐字符比对;
- 确认机器人没有被移出群。群里如果有人把机器人移除,旧的 Webhook 不会马上失效,但消息发过去可能没有明确报错,具体表现因版本而异;
- 检查群内是否开启了全员禁言。机器人消息也会被禁言机制挡住;
- 如果用的是 markdown 类型,临时换成最简单的 text 试一次,排除格式问题。
一个很实用的定位技巧:找一台能访问外网的机器,用 curl 手动发一条测试消息。这样可以直接排除代码逻辑、网络代理、证书等环节。
curl 'https://oapi.dingtalk.com/robot/send?access_token=你的token' \ -H 'Content-Type: application/json' \ -d '{"msgtype":"text","text":{"content":"test"}}'如果手动 curl 能收到,问题基本确定在程序逻辑;如果手动 curl 也收不到,问题就在配置或群里。
5.2 加签验证失败:时间戳和 URL 编码
加签失败是最典型的配置型坑。接口返回的提示往往是签名不匹配,但根因通常在三处:
- 服务器系统时间和真实时间偏差过大。钉钉服务端会校验时间窗口,时间偏太多,合法签名也会被判为非法。用 NTP 校准服务器时间是必须的;
- 生成 timestamp 时用了秒而不是毫秒。秒级时间戳和签名串不匹配;
- Base64 之后没做 URL 编码,或者编码方式不对。
+在 URL 里会被当成空格,/会影响路径解析,=会被当成参数分隔符。
建议把生成签名这一步单独抽出来写单元测试,输入固定 secret 和时间戳,断言输出固定签名。这样后面排查问题时,可以直接排除签名算法的嫌疑。
5.3 被限流之后:分清可重试和不可重试
限流错误在日志里非常明显。短时间大量推送时,前面几个请求成功,后面陆续开始返回频控错误。这时候最忌讳的就是盲目加 sleep 重试。
我曾经处理过一个案例:某个报表任务每天上午集中推送,一次推几百条,刚开始正常,过几分钟后所有消息全部失败。后来我把推送改成异步队列,同时在业务层做聚合,把多条内容合并成一条 markdown,推送次数直接降了一个数量级,问题才彻底消失。
通用做法是给每个机器人维护一个简单令牌桶,发送前先检查是否还有配额。超出配额的消息进入本地队列,而不是立刻丢弃。这样既不会触顶限流,也不会丢告警。
5.4 密钥泄漏和机器人重置
最后一个坑不是技术问题,而是安全习惯。Webhook、AppSecret、access_token 任何一个落到公网仓库或者聊天记录里,都意味着别人可以替你的机器人发消息。
我见过因为把 access_token 打进日志,导致线上机器人被刷屏的案例。处理办法是:
- secret、token 全部从环境变量或配置中心读取,不写死在代码里;
- 日志输出时做脱敏,只保留最后几位;
- 一旦怀疑泄漏,立刻到群里删除机器人重新创建,或者在企业后台重置密钥;
- 公网仓库的提交历史也要清理,否则只改当前代码不解决问题,旧的提交记录里仍有密钥。
这些习惯越早建立越好,等真正出问题再补救,损失往往已经造成。
6. 几个值得抄作业的落地场景
最后分享三个我实际搭建过的场景。它们不是玩具 Demo,而是可以直接映射到日常工作中的模式。
6.1 用机器人把监控告警和值班表串起来
这是自定义机器人最经典的应用。监控系统探测到服务异常后,把事件发到消息队列,推送程序从队列消费事件,然后根据级别决定推送策略:
- P0 级别:立即推送,@ 当前值班人,同时发一张 ActionCard,卡片按钮是"我要处理";
- P1 级别:推送 markdown 汇总,不 @ 所有人;
- P2 级别:聚合后每小时推送一次,不单独打扰。
值班表单独维护一个映射表,把日期、时段和值班人手机号对应起来。推送程序读取这张表,自动决定该 @ 谁。这里的关键是不要写死手机号,否则轮值换班时你改代码都改不过来。
6.2 群内 @ 机器人执行查询命令
这个场景需要企业内部应用机器人加 Stream 模式。我在一个技术群里部署了查询机器人,支持几个固定命令,比如查当前服务状态、查最近一次发布记录、查线上配置项。
消息进来后,先判断发送者是否在白名单里,再解析命令,查询内部系统,最后把结果以 markdown 返回。看似简单,但有几个细节值得注意:
- 命令解析要支持"精确匹配"和"带参数"两种形态;
- 查询操作要加超时,避免内部接口响应慢导致机器人长时间不回复;
- 敏感信息不能直接回在群里,只返回摘要,具体详情跳转内部页面。
这个机器人上线后,群里问"现在服务什么状态"的次数明显少了,因为答案可以直接从聊天记录里翻。它最大的价值不是省了多少时间,而是把常见问题的回答标准化了。
6.3 定时报表和审批提醒的组合用法
定时任务是另一个很适合机器人的场景。每天固定时间,程序从数据库汇总前一天的数据,生成一张 markdown 表格,包括完成项、异常项、明日计划,推送管理群。对于需要审批的事项,额外推送一张 ActionCard,按钮直达审批页面。
这里我会特别注意消息模板的统一。把所有模板放在一个配置文件夹里,统一维护,而不是散落在各个脚本中。因为业务人员会频繁改文案、改字段,每次都要到代码里找字符串,既累又容易改坏。
如果产品和运营希望机器人更"智能",还可以把历史报表数据存起来,向机器人提问"上周告警总数"时,它就能从存储中统计并回复。这已经进入了对话式数据分析的范畴,但基础仍然是先跑通消息闭环。每一步都走扎实,后面才有扩展的空间。