Dify API 集成完整指南#

概述#

Dify 的 API 让你将 AI 能力集成到现有应用程序中，而无需从零开始构建 AI 基础设施。你可以获得大型语言模型的全部功能，同时拥有创建自定义用户体验的灵活性。

为什么使用 Dify API#

核心优势#

优势	说明
跳过后端复杂性	直接从前端应用访问 LLM 能力，无需管理 AI 基础设施
可视化应用管理	可视化设计和更新 AI 行为——更改会立即同步到所有 API 消费者
提供商灵活性	在 AI 提供商之间切换并集中管理 API 密钥，无需更改代码

API 集成的工作原理#

PRTCL // PLAINTEXT

1
1. 构建应用
2
   └─ 在 Dify Studio 中构建具有所需 AI 能力的应用
3

4
2. 生成 API 凭据
5
   └─ 安全访问应用功能
6

7
3. 调用 API
8
   └─ 从你的应用程序调用 API 获取 AI 驱动的响应
9

10
4. 用户交互
11
   └─ 用户通过你的自定义界面交互，Dify 处理 AI 处理过程

功能继承#

你的 API 会自动继承 Dify 应用的所有功能：

提示词
知识库
工具
模型配置

API 访问设置#

访问 API 设置页面#

进入你的应用
在左侧边栏找到 API Access
进入 API 设置页面

创建 API 凭据#

步骤：

点击 创建凭据
输入凭据名称（可选）
选择环境（开发、测试、生产）
点击创建
保存生成的 API Key

重要提示：

永远不要在前端代码或客户端请求中暴露 API 密钥
始终从后端调用 Dify API
定期轮换密钥

查看 API 文档#

Dify 会自动生成针对你应用配置的完整 API 文档，包括：

端点 URL
请求方法
请求参数
响应格式
示例代码

API 安全#

凭据管理#

最佳实践：

PRTCL // YAML

1
凭据管理:
2
  environment:
3
    - name: "development"
4
      api_key: "${DIFY_DEV_API_KEY}"
5
    - name: "testing"
6
      api_key: "${DIFY_TEST_API_KEY}"
7
    - name: "production"
8
      api_key: "${DIFY_PROD_API_KEY}"
9

10
  lifecycle:
11
    rotate: true
12
    rotation_period: "90days"
13
    revoke_unused: true

定期轮换：

每 90 天轮换一次密钥
撤销未使用的凭据
监控 API 使用情况

安全措施#

后端存储：

PRTCL // PYTHON

1
# 使用环境变量存储 API 密钥
2
import os
3

4
DIFY_API_KEY = os.getenv("DIFY_API_KEY")

速率限制：

PRTCL // PYTHON

1
from flask_limiter import Limiter
2

3
limiter = Limiter(
4
    app=app,
5
    key_func=get_remote_address,
6
    default_limits=["100 per hour"]
7
)
8

9
@app.route("/api/chat")
10
@limiter.limit("20 per minute")
11
def chat():
12
    pass

请求验证：

PRTCL // PYTHON

1
def validate_request(request):
2
    # 验证请求来源
3
    if not verify_origin(request):
4
        return False
5

6
    # 验证请求格式
7
    if not validate_format(request.json):
8
        return False
9

10
    return True

日志记录：

PRTCL // PYTHON

1
import logging
2

3
logging.basicConfig(level=logging.INFO)
4

5
def log_api_call(endpoint, user_id, status):
6
    logging.info(f"API Call: {endpoint} by {user_id} - Status: {status}")

文本生成应用 API#

端点信息#

URL：POST https://api.dify.ai/v1/completion-messages

请求参数#

参数	类型	必填	说明
inputs	object	是	应用输入变量
response_mode	string	是	响应模式：streaming, blocking
user	string	是	用户标识
files	array	否	文件列表

cURL 示例#

PRTCL // BASH

1
curl --location --request POST 'https://api.dify.ai/v1/completion-messages' \
2
  --header 'Authorization: Bearer ENTER-YOUR-SECRET-KEY' \
3
  --header 'Content-Type: application/json' \
4
  --data-raw '{
5
    "inputs": {
6
      "text": "Hello, how are you?"
7
    },
8
    "response_mode": "streaming",
