# Dify Plugin 开发教程
本教程详细介绍如何开发一个完整的 Dify Plugin 工具。
## 1. 工具类基本结构
每个工具都需要继承 Tool 基类，并实现 _invoke 方法：
```python
from collections.abc import Generator
from typing import Any
from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage
class ExampleTool(Tool):
def invoke(self, toolparameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
# 工具逻辑实现
pass
```
## 2. 参数定义 (YAML 配置)
### 2.1 基本信息定义
```yaml
identity:
name: "example_tool" # 工具唯一标识
author: "your_name"
label:
en_US: "Example Tool"
zh_Hans: "示例工具"
description:
human:
en_US: "This is an example tool description."
zh_Hans: "这是一个示例工具描述。"
llm: "This tool is used for demonstration purposes."
```
### 2.2 参数字段定义
```yaml
parameters:
- name: required_param
type: string
required: true
label:
en_US: Required Parameter
zh_Hans: 必需参数
human_description:
en_US: "This is a required string parameter"
zh_Hans: "这是一个必需的字符串参数"
llm_description: "A required string parameter for the tool"
form: llm
- name: optional_param
type: string
required: false
label:
en_US: Optional Parameter
zh_Hans: 可选参数
human_description:
en_US: "This is an optional parameter"
zh_Hans: "这是一个可选参数"
llm_description: "An optional parameter with default value"
form: llm
- name: number_param
type: number
required: false
default: 100
label:
en_US: Number Parameter
zh_Hans: 数字参数
- name: boolean_param
type: boolean
required: false
default: false
label:
en_US: Boolean Switch
zh_Hans: 布尔开关
- name: select_param
type: select
required: true
options:
- value: "option1"
label:
en_US: "Option One"
zh_Hans: "选项一"
- value: "option2"
label:
en_US: "Option Two"
zh_Hans: "选项二"
label:
en_US: "Select Option"
zh_Hans: "选择选项"
```
参数类型说明：
- string: 字符串类型
- number: 数字类型
- boolean: 布尔类型
- select: 下拉选择，需要配合 options 字段
- file: 文件类型
## 3. 工具逻辑实现
### 3.1 参数获取和验证
```python
def invoke(self, toolparameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
try:
# 获取必需参数
required_param = tool_parameters.get('required_param')
optional_param = tool_parameters.get('optional_param', 'default_value')
number_param = tool_parameters.get('number_param', 100)
boolean_param = tool_parameters.get('boolean_param', False)
select_param = tool_parameters.get('select_param')
# 验证必要参数
if not required_param:
raise ValueError("required_param 是必需的参数")
# 参数类型转换和验证
try:
number_value = int(number_param)
except (TypeError, ValueError):
raise ValueError("number_param 必须是有效的数字")
# 继续业务逻辑...
```
### 3.2 返回成功结果
使用 self.create_json_message() 返回结构化数据：
```python
# 返回成功结果
result_data = {
"processed_items": 10,
"status": "completed",
"details": {"param1": required_param, "option": select_param}
}
yield self.create_json_message({
"success": True,
"message": "操作成功完成",
"data": result_data
})
```
返回格式建议：
- success: 操作是否成功 (boolean)
- message: 人类可读的消息 (string)
- data: 实际返回的数据 (any)
### 3.3 返回文本消息
如果需要返回纯文本：
```python
yield self.create_text_message("操作成功完成")
```
### 3.4 返回多条消息
可以多次使用 yield 返回多条消息：
```python
# 返回进度消息
yield self.create_text_message("开始处理数据...")
# 处理逻辑
processed_data = self.process_data(required_param)
# 返回最终结果
yield self.create_json_message({
"success": True,
"message": "数据处理完成",
"data": processed_data
})
```
## 4. 异常处理
### 4.1 参数验证异常
```python
# 验证必要参数
if not required_param:
raise ValueError("required_param 是必需的参数")
# 验证参数格式
if select_param not in ['option1', 'option2']:
raise ValueError("select_param 必须是有效的选项")
```
### 4.2 业务逻辑异常
```python
try:
# 业务逻辑
result = self._complex_operation(required_param, number_param)
except SpecificException as e:
# 处理特定异常
raise Exception(f"业务操作失败: {str(e)}")
except Exception as e:
# 处理其他异常
raise Exception(f"处理过程中发生错误: {str(e)}")
```
### 4.3 统一异常处理
```python
def invoke(self, toolparameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
try:
# 主要逻辑
# ...
except ValueError as e:
# 参数错误
raise Exception(f"参数错误: {str(e)}")
except Exception as e:
# 其他所有异常
raise Exception(f"工具执行失败: {str(e)}")
```
## 5. 完整示例
### 5.1 简单数据处理工具
```python
from collections.abc import Generator
from typing import Any
from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage
class DataProcessorTool(Tool):
def invoke(self, toolparameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
try:
# 获取参数
input_data = tool_parameters.get('input_data')
operation = tool_parameters.get('operation', 'uppercase')
repeat_times = tool_parameters.get('repeat_times', 1)
# 参数验证
if not input_data:
raise ValueError("input_data 是必需的参数")
if operation not in ['uppercase', 'lowercase', 'reverse']:
raise ValueError("operation 必须是 uppercase、lowercase 或 reverse")
# 业务逻辑
result = self._process_data(input_data, operation, repeat_times)
# 返回结果
yield self.create_json_message({
"success": True,
"message": f"数据处理完成，操作类型: {operation}",
"data": {
"original": input_data,
"processed": result,
"operation": operation,
"length": len(result)
}
})
except Exception as e:
raise Exception(f"数据处理工具执行失败: {str(e)}")
def processdata(self, data: str, operation: str, repeat_times: int) -> str:
"""处理数据的内部方法"""
result = data
if operation == 'uppercase':
result = data.upper()
elif operation == 'lowercase':
result = data.lower()
elif operation == 'reverse':
result = data[::-1]
return result * repeat_times
```
对应的 YAML 配置：
```yaml
identity:
name: "data_processor"
author: "developer"
label:
en_US: "Data Processor"
zh_Hans: "数据处理器"
description:
human:
en_US: "Process text data with various operations."
zh_Hans: "使用多种操作处理文本数据。"
llm: "A tool for processing text data with operations like uppercase, lowercase, and reverse."
parameters:
- name: input_data
type: string
required: true
label:
en_US: Input Data
zh_Hans: 输入数据
human_description:
en_US: "The text data to process"
zh_Hans: "要处理的文本数据"
llm_description: "The input text data that will be processed"
form: llm
- name: operation
type: select
required: false
default: "uppercase"
options:
- value: "uppercase"
label:
en_US: "Uppercase"
zh_Hans: "大写"
- value: "lowercase"
label:
en_US: "Lowercase"
zh_Hans: "小写"
- value: "reverse"
label:
en_US: "Reverse"
zh_Hans: "反转"
label:
en_US: "Operation Type"
zh_Hans: "操作类型"
human_description:
en_US: "The operation to perform on the data"
zh_Hans: "对数据执行的操作类型"
llm_description: "The operation type: uppercase, lowercase, or reverse"
form: llm
- name: repeat_times
type: number
required: false
default: 1
label:
en_US: "Repeat Times"
zh_Hans: "重复次数"
human_description:
en_US: "Number of times to repeat the processed result"
zh_Hans: "重复处理结果的次数"
llm_description: "How many times to repeat the processed result"
form: llm
extra:
python:
source: tools/data_processor.py
```
### 5.2 API 调用工具示例
```python
import requests
from collections.abc import Generator
from typing import Any
from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage
class APICallTool(Tool):
def invoke(self, toolparameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
try:
# 获取参数
url = tool_parameters.get('api_url')
method = tool_parameters.get('method', 'GET')
headers = tool_parameters.get('headers', {})
timeout = tool_parameters.get('timeout', 30)
# 参数验证
if not url:
raise ValueError("api_url 是必需的参数")
if method not in ['GET', 'POST', 'PUT', 'DELETE']:
raise ValueError("method 必须是 GET、POST、PUT 或 DELETE")
# 执行 API 调用
response = self._call_api(url, method, headers, timeout)
# 返回结果
yield self.create_json_message({
"success": True,
"message": f"API 调用成功，状态码: {response.status_code}",
"data": {
"status_code": response.status_code,
"headers": dict(response.headers),
"response": response.json() if response.content else None
}
})
except requests.exceptions.RequestException as e:
raise Exception(f"API 调用失败: {str(e)}")
except Exception as e:
raise Exception(f"API 工具执行失败: {str(e)}")
def callapi(self, url: str, method: str, headers: dict, timeout: int) -> requests.Response:
"""执行 API 调用的内部方法"""
response = requests.request(
method=method,
url=url,
headers=headers,
timeout=timeout
)
response.raise_for_status()
return response
```
## 6. 文件配置
在 extra 部分指定 Python 源文件位置：
```yaml
extra:
python:
source: tools/your_tool.py # 相对于插件根目录的路径
```
## 7. 开发最佳实践
1. 参数验证: 始终验证输入参数的有效性
2. 错误处理: 提供清晰具体的错误消息
3. 资源清理: 如果有资源需要清理，使用 try-finally 确保清理
4. 类型安全: 进行适当的类型检查和转换
5. 文档完整: 为所有参数提供完整的中英文文档
6. 性能考虑: 避免在工具中进行耗时操作，必要时提供进度反馈
## 8. 测试建议
- 测试正常流程和边界情况
- 测试参数缺失和无效的情况
- 测试异常处理流程
- 验证返回格式是否符合预期
通过以上教程，您可以快速掌握 Dify Plugin 的开发方法，创建自己的工具插件。