Dify 接口怎么接入项目？从 API Key 到 Python/Node 源码实战指南

发布人：慈云数据-客服中心发布时间：20小时前阅读量：0

Dify API接口调用教程｜附源码

在 AI 应用开发中，很多团队会使用 Dify 来快速搭建智能客服、知识库问答、Agent、工作流自动化应用等。Dify 的优势在于：它既提供了可视化编排能力，也提供了标准化的 API 接口，方便开发者把已经搭建好的 AI 应用集成到网站、后台系统、小程序、企业微信、飞书、钉钉或自研业务系统中。

本文将以实战方式介绍 Dify API 接口调用方法，包括 API Key 获取、接口地址说明、Chat 应用调用、Workflow 工作流调用、流式响应处理，以及 Python、Node.js 示例源码。

一、Dify API 是什么？

Dify API 是 Dify 为每个应用提供的 HTTP 接口能力。开发者可以通过 API 调用 Dify 中已经配置好的应用，比如：

聊天助手应用
文本生成应用
Agent 应用
工作流应用
知识库问答应用
多轮对话应用

通过 Dify API，你不需要在业务系统里重复实现 Prompt 编排、模型参数配置、知识库检索、插件调用、工作流节点逻辑等能力，只需要把用户问题或业务参数传给 Dify，由 Dify 完成 AI 处理并返回结果。

二、调用 Dify API 前的准备工作

在正式调用接口之前，需要先完成以下准备。

1. 创建 Dify 应用

Chatbot：聊天助手
Agent：具备工具调用能力的智能体
Workflow：工作流应用
Text Generator：文本生成应用

如果你想实现智能客服或知识库问答，一般可以选择 Chatbot。

如果你想实现更复杂的流程，例如“输入产品名称 → 查询资料 → 生成文案 → 输出结构化 JSON”，则可以选择 Workflow。

2. 获取 API Key

进入指定应用后，通常可以在以下位置找到 API 访问配置：

应用详情 → API 访问 → API Key

点击创建或复制 API Key。

API Key 是调用 Dify API 的凭证，调用时需要放在请求头中：

Authorization: Bearer your-api-key

注意：API Key 不要暴露在前端页面中，尤其不要直接写在浏览器 JavaScript 代码里。更推荐通过后端服务中转调用 Dify API。

3. 确认 API Base URL

如果你使用的是 Dify 云服务，默认接口地址一般为：

https://api.dify.ai/v1

如果你是私有化部署，例如部署在自己的服务器上，则接口地址通常类似：

https://your-domain.com/v1

或：

http://your-server-ip/v1

最终地址以你的 Dify 部署配置为准。

三、Dify API 通用请求格式

Dify API 一般采用 RESTful 风格，请求方式通常为 POST，数据格式为 JSON。

请求头示例：

Authorization: Bearer YOUR_API_KEY
Content-Type: application/json

常用请求参数包括：

参数名	说明
`inputs`	应用输入变量，通常是一个对象
`query`	用户输入的问题，常用于 Chat 应用
`response_mode`	响应模式，支持 `blocking` 或 `streaming`
`user`	用户唯一标识，用于区分不同终端用户
`conversation_id`	会话 ID，用于多轮对话
`files`	文件参数，部分应用支持上传文件

其中：

blocking 表示阻塞式响应，接口会等结果生成完后一次性返回。
streaming 表示流式响应，适合聊天机器人逐字输出的场景。

四、调用 Chat 应用接口

如果你的 Dify 应用是聊天助手，可以使用以下接口：

POST /chat-messages

完整 URL 示例：

https://api.dify.ai/v1/chat-messages

1. cURL 调用示例

curl -X POST 'https://api.dify.ai/v1/chat-messages' \
  -H 'Authorization: Bearer YOUR_API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "inputs": {},
    "query": "请介绍一下 Dify 的主要功能",
    "response_mode": "blocking",
    "user": "user-001"
  }'

2. 返回结果示例

阻塞模式下，接口可能返回类似结果：

{
  "event": "message",
  "task_id": "abc123",
  "id": "msg_001",
  "message_id": "msg_001",
  "conversation_id": "conv_001",
  "mode": "chat",
  "answer": "Dify 是一个开源的大语言模型应用开发平台，可以用于构建聊天机器人、Agent、工作流应用和知识库问答系统。",
  "metadata": {
    "usage": {
      "prompt_tokens": 120,
      "completion_tokens": 80,
      "total_tokens": 200
    }
  },
  "created_at": 1710000000
}

