Dify 插件开发：Prompt

your_plugin/
├── _assets/             # 图标和视觉资源
├── provider/            # 提供者定义和验证
│   ├── your_plugin.py   # 凭证验证逻辑
│   └── your_plugin.yaml # 提供者配置
├── tools/               # 工具实现
│   ├── feature_one.py   # 工具功能实现
│   ├── feature_one.yaml # 工具参数和描述
│   ├── feature_two.py   # 另一个工具实现
│   └── feature_two.yaml # 另一个工具配置
├── utils/               # 辅助函数
│   └── helpers.py       # 通用功能逻辑
├── working/             # 进度记录和工作文件
├── .env.example         # 环境变量模板
├── main.py              # 入口文件
├── manifest.yaml        # 插件主配置
├── README.md            # 文档
└── requirements.txt     # 依赖列表

./dify plugin init

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS/Linux:
source venv/bin/activate

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

# 复制并编辑环境变量
cp .env.example .env

# 启动本地服务
python -m main

# 打包插件
./dify plugin package ./your_plugin_dir

your_plugin/
├── _assets/             # 图标和视觉资源
├── provider/            # 提供者定义和验证
│   ├── your_plugin.py   # 凭证验证逻辑
│   └── your_plugin.yaml # 提供者配置
├── tools/               # 工具实现
│   ├── your_plugin.py   # 工具功能实现
│   └── your_plugin.yaml # 工具参数和描述
├── utils/               # (可选) 辅助函数
├── working/             # 进度记录和工作文件
├── .env.example         # 环境变量模板
├── main.py              # 入口文件
├── manifest.yaml        # 插件主配置
├── README.md            # 文档
└── requirements.txt     # 依赖列表

version: 0.0.1
type: plugin
author: your_name
name: your_plugin_name  # 不要修改此字段
label:
  en_US: Your Plugin Display Name
  zh_Hans: 你的插件显示名称
description:
  en_US: Detailed description of your plugin functionality
  zh_Hans: 插件功能的详细描述
icon: icon.svg
resource:
  memory: 268435456  # 256MB
  permission: {}
plugins:
  tools:
    - provider/your_plugin.yaml
meta:
  version: 0.0.1
  arch:
    - amd64
    - arm64
  runner:
    language: python
    version: "3.12"
    entrypoint: main
created_at: 2025-04-19T00:00:00.000000+08:00
privacy: PRIVACY.md
tags:
  - utilities  # 只使用预定义的标签

identity:
  author: your_name
  name: your_plugin_name  # 不要修改此字段
  label:
    en_US: Your Plugin Display Name
    zh_Hans: 你的插件显示名称
  description:
    en_US: Detailed description of your plugin functionality
    zh_Hans: 插件功能的详细描述
  icon: icon.svg
credentials_for_provider:  # 仅在需要API密钥等凭证时添加
  api_key:
    type: secret-input
    required: true
    label:
      en_US: API Key
      zh_Hans: API密钥
    placeholder:
      en_US: Enter your API key
      zh_Hans: 输入你的API密钥
    help:
      en_US: How to get your API key
      zh_Hans: 如何获取API密钥
    url: https://example.com/get-api-key
tools:  # 工具列表，添加新工具时在此更新
  - tools/feature_one.yaml
  - tools/feature_two.yaml
extra:
  python:
    source: provider/your_plugin.py

identity:
  name: feature_name  # 与文件名对应
  author: your_name
  label:
    en_US: Feature Display Name
    zh_Hans: 功能显示名称
description:
  human:  # 给人类用户看的描述
    en_US: Description for human users
    zh_Hans: 面向用户的功能描述
  llm: Description for AI models to understand when to use this tool.  # 给AI看的描述
parameters:  # 参数定义
  - name: param_name
    type: string  # string, number, boolean, file等
    required: true
    label:
      en_US: Parameter Display Name
      zh_Hans: 参数显示名称
    human_description:
      en_US: Parameter description for users
      zh_Hans: 面向用户的参数描述
    llm_description: Detailed parameter description for AI models
    form: llm  # llm表示可由AI从用户输入中提取，form表示需要在UI中配置
  # 其他参数...
extra:
  python:
    source: tools/feature.py  # 对应的Python实现文件
# 可选：定义输出的JSON Schema
output_schema:
  type: object
  properties:
    result:
      type: string
      description: Description of the result

from collections.abc import Generator
from typing import Any

from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage

# 导入工具函数，确保函数名称拼写正确
from utils.helpers import process_data

