02 Plugin

Dify 插件开发 SKILL

概述

Dify 插件系统采用创新的 Beehive（蜂巢）架构，支持三种运行时模式，提供四种插件类型，为开发者提供强大的扩展能力。本 SKILL 提供从入门到精通的完整插件开发指南。

适用场景

当你需要以下场景时使用本 SKILL：

• 开发 Dify 插件：工具、模型、数据源、扩展四种类型
• 集成第三方服务：API 包装、OAuth 授权
• 实现自定义工具：扩展 AI 能力
• 扩展 Dify 平台功能：触发器、Agent 策略
• 将现有 API 迁移为插件：API 到插件的转换
• 调试和优化插件性能：性能调优和问题排查

核心价值

• Beehive 架构：去中心化调度、主从选举、故障自愈
• 多运行时支持：Local、Debug、Serverless 三种模式
• 完整工具链：dify CLI、Python SDK、调试运行时
• 丰富生态：70+ 模型插件、20+ 工具插件、18+ 数据源插件
• 生产级质量：OAuth 2.0、权限控制、监控追踪

快速开始

1. 环境准备

# 安装 Python 3.12+
python --version  # 确保 >= 3.12

# 安装 uv 包管理器（推荐）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 安装 dify CLI
pip install dify-cli

# 验证安装
dify --version

2. 创建第一个插件

# 初始化插件项目
dify plugin init

# 选择插件类型
? Select plugin type: Tool
? Plugin name: my_first_tool
? Author: your_name
? Description: My first Dify tool plugin

# 进入项目目录
cd my_first_tool

# 安装依赖
uv pip install -r requirements.txt

3. 实现插件逻辑

编辑 tools/my_tool.py：

from typing import Any, Generator
from dify_plugin.entities.tool import ToolInvokeMessage
from dify_plugin import Tool


class MyTool(Tool):
    def _invoke(
        self, tool_parameters: dict[str, Any]
    ) -> Generator[ToolInvokeMessage, None, None]:
        """
        工具调用入口
        """
        # 获取参数
        text = tool_parameters.get("text", "")

        # 处理逻辑
        result = f"处理结果: {text.upper()}"

        # 返回结果
        yield self.create_text_message(result)

4. 本地调试

# 启动 Debug Runtime
dify plugin run

# 在 Dify 界面中测试插件
# 访问 http://localhost:5001

5. 打包部署

# 打包插件
dify plugin package

# 生成 my_first_tool.difypkg 文件
# 上传到 Dify Marketplace 或私有部署

核心概念

Beehive 架构

Dify 插件系统采用 Beehive（蜂巢）架构，核心特点：

1. 去中心化调度：每个守护进程节点独立管理本地插件
2. 主从选举：通过 Redis 分布式锁实现自动主节点选举
3. 状态同步：所有节点状态存储在 Redis，支持跨节点访问
4. 故障自愈：节点宕机时自动重新分配插件

详细架构说明请参考：types/architecture.md

三种运行时模式

详细对比请参考：development/runtime-modes.md

四种插件类型

1. Tool（工具插件）

扩展 AI 能力的工具，如翻译、搜索、数据处理等。

# 示例：Google Translate
type: plugin
plugins:
  tools:
    - provider/google_translate.yaml

使用场景：

• API 包装（REST API → Tool）
• 数据处理（格式转换、计算）
• 外部服务集成（搜索、翻译）

2. Model（模型插件）

接入 LLM 提供商，如 Anthropic、OpenAI、Gemini 等。

# 示例：Anthropic Claude
type: plugin
plugins:
  models:
    - provider/anthropic.yaml

使用场景：

• 接入新的 LLM 提供商
• 自定义模型配置
• 多模态模型支持

3. Datasource（数据源插件）

连接存储系统，如 S3、Google Drive、Notion 等。

# 示例：AWS S3
type: plugin
plugins:
  datasources:
    - provider/aws_s3.yaml

使用场景：

• 云存储集成
• 文档平台连接
• 数据库访问

4. Extension（扩展插件）

平台功能扩展，如 Slack、企业微信等第三方平台集成。

# 示例：Slack
type: plugin
plugins:
  extensions:
    - provider/slack.yaml

使用场景：

• 第三方平台集成
• 触发器实现
• 自定义 Agent 策略

详细类型说明请参考：types/

项目结构

标准的 Dify 插件项目结构：