其中最常用的字段是：

字段	说明
`answer`	AI 返回的最终回答
`conversation_id`	当前会话 ID，多轮对话时需要保存
`message_id`	消息 ID
`metadata.usage`	Token 使用情况

五、Python 调用 Dify Chat API 源码

下面是一个完整的 Python 示例，使用 requests 调用 Dify Chat API。

1. 安装依赖

pip install requests

2. Python 阻塞式调用源码

import requests


DIFY_API_KEY = "YOUR_API_KEY"
DIFY_API_URL = "https://api.dify.ai/v1/chat-messages"


def call_dify_chat(query, user="user-001", conversation_id=None):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "inputs": {},
        "query": query,
        "response_mode": "blocking",
        "user": user
    }

    if conversation_id:
        payload["conversation_id"] = conversation_id

    response = requests.post(DIFY_API_URL, headers=headers, json=payload, timeout=60)

    if response.status_code != 200:
        raise Exception(f"请求失败，状态码：{response.status_code}，响应：{response.text}")

    return response.json()


if __name__ == "__main__":
    result = call_dify_chat("请用三句话介绍 Dify")
    print("AI回答：")
    print(result.get("answer"))

    print("会话ID：")
    print(result.get("conversation_id"))

3. 多轮对话示例

多轮对话的关键是保存第一次接口返回的 conversation_id，后续请求继续传入该 ID。

import requests


DIFY_API_KEY = "YOUR_API_KEY"
DIFY_API_URL = "https://api.dify.ai/v1/chat-messages"


def chat(query, conversation_id=None):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }

    data = {
        "inputs": {},
        "query": query,
        "response_mode": "blocking",
        "user": "user-001"
    }

    if conversation_id:
        data["conversation_id"] = conversation_id

    response = requests.post(DIFY_API_URL, headers=headers, json=data)
    response.raise_for_status()

    return response.json()


if __name__ == "__main__":
    conversation_id = None

    result1 = chat("你是谁？", conversation_id)
    print(result1["answer"])
    conversation_id = result1["conversation_id"]

    result2 = chat("请继续用更简单的话解释", conversation_id)
    print(result2["answer"])

六、Python 流式响应调用源码

如果你希望像 ChatGPT 一样逐字显示回答，可以使用 streaming 模式。

1. 流式调用示例

import requests
import json


DIFY_API_KEY = "YOUR_API_KEY"
DIFY_API_URL = "https://api.dify.ai/v1/chat-messages"


def stream_chat(query):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "inputs": {},
        "query": query,
        "response_mode": "streaming",
        "user": "user-001"
    }

    response = requests.post(
        DIFY_API_URL,
        headers=headers,
        json=payload,
        stream=True,
        timeout=60
    )

    if response.status_code != 200:
        raise Exception(f"请求失败：{response.status_code}，{response.text}")

    for line in response.iter_lines():
        if not line:
            continue

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

        if decoded_line.startswith("data:"):
            data = decoded_line.replace("data:", "").strip()

            if data == "[DONE]":
                break

            try:
                event = json.loads(data)
            except json.JSONDecodeError:
                continue

            event_type = event.get("event")

            if event_type == "message":
                answer = event.get("answer", "")
                print(answer, end="", flush=True)

            elif event_type == "message_end":
                print("\n\n回答结束")
                print("会话ID：", event.get("conversation_id"))


if __name__ == "__main__":
    stream_chat("请详细介绍 Dify 的 API 调用方式")

七、Node.js 调用 Dify Chat API 源码

如果你的项目是 Node.js 后端，也可以使用 fetch 或 axios 调用 Dify API。

下面以 Node.js 原生 fetch 为例。

Node.js 18 及以上版本内置 fetch，如果版本较低，可以安装 node-fetch。

1. 阻塞式调用源码

const DIFY_API_KEY = "YOUR_API_KEY";
const DIFY_API_URL = "https://api.dify.ai/v1/chat-messages";

async function callDifyChat(query, conversationId = null) {
  const payload = {
    inputs: {},
    query,
    response_mode: "blocking",
    user: "user-001"
  };

  if (conversationId) {
    payload.conversation_id = conversationId;
  }

  const response = await fetch(DIFY_API_URL, {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${DIFY_API_KEY}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify(payload)
  });

  if (!response.ok) {
    const errorText = await response.text();
    throw new Error(`Dify API 请求失败：${response.status} ${errorText}`);
  }

  return await response.json();
}

