Dify对接钉钉开发者后台添加的机器人

你说的这个场景和前面假设的**“钉钉群自定义机器人”**不是一类。

你现在使用的是：

钉钉开发者后台创建的机器人（企业内部机器人 / 钉钉应用机器人）

这种机器人没有固定 Webhook 地址，消息链路也不同。

正确架构应该是：

钉钉用户
   |
   | @机器人 / 私聊 / 群聊消息
   ↓
钉钉开放平台机器人服务
   |
   | Stream模式（长连接）
   ↓
你的 FastAPI 服务
   |
   | Dify Streaming API
   ↓
Dify
   |
   | SSE流
   ↓
FastAPI
   |
   | Stream回复
   ↓
钉钉机器人

也就是说：

• ❌ 不使用 oapi.dingtalk.com/robot/send

• ❌ 不使用 Webhook 回调发送

• ✅ 使用钉钉 Stream模式

• ✅ 你的服务主动连接钉钉长连接网关

• ✅ 收到事件后调用 Dify

• ✅ 调用钉钉 SDK 回复

一、需要准备的信息

在钉钉开发者后台创建机器人应用：

进入：

钉钉开放平台

获取：

Client ID
Client Secret
Robot Code

通常类似：

client_id:
dingxxxxxxxx

client_secret:
xxxxxxxxxxxx

robotCode:
dingxxxxxxxx

二、安装钉钉 Stream SDK

官方推荐 Python：

pip install dingtalk-stream
pip install fastapi
pip install requests
pip install python-dotenv

三、项目结构

dify-dingtalk-stream/

├── main.py
├── dify.py
├── dingtalk_bot.py
├── config.py
└── .env

1. 配置

.env

DING_CLIENT_ID=dingxxxxxxxx
DING_CLIENT_SECRET=xxxxxxxx

DIFY_API_KEY=app-xxxxxxxx
DIFY_URL=https://api.dify.ai/v1/chat-messages

2. Dify Streaming调用

dify.py

import requests
import json

from config import (
    DIFY_API_KEY,
    DIFY_URL
)


def ask_dify(question,user):

    headers={
        "Authorization":
        f"Bearer {DIFY_API_KEY}",

        "Content-Type":
        "application/json"
    }


    body={

        "query":
        question,

        "inputs":{},

        "response_mode":
        "streaming",

        "user":
        user
    }


    r=requests.post(
        DIFY_URL,
        headers=headers,
        json=body,
        stream=True
    )


    answer=""


    for line in r.iter_lines():

        if line:

            line=line.decode()


            if line.startswith("data:"):

                data=json.loads(
                    line[5:]
                )


                if data.get("event")=="message":

                    answer += data.get(
                        "answer",
                        ""
                    )


    return answer

3. 钉钉 Stream机器人

dingtalk_bot.py

核心代码：

import dingtalk_stream

from dify import ask_dify



class RobotHandler(
    dingtalk_stream.ChatbotHandler
):


    async def process(
        self,
        message
    ):


        # 用户消息

        text = (
            message.content
            .get("text","")
        )


        sender = (
            message.sender_staff_id
        )


        # 调用Dify

        answer = ask_dify(
            text,
            sender
        )


        # 回复钉钉

        self.reply_text(
            answer,
            message
        )



def start_bot():

    credential = (
        dingtalk_stream.Credential(
            client_id=
            "你的client_id",

            client_secret=
            "你的secret"
        )
    )


    client = (
        dingtalk_stream.DingTalkStreamClient(
            credential
        )
    )


    handler=RobotHandler()


    client.register_callback_handler(
        handler
    )


    client.start()

4. 启动入口

main.py

from dingtalk_bot import start_bot


if __name__=="__main__":

    start_bot()

运行：

python main.py

启动后：

connected to DingTalk Stream server

表示机器人在线。

五、Stream模式和Webhook模式区别

六、如果要真正实现“流式回复”

你的场景反而更适合。

可以做：

Dify SSE:

chunk1:
正在

chunk2:
查询

chunk3:
知识库


↓

钉钉 Stream:

机器人：
正在查询知识库...

机器人：
正在生成答案...

机器人：
最终结果...

因为开发者机器人支持主动发送消息，不受 Webhook 限制。

七、生产建议架构

建议改成：

                 钉钉Stream
                     |
                     |
              FastAPI + SDK
                     |
        ┌────────────┴────────────┐
        |
     Redis
   会话管理
        |
     Dify Workflow
        |
   企业知识库/RAG

你前面的需求（Dify + 钉钉开发者机器人 + Stream模式）应该采用这一套，而不是 Webhook 机器人方案。

如果你需要，我可以继续给你补一版真正可运行的钉钉 Stream SDK 最新版代码（包含 Card Stream 流式卡片回复，实现类似 ChatGPT 打字效果）。