ベストプラクティス

Result 型を一貫して使用する

エントリーポイントでは例外を発生させる代わりに、常に Ok/Err を返してください：

from plugin.sdk.plugin import Ok, Err, SdkError

@plugin_entry(id="process")
def process(self, data: str, **_):
    if not data:
        return Err(SdkError("data is required"))

    try:
        result = self._do_work(data)
        return Ok({"result": result})
    except ValueError as e:
        return Err(SdkError(f"Validation error: {e}"))
    except Exception as e:
        self.logger.exception(f"Unexpected error: {e}")
        return Err(SdkError(f"Internal error"))

コード構成

初期化、ヘルパー、パブリックエントリーポイントを分離してください：

@neko_plugin
class WellOrganizedPlugin(NekoPluginBase):
    def __init__(self, ctx):
        super().__init__(ctx)
        self._initialize()

    # --- ライフサイクル ---
    @lifecycle(id="startup")
    def on_startup(self, **_):
        return Ok({"status": "ready"})

    # --- プライベートヘルパー ---
    def _initialize(self):
        """リソースのセットアップ。"""
        pass

    def _validate(self, data):
        """内部バリデーション。"""
        pass

    # --- パブリックエントリーポイント ---
    @plugin_entry(id="process")
    def process(self, data: str, **_):
        self._validate(data)
        return Ok({"result": self._do_work(data)})

ロギング

適切なログレベルを使用してください：

レベル	使用場面
debug	詳細な診断情報
info	通常動作のマイルストーン
warning	予期しないが処理された状況
error	注意が必要なエラー
exception	スタックトレース付きエラー

self.logger.debug(f"Processing item {item_id}")
self.logger.info(f"Plugin started successfully")
self.logger.warning(f"Retry attempt {attempt}/3")
self.logger.error(f"Failed to connect: {err}")
self.logger.exception(f"Unexpected error in process()")

ステータス更新

長時間実行される操作中は進捗を報告してください：

@plugin_entry(id="batch_job")
def batch_job(self, items: list, **_):
    total = len(items)
    for i, item in enumerate(items):
        self._process(item)
        self.report_status({
            "status": "processing",
            "progress": (i + 1) / total * 100,
            "message": f"Processing {i+1}/{total}"
        })

    self.report_status({"status": "completed", "progress": 100})
    return Ok({"processed": total})

入力バリデーション

自動 JSON Schema バリデーションのために input_schema を使用するか、Pydantic モデルのために params を使用してください：

# オプション A: JSON Schema
@plugin_entry(
    id="validated",
    input_schema={
        "type": "object",
        "properties": {
            "email": {"type": "string", "format": "email"},
            "age": {"type": "integer", "minimum": 0, "maximum": 150}
        },
        "required": ["email", "age"]
    }
)
def validated(self, email: str, age: int, **_):
    return Ok({"email": email, "age": age})

# オプション B: Pydantic モデル（スキーマを自動生成）
from pydantic import BaseModel, Field

class UserInput(BaseModel):
    email: str = Field(..., description="User email")
    age: int = Field(..., ge=0, le=150)

@plugin_entry(id="validated_v2", params=UserInput)
def validated_v2(self, email: str, age: int, **_):
    return Ok({"email": email, "age": age})

作業ディレクトリ

ハードコードされたパスの代わりに self.config_dir と self.data_path() を使用してください：

# プラグインディレクトリ（plugin.toml がある場所）
config_file = self.config_dir / "config.json"

# データディレクトリ（自動作成されるサブディレクトリ）
db_path = self.data_path("cache.db")       # → <plugin_dir>/data/cache.db
logs_dir = self.data_path("logs")          # → <plugin_dir>/data/logs/

プラグイン間呼び出しのエラーハンドリング

他のプラグインを呼び出す際は常に Err を処理してください：

@plugin_entry(id="orchestrate")
async def orchestrate(self, **_):
    # まず依存関係を確認
    dep = await self.plugins.require_enabled("dependency_plugin")
    if isinstance(dep, Err):
        return Err(SdkError("Required plugin 'dependency_plugin' is not available"))

    # 呼び出しを実行
    result = await self.plugins.call_entry("dependency_plugin:do_work", {"key": "val"})
    if isinstance(result, Err):
        self.logger.error(f"Cross-plugin call failed: {result.error}")
        return Err(SdkError("Dependency call failed"))

    return Ok({"combined": result.value})

グレースフルシャットダウン

shutdown ライフサイクルでリソースをクリーンアップしてください：

@lifecycle(id="shutdown")
async def on_shutdown(self, **_):
    # ネットワーク接続を閉じる
    if self.session:
        await self.session.close()

    # 保留中のデータをフラッシュ
    await self.store.flush()

    # タイマーをキャンセル（自動的に処理されますが、ログに記録）
    self.logger.info("Plugin shutting down gracefully")
    return Ok({"status": "stopped"})

プラグインチェックリスト

プラグインをリリースする前に確認してください：

• [ ] すべてのエントリーポイントが Ok/Err を返している（生の dict や例外ではなく）
• [ ] @lifecycle(id="startup") と @lifecycle(id="shutdown") が実装されている
• [ ] パラメーターを受け取るすべてのエントリーポイントに input_schema が定義されている
• [ ] すべてのエントリーポイントのシグネチャに **_ が含まれている
• [ ] print() の代わりにロガーが使用されている
• [ ] タイマーを使用する場合、共有状態がロックで保護されている
• [ ] プラグイン間呼び出しが Err 結果を処理している
• [ ] plugin.toml に正しい entry パスと SDK バージョン制約が設定されている