async function main() {
  const result = await callDifyChat("请介绍一下 Dify");
  console.log("AI回答：", result.answer);
  console.log("会话ID：", result.conversation_id);
}

main().catch(console.error);

2. Node.js 流式响应源码

const DIFY_API_KEY = "YOUR_API_KEY";
const DIFY_API_URL = "https://api.dify.ai/v1/chat-messages";

async function streamDifyChat(query) {
  const response = await fetch(DIFY_API_URL, {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${DIFY_API_KEY}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      inputs: {},
      query,
      response_mode: "streaming",
      user: "user-001"
    })
  });

  if (!response.ok) {
    const errorText = await response.text();
    throw new Error(`请求失败：${response.status} ${errorText}`);
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder("utf-8");

  while (true) {
    const { value, done } = await reader.read();

    if (done) {
      break;
    }

    const chunk = decoder.decode(value, { stream: true });
    const lines = chunk.split("\n");

    for (const line of lines) {
      if (!line.startsWith("data:")) {
        continue;
      }

      const data = line.replace("data:", "").trim();

      if (!data || data === "[DONE]") {
        continue;
      }

      try {
        const event = JSON.parse(data);

        if (event.event === "message") {
          process.stdout.write(event.answer || "");
        }

        if (event.event === "message_end") {
          console.log("\n回答结束");
          console.log("会话ID：", event.conversation_id);
        }
      } catch (err) {
        // 忽略不完整数据片段
      }
    }
  }
}

streamDifyChat("请用通俗语言介绍 Dify 的工作流能力").catch(console.error);

八、调用 Workflow 工作流接口

如果你创建的是 Dify Workflow 应用，可以调用工作流运行接口。

接口地址：

POST /workflows/run

完整 URL：

https://api.dify.ai/v1/workflows/run

Workflow 应用通常不使用 query 字段，而是通过 inputs 传入工作流开始节点中定义的变量。

1. Workflow cURL 示例

假设你的工作流开始节点中定义了两个变量：

topic：文章主题
style：写作风格

那么可以这样调用：

curl -X POST 'https://api.dify.ai/v1/workflows/run' \
  -H 'Authorization: Bearer YOUR_API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "inputs": {
      "topic": "Dify API接口调用教程",
      "style": "技术教程风格"
    },
    "response_mode": "blocking",
    "user": "user-001"
  }'

2. Workflow 返回示例

{
  "workflow_run_id": "run_001",
  "task_id": "task_001",
  "data": {
    "id": "run_001",
    "workflow_id": "workflow_001",
    "status": "succeeded",
    "outputs": {
      "text": "这是一篇由 Dify Workflow 生成的文章内容。"
    },
    "elapsed_time": 3.25,
    "total_tokens": 1200,
    "created_at": 1710000000,
    "finished_at": 1710000003
  }
}

其中 data.outputs 是最常用的结果字段。

3. Python 调用 Workflow 源码

import requests


DIFY_API_KEY = "YOUR_API_KEY"
DIFY_WORKFLOW_URL = "https://api.dify.ai/v1/workflows/run"


def run_workflow(topic, style):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "inputs": {
            "topic": topic,
            "style": style
        },
        "response_mode": "blocking",
        "user": "user-001"
    }

    response = requests.post(DIFY_WORKFLOW_URL, headers=headers, json=payload, timeout=120)

    if response.status_code != 200:
        raise Exception(f"Workflow 调用失败：{response.status_code}，{response.text}")

    return response.json()


if __name__ == "__main__":
    result = run_workflow("Dify API接口调用教程", "技术博客风格")

    outputs = result.get("data", {}).get("outputs", {})
    print(outputs)

九、文本生成应用接口调用

如果你创建的是 Text Generator 类型应用，可以调用文本生成接口：

POST /completion-messages

示例：

curl -X POST 'https://api.dify.ai/v1/completion-messages' \
  -H 'Authorization: Bearer YOUR_API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "inputs": {
      "topic": "人工智能在企业中的应用"
    },
    "response_mode": "blocking",
    "user": "user-001"
  }'

Python 示例：

import requests


API_KEY = "YOUR_API_KEY"
URL = "https://api.dify.ai/v1/completion-messages"