9
    "user": "abc-123"
10
  }'

Python 示例#

PRTCL // PYTHON

1
import requests
2
import json
3

4
url = "https://api.dify.ai/v1/completion-messages"
5
headers = {
6
    'Authorization': 'Bearer ENTER-YOUR-SECRET-KEY',
7
    'Content-Type': 'application/json',
8
}
9
data = {
10
    "inputs": {
11
        "text": 'Hello, how are you?'
12
    },
13
    "response_mode": "streaming",
14
    "user": "abc-123"
15
}
16
response = requests.post(url, headers=headers, data=json.dumps(data))
17
print(response.text)

响应格式#

PRTCL // JSON

1
{
2
  "result": "Hello! I'm doing well, thank you for asking. How can I assist you today?",
3
  "message_id": "550e8400-e29b-41d4-a716-446655440000",
4
  "created_at": 1678901234,
5
  "metadata": {
6
    "usage": {
7
      "prompt_tokens": 10,
8
      "prompt_unit_price": "0.00001",
9
      "prompt_price_unit": "1K tokens",
10
      "prompt_price": "0.0001",
11
      "completion_tokens": 20,
12
      "completion_unit_price": "0.00002",
13
      "completion_price_unit": "1K tokens",
14
      "completion_price": "0.0004",
15
      "total_tokens": 30,
16
      "total_price": "0.0005",
17
      "currency": "USD",
18
      "latency": 1.5
19
    }
20
  }
21
}

对话应用 API#

端点信息#

URL：POST https://api.dify.ai/v1/chat-messages

conversation_id 的关键注意事项#

重要说明：

API 不会共享 WebApp 创建的对话
通过 API 创建的对话与在 WebApp 界面中创建的对话是隔离的

生成 conversation_id：

PRTCL // PYTHON

1
# 开始新对话时，将 conversation_id 字段留空
2
data = {
3
    "inputs": {},
4
    "query": "Hello",
5
    "response_mode": "streaming",
6
    "user": "abc-123"
7
    # conversation_id 不设置，系统将生成新的
8
}

在现有会话中处理 conversation_id：

PRTCL // PYTHON

1
# 一旦生成 conversation_id，后续调用应包含此 ID
2
data = {
3
    "inputs": {},  # 会被忽略
4
    "query": "How are you?",
5
    "response_mode": "streaming",
6
    "conversation_id": "1c7e55fb-1ba2-4e10-81b5-30addcea2276",
7
    "user": "abc-123"
8
}

管理动态变量：

PRTCL // PYTHON

1
# 使用会话变量调整机器人行为
2
data = {
3
    "query": "Change your language to Chinese",
4
    "response_mode": "streaming",
5
    "conversation_id": "1c7e55fb-1ba2-4e10-81b5-30addcea2276",
6
    "user": "abc-123",
7
    "variables": {
8
        "language": "Chinese"
9
    }
10
}

请求参数#

参数	类型	必填	说明
inputs	object	是	应用输入变量
query	string	是	用户查询
response_mode	string	是	响应模式：streaming, blocking
conversation_id	string	否	对话 ID
user	string	是	用户标识
files	array	否	文件列表

cURL 示例#

PRTCL // BASH

1
curl --location --request POST 'https://api.dify.ai/v1/chat-messages' \
2
  --header 'Authorization: Bearer ENTER-YOUR-SECRET-KEY' \
3
  --header 'Content-Type: application/json' \
4
  --data-raw '{
5
    "inputs": {},
6
    "query": "Hello",
7
    "response_mode": "streaming",
8
    "conversation_id": "1c7e55fb-1ba2-4e10-81b5-30addcea2276",
9
    "user": "abc-123"
10
  }'

Python 示例#

PRTCL // PYTHON

1
import requests
2
import json
3

4
url = 'https://api.dify.ai/v1/chat-messages'
5
headers = {
6
    'Authorization': 'Bearer ENTER-YOUR-SECRET-KEY',
7
    'Content-Type': 'application/json',
8
}
9
data = {
10
    "inputs": {},
11
    "query": "Hello",
12
    "response_mode": "streaming",
13
    "conversation_id": "1c7e55fb-1ba2-4e10-81b5-30addcea2276",
14
    "user": "abc-123"
15
}
16
response = requests.post(url, headers=headers, data=json.dumps(data))
17
print(response.text)

