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模式区别 

 

 

 

 项目 

 群自定义机器人 

 开发者机器人 

 

 

 

 

 创建位置 

 群设置 

 开放平台 

 

 

 Webhook 

 有 

 无 

 

 

 长连接 

 无 

 有 

 

 

 Stream模式 

 ❌ 

 ✅ 

 

 

 SDK 

 不需要 

 需要 

 

 

 企业应用 

 弱 

 强 

 

 

 权限控制 

 少 

 完整 

 

 

 私聊 

 ❌ 

 ✅ 

 

 

 群@机器人 

 ✅ 

 ✅ 

 

 

 

 

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

 你的场景反而更适合。 

 可以做： 

 Dify SSE:

chunk1:

正在

chunk2:

查询

chunk3:

知识库

↓

钉钉 Stream:

机器人：

正在查询知识库...

机器人：

正在生成答案...

机器人：

最终结果...

 

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

 

 七、生产建议架构 

 建议改成： 

 钉钉Stream

 |

 |

 FastAPI + SDK

 |

 ┌────────────┴────────────┐

 |

 Redis

 会话管理

 |

 Dify Workflow

 |

 企业知识库/RAG

 

 

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

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