Dify プラグイン開発：プロンプト

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
  ja: あなたのプラグイン表示名
description:
  en_US: Detailed description of your plugin functionality
  ja: プラグイン機能の詳細な説明
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
    ja: あなたのプラグイン表示名
  description:
    en_US: Detailed description of your plugin functionality
    ja: プラグイン機能の詳細な説明
  icon: icon.svg
credentials_for_provider:  # APIキーなどの認証情報が必要な場合のみ追加
  api_key:
    type: secret-input
    required: true
    label:
      en_US: API Key
      ja: APIキー
    placeholder:
      en_US: Enter your API key
      ja: APIキーを入力してください
    help:
      en_US: How to get your API key
      ja: 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
    ja: 機能表示名
description:
  human:  # 人間ユーザー向けの説明
    en_US: Description for human users
    ja: ユーザー向け機能説明
  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
      ja: パラメータ表示名
    human_description:
      en_US: Parameter description for users
      ja: ユーザー向けパラメータ説明
    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
    ja: TOTP 検証ツール
description:
  human:
    en_US: Time-based one-time password (TOTP) validator
    ja: 時間ベースのワンタイムパスワード (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
      ja: TOTP シークレットキーまたはプロビジョニングURI
    human_description:
      en_US: The secret key or provisioning URI used to generate the TOTP
      ja: 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
      ja: 検証する6桁のTOTPコード
    human_description:
      en_US: 6 digit TOTP code to validate
      ja: 検証する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

# 1ファイルに1つのToolサブクラスのみを含む
class TotpTool(Tool):
    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage, None, None]:
        """時間ベースのワンタイムパスワード(TOTP)を検証します"""
        # パラメータを取得、KeyErrorを避けるためにget()を使用
        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

# 注意：1ファイルに1つの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