class FeatureTool(Tool):
    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage, None, None]:
        try:
            # 1. 获取必需参数
            param = tool_parameters.get("param_name", "")
            
            # 2. 获取可选参数 - 使用try-except方式
            try:
                optional_param1 = tool_parameters["optional_param1"]
                optional_param2 = tool_parameters["optional_param2"]
            except KeyError:
                optional_param1 = None
                optional_param2 = None
                
            # 另一种可选参数获取方式 - 使用.get()方法
            another_optional = tool_parameters.get("another_optional")  # 如果不存在返回None
            
            # 3. 验证必需参数
            if not param:
                yield self.create_text_message("Parameter is required.")
                return
                
            # 4. 实现业务逻辑
            result = self._process_data(param, optional_param1, optional_param2)
            
            # 5. 返回结果
            # 文本输出
            yield self.create_text_message(f"Processed result: {result}")
            # JSON输出
            yield self.create_json_message({"result": result})
            # 变量输出 (用于工作流)
            yield self.create_variable_message("result_var", result)
                
        except Exception as e:
            # 错误处理
            yield self.create_text_message(f"Error: {str(e)}")
            
    def _process_data(self, param: str, opt1=None, opt2=None) -> str:
        """
        实现具体的业务逻辑
        
        Args:
            param: 必需的参数
            opt1: 可选参数1
            opt2: 可选参数2
            
        Returns:
            处理结果
        """
        # 根据参数是否存在执行不同的逻辑
        if opt1 and opt2:
            return f"Processed with all options: {param}, {opt1}, {opt2}"
        elif opt1:
            return f"Processed with option 1: {param}, {opt1}"
        elif opt2:
            return f"Processed with option 2: {param}, {opt2}"
        else:
            return f"Processed basic: {param}"

import requests
from typing import Dict, Any, Optional

def call_external_api(endpoint: str, params: Dict[str, Any], api_key: str) -> Dict[str, Any]:
    """
    调用外部API的通用函数
    
    Args:
        endpoint: API端点URL
        params: 请求参数
        api_key: API密钥
        
    Returns:
        API响应的JSON数据
    
    Raises:
        Exception: 如果API调用失败
    """
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.get(endpoint, params=params, headers=headers, timeout=10)
        response.raise_for_status()  # 如果状态码不是200，抛出异常
        return response.json()
    except requests.RequestException as e:
        raise Exception(f"API调用失败: {str(e)}")

dify_plugin~=0.0.1b76
requests~=2.31.0
# 其他依赖...

# 复制工具YAML文件作为模板
cp tools/existing_tool.yaml tools/new_tool.yaml

# 复制工具Python实现
cp tools/existing_tool.py tools/new_tool.py

# 同理适用于provider文件
cp provider/existing.yaml provider/new.yaml

tools/
├── search.py          # 搜索功能
├── search.yaml
├── create.py          # 创建功能
├── create.yaml
├── update.py          # 更新功能
├── update.yaml
├── delete.py          # 删除功能
└── delete.yaml

try:
    # 尝试执行操作
    result = some_operation()
    yield self.create_text_message("操作成功")
except ValueError as e:
    # 参数错误
    yield self.create_text_message(f"参数错误: {str(e)}")
except requests.RequestException as e:
    # API调用错误
    yield self.create_text_message(f"API调用失败: {str(e)}")
except Exception as e:
    # 其他未预期错误
    yield self.create_text_message(f"发生错误: {str(e)}")

# 在工具实现中
from utils.api_client import ApiClient

class SearchTool(Tool):
    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage, None, None]:
        client = ApiClient(self.runtime.credentials["api_key"])
        results = client.search(tool_parameters["query"])
        yield self.create_json_message(results)

# 文本输出
yield self.create_text_message("这是文本消息")

# JSON输出
yield self.create_json_message({"key": "value"})

# 链接输出
yield self.create_link_message("https://example.com")

# 变量输出 (用于工作流)
yield self.create_variable_message("variable_name", "variable_value")

import pyotp
import time