响应格式#

PRTCL // JSON

1
{
2
  "result": "Hello! I'm doing well, thank you for asking. How can I assist you today?",
3
  "conversation_id": "1c7e55fb-1ba2-4e10-81b5-30addcea2276",
4
  "message_id": "550e8400-e29b-41d4-a716-446655440001",
5
  "created_at": 1678901234,
6
  "metadata": {
7
    "usage": {
8
      "prompt_tokens": 15,
9
      "prompt_unit_price": "0.00001",
10
      "prompt_price_unit": "1K tokens",
11
      "prompt_price": "0.00015",
12
      "completion_tokens": 25,
13
      "completion_unit_price": "0.00002",
14
      "completion_price_unit": "1K tokens",
15
      "completion_price": "0.0005",
16
      "total_tokens": 40,
17
      "total_price": "0.00065",
18
      "currency": "USD",
19
      "latency": 2.0
20
    }
21
  }
22
}

流式响应处理#

流式响应格式#

流式响应以 Server-Sent Events (SSE) 格式返回：

PRTCL // PLAINTEXT

1
data: {"event": "message", "task_id": "xxx", "id": "xxx", "answer": "Hello"}
2

3
data: {"event": "message", "task_id": "xxx", "id": "xxx", "answer": "Hello, how"}
4

5
data: {"event": "message", "task_id": "xxx", "id": "xxx", "answer": "Hello, how are"}
6

7
data: {"event": "message_end", "task_id": "xxx", "id": "xxx", "metadata": {...}}

Python 流式处理示例#

PRTCL // PYTHON

1
import requests
2
import json
3

4
def stream_chat():
5
    url = 'https://api.dify.ai/v1/chat-messages'
6
    headers = {
7
        'Authorization': 'Bearer ENTER-YOUR-SECRET-KEY',
8
        'Content-Type': 'application/json',
9
    }
10
    data = {
11
        "inputs": {},
12
        "query": "Tell me a story",
13
        "response_mode": "streaming",
14
        "user": "abc-123"
15
    }
16

17
    response = requests.post(url, headers=headers, json=data, stream=True)
18

19
    for line in response.iter_lines():
20
        if line:
21
            line = line.decode('utf-8')
22
            if line.startswith('data: '):
23
                data = json.loads(line[6:])
24
                if data.get('event') == 'message':
25
                    print(data.get('answer'), end='', flush=True)
26
                elif data.get('event') == 'message_end':
27
                    print('\n\nResponse completed!')
28
                    print(f"Metadata: {data.get('metadata')}")
29

30
stream_chat()

JavaScript 流式处理示例#

PRTCL // JAVASCRIPT

1
async function streamChat() {
2
    const response = await fetch('https://api.dify.ai/v1/chat-messages', {
3
        method: 'POST',
4
        headers: {
5
            'Authorization': 'Bearer ENTER-YOUR-SECRET-KEY',
6
            'Content-Type': 'application/json',
7
        },
8
        body: JSON.stringify({
9
            inputs: {},
10
            query: "Tell me a story",
11
            response_mode: "streaming",
12
            user: "abc-123"
13
        })
14
    });
15

16
    const reader = response.body.getReader();
17
    const decoder = new TextDecoder();
18

19
    while (true) {
20
        const { done, value } = await reader.read();
21
        if (done) break;
22

23
        const chunk = decoder.decode(value);
24
        const lines = chunk.split('\n');
25

26
        for (const line of lines) {
27
            if (line.startsWith('data: ')) {
28
                const data = JSON.parse(line.substring(6));
29
                if (data.event === 'message') {
30
                    process.stdout.write(data.answer);
31
                } else if (data.event === 'message_end') {
32
                    console.log('\n\nResponse completed!');
33
                    console.log('Metadata:', data.metadata);
34
                }
35
            }
36
        }
37
    }
38
}
39

40
streamChat();

错误处理#

常见错误代码#

错误代码	说明	解决方案
401	未授权	检查 API 密钥
429	速率限制	减少请求频率
500	服务器错误	稍后重试
400	请求错误	检查请求参数

