62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Callable
|
|
|
|
from app.tools.base import BaseTool
|
|
from app.tools.discover import ToolDiscovery
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ToolRegistry:
|
|
def __init__(self) -> None:
|
|
self._tools: dict[str, BaseTool] = {}
|
|
self._schemas: dict[str, dict[str, Any]] = {}
|
|
|
|
def register(self, tool: BaseTool) -> None:
|
|
self._tools[tool.name] = tool
|
|
|
|
def discover_and_init(
|
|
self,
|
|
init_factory: Callable[[dict], BaseTool] | None = None,
|
|
) -> None:
|
|
"""Discover tools from plugins and initialize them."""
|
|
discovery = ToolDiscovery()
|
|
discovered = discovery.discover()
|
|
|
|
for name, data in discovered.items():
|
|
manifest = data.get("manifest", {})
|
|
|
|
if init_factory:
|
|
tool = init_factory({"name": name, "manifest": manifest})
|
|
else:
|
|
tool_instance = data.get("instance")
|
|
if tool_instance:
|
|
self._tools[name] = tool_instance
|
|
self._schemas[name] = {
|
|
"description": manifest.get("description", ""),
|
|
"args_schema": manifest.get("args_schema", {}),
|
|
"requires_permission": manifest.get("requires_permission", False),
|
|
}
|
|
logger.info(f"Registered tool: {name}")
|
|
logger.warning(f"No init_factory provided for {name}")
|
|
|
|
def get(self, name: str) -> BaseTool:
|
|
if name not in self._tools:
|
|
raise KeyError(f"Tool {name} is not registered")
|
|
return self._tools[name]
|
|
|
|
def list_names(self) -> list[str]:
|
|
return list(self._tools.keys())
|
|
|
|
def get_schema(self, name: str) -> dict[str, Any]:
|
|
return self._schemas.get(name, {})
|
|
|
|
def list_schemas(self) -> list[dict[str, Any]]:
|
|
return [
|
|
{"name": name, **schema}
|
|
for name, schema in self._schemas.items()
|
|
]
|
|
|