83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PLUGINS_DIR = Path(__file__).parent / "plugins"
|
|
|
|
|
|
class ToolDiscovery:
|
|
"""Decentralized tool discovery system."""
|
|
|
|
def __init__(self, plugins_dir: Path | None = None) -> None:
|
|
self._plugins_dir = plugins_dir or PLUGINS_DIR
|
|
|
|
def discover(self) -> dict[str, Any]:
|
|
"""Discover all tools from plugins directory."""
|
|
tools = {}
|
|
|
|
if not self._plugins_dir.exists():
|
|
logger.warning(f"Plugins directory not found: {self._plugins_dir}")
|
|
return tools
|
|
|
|
for folder in self._plugins_dir.iterdir():
|
|
if not folder.is_dir():
|
|
continue
|
|
|
|
manifest_file = folder / "manifest.json"
|
|
if not manifest_file.exists():
|
|
logger.warning(f"Missing manifest.json in {folder.name}")
|
|
continue
|
|
|
|
try:
|
|
manifest = self._load_manifest(manifest_file)
|
|
|
|
tool_name = manifest.get("name", folder.name)
|
|
tools[tool_name] = {
|
|
"manifest": manifest,
|
|
"tool_class": folder.name,
|
|
}
|
|
logger.info(f"Discovered tool: {tool_name}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load tool {folder.name}: {e}")
|
|
continue
|
|
|
|
return tools
|
|
|
|
def _load_manifest(self, manifest_file: Path) -> dict[str, Any]:
|
|
with open(manifest_file) as f:
|
|
return json.load(f)
|
|
|
|
def _load_tool_class(self, tool_name: str, manifest: dict[str, Any]) -> Any:
|
|
entrypoint = manifest.get("entrypoint", "Tool")
|
|
module = importlib.import_module(f"app.tools.plugins.{tool_name}")
|
|
tool_class = getattr(module, entrypoint)
|
|
return tool_class
|
|
|
|
def get_tool_schemas(self) -> list[dict[str, Any]]:
|
|
"""Get schemas for all discovered tools."""
|
|
tools = self.discover()
|
|
schemas = []
|
|
|
|
for name, data in tools.items():
|
|
manifest = data.get("manifest", {})
|
|
schemas.append({
|
|
"name": name,
|
|
"description": manifest.get("description", ""),
|
|
"args_schema": manifest.get("args_schema", {}),
|
|
"requires_permission": manifest.get("requires_permission", False),
|
|
})
|
|
|
|
return schemas
|
|
|
|
|
|
def discover_tools() -> dict[str, Any]:
|
|
"""Convenience function for quick tool discovery."""
|
|
discovery = ToolDiscovery()
|
|
return discovery.discover() |