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

前面给你的版本偏“概念代码”，这里改成**工程化 services 分层结构**。

目标：

- 钉钉 **开发者后台机器人**
- **Stream 模式**
- Python `dingtalk-stream`
- FastAPI 项目结构
- Dify Chat API Streaming
- Service 层隔离
- 后续可扩展 Redis、知识库、权限

`dingtalk-stream` SDK 本身就是针对钉钉 Stream 模式的机器人收消息、事件回调设计的。当前 Python 包版本可直接通过 pip 安装。([PyPI](https://pypi.org/project/dingtalk-stream/?utm_source=chatgpt.com "dingtalk-stream · PyPI"))

---

# 一、项目结构

```text
dify-dingtalk-bot/

├── app/
│
│   ├── main.py
│   │
│   ├── config/
│   │   └── settings.py
│   │
│   ├── services/
│   │   │
│   │   ├── dify_service.py
│   │   │
│   │   ├── dingtalk_service.py
│   │   │
│   │   └── chat_service.py
│   │
│   ├── handlers/
│   │   └── robot_handler.py
│   │
│   └── utils/
│       └── logger.py
│
├── requirements.txt
└── .env

```

---

# 二、依赖

requirements.txt

```txt
fastapi
uvicorn
python-dotenv
requests
dingtalk-stream

```

安装：

```bash
pip install -r requirements.txt

```

---

# 三、配置

## .env

```env
# 钉钉Stream
DING_CLIENT_ID=dingxxxx
DING_CLIENT_SECRET=xxxx


# Dify
DIFY_API_KEY=app-xxxx
DIFY_API_URL=https://api.dify.ai/v1/chat-messages


# 服务
APP_NAME=dify-dingtalk-bot

```

---

# 四、配置服务

## app/config/settings.py

```python
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 Service

负责：

- 调用 Dify
- 消费 SSE
- 返回完整答案

## app/services/dify\_service.py

```python
import requests
import json

from app.config.settings import settings



class DifyService:


    def __init__(self):

        self.url = (
            settings.DIFY_API_URL
        )

        self.key = (
            settings.DIFY_API_KEY
        )



    def chat(
        self,
        question:str,
        user:str
    ):


        headers={

            "Authorization":
            f"Bearer {self.key}",

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


        payload={

            "inputs":{},

            "query":
            question,


            "response_mode":
            "streaming",


            "user":
            user

        }



        response=requests.post(

            self.url,

            headers=headers,

            json=payload,

            stream=True,

            timeout=120

        )


        answer=""


        for line in 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 event=="message":

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


            if event=="message_end":

                break



        return answer

```

---

# 六、聊天业务 Service

以后加：

- 用户权限
- Redis
- 会话ID
- 日志

都放这里。

## app/services/chat\_service.py

```python
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

```python
import dingtalk_stream



class DingTalkService:



    def reply(
        self,
        handler,
        text,
        message
    ):


        handler.reply_text(

            text,

            message

        )

```

---

# 八、机器人 Handler

这是核心。

## app/handlers/robot\_handler.py

```python
import logging

import dingtalk_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,
        callback:
        dingtalk_stream.CallbackMessage
    ):


        incoming = (
            dingtalk_stream
            .ChatbotMessage
            .from_dict(
                callback.data
            )
        )


        question=(

            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 (
            AckMessage.STATUS_OK,
            "OK"
        )

```

---

# 九、启动 Stream Client

## app/main.py

```python
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

        )

    )



    client.register_callback_handler(

        dingtalk_stream
        .chatbot
        .ChatbotMessage
        .TOPIC,


        RobotHandler()

    )



    client.start_forever()



if __name__=="__main__":

    start()

```

---

# 十、启动

```bash
python -m app.main

```

正常日志：

```text
INFO 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. 推荐生产目录

大型项目：

```text
services

├── dingtalk
│    ├── stream_client.py
│    ├── message.py
│    └── card.py
│
├── dify
│    ├── client.py
│    ├── stream.py
│    └── workflow.py
│
├── memory
│    └── redis.py
│
└── security
     └── permission.py

```

这个结构可以直接扩展成企业级 **钉钉 AI 助手 + Dify RAG 知识库平台**。

现在信息完整了，可以确定有 **两个独立问题**：

1. **`.env` 没有生效（最关键）**
2. **websockets 版本和 dingtalk-stream 0.24.3 不兼容**

我们逐个修。

---

# 一、你的 `.env` 当前全部被注释了

你贴出来：

```env
# DING_CLIENT_ID=ding09cjpitrovdwlax0
# DING_CLIENT_SECRET=xxxxx

```

注意前面的：

```text
#

```

意味着 Python 根本读不到。

所以：

```python
settings.DING_CLIENT_ID

```

实际应该是：

```text
None

```

SDK 发起：

```
POST https://api.dingtalk.com/v1.0/gateway/connections/open

```

时，没有合法 credential，钉钉返回：

```json
{
 "code":"systemError"
}

```

---

## 修改 .env

去掉所有 `#`：

```bash
vim /opt/dify-dingtalk-bot/.env

```

改成：

```env
# 钉钉Stream
DING_CLIENT_ID=xxxxxxxxxxxxx

DING_CLIENT_SECRET=你的secret


# Dify
DIFY_API_KEY=app-xxxxxxxxxxxxxxxxxxxxx

DIFY_API_URL=http://192.168.4.20/v1/chat-messages


APP_NAME=dify-dingtalk-bot

```

注意：

你的 Dify 地址之前：

```env
DIFY_API_URL=http://192.168.4.20/v1

```

少了：

```
/chat-messages

```

应该：

```
http://192.168.4.20/v1/chat-messages

```

---

# 二、验证环境变量

不要启动机器人，先测试：

```bash
python

```

执行：

```python
from app.config.settings import settings

print("client:", settings.DING_CLIENT_ID)

print("secret:",
      bool(settings.DING_CLIENT_SECRET))

print("dify:",
      settings.DIFY_API_URL)

```

正确应该：

```text
client: xxxxxxxxxxxx

secret: True

dify: http://192.168.4.20/v1/chat-messages

```

如果还是：

```text
client: None

```

说明 dotenv 路径问题。

---

# 三、修复 websockets 兼容问题

你的错误：

```
AttributeError:
module 'websockets' has no attribute 'exceptions'

```

原因：

你现在：

```text
dingtalk-stream 0.24.3

```

依赖：

```text
websockets

```

但是装到了新版：

大概率：

```
websockets 15.x

```

新版删除了一些旧接口。

---

查看：

```bash
pip show websockets

```

应该类似：

```text
Version: 15.x

```

---

降级：

```bash
pip uninstall websockets -y

pip install websockets==10.4

```

然后确认：

```bash
pip show websockets

```

应该：

```
Version: 10.4

```

---

# 四、修改 requirements.txt

否则以后重新部署还会坏。

你的：

`app/requirements.txt`

改：

```txt
fastapi
uvicorn
python-dotenv
requests
dingtalk-stream==0.24.3
websockets==10.4

```

---

# 五、重新测试

执行：

```bash
cd /opt/dify-dingtalk-bot

source /opt/ding_dify/bin/activate

python -m app.main

```

正常应该看到：

类似：

```
INFO:dingtalk_stream.client:
open connection

INFO:dingtalk_stream.client:
connection established

INFO:dingtalk_stream.client:
websocket connected

```

而不是循环：

```
400 systemError

```

---

# 六、如果仍然 400

那就只剩钉钉后台配置。

重点检查：

## 1. Client ID

必须来自：

钉钉开放平台：

```
应用开发
 ↓
企业内部应用
 ↓
凭证与基础信息
 ↓
Client ID

```

不是：

- AppKey
- AgentId
- RobotCode

---

## 2. Stream模式开启

后台：

```
机器人管理
 ↓
消息接收模式

```

必须：

```
Stream模式

```

---

## 3. 发布应用

必须：

```
版本管理与发布
 ↓
上线

```

测试应用也需要发布。

---

## 4. 机器人权限

至少：

```
机器人消息接收
机器人发送消息

```

---

你现在先做两个动作：

### ① 取消 `.env` 注释

### ② 降级 websocket

执行：

```bash
pip uninstall websockets -y
pip install websockets==10.4

```

然后重新跑。

如果还有错误，把新的启动日志贴出来。你现在距离通已经很近了。

不对。你现在加日志的位置有问题，会导致程序启动时直接报错。

你写在文件顶部：

```python
logger.info(
    "Dify request question=%s user=%s",
    question,
    user
)

logger.info(
    "Dify response=%s",
    answer
)

```

但是此时：

```python
question
user
answer

```

三个变量都还不存在。

Python 加载 `dify_service.py` 时就会执行这里，所以会报：

```text
NameError: name 'question' is not defined

```

---

正确方式：**日志应该放到 `chat()` 方法内部。**

修改后的 `dify_service.py`：

```python
import requests
import json
import logging

from app.config.settings import settings


logger = logging.getLogger(__name__)


class DifyService:


    def __init__(self):

        self.url = settings.DIFY_API_URL

        self.key = settings.DIFY_API_KEY



    def chat(
        self,
        question: str,
        user: str
    ):


        logger.info(
            "Dify request question=%s user=%s",
            question,
            user
        )


        headers = {

            "Authorization":
            f"Bearer {self.key}",

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


        payload = {

            "inputs": {},

            "query": question,

            "response_mode":
            "streaming",

            "conversation_id":"",

            "user": user
        }


        logger.info(
            "Dify url=%s",
            self.url
        )


        response = requests.post(

            self.url,

            headers=headers,

            json=payload,

            stream=True,

            timeout=120

        )


        logger.info(
            "Dify http status=%s",
            response.status_code
        )


        answer = ""


        for line in response.iter_lines():


            if not line:
                continue


            line = line.decode(
                "utf-8"
            )


            logger.debug(
                "Dify stream=%s",
                line
            )


            if not line.startswith(
                "data:"
            ):
                continue



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


            event=data.get(
                "event"
            )


            if event=="message":

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

                answer += chunk


                logger.info(
                    "Dify chunk=%s",
                    chunk
                )


            elif event=="message_end":

                break


            elif event=="error":

                logger.error(
                    "Dify error=%s",
                    data
                )


        logger.info(
            "Dify response=%s",
            answer
        )


        return answer

```

---

## 另外，你现在排查钉钉无回答，建议再加一个关键日志

在：

```python
response=requests.post(...)

```

后面增加：

```python
logger.info(
    "Dify headers=%s",
    response.headers
)

```

以及：

```python
if response.status_code != 200:
    logger.error(
        "Dify error body=%s",
        response.text
    )

```

否则如果 API Key 错误，你只看到：

```text
answer=""

```

不知道原因。

---

## 重启服务

如果你用 systemd：

```bash
systemctl restart dify-dingtalk

```

查看：

```bash
journalctl -u dify-dingtalk -f

```

然后钉钉发送问题。

你应该看到类似：

```text
Dify request question=公司报销流程是什么 user=xxx

Dify url=http://192.168.4.20/v1/chat-messages

Dify http status=200

Dify chunk=根据公司制度...

Dify response=根据公司制度...

```

---

如果日志显示：

```text
Dify http status=200
Dify response=

```

那就是 **Dify 应用配置问题**。

如果显示：

```text
401

```

就是 API Key。

如果显示：

```text
404

```

就是 API URL。

如果显示：

```text
200 有answer

```

但是钉钉没回复，就是 `dingtalk_service.py / handler` 的问题。你下一步把日志贴出来即可。