钉钉 AI 助手怎么接入？三种方案对比（附 Stream 模式代码）

钉钉接 AI 有三条路，选错不是"多写点代码"，而是上线后收不到消息、被限流、或者数据边界说不清

2026-09-21 · Leo · 接入教程 · 10 分钟阅读
TL;DR — 钉钉接 AI 助手有三条路：方案 A：企业内部应用机器人 + Stream 长连接（不用公网回调地址，收发都可控，正式场景首选）；方案 B：自定义群机器人 Webhook（10 分钟可用，但只能发不能收，做不了对话）；方案 C：第三方 SaaS 托管（最快上线，但对话内容要过第三方平台）。本文给出三种方案的完整步骤和代码，以及 Stream 断连、机器人收不到消息、群 Webhook 被限流、卡片回调失败等 5 个高频坑的修法和上线前检查清单。
⚠️ 本文是通用接入教程，接口字段以钉钉官方文档为准；代码为示例配置，上线前请按自己公司的安全与合规要求调整。不涉及任何具体客户案例。

先看结论：三种方案一张表

维度方案 A：企业内部机器人 + Stream方案 B：自定义群机器人 Webhook方案 C：第三方 SaaS 托管
收消息✅ 单聊/群聊都可收可发❌ 只能往群里发✅ 平台代收代发
要不要公网地址不用：服务主动连钉钉长连接不用（只出不进）不用
数据边界消息与回复都在自己服务器推送内容经钉钉服务器对话内容经第三方平台
上线耗时1–3 天（主要是长连接与卡片调试）约 10 分钟半天到 1 天（配置为主）
适合场景正式 AI 助手、客服答疑、流程查询告警播报、日报推送、CI/CD 通知快速验证价值、无开发人力

一句话：做对话助手选 A，做通知播报选 B，先验证价值选 C。最常见的错误是拿方案 B 做对话（收不到群里的消息，做不了），或者没评估数据边界就直接上方案 C。

方案 A：企业内部应用机器人 + Stream 长连接（推荐）

这是钉钉官方推荐的接入方式：你的服务用 Stream 模式主动和钉钉建立 WebSocket 长连接，员工在机器人会话里发消息 → 钉钉通过这条连接把消息推给你的服务 → 服务调 AI → 再通过连接把回复发回去。因为没有入站 HTTP 回调，不需要公网 IP、不需要备案域名、不需要在防火墙上开端口，内网机器也能跑，这是它和"HTTP 回调"路线最大的区别。

第 1 步：创建企业内部应用与机器人（约 10 分钟）

钉钉开发者后台 →「应用开发」→「企业内部应用」→ 创建应用，然后在应用里「添加机器人」，消息接收模式选 Stream 模式。记下三个值：

发布前先把可见范围设成一个小部门做灰度，验证通过再放开。

第 2 步：用 Stream 模式收消息（约 30 分钟，含调试）

Stream 模式的核心是：服务启动时用凭证换一个临时连接地址，然后连上去、订阅机器人消息主题。连接断了要重连——这是后面坑 1 的根源。

# dingtalk_stream_bot.py —— 建立长连接并订阅机器人消息
from dingtalk_stream import DingTalkStreamClient, Credential
from dingtalk_stream.chatbot import ChatbotHandler, ChatbotMessage

class AIHandler(ChatbotHandler):
    async def process(self, callback):
        msg = ChatbotMessage.from_dict(callback.data)
        # 先回执，再异步处理，别在回调里等 AI（见坑 3）
        self.reply_text("收到，正在处理…", msg)
        question = msg.text.content.strip()
        answer = ask_ai(question)          # 你自己的 AI 服务 / 知识库
        self.reply_text(answer, msg)
        return 200, "OK"

credential = Credential(CLIENT_ID, CLIENT_SECRET)      # 从环境变量读，别写死
client = DingTalkStreamClient(credential)
client.register_callback_handler("/v1.0/im/bot/messages/get", AIHandler())
client.start_forever()                 # 内部含自动重连，但要有进程级守护