my_plugin/
├── manifest.yaml           # 插件清单（必需）
├── main.py                 # 入口文件（必需）
├── requirements.txt        # Python 依赖
├── pyproject.toml         # 项目配置
├── README.md              # 插件说明
├── PRIVACY.md             # 隐私政策（如需 OAuth）
├── .env.example           # 环境变量示例
│
├── _assets/               # 资源文件
│   └── icon.svg          # 插件图标
│
├── provider/              # 提供者配置
│   ├── provider_name.yaml # 提供者配置文件
│   └── provider_name.py   # 提供者实现
│
└── tools/                 # 工具实现（Tool 插件）
    ├── tool_name.yaml    # 工具配置文件
    └── tool_name.py      # 工具实现

核心文件说明

manifest.yaml

插件清单文件，定义插件的基本信息和配置：

author: your_name
name: my_plugin
version: 0.0.1
type: plugin

description:
  en_US: Plugin description
  zh_Hans: 插件描述

label:
  en_US: My Plugin
  zh_Hans: 我的插件

icon: icon.svg

meta:
  version: 0.0.1
  arch:
    - amd64
    - arm64
  runner:
    language: python
    version: '3.12'
    entrypoint: main

plugins:
  tools:
    - provider/my_provider.yaml

resource:
  memory: 1048576  # 1MB = 1048576 bytes
  permission:
    tool:
      enabled: true
    model:
      enabled: true
      llm: true

tags:
  - utilities

详细配置说明请参考：development/manifest.md

main.py

插件入口文件：

from dify_plugin import Plugin, DifyPluginEnv

# 创建插件实例
plugin = Plugin(DifyPluginEnv(MAX_REQUEST_TIMEOUT=120))

if __name__ == '__main__':
    # 启动插件
    plugin.run()

开发指南

Tool 插件开发

1. 创建 Provider

编辑 provider/my_provider.yaml：

identity:
  author: your_name
  name: my_provider
  label:
    en_US: My Provider
    zh_Hans: 我的提供者
  description:
    en_US: Provider description
    zh_Hans: 提供者描述
  icon: icon.svg
  tags:
    - utilities

extra:
  python:
    source: provider/my_provider.py

tools:
  - tools/my_tool.yaml

编辑 provider/my_provider.py：

from typing import Any
from dify_plugin.errors.tool import ToolProviderCredentialValidationError
from dify_plugin import ToolProvider
from tools.my_tool import MyTool


class MyProvider(ToolProvider):
    def _validate_credentials(self, credentials: dict[str, Any]) -> None:
        """
        验证凭证（如果需要）
        """
        try:
            # 验证 API Key 或其他凭证
            api_key = credentials.get("api_key")
            if not api_key:
                raise ToolProviderCredentialValidationError("API Key is required")

            # 测试凭证有效性
            MyTool().invoke(
                tool_parameters={"text": "test"}
            )
        except Exception as e:
            raise ToolProviderCredentialValidationError(str(e))

2. 创建 Tool

编辑 tools/my_tool.yaml：

identity:
  author: your_name
  name: my_tool
  label:
    en_US: My Tool
    zh_Hans: 我的工具

description:
  human:
    en_US: Tool description for humans
    zh_Hans: 给人类看的工具描述
  llm: Tool description for LLM

extra:
  python:
    source: tools/my_tool.py

parameters:
  - name: text
    type: string
    required: true
    label:
      en_US: Input Text
      zh_Hans: 输入文本
    human_description:
      en_US: The text to process
      zh_Hans: 要处理的文本
    llm_description: The input text to be processed
    form: llm

编辑 tools/my_tool.py：

from typing import Any, Generator
from dify_plugin.entities.tool import ToolInvokeMessage
from dify_plugin import Tool


class MyTool(Tool):
    def _invoke(
        self, tool_parameters: dict[str, Any]
    ) -> Generator[ToolInvokeMessage, None, None]:
        """
        工具调用入口

        Args:
            tool_parameters: 工具参数字典

        Yields:
            ToolInvokeMessage: 工具调用消息
        """
        # 1. 获取参数
        text = tool_parameters.get("text", "")

        if not text:
            yield self.create_text_message("Error: text parameter is required")
            return

        try:
            # 2. 处理逻辑
            result = self._process_text(text)

            # 3. 返回结果
            yield self.create_text_message(result)

        except Exception as e:
            yield self.create_text_message(f"Error: {str(e)}")

    def _process_text(self, text: str) -> str:
        """
        处理文本的具体逻辑
        """
        return text.upper()