错误处理示例#

PRTCL // PYTHON

1
import requests
2
from requests.exceptions import RequestException
3

4
def chat_with_retry(url, headers, data, max_retries=3):
5
    for attempt in range(max_retries):
6
        try:
7
            response = requests.post(url, headers=headers, json=data)
8
            response.raise_for_status()
9
            return response.json()
10
        except RequestException as e:
11
            if response.status_code == 429:
12
                wait_time = 2 ** attempt  # 指数退避
13
                time.sleep(wait_time)
14
                continue
15
            elif response.status_code >= 500:
16
                if attempt < max_retries - 1:
17
                    time.sleep(wait_time)
18
                    continue
19
            raise e
20
    return None

最佳实践#

性能优化#

批量请求：

PRTCL // PYTHON

1
def batch_chat(messages):
2
    # 批量处理多个请求
3
    results = []
4
    for message in messages:
5
        result = chat_with_retry(url, headers, {
6
            "inputs": {},
7
            "query": message,
8
            "response_mode": "blocking",
9
            "user": "abc-123"
10
        })
11
        results.append(result)
12
    return results

缓存响应：

PRTCL // PYTHON

1
from functools import lru_cache
2

3
@lru_cache(maxsize=100)
4
def cached_chat(query):
5
    return chat_with_retry(url, headers, {
6
        "inputs": {},
7
        "query": query,
8
        "response_mode": "blocking",
9
        "user": "abc-123"
10
    })

监控和日志#

监控 API 使用：

PRTCL // PYTHON

1
import logging
2

3
def log_api_call(endpoint, status, response_time):
4
    logging.info(f"API: {endpoint} | Status: {status} | Time: {response_time}ms")
5

6
def monitor_chat():
7
    start_time = time.time()
8
    response = chat_with_retry(url, headers, data)
9
    response_time = (time.time() - start_time) * 1000
10
    log_api_call('chat', response.status_code, response_time)
11
    return response

测试策略#

单元测试：

PRTCL // PYTHON

1
import unittest
2

3
class TestDifyAPI(unittest.TestCase):
4
    def test_chat(self):
5
        response = chat_with_retry(url, headers, {
6
            "inputs": {},
7
            "query": "Hello",
8
            "response_mode": "blocking",
9
            "user": "test"
10
        })
11
        self.assertIsNotNone(response)
12
        self.assertIn('result', response)

集成测试：

PRTCL // PYTHON

1
def test_conversation_flow():
2
    # 开始对话
3
    response1 = chat_with_retry(url, headers, {
4
        "inputs": {},
5
        "query": "Hello",
6
        "response_mode": "blocking",
7
        "user": "test"
8
    })
9
    conversation_id = response1['conversation_id']
10

11
    # 继续对话
12
    response2 = chat_with_retry(url, headers, {
13
        "inputs": {},
14
        "query": "How are you?",
15
        "response_mode": "blocking",
16
        "conversation_id": conversation_id,
17
        "user": "test"
18
    })
19

20
    self.assertIsNotNone(response2)

资源链接#

API 文档: https://docs.dify.ai/zh/use-dify/publish/developing-with-apis
API 参考: https://docs.dify.ai/zh/reference/api
示例代码: https://github.com/langgenius/dify-api-examples

最后更新: 2026-03-12 作者: EchoHaoRan

Dify API 集成完整指南#

概述#

为什么使用 Dify API#

核心优势#

API 集成的工作原理#

功能继承#

API 访问设置#

访问 API 设置页面#

创建 API 凭据#

查看 API 文档#

API 安全#

凭据管理#

安全措施#

文本生成应用 API#

端点信息#

请求参数#

cURL 示例#

Python 示例#

响应格式#

对话应用 API#

端点信息#

conversation_id 的关键注意事项#

请求参数#

cURL 示例#

Python 示例#

响应格式#

流式响应处理#

流式响应格式#

Python 流式处理示例#

JavaScript 流式处理示例#

错误处理#

常见错误代码#

错误处理示例#

最佳实践#

性能优化#

监控和日志#

测试策略#

资源链接#

Dify API集成完整指南

OpenClaw模型配置完整指南

RAG检索增强生成技术完全指南

Dify知识库使用指南

评论