调试期最容易踩的坑：话题（topic）写错——机器人消息的 topic 是固定的，写错会表现为"连上了但永远收不到消息"，日志里却没有任何报错。

第 3 步：发消息与卡片（主动推送 + 会话回复）

回复会话用机器人消息接口；如果要做"审批进度""工单状态"这类可点击的交互，用互动卡片。两个接口都走同一个 access_token，必须集中缓存（有效期 7200 秒，有获取频率限制）。

# dingtalk_client.py —— token 集中缓存 + 发送机器人消息
import time, requests

_token, _ts = "", 0

def get_token():
    global _token, _ts
    if _token and time.time() - _ts < 7000:          # 留 200 秒余量
        return _token
    r = requests.post("https://api.dingtalk.com/v1.0/oauth2/accessToken",
                      json={"appKey": CLIENT_ID, "appSecret": CLIENT_SECRET},
                      timeout=5).json()
    _token, _ts = r["accessToken"], time.time()
    return _token

def reply_text(user_id, text):
    requests.post(
        "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend",
        headers={"x-acs-dingtalk-access-token": get_token()},
        json={"robotCode": ROBOT_CODE, "userIds": [user_id],
              "msgKey": "sampleText", "msgParam": '{"content": "%s"}' % text[:1800]},
        timeout=5)

注意最后一行：钉钉的文本消息内容是包在 msgParam 的 JSON 字符串里的，超长要先截断或分段，否则接口直接报错（见坑 5）。

第 4 步：接上你的 AI 与知识库

把 ask_ai() 换成你的 RAG 知识库或大模型服务即可。三个工程要点：

方案 B：自定义群机器人 Webhook（10 分钟，但只能发）

群设置 →「智能群助手」→「添加机器人」→「自定义」，拿到一个 Webhook 地址，之后往这个地址 POST JSON 就能往群里发消息（可选加签校验）：

curl 'https://oapi.dingtalk.com/robot/send?access_token=你的TOKEN' \
  -H 'Content-Type: application/json' \
  -d '{"msgtype":"text","text":{"content":"部署完成：订单服务 v2.3 已上线"}}'

它的定位是单向通知：监控告警、CI/CD 结果、每日播报。三个硬限制：收不到群里的消息（做不了对话）、每个机器人每分钟发送条数有限（高频告警必须做聚合）、@ 人要填手机号或 userId（不是昵称）。支持 text、markdown、link、actionCard 等类型，适合"把系统里的事件告诉人"，不适合"人问系统答"。

方案 C：第三方 SaaS 托管（最快，但先想清楚数据边界）

市面上有不少托管平台：授权钉钉后网页配置知识库和话术，不用自己写长连接代码，半天就能上线。适合两种情况：没有开发人力、想先小成本验证 AI 助手有没有价值。

但要在签约前问清四件事，不然日后很难受：

  1. 数据存哪：员工提问和上传的文档是否存在对方平台？存在哪个区域？
  2. 能不能导出：哪天换方案，知识库和会话记录能不能完整带走？
  3. 支不支持私有化：验证成功后能否迁到方案 A 的自建形态？
  4. 计费方式：按坐席还是按调用量？底层大模型的调用费是谁出？

横向对比：钉钉 / 飞书 / 企业微信，接入难度差在哪

维度钉钉飞书企业微信
主流接入方式企业内部机器人 + Stream 长连接事件订阅（长连接或 HTTP 回调）自建应用 + HTTP 回调
要不要公网地址不用（长连接主动外连）二选一，长连接可免要：必须 HTTPS 回调
回调时效要求长连接无 5 秒硬限HTTP 回调同样要求快速返回5 秒内必须响应，否则重试
消息去重按消息 ID 去重按事件 ID 去重按 MsgId 去重
内网部署友好度高：不需要入站端口中低：要暴露回调地址

结论：如果公司内网出网受限、又不想为回调单独申请公网域名，钉钉的 Stream 模式是三家里面最好落地的；企业微信的 5 秒回调硬限最需要工程兜底；飞书介于两者之间。多平台一起接的时候，把"收消息 → 调 AI → 回复"抽成一层统一网关，各平台的差异只留在适配层，后面加平台才不会推倒重来。

