"""Tool registration and schema export.""" from __future__ import annotations import logging from pathlib import Path from typing import TYPE_CHECKING, Any from app.models.config import AppConfig from app.tools.base import BaseTool if TYPE_CHECKING: from app.services.skills import SkillsManager logger = logging.getLogger(__name__) class ToolRegistry: """Registry of available tools, keyed by name.""" def __init__(self) -> None: self._tools: dict[str, BaseTool] = {} self._disabled: set[str] = set() def register(self, tool: BaseTool) -> None: """Register a tool instance. Raises ValueError on duplicate name.""" if tool.name in self._tools: raise ValueError(f"Duplicate tool name: '{tool.name}'") self._tools[tool.name] = tool logger.debug("Registered tool: %s", tool.name) def get(self, name: str) -> BaseTool | None: """Look up a tool by name. Returns None if disabled or not found.""" if name in self._disabled: return None return self._tools.get(name) def get_all(self) -> dict[str, BaseTool]: """Return all registered tools (excluding disabled).""" return {k: v for k, v in self._tools.items() if k not in self._disabled} def get_openai_tools_schema(self) -> list[dict[str, Any]]: """Return OpenAI function-calling schemas for all active tools.""" return [ tool.get_openai_schema() for tool in self._tools.values() if tool.name not in self._disabled ] def apply_filter( self, *, enable: list[str] | None = None, disable: list[str] | None = None, ) -> set[str]: """Apply a tool filter, returning the previous disabled set for restoration. Args: enable: If set, only these tools (plus always-on tools) are available. disable: Specific tools to disable. Returns: The previous disabled set (snapshot for restore). """ previous = set(self._disabled) if enable is not None: # Whitelist mode: disable everything not in the enable list self._disabled = {name for name in self._tools if name not in enable} elif disable is not None: # Blacklist mode: add to existing disabled set (preserves global disables) self._disabled = set(self._disabled) | set(disable) else: self._disabled = set() return previous def restore_filter(self, previous: set[str]) -> None: """Restore a previous filter state.""" self._disabled = previous def all_tool_names(self) -> list[str]: """Return all registered tool names (including disabled).""" return list(self._tools.keys()) def create_default_registry( workspace_root: Path, config: AppConfig, skills_manager: SkillsManager | None = None, skill_runner: object | None = None, ) -> ToolRegistry: """Create a ToolRegistry populated with all built-in tools. Args: workspace_root: Workspace root path. config: Application configuration. skills_manager: Optional skills manager for skill tools. skill_runner: Optional SkillRunner for package skill activation. """ # Read tools from app.tools.filesystem import ListDirTool, ReadFileTool, ReadManyFilesTool # Write tools from app.tools.filesystem import DeleteFileTool, MakeDirTool, WriteFileTool # Edit tools from app.tools.edit import PatchApplyTool, StrReplaceTool # Shell tools from app.tools.shell import RunCommandTool # Control flow from app.tools.finish import FinishTool # Search tools from app.tools.search import FindFilesTool, GrepFilesTool registry = ToolRegistry() # Read registry.register(ReadFileTool(workspace_root, config)) registry.register(ReadManyFilesTool(workspace_root, config)) registry.register(ListDirTool(workspace_root, config)) # Search registry.register(GrepFilesTool(workspace_root, config)) registry.register(FindFilesTool(workspace_root, config)) # Write registry.register(WriteFileTool(workspace_root, config)) registry.register(MakeDirTool(workspace_root, config)) registry.register(DeleteFileTool(workspace_root, config)) # Edit registry.register(StrReplaceTool(workspace_root, config)) registry.register(PatchApplyTool(workspace_root, config)) # Shell registry.register(RunCommandTool(workspace_root, config)) # Control flow registry.register(FinishTool(workspace_root, config)) # Skills (conditional) if skills_manager is not None: from app.services.skill_runner import SkillRunner as SkillRunnerType from app.tools.skills import FinishSkillTool, LoadSkillTool runner = skill_runner if isinstance(skill_runner, SkillRunnerType) else None registry.register(LoadSkillTool(workspace_root, config, skills_manager, runner)) registry.register(FinishSkillTool(workspace_root, config, runner)) return registry