Dify对接钉钉开发者后台添加的机器人
你说的这个场景和前面假设的**“钉钉群自定义机器人”**不是一类。前面给你的版本偏“概念代码”，这里改成工程化 services 分层结构。
你现在使用的是：
钉钉开发者后台创建的机器人（企业内部机器人 / 钉钉应用机器人）
这种机器人没有固定 Webhook 地址，消息链路也不同。
正确架构应该是：
钉钉用户
|
| @机器人 / 私聊 / 群聊消息
↓
钉钉开放平台机器人服务
|
| Stream模式（长连接）
↓
你的 FastAPI 服务
|
| Dify Streaming API
↓
Dify
|
| SSE流
↓
FastAPI
|
| Stream回复
↓
钉钉机器人
也就是说：目标：
-
❌钉钉不使用开发者后台机器人
Stream 模式
Python oapi.dingtalk.com/robot/senddingtalk-stream
❌FastAPI 不使用 Webhook 回调发送项目结构
✅Dify 使用钉钉Chat Stream模式API Streaming
✅Service 你的服务主动连接钉钉长连接网关层隔离
✅后续可扩展 收到事件后调用 Dify
✅ 调用钉钉 SDK 回复Redis、知识库、权限
dingtalk-stream
一、需要准备的信息
SDK 在钉钉开发者后台创建机器人应用：本身就是针对钉钉 Stream 模式的机器人收消息、事件回调设计的。当前 Python 包版本可直接通过 pip 安装。(PyPI)
进入：
钉钉开放平台
获取：
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/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. 配置二、依赖
.envrequirements.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.endpoint is wss://xxxx
INFO websocket connected
钉钉开发者后台机器人上线后：
群里：
@机器人 公司的年假政策是什么？
流程：
ChatbotMessage
|
↓
RobotHandler
|
↓
ChatService
|
↓
DifyService
|
↓
Dify SSE
|
↓
reply_text()
|
↓
钉钉群
十一、下一步建议改造（生产版）
你这个架构建议继续升级：
1. 增加 Redis 会话
现在：
用户
|
Dify
|
新会话
升级：
staff_id
|
Redis
|
conversation_id
|
Dify上下文
2. 增加真正 Stream 卡片回复
当前：
Dify:
chunk1
chunk2
chunk3
钉钉:
一次回复
升级：
Dify chunk
|
|
钉钉AI卡片stream
|
|
逐字刷新
可以做到类似 ChatGPT 的效果。
3. 推荐生产目录
大型项目：
services
├── dingtalk
│ ├── stream_client.py
│ ├── message.py
│ └── card.py
│
├── dify
│ ├── client.py
│ ├── stream.py
│ └── workflow.py
│
├── memory
│ └── redis.py
│
└── security
└── permission.py
启动后：
connected to DingTalk Stream server
表示机器人在线。
五、Stream模式和Webhook模式区别
六、如果要真正实现“流式回复”
你的场景反而更适合。
可以做：
Dify SSE:
chunk1:
正在
chunk2:
查询
chunk3:
知识库
↓
钉钉 Stream:AI 机器人：
正在查询知识库...
机器人：
正在生成答案...
机器人：
最终结果...
因为开发者机器人支持主动发送消息，不受 Webhook 限制。
七、生产建议架构
建议改成：
钉钉Stream
|
|
FastAPI助手 + SDK
|
┌────────────┴────────────┐
|
Redis
会话管理
|
Dify Workflow
|
企业知识库/RAG 你前面的需求（Dify + 钉钉开发者机器人 + Stream模式）应该采用这一套，而不是 Webhook 机器人方案。
如果你需要，我可以继续给你补一版真正可运行的钉钉 Stream SDK 最新版代码（包含 Card Stream 流式卡片回复，实现类似 ChatGPT 打字效果）知识库平台。