五个高频坑：现象与修法

坑 1：Stream 长连接半夜断掉，白天没人发现

现象：早上员工问机器人没反应，重启服务就好了。

根因：长连接被中间网络设备掐断，客户端没有重连成功，或者进程因为未捕获异常退出了。

✅ 修法：客户端开启自动重连 + 用 systemd/容器把进程守护起来（异常退出自动拉起）；加一条"存活探针"——定时发一条心跳消息给自己，超过 N 分钟没收到就告警。
坑 2：连接建立成功，但永远收不到消息

现象：日志显示已连接，员工发消息却毫无反应，也没有报错。

根因：订阅的 topic 写错；或者机器人没有发布、可见范围没包含提问人；或者注册的回调路径与消息类型不匹配。

✅ 修法：核对 topic 常量（机器人消息有固定 topic）；确认应用已发布且可见范围覆盖测试账号；把收到的原始 callback 数据整体打日志，先确认"有没有到"，再查"为什么没回"。
坑 3：AI 处理慢，员工收到重复回答或空回复

现象：问题复杂时，同一句话被回答两三次，或者只回了个"收到"。

根因：在回调里同步等 AI，超时后平台重推；或者先回了"收到"却没有后续消息。

✅ 修法：回调里只做去重 + 入队，立刻返回；AI 处理放到异步任务，算完主动推送；按消息 ID 去重（Redis SETNX 带过期时间），多实例部署共享去重状态。
坑 4：群机器人被限流，告警反而没发出去

现象：故障期间告警刷屏，最关键的几条反而没到群里。

根因：自定义群机器人有每分钟发送条数上限，超了直接丢弃或报错。

✅ 修法：告警先做聚合（相同告警合并、按 1 分钟窗口汇总），失败重试要有退避；关键告警走企业内部机器人单聊通道，别和群播报抢同一个出口。
坑 5：长回复发不出去或被截断

现象：AI 回答长一点，员工端收不到或只有前半截，接口报参数错误。

根因：文本内容包在 msgParam 的 JSON 字符串里，超长或引号未转义都会失败。

✅ 修法：用 json.dumps 生成 msgParam 而不是手拼字符串；超长内容按段落切分成多条，或只回摘要 + 链接到知识库页面。

上线前检查清单（8 条）

  1. 长连接是否有自动重连 + 进程守护（异常退出能自动拉起）？
  2. 是否有存活探针：定时心跳，超过阈值没收到就告警？
  3. 消息去重是否落地（Redis/内存集合），多实例是否共享去重状态？
  4. access_token 是否集中缓存、加锁刷新、留过期余量？
  5. AI 调用是否有超时兜底（先回执，超时给提示，而不是让用户干等）？
  6. 长回复是否分段或摘要化，msgParam 是否用序列化生成而非手拼？
  7. 发送前是否有敏感词/权限过滤，谁能用这个机器人是否按部门收敛？
  8. 日志是否脱敏，Client Secret 是否全部走环境变量而非代码库？

写在最后

三种方案没有绝对优劣，选择取决于三件事：数据边界要求多高、有多少开发人力、要解决什么场景。通知播报别上方案 A 浪费人力；正式对话助手别用方案 B（收不到消息）；上方案 C 之前把数据边界四问问清楚。多数团队的真实路径是：先用 C 或 B 快速验证，验证成功后迁到方案 A 做长期自营。

如果只需要一个结论：做对话选 A，做通知选 B，做验证选 C——别混用。

💡 延伸阅读：其他 IM 平台怎么接？看《企业微信 AI 机器人怎么搭建？三种方案对比》和《飞书 AI 机器人 30 分钟快速接入》；多平台统一接入的整体思路见《企业 IM 里直接调 AI Agent：4 步搞定飞书/钉钉/企微接入》。

有疑问？邮箱咨询：278946228@qq.com

← 返回博客列表

延伸阅读