backend/src/tools/registry_loader.py
"""Tool Registry loader — YAML → TOOL_CLASS_MAP → constructed Tool instances. ``class:`` values resolve against an explicit hand-maintained map — never ``importlib`` or filesystem scanning (tools-design.md §4). """ from __future__ import annotations from pathlib import Path from typing import Any import yaml from config.paths import resolve_repo_path from config.schemas import ToolRegistryEntry, ToolsRegistryConfig from config.settings import Settings, get_settings from observability import log_event from tools.base_tool import BaseTool from tools.company_profile_tool import CompanyProfileTool from tools.financial_metrics_tool import FinancialMetricsTool from tools.news_tool import NewsTool from tools.sec_filings_tool import SECFilingsTool TOOL_CLASS_MAP: dict[str, type[BaseTool]] = { "CompanyProfileTool": CompanyProfileTool, "FinancialMetricsTool": FinancialMetricsTool, "NewsTool": NewsTool, "SECFilingsTool": SECFilingsTool, } class ToolRegistryError(RuntimeError): """Raised when the tools registry cannot be loaded or a class cannot be resolved.""" class ToolRegistry: """In-memory map of registry tool name → constructed ``BaseTool`` instance.""" def __init__(self, tools: dict[str, BaseTool]) -> None: self._tools = dict(tools) def get(self, name: str) -> BaseTool: try: return self._tools[name] except KeyError as exc: raise ToolRegistryError(f"Unknown tool '{name}'") from exc def names(self) -> list[str]: return sorted(self._tools.keys()) def __contains__(self, name: str) -> bool: return name in self._tools def __len__(self) -> int: return len(self._tools) def resolve_tool_class(class_name: str) -> type[BaseTool]: """Resolve a registry ``class:`` value via ``TOOL_CLASS_MAP`` (fail fast).""" try: return TOOL_CLASS_MAP[class_name] except KeyError as exc: known = ", ".join(sorted(TOOL_CLASS_MAP)) or "(none)" raise ToolRegistryError( f"Unknown tool class '{class_name}'. " f"Add it to TOOL_CLASS_MAP in tools/registry_loader.py. Known: {known}" ) from exc def _provider_source(settings: Settings, config_key: str) -> str: if not hasattr(settings, config_key): raise ToolRegistryError( f"Tool config_key '{config_key}' is not a Settings field. " f"Add it to backend/src/config/settings.py." ) value = getattr(settings, config_key) if value is None or (isinstance(value, str) and not value.strip()): return "mock" return str(value) def _build_tool(entry: ToolRegistryEntry, settings: Settings) -> BaseTool: tool_cls = resolve_tool_class(entry.class_) source = _provider_source(settings, entry.config_key) # Phase 1: all providers default to mock clients; ``source`` labels ToolResult. return tool_cls(source=source) def load_tools_config(registry_path: Path | None = None) -> ToolsRegistryConfig: """Load and schema-validate ``config/tools/registry.yaml``.""" settings = get_settings() path = registry_path or resolve_repo_path(settings.tools_registry_path) log_event( "tools_registry_load", component="ToolRegistry", fields={"path": str(path)}, ) try: with path.open(encoding="utf-8") as handle: data: Any = yaml.safe_load(handle) except OSError as exc: raise ToolRegistryError(f"Cannot read tools registry '{path}': {exc}") from exc except yaml.YAMLError as exc: raise ToolRegistryError(f"Invalid YAML in tools registry '{path}': {exc}") from exc if data is None: data = {} try: return ToolsRegistryConfig.model_validate(data) except Exception as exc: raise ToolRegistryError( f"Schema validation failed for tools registry '{path}': {exc}" ) from exc def load_tool_registry( settings: Settings | None = None, *, registry_path: Path | None = None, ) -> ToolRegistry: """Build a ``ToolRegistry`` from YAML + ``TOOL_CLASS_MAP`` + Settings ``config_key``s.""" resolved_settings = settings or get_settings() path = registry_path or resolve_repo_path(resolved_settings.tools_registry_path) config = load_tools_config(path) tools: dict[str, BaseTool] = {} for name, entry in config.tools.items(): tool = _build_tool(entry, resolved_settings) if tool.name != name: raise ToolRegistryError( f"Registry key '{name}' does not match Tool.name '{tool.name}' " f"for class '{entry.class_}'." ) tools[name] = tool log_event( "tools_registry_loaded", component="ToolRegistry", fields={"tool_count": len(tools), "tools": sorted(tools.keys())}, ) return ToolRegistry(tools)
Follow the work.
Occasional updates on SignalFoundry, MarketCompass, and what we are building at CompassFoundry Labs.