Skip to content
← How we build

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.

No spam. Unsubscribe anytime.