Plugin API

Plugin

class Plugin:
    """Abstract base class for plugins."""

    def __init__(self, manifest: PluginManifest) -> None:
        self._manifest = manifest

    @property
    def manifest(self) -> PluginManifest:
        return self._manifest

生命周期方法

    async def on_worker_startup(self) -> None:
        """Worker 启动时调用"""

    async def on_worker_shutdown(self) -> None:
        """Worker 关闭时调用"""

    async def on_task_start(self, context: AgentContext) -> None:
        """任务开始时调用"""

    async def on_task_complete(
        self,
        context: AgentContext,
        result: Any,
    ) -> None:
        """任务成功完成时调用"""

    async def on_task_error(
        self,
        context: AgentContext,
        error: Exception,
    ) -> None:
        """任务出错时调用"""

    async def on_task_cancel(self, context: AgentContext) -> None:
        """任务取消时调用"""

    async def register_agent_configs(
        self,
        build_context: PluginBuildContext,
    ) -> list[AgentConfig]:
        """注册 Agent 配置"""

PluginManifest

@dataclass
class PluginManifest:
    plugin_id: str
    version: str
    name: Optional[str] = None
    description: Optional[str] = None

PluginBuildContext

class PluginBuildContext:
    """Plugin build context."""

    @property
    def redis_client(self) -> Redis:
        """获取 Redis 客户端"""

AgentConfig

@dataclass
class AgentConfig:
    agent_id: str
    tools: Optional[Dict[str, Callable]] = None
    prompts: Optional[Dict[str, str]] = None
    skills: Optional[List[str]] = None

PluginRegistry

class PluginRegistry:
    def register_bundle(self, plugin: Plugin) -> None:
        """注册插件"""

    def get_plugin(self, plugin_id: str) -> Optional[Plugin]:
        """获取插件"""

    def list_plugins(self) -> List[Plugin]:
        """列出所有插件"""

    def build_agent_configs(self) -> Dict[str, AgentConfig]:
        """构建所有 Agent 配置"""