Dify对接钉钉开发者后台添加的机器人
你说的这个场景和前面假设给你的**版本偏“钉钉群自定义机器人概念代码”**不是一类，这里改成工程化 services 分层结构。
你现在使用的是：
钉钉开发者后台创建的机器人（企业内部机器人 / 钉钉应用机器人）
这种机器人没有固定 Webhook 地址，消息链路也不同。
正确架构应该是：
钉钉用户
|
| @机器人 / 私聊 / 群聊消息
↓
钉钉开放平台机器人服务
|
| Stream模式（长连接）
↓
你的 FastAPI 服务
|
| Dify Streaming API
↓
Dify
|
| SSE流
↓
FastAPI
|
| Stream回复
↓
钉钉机器人
也就是说目标：
-
❌钉钉不使用oapi.dingtalk.com/robot/send -
❌Stream不使用 Webhook 回调发送 ✅ 使用钉钉Stream模模式-
✅Python你的服务主动连接钉钉长连接网关dingtalk-stream -
✅FastAPI收到事件后调用 Dify项目结构 -
✅Dify调用钉钉ChatSDKAPI回复Streaming -
Service 层隔离
-
后续可扩展 Redis、知识库、权限
dingtalk-stream
一、需要准备的信息
SDK 在钉钉开发者后台创建机器人应用：
进入：
钉钉开放平台
获取：
Client ID
Client Secret
Robot Code
通常类似：
client_id:
dingxxxxxxxx
client_secret:
xxxxxxxxxxxx
robotCode:
dingxxxxxxxx
二、安装本身就是针对钉钉 Stream SDK
模式的机器人收消息、事件回调设计的。当前 官方推荐Python Python：
pip install dingtalk-stream包版本可直接通过 pip install fastapi
pip install requests
pip install python-dotenv
三一、项目结构
dify-dingtalk-stream/bot/
├── app/
│
│ ├── main.py
├──│ dify.py│
│ ├── dingtalk_bot.config/
│ │ └── settings.py
│ │
│ ├── config.services/
│ │ │
│ │ ├── dify_service.py
│ │ │
│ │ ├── dingtalk_service.py
│ │ │
│ │ └── chat_service.py
│ │
│ ├── handlers/
│ │ └── robot_handler.py
│ │
│ └── utils/
│ └── logger.py
│
├── requirements.txt
└── .env
1.二、依赖
requirements.txt
fastapi
uvicorn
python-dotenv
requests
dingtalk-stream
安装：
pip install -r requirements.txt
三、配置
.env
# 钉钉Stream
DING_CLIENT_ID=dingxxxxxxxxdingxxxx
DING_CLIENT_SECRET=xxxxxxxxxxxx
# Dify
DIFY_API_KEY=app-xxxxxxxxxxxx
DIFY_URL=DIFY_API_URL=https://api.dify.ai/v1/chat-messages
# 服务
APP_NAME=dify-dingtalk-bot
2.四、配置服务
app/config/settings.py
import os
from dotenv import load_dotenv
load_dotenv()
class Settings:
DING_CLIENT_ID = os.getenv(
"DING_CLIENT_ID"
)
DING_CLIENT_SECRET = os.getenv(
"DING_CLIENT_SECRET"
)
DIFY_API_KEY=os.getenv(
"DIFY_API_KEY"
)
DIFY_API_URL=os.getenv(
"DIFY_API_URL"
)
settings=Settings()
五、Dify Streaming调用Service
dify.负责：
-
调用 Dify
-
消费 SSE
-
返回完整答案
app/services/dify_service.py
import requests
import json
from configapp.config.settings import settings
class DifyService:
def __init__(self):
self.url = (
DIFY_API_KEY,settings.DIFY_API_URL
DIFY_URL)
self.key = (
settings.DIFY_API_KEY
)
def ask_dify(question,user)chat(
self,
question:str,
user:str
):
headers={
"Authorization":
f"Bearer {DIFY_API_KEY}self.key}",
"Content-Type":
"application/json"
}
body=payload={
"query":
question,
"inputs":{},
"query":
question,
"response_mode":
"streaming",
"user":
user
}
r=response=requests.post(
DIFY_URL,self.url,
headers=headers,
json=body,payload,
stream=TrueTrue,
timeout=120
)
answer=""
for line in r.response.iter_lines():
if not line:
continue
line=line.decode(
"utf-8"
)
if not line.startswith(
"data:"
):
continue
data=json.loads(
line[5:]
)
event=data.get(
"event"
)
if data.get("event")=event=="message":
answer += data.get(
"answer",
""
)
if event=="message_end":
break
return answer
3.六、聊天业务 钉钉 Stream机器人Service
dingtalk_bot.py以后加：
-
核心代码：用户权限 -
Redis
-
会话ID
-
日志
都放这里。
app/services/chat_service.py
from app.services.dify_service import DifyService
class ChatService:
def __init__(self):
self.dify=DifyService()
def ask(
self,
message,
user
):
return self.dify.chat(
question=message,
user=user
)
七、钉钉发送 Service
app/services/dingtalk_service.py
import dingtalk_stream
fromclass difyDingTalkService:
def reply(
self,
handler,
text,
message
):
handler.reply_text(
text,
message
)
八、机器人 Handler
这是核心。
app/handlers/robot_handler.py
import logging
import ask_difydingtalk_stream
from dingtalk_stream import AckMessage
from app.services.chat_service import ChatService
from app.services.dingtalk_service import DingTalkService
class RobotHandler(
dingtalk_stream.ChatbotHandler
):
def __init__(self):
super().__init__()
self.chat_service=(
ChatService()
)
self.ding_service=(
DingTalkService()
)
async def process(
self,
messagecallback:
dingtalk_stream.CallbackMessage
):
# 用户消息
textincoming = (
message.contentdingtalk_stream
.get("text","")ChatbotMessage
).from_dict(
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"callback.data
)
)
clientquestion=(
=incoming
.text
.content
.strip()
)
user=(
incoming
.sender_staff_id
)
logging.info(
f"user={user}, msg={question}"
)
try:
answer=(
self.chat_service.ask(
question,
user
)
)
self.ding_service.reply(
self,
answer,
incoming
)
except Exception as e:
self.reply_text(
"服务异常："+str(e),
incoming
)
return (
dingtalk_stream.AckMessage.STATUS_OK,
"OK"
)
九、启动 Stream Client
app/main.py
import logging
import dingtalk_stream
from app.config.settings import settings
from app.handlers.robot_handler import RobotHandler
def start():
logging.basicConfig(
level=logging.INFO
)
credential=(
dingtalk_stream
.Credential(
settings.DING_CLIENT_ID,
settings.DING_CLIENT_SECRET
)
)
client=(
dingtalk_stream
.DingTalkStreamClient(
credential
)
)
handler=RobotHandler()
client.register_callback_handler(
handlerdingtalk_stream
.chatbot
.ChatbotMessage
.TOPIC,
RobotHandler()
)
client.start_forever()
if __name__=="__main__":
start()
4. 十、启动入口
main.py
frompython dingtalk_bot-m import start_bot
if __name__=="__main__":
start_bot()app.main
运行正常日志：
pythonINFO main.pyendpoint is wss://xxxx
INFO websocket connected
启动钉钉开发者后台机器人上线后：
群里：
connected@机器人 to DingTalk Stream server公司的年假政策是什么？
表示机器人在线。流程：
ChatbotMessage
|
↓
RobotHandler
|
↓
ChatService
|
↓
DifyService
|
↓
Dify SSE
|
↓
reply_text()
|
↓
钉钉群
五十一、Stream模式和Webhook模式区别
六、如果要真正实现“流式回复”议改造（生产版）
你的场景反而更适合。这个架构建议继续升级：
1. 增加 Redis 会话
可以做现在：
用户
|
Dify
|
新会话
升级：
staff_id
|
Redis
|
conversation_id
|
Dify上下文
2. 增加真正 Stream 卡片回复
当前：
Dify:
chunk1
chunk2
chunk3
钉钉:
一次回复
升级：
Dify SSE:chunk
chunk1:|
正在
chunk2:
查询
chunk3:
知识库
↓|
钉钉AI卡片stream
Stream:|
机器人：|
正在查询知识库...
机器人：
正在生成答案...
机器人：
最终结果...逐字刷新
因为开发者机器人支持主动发送消息，不受可以做到类似 WebhookChatGPT 限制的效果。
七、3. 推荐生产建议架构目录
建议改成大型项目：
services
钉钉Stream
|
|
FastAPI + SDK
|
┌├── dingtalk
│ ├── stream_client.py
│ ├── message.py
│ └── card.py
│
├── dify
│ ├──┴ client.py
│ ├── stream.py
│ └── workflow.py
│
├── memory
│ └── redis.py
│
└── security
└──┐ |
Redis
会话管理
|
Dify Workflow
|
企业知识库/RAGpermission.py
这个结构可以直接扩展成企业级
你前面的需求（Dify钉钉 AI 助手 + 钉钉开发者机器人Dify +RAG Stream模式）应该采用这一套，而不是 Webhook 机器人方案。
如果你需要，我可以继续给你补一版真正可运行的钉钉 Stream SDK 最新版代码（包含 Card Stream 流式卡片回复，实现类似 ChatGPT 打字效果）知识库平台。