详细开发指南请参考：development/tool.md

OAuth 2.0 集成

OAuth 流程概述

Dify 支持完整的 OAuth 2.0 授权流程：

1. 获取授权 URL：用户点击授权按钮
2. 用户授权：跳转到第三方服务授权页面
3. 获取授权码：用户授权后返回授权码
4. 交换访问令牌：使用授权码交换访问令牌
5. 刷新令牌：访问令牌过期时使用刷新令牌获取新令牌

配置 OAuth Schema

在 provider/my_provider.yaml 中添加 OAuth 配置：

oauth_schema:
  client_schema:
    - name: client_id
      type: secret-input
      required: true
      label:
        en_US: Client ID
        zh_Hans: 客户端 ID
      placeholder:
        en_US: Enter your OAuth Client ID
        zh_Hans: 输入您的 OAuth 客户端 ID
      help:
        en_US: Get your client ID from the service provider
        zh_Hans: 从服务提供商获取客户端 ID
      url: https://console.example.com/credentials

    - name: client_secret
      type: secret-input
      required: true
      label:
        en_US: Client Secret
        zh_Hans: 客户端密钥
      placeholder:
        en_US: Enter your OAuth Client Secret
        zh_Hans: 输入您的 OAuth 客户端密钥

  credentials_schema:
    - name: access_token
      type: secret-input
      label:
        en_US: Access Token
        zh_Hans: 访问令牌

    - name: refresh_token
      type: secret-input
      label:
        en_US: Refresh Token
        zh_Hans: 刷新令牌

实现 OAuth Provider

from typing import Any
from dify_plugin import ToolProvider
from dify_plugin.errors.tool import ToolProviderCredentialValidationError
import requests


class MyOAuthProvider(ToolProvider):
    def _validate_credentials(self, credentials: dict[str, Any]) -> None:
        """
        验证 OAuth 凭证
        """
        access_token = credentials.get("access_token")
        if not access_token:
            raise ToolProviderCredentialValidationError(
                "Access token is required"
            )

        # 测试访问令牌有效性
        try:
            response = requests.get(
                "https://api.example.com/user",
                headers={"Authorization": f"Bearer {access_token}"}
            )
            response.raise_for_status()
        except Exception as e:
            raise ToolProviderCredentialValidationError(
                f"Invalid access token: {str(e)}"
            )

使用 OAuth 令牌

在 Tool 中使用 OAuth 令牌：

class MyOAuthTool(Tool):
    def _invoke(
        self, tool_parameters: dict[str, Any]
    ) -> Generator[ToolInvokeMessage, None, None]:
        # 获取 OAuth 凭证
        credentials = self.runtime.credentials
        access_token = credentials.get("access_token")

        # 调用 API
        response = requests.get(
            "https://api.example.com/data",
            headers={"Authorization": f"Bearer {access_token}"}
        )

        yield self.create_text_message(response.text)

完整 OAuth 集成指南请参考：development/oauth-integration.md

测试和调试

单元测试

使用 pytest 编写单元测试：

# tests/test_my_tool.py
import pytest
from tools.my_tool import MyTool


def test_my_tool():
    tool = MyTool()
    result = list(tool.invoke(tool_parameters={"text": "hello"}))

    assert len(result) == 1
    assert "HELLO" in result[0].message

运行测试：

pytest tests/

集成测试

测试插件与 Dify 系统的集成：

# tests/test_integration.py
def test_plugin_integration():
    # 测试插件安装
    # 测试工具调用
    # 测试 OAuth 流程
    pass

Debug Runtime 调试

启动 Debug Runtime 进行远程调试：

# 启动 Debug Runtime
dify plugin run --debug

# 在 Dify 界面中测试
# 查看实时日志和错误信息

常见问题排查

插件启动失败

• 检查 Python 版本（需要 3.12+）
• 检查依赖安装（requirements.txt）
• 检查 manifest.yaml 配置
• 检查入口文件路径（main.py）

权限验证失败

• 检查 credentials 配置
• 检查 OAuth 令牌是否过期
• 检查 API Key 是否有效
• 检查权限声明（resource.permission）

运行时错误

• 调整超时时间（MAX_REQUEST_TIMEOUT）
• 优化内存使用（resource.memory）
• 检查网络连接
• 检查反向调用权限

详细排查指南请参考：testing/troubleshooting.md

从 API 到插件迁移

迁移策略

