做后端开发和运维的人大概率都有过这样的经历监控系统报警了邮件发出去没人及时看值班群里的消息又被聊天记录淹没等真正有人处理的时候已经过去了半小时。我自己是在一次线上变更踩坑之后才认真把钉钉机器人开发这件事从头到尾折腾了一遍。它看起来只是往群里推一条消息的小功能实际牵扯到机器人选型、安全签名、消息类型选择、回调接入和限流处理每个环节都有让人意外的地方。这篇文章不是官方文档的复述而是我从自定义机器人、企业内部应用机器人一直到 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), digestmodhashlib.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}timestamp{timestamp}sign{sign} resp requests.post(url, jsonpayload, timeout5) 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}, timeout5, ) data resp.json() if data.get(errcode) ! 0: raise RuntimeError(fgettoken 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, jsonpayload, timeouttimeout) result resp.json() if result.get(errcode) ! 0: raise RuntimeError(fpush 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?id123 }, { title: 取消, action_url: https://example.com/cancel?id123 } ] } }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按钮直达审批页面。这里我会特别注意消息模板的统一。把所有模板放在一个配置文件夹里统一维护而不是散落在各个脚本中。因为业务人员会频繁改文案、改字段每次都要到代码里找字符串既累又容易改坏。如果产品和运营希望机器人更智能还可以把历史报表数据存起来向机器人提问上周告警总数时它就能从存储中统计并回复。这已经进入了对话式数据分析的范畴但基础仍然是先跑通消息闭环。每一步都走扎实后面才有扩展的空间。 
长河君 · 2026/10/17 2:06:24 · 阅读 1.2k 钉钉机器人开发实践：从Webhook到Stream模式的全链路指南
做后端开发和运维的人大概率都有过这样的经历监控系统报警了邮件发出去没人及时看值班群里的消息又被聊天记录淹没等真正有人处理的时候已经过去了半小时。我自己是在一次线上变更踩坑之后才认真把钉钉机器人开发这件事从头到尾折腾了一遍。它看起来只是往群里推一条消息的小功能实际牵扯到机器人选型、安全签名、消息类型选择、回调接入和限流处理每个环节都有让人意外的地方。这篇文章不是官方文档的复述而是我从自定义机器人、企业内部应用机器人一直到 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), digestmodhashlib.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}timestamp{timestamp}sign{sign} resp requests.post(url, jsonpayload, timeout5) 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}, timeout5, ) data resp.json() if data.get(errcode) ! 0: raise RuntimeError(fgettoken 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, jsonpayload, timeouttimeout) result resp.json() if result.get(errcode) ! 0: raise RuntimeError(fpush 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?id123 }, { title: 取消, action_url: https://example.com/cancel?id123 } ] } }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按钮直达审批页面。这里我会特别注意消息模板的统一。把所有模板放在一个配置文件夹里统一维护而不是散落在各个脚本中。因为业务人员会频繁改文案、改字段每次都要到代码里找字符串既累又容易改坏。如果产品和运营希望机器人更智能还可以把历史报表数据存起来向机器人提问上周告警总数时它就能从存储中统计并回复。这已经进入了对话式数据分析的范畴但基础仍然是先跑通消息闭环。每一步都走扎实后面才有扩展的空间。