def verify_totp(secret_key, totp_code, offset=5, strict=False):
    """
    验证基于时间的一次性密码(TOTP)。
    
    Args:
        secret_key: 用于生成TOTP的密钥或配置URL
        totp_code: 用户提交的动态令牌
        offset: 允许提前或延迟验证的秒数
        strict: 是否使用严格验证(仅在精确匹配时返回成功)
        
    Returns:
        包含以下内容的字典:
        - 'status': 'success' 或 'fail'
        - 'detail': 内部消息(不面向终端用户)
    """
    try:
        # 检测是否为配置URL
        if secret_key.startswith('otpauth://'):
            totp = pyotp.parse_uri(secret_key)
        else:
            totp = pyotp.TOTP(secret_key)
            
        current_time = time.time()

        # 精确时间验证
        if totp.verify(totp_code):
            return {'status': 'success', 'detail': 'Token is valid'}

        # 偏移验证
        early_valid = totp.verify(totp_code, for_time=current_time + offset)
        late_valid = totp.verify(totp_code, for_time=current_time - offset)
        off_time_valid = early_valid or late_valid

        detail_message = (
            f"Token is valid but not on time. "
            f"{'Early' if early_valid else 'Late'} within {offset} seconds"
            if off_time_valid else
            "Token is invalid"
        )

        if strict:
            return {'status': 'fail', 'detail': detail_message}
        else:
            return (
                {'status': 'success', 'detail': detail_message}
                if off_time_valid
                else {'status': 'fail', 'detail': detail_message}
            )
    except Exception as e:
        return {'status': 'fail', 'detail': f'Verification error: {str(e)}'}

identity:
  name: totp
  author: alterxyz
  label:
    en_US: TOTP Validator
    zh_Hans: TOTP 验证器
description:
  human:
    en_US: Time-based one-time password (TOTP) validator
    zh_Hans: 基于时间的一次性密码 (TOTP) 验证器
  llm: Time-based one-time password (TOTP) validator, this tool is used to validate a 6 digit TOTP code with a secret key or provisioning URI.
parameters:
  - name: secret_key
    type: string
    required: true
    label:
      en_US: TOTP secret key or provisioning URI
      zh_Hans: TOTP 私钥或 URI
    human_description:
      en_US: The secret key or provisioning URI used to generate the TOTP
      zh_Hans: 用于生成 TOTP 的私钥或 URI
    llm_description: The secret key or provisioning URI (starting with 'otpauth://') used to generate the TOTP, this is highly sensitive and should be kept secret.
    form: llm
  - name: user_code
    type: string
    required: true
    label:
      en_US: 6 digit TOTP code to validate
      zh_Hans: 要验证的 6 位 TOTP 代码
    human_description:
      en_US: 6 digit TOTP code to validate
      zh_Hans: 要验证的 6 位 TOTP 代码
    llm_description: 6 digit TOTP code to validate
    form: llm
extra:
  python:
    source: tools/totp.py
output_schema:
  type: object
  properties:
    True_or_False:
      type: string
      description: Whether the TOTP is valid or not, return in string format, "True" or "False".

from collections.abc import Generator
from typing import Any

# 正确导入工具函数
from utils.totp_verify import verify_totp

from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage

# 一个文件只包含一个Tool子类
class TotpTool(Tool):
    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage, None, None]:
        """验证基于时间的一次性密码(TOTP)"""
        # 获取参数，使用get()避免KeyError
        secret_key = tool_parameters.get("secret_key")
        totp_code = tool_parameters.get("user_code")
        
        # 参数验证
        if not secret_key:
            yield self.create_text_message("Error: Secret key is required.")
            return
        if not totp_code:
            yield self.create_text_message("Error: TOTP code is required.")
            return
        
        try:
            # 调用工具函数
            result = verify_totp(secret_key, totp_code)
            
            # 返回结果
            yield self.create_json_message(result)
            
            # 基于验证结果返回不同的消息
            if result["status"] == "success":
                yield self.create_text_message("Valid")
                yield self.create_variable_message("True_or_False", "True")
            else:
                yield self.create_text_message("Invalid")
                yield self.create_variable_message("True_or_False", "False")
                
        except Exception as e:
            # 错误处理
            yield self.create_text_message(f"Verification error: {str(e)}")

from collections.abc import Generator
from typing import Any

import pyotp

from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage

# 注意：一个文件只包含一个Tool子类
class SecretGenerator(Tool):
    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage, None, None]:
        """生成TOTP密钥"""
        try:
            # 生成随机密钥
            secret_key = pyotp.random_base32()
            yield self.create_text_message(secret_key)
            
            # 安全获取可选参数
            name = tool_parameters.get("name")
            issuer_name = tool_parameters.get("issuer_name")
            
            # 如果提供了名称或发行方，生成配置URI
            if name or issuer_name:
                provisioning_uri = pyotp.totp.TOTP(secret_key).provisioning_uri(
                    name=name, 
                    issuer_name=issuer_name
                )
                yield self.create_variable_message("provisioning_uri", provisioning_uri)
                
        except Exception as e:
            yield self.create_text_message(f"Error generating secret: {str(e)}")

dify_plugin~=0.0.1b76
pyotp~=2.9.0