Source code for tool_loader

"""Auto-discover and load tools from a directory of Python scripts.

Each ``.py`` file in the tools directory must expose **either**:

**Single-tool format** (one tool per file):

* ``TOOL_NAME`` -- unique name for the tool (str).
* ``TOOL_DESCRIPTION`` -- human-readable description (str).
* ``TOOL_PARAMETERS`` -- JSON Schema ``object`` for accepted args (dict).
* ``async def run(**kwargs) -> str`` -- the tool handler.

**Multi-tool format** (multiple tools per file):

* ``TOOLS`` -- a list of dicts, each with keys:
  ``name``, ``description``, ``parameters``, ``handler``.

Malformed files are logged as warnings but do **not** prevent the rest of
the tools from loading.
"""

from __future__ import annotations

import importlib.util
import logging
import sys
from pathlib import Path

from tools import ToolDefinition, ToolRegistry

logger = logging.getLogger(__name__)

_REQUIRED_ATTRS = ("TOOL_NAME", "TOOL_DESCRIPTION", "TOOL_PARAMETERS", "run")




def load_tools(directory: str | Path, registry: ToolRegistry) -> None:
    """Scan *directory* for ``.py`` tool scripts and register them.

    Parameters
    ----------
    directory:
        Path to the tools directory (e.g. ``"tools"``).
    registry:
        The :class:`ToolRegistry` to register discovered tools into.
    """
    tools_path = Path(directory)
    if not tools_path.is_dir():
        logger.warning(
            "Tools directory does not exist: %s – skipped",
            tools_path,
        )
        return

    for py_file in sorted(tools_path.glob("*.py")):
        if py_file.name.startswith("_"):
            continue  # skip __init__.py, __pycache__ artefacts, etc.

        module_name = f"_tool_{py_file.stem}"
        try:
            spec = importlib.util.spec_from_file_location(module_name, py_file)
            if spec is None or spec.loader is None:
                logger.warning(
                    "Could not create module spec for %s",
                    py_file,
                )
                continue

            canonical = f"tools.{py_file.stem}"
            if canonical in sys.modules:
                module = sys.modules[canonical]
                sys.modules[module_name] = module  # alias only, no re-exec
            else:
                module = importlib.util.module_from_spec(spec)
                sys.modules[module_name] = module
                spec.loader.exec_module(module)

            # --- Multi-tool format: TOOLS list ---------------------------
            tools_list = getattr(module, "TOOLS", None)
            if tools_list and isinstance(tools_list, list):
                count = 0
                for entry in tools_list:
                    t_name = entry.get("name")
                    t_desc = entry.get("description")
                    t_params = entry.get("parameters")
                    t_handler = entry.get("handler")
                    valid = (
                        t_name and t_desc
                        and t_params is not None
                        and callable(t_handler)
                    )
                    if not valid:
                        logger.warning(
                            "Tool %s: malformed TOOLS entry %r – skip",
                            py_file.name, entry.get("name", "?"),
                        )
                        continue
                    registry._tools[t_name] = ToolDefinition(
                        name=t_name,
                        description=t_desc,
                        parameters=t_params,
                        handler=t_handler,
                        no_background=bool(entry.get("no_background")),
                        allow_repeat=bool(entry.get("allow_repeat")),
                    )
                    count += 1
                logger.info(
                    "Loaded %d tool(s) from %s (multi-tool)",
                    count, py_file.name,
                )
                continue

            # --- Single-tool format --------------------------------------
            missing = [
                a for a in _REQUIRED_ATTRS
                if not hasattr(module, a)
            ]
            if missing:
                logger.warning(
                    "Tool %s missing attrs: %s – skip",
                    py_file.name,
                    ", ".join(missing),
                )
                continue

            handler = getattr(module, "run")
            if not callable(handler):
                logger.warning(
                    "Tool %s: 'run' not callable – skip",
                    py_file.name,
                )
                continue

            tool_name: str = getattr(module, "TOOL_NAME")
            tool_desc: str = getattr(module, "TOOL_DESCRIPTION")
            tool_params: dict = getattr(module, "TOOL_PARAMETERS")

            no_bg = bool(getattr(module, "TOOL_NO_BACKGROUND", False))
            allow_repeat = bool(getattr(module, "TOOL_ALLOW_REPEAT", False))

            registry._tools[tool_name] = ToolDefinition(
                name=tool_name,
                description=tool_desc,
                parameters=tool_params,
                handler=handler,
                no_background=no_bg,
                allow_repeat=allow_repeat,
            )
            logger.info(
                "Loaded tool: %s (%s)", tool_name, py_file.name,
            )

        except Exception:
            logger.exception(
                "Failed to load tool from %s", py_file.name,
            )

    logger.info(
        "Tool loading complete – %d tool(s) registered",
        len(registry),
    )