1. API 包装模式：将 REST API 包装为 Tool 插件
2. SDK 集成模式：使用第三方 Python SDK
3. OAuth 迁移：实现 OAuth 2.0 授权流程
4. 批量操作优化：使用流式响应处理大数据

迁移步骤

1. 分析现有 API

   • API 端点和参数
   • 认证方式（API Key / OAuth）
   • 响应格式和错误处理

2. 选择插件类型

   • REST API → Tool 插件
   • LLM API → Model 插件
   • 存储 API → Datasource 插件

3. 创建插件项目

   dify plugin init
   

4. 实现 API 调用

   import requests
   
   class APIWrapperTool(Tool):
       def _invoke(self, tool_parameters: dict[str, Any]):
           response = requests.post(
               "https://api.example.com/endpoint",
               json=tool_parameters,
               headers={"Authorization": f"Bearer {api_key}"}
           )
           yield self.create_text_message(response.text)
   

5. 配置认证

   • API Key：在 provider.yaml 中配置 credentials
   • OAuth：配置 oauth_schema

6. 添加错误处理

   try:
       response = requests.post(...)
       response.raise_for_status()
   except requests.exceptions.RequestException as e:
       yield self.create_text_message(f"API Error: {str(e)}")
   

7. 编写测试

   def test_api_wrapper():
       tool = APIWrapperTool()
       result = list(tool.invoke(tool_parameters={...}))
       assert result[0].message
   

8. 打包部署

   dify plugin package
   

实际案例

• REST API → Tool：Google Translate（无认证）
• OAuth API → Tool：Google Contacts（OAuth 2.0）
• LLM API → Model：Anthropic Claude（API Key）
• 存储 API → Datasource：AWS S3（IAM 认证）

详细迁移指南请参考：development/migration-guide.md

最佳实践

代码组织

1. 单一职责：每个 Tool 只做一件事
2. 错误处理：捕获所有异常并返回友好错误信息
3. 参数验证：在 _invoke 开始时验证所有参数
4. 日志记录：使用结构化日志记录关键操作

性能优化

1. 流式响应：对于大数据使用 Generator 流式返回
2. 缓存策略：缓存频繁访问的数据
3. 异步处理：使用 async/await 处理 I/O 密集操作
4. 资源限制：合理配置 resource.memory

安全考虑

1. 输入验证：验证所有用户输入
2. 敏感数据：使用 secret-input 类型存储敏感信息
3. 权限控制：最小权限原则，只声明必需的权限
4. 错误信息：不要在错误信息中泄露敏感数据

用户体验

1. 清晰的描述：提供详细的 description 和 help
2. 合理的默认值：为参数提供合理的默认值
3. 多语言支持：至少支持 en_US 和 zh_Hans
4. 友好的错误：返回用户可理解的错误信息

部署和运维

运行时模式选择

Serverless 部署

部署到 AWS Lambda：

# 1. 打包插件
dify plugin package

# 2. 创建 Lambda 函数
aws lambda create-function \
  --function-name my-plugin \
  --runtime python3.12 \
  --handler main.handler \
  --zip-file fileb://my_plugin.difypkg

# 3. 配置环境变量
aws lambda update-function-configuration \
  --function-name my-plugin \
  --environment Variables={DIFY_PLUGIN_DAEMON_URL=https://...}

# 4. 在 Dify 中注册端点
# 访问 Dify 管理界面，添加 Serverless 端点

监控和日志

1. OpenTelemetry 追踪：自动集成分布式追踪
2. Sentry 错误追踪：捕获和报告错误
3. 结构化日志：使用 Python logging 模块
4. 性能指标：监控响应时间、错误率、资源使用

故障恢复

1. 自动重启：WatchDog 监控插件健康状态
2. 重试机制：Serverless 模式支持指数退避重试
3. 降级处理：故障时返回友好错误信息
4. 备份恢复：定期备份插件配置和状态

参考资源

官方文档

• Dify Plugin Development Cheatsheet
• Dify Plugin System Design
• Initialize Development Tools

开源项目

• dify-plugin-daemon - 插件守护进程
• dify-official-plugins - 官方插件集合
• dify-plugin-sdks - Python SDK

社区资源

• Dify Discord - 官方 Discord 社区
• Dify Reddit - Reddit 讨论区
• Dify Marketplace - 插件市场

详细文档

• types/ - 四种插件类型详解
• development/ - 开发指南
• templates/ - 插件模板
• testing/ - 测试指南

版本: v1.0.0 最后更新: 2026-03-04 维护者: Dify SKILL Team