def generate_text(topic):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "inputs": {
            "topic": topic
        },
        "response_mode": "blocking",
        "user": "user-001"
    }

    response = requests.post(URL, headers=headers, json=payload)
    response.raise_for_status()

    return response.json()


if __name__ == "__main__":
    result = generate_text("人工智能在企业数字化转型中的价值")
    print(result.get("answer"))

十、在后端项目中封装 Dify API

实际项目中，不建议每个业务接口都直接写一遍 Dify 请求逻辑。更推荐封装一个统一的 Dify 客户端类。

下面给出一个 Python 封装示例。

import requests


class DifyClient:
    def __init__(self, api_key, base_url="https://api.dify.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")

    def _headers(self):
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def chat(self, query, user, inputs=None, conversation_id=None, response_mode="blocking"):
        url = f"{self.base_url}/chat-messages"

        payload = {
            "inputs": inputs or {},
            "query": query,
            "response_mode": response_mode,
            "user": user
        }

        if conversation_id:
            payload["conversation_id"] = conversation_id

        response = requests.post(url, headers=self._headers(), json=payload, timeout=60)
        response.raise_for_status()

        return response.json()

    def completion(self, inputs, user, response_mode="blocking"):
        url = f"{self.base_url}/completion-messages"

        payload = {
            "inputs": inputs,
            "response_mode": response_mode,
            "user": user
        }

        response = requests.post(url, headers=self._headers(), json=payload, timeout=60)
        response.raise_for_status()

        return response.json()

    def workflow_run(self, inputs, user, response_mode="blocking"):
        url = f"{self.base_url}/workflows/run"

        payload = {
            "inputs": inputs,
            "response_mode": response_mode,
            "user": user
        }

        response = requests.post(url, headers=self._headers(), json=payload, timeout=120)
        response.raise_for_status()

        return response.json()


if __name__ == "__main__":
    client = DifyClient(api_key="YOUR_API_KEY")

    chat_result = client.chat(
        query="请解释什么是 Dify",
        user="user-001"
    )

    print(chat_result.get("answer"))

这种封装方式有几个好处：

统一管理 API Key
统一处理请求头
统一处理接口地址
方便后续增加日志、重试、异常捕获
便于集成到 Flask、FastAPI、Django 等后端框架中

十一、FastAPI 集成 Dify 示例

下面给出一个简单的 FastAPI 后端接口示例。前端请求你的后端接口，由后端再调用 Dify，这样可以避免 API Key 暴露在前端。

1. 安装依赖

pip install fastapi uvicorn requests

2. FastAPI 源码

from fastapi import FastAPI
from pydantic import BaseModel
import requests


app = FastAPI()

DIFY_API_KEY = "YOUR_API_KEY"
DIFY_API_URL = "https://api.dify.ai/v1/chat-messages"


class ChatRequest(BaseModel):
    query: str
    conversation_id: str | None = None
    user: str = "web-user-001"


@app.post("/api/chat")
def chat(req: ChatRequest):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "inputs": {},
        "query": req.query,
        "response_mode": "blocking",
        "user": req.user
    }

    if req.conversation_id:
        payload["conversation_id"] = req.conversation_id

    response = requests.post(DIFY_API_URL, headers=headers, json=payload, timeout=60)

    if response.status_code != 200:
        return {
            "success": False,
            "error": response.text
        }

    data = response.json()

    return {
        "success": True,
        "answer": data.get("answer"),
        "conversation_id": data.get("conversation_id"),
        "message_id": data.get("message_id")
    }

启动服务：

uvicorn main:app --reload --port 8000

前端或接口测试工具可以请求：

POST http://localhost:8000/api/chat

请求体：

{
  "query": "请介绍一下 Dify",
  "user": "user-001"
}

十二、常见错误与解决方法

1. 401 Unauthorized

常见原因：

API Key 写错
请求头中没有携带 Authorization
Authorization 格式不正确

正确格式：

Authorization: Bearer YOUR_API_KEY

注意 Bearer 后面有一个空格。

2. 404 Not Found

常见原因：

API 地址写错
私有化部署时 Base URL 不正确
应用类型和接口不匹配

例如 Chat 应用应该调用：

/chat-messages

Workflow 应用应该调用：

/workflows/run

Text Generator 应用应该调用：

/completion-messages

3. 400 Bad Request

常见原因：

请求参数缺失
inputs 中的变量名与 Dify 应用中定义的不一致
JSON 格式错误
response_mode 填写错误

如果 Workflow 开始节点定义的是 topic，那么请求中也必须传：

{
  "inputs": {
    "topic": "测试主题"
  }
}

不能写成：

{
  "inputs": {
    "title": "测试主题"
  }
}

4. 流式响应无法解析

流式响应通常是 Server-Sent Events，返回数据不是一个完整 JSON，而是一行一行的事件流。

因此不能直接使用：

response.json()

而应该使用：

response.iter_lines()

并解析以 data: 开头的内容。

十三、实际项目中的最佳实践

1. API Key 放在服务端

不要把 Dify API Key 写在前端代码中。正确做法是：

前端页面 → 你的后端服务 → Dify API

这样可以避免 Key 泄露。

2. 保存 conversation_id

如果你要实现连续对话，需要保存 Dify 返回的 conversation_id。可以保存到：

浏览器 localStorage
后端数据库
Redis
用户会话表

但如果涉及隐私数据，更推荐保存在后端。

3. 为不同用户传不同 user

user 字段建议传入业务系统中的用户唯一标识，例如：

user_10001

这样方便 Dify 区分不同终端用户，也便于后续查看日志和统计。

4. 给接口设置超时时间

AI 生成可能会比较慢，后端调用时建议设置合理的 timeout。例如：

requests.post(url, headers=headers, json=payload, timeout=60)

Workflow 如果步骤较多，可以设置更长时间，例如 120 秒。

5. 增加异常处理与日志

生产环境中建议记录：

请求时间
用户 ID
conversation_id
message_id
响应耗时
错误状态码
错误详情

这样可以快速定位问题。

十四、总结

Dify API 的调用整体并不复杂，核心步骤可以概括为：

在 Dify 中创建应用；
获取应用 API Key；
确认 API Base URL；
根据应用类型选择接口；
在请求头中携带 Authorization: Bearer API_KEY；
使用 inputs、query、user 等参数发起请求；
根据 blocking 或 streaming 选择不同的响应处理方式。

如果是聊天类应用，优先使用：

POST /chat-messages

如果是文本生成应用，使用：

POST /completion-messages

如果是工作流应用，使用：

POST /workflows/run

在企业项目中，推荐通过后端服务封装 Dify API，再提供给前端调用。这样既能保护 API Key，也方便统一处理日志、权限、限流和异常。

通过本文中的 Python、Node.js、FastAPI 示例源码，你可以快速将 Dify AI 应用接入自己的业务系统，实现智能客服、知识库问答、AI 文案生成、工作流自动化等能力。

文章标签： DifyAPI 接口调用流式响应 Workflow工作流

上一篇：跨境电商接入 Dify API：从客服自动回复到商品文案生成实战指南

下一篇：Dify API 接入实战：从获取 Key 到聊天、工作流调用全流程源码示例

更多栏目

新闻动态

文档中心

下载中心

目录结构

全文

产品与服务

新闻帮助

生态合作

了解我们

Dify 接口怎么接入项目？从 API Key 到 Python/Node 源码实战指南

Dify API接口调用教程｜附源码

一、Dify API 是什么？

二、调用 Dify API 前的准备工作

1. 创建 Dify 应用

2. 获取 API Key

3. 确认 API Base URL

三、Dify API 通用请求格式

四、调用 Chat 应用接口

1. cURL 调用示例

2. 返回结果示例

五、Python 调用 Dify Chat API 源码

1. 安装依赖

2. Python 阻塞式调用源码

3. 多轮对话示例

六、Python 流式响应调用源码

1. 流式调用示例

七、Node.js 调用 Dify Chat API 源码

1. 阻塞式调用源码

2. Node.js 流式响应源码

八、调用 Workflow 工作流接口

1. Workflow cURL 示例

2. Workflow 返回示例

3. Python 调用 Workflow 源码

九、文本生成应用接口调用

十、在后端项目中封装 Dify API

十一、FastAPI 集成 Dify 示例

1. 安装依赖

2. FastAPI 源码

十二、常见错误与解决方法

1. 401 Unauthorized

2. 404 Not Found

3. 400 Bad Request

4. 流式响应无法解析

十三、实际项目中的最佳实践

1. API Key 放在服务端

2. 保存 conversation_id

3. 为不同用户传不同 user

4. 给接口设置超时时间

5. 增加异常处理与日志

十四、总结