orchestrata_ultimate.tools.registry

Reference › orchestrata_ultimate › tools › registry

This module is the hub of the tool framework: a thread-safe index of BaseTool instances keyed by name, which supports register, discover, describe and dispatch. It replaces a hard-coded dispatch chain in the agents layer. Its importers include agents.tool_registry, kernel.cli_main, plugin.registries and every tool module that has a register_* function. Tests are in tests/tools/test_registry.py and several other tool test modules.

ApprovalGate is a runtime-checkable protocol with one method, approve(tool_name, arguments), returning a boolean. ToolRegistry takes an optional gate. register rejects a duplicate name with ValueError unless replace=True. unregister, clear, get, has, names, and the container dunder methods manage and inspect the index. list filters by category, tag or priority, metadata returns every tool's metadata (the corpus the search tool indexes), and to_function_schemas renders all or named tools as LLM function definitions.

execute dispatches by name. An unknown tool name yields a failed ToolResult and not an exception. A gate, either the registry's own or one passed per call, is consulted for tools whose metadata sets requires_approval; the Prolog-backed gate is approval. default_registry returns the process-wide singleton and register_tool registers into it.

At a glance

Class diagram

Dependency diagram

API

Classes

ApprovalGate (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 29

Decides whether an approval-requiring tool call may proceed.

ToolRegistry

Line 43

Thread-safe registry of tools, indexed by name.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/registry.py

"""Unified tool registry — register, discover, describe, dispatch.

A thread-safe index of :class:`~orchestrata_ultimate.tools.base.BaseTool`
instances keyed by tool name. Replaces the pre-existing hardcoded ``if/elif``
dispatch with dynamic registration + metadata-driven discovery, and is the
index SPEC-072 (ToolSearchTool) searches over.

Spec: SPEC-062 (CheetahClaws Tools Gap) — unified tool framework; enables
SPEC-072 (ToolSearchTool).
"""

from __future__ import annotations

import builtins
import threading
from collections.abc import Iterable, Iterator, Mapping
from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)


@runtime_checkable
class ApprovalGate(Protocol):
    """Decides whether an approval-requiring tool call may proceed.

    The seam a human-in-the-loop channel implements. `ToolRegistry` fails
    closed when none is installed, so a tool declaring
    ``ToolMetadata.requires_approval`` cannot run until something here says
    yes.
    """

    def approve(self, tool_name: str, arguments: Mapping[str, Any]) -> bool:
        """Return True to let the call proceed, False to refuse it."""
        ...


class ToolRegistry:
    """Thread-safe registry of tools, indexed by name.

    Registration is explicit (``register``); discovery is by name (``get``),
    filter (``list``), or full metadata index (``metadata``). Dispatch goes
    through ``execute`` so callers never touch the tool objects directly.
    """

    def __init__(self, *, approval_gate: ApprovalGate | None = None) -> None:
        self._tools: dict[str, BaseTool] = {}
        self._lock = threading.Lock()
        self._approval_gate = approval_gate

    # ── registration ────────────────────────────────────────────────

    def register(self, tool: BaseTool, *, replace: bool = False) -> None:
        """Register ``tool``. Raises ``ValueError`` on a duplicate name unless ``replace``."""
        name = tool.metadata.name
        with self._lock:
            if name in self._tools and not replace:
                raise ValueError(f"Tool already registered: {name!r}")
            self._tools[name] = tool

    def unregister(self, name: str) -> bool:
        """Remove a tool by name. Returns ``True`` if it was present."""
        with self._lock:
            return self._tools.pop(name, None) is not None

    def clear(self) -> None:
        """Remove all registered tools."""
        with self._lock:
            self._tools.clear()

    # ── lookup / discovery ──────────────────────────────────────────

    def get(self, name: str) -> BaseTool | None:
        """Return the tool registered under ``name``, or ``None``."""
        return self._tools.get(name)

    def has(self, name: str) -> bool:
        return name in self._tools

    def names(self) -> list[str]:
        """Registered tool names, sorted."""
        return sorted(self._tools)

    def list(
        self,
        *,
        category: ToolCategory | None = None,
        tag: str | None = None,
        priority: ToolPriority | None = None,
    ) -> list[BaseTool]:
        """Return registered tools, optionally filtered by category/tag/priority."""
        tools = [self._tools[n] for n in self.names()]
        if category is not None:
            tools = [t for t in tools if t.metadata.category == category]
        if tag is not None:
            tools = [t for t in tools if tag in t.metadata.tags]
        if priority is not None:
            tools = [t for t in tools if t.metadata.priority == priority]
        return tools

    def metadata(self) -> builtins.list[ToolMetadata]:
        """All registered tools' metadata (the index SPEC-072 searches)."""
        return [self._tools[n].metadata for n in self.names()]

    def to_function_schemas(
        self, names: Iterable[str] | None = None
    ) -> builtins.list[dict[str, Any]]:
        """Render tools as LLM function-calling definitions.

        ``names=None`` renders all registered tools; otherwise only the named
        ones that are registered (unknown names are skipped).
        """
        if names is None:
            selected = [self._tools[n] for n in self.names()]
        else:
            selected = [self._tools[n] for n in names if n in self._tools]
        return [t.to_function_schema() for t in selected]

    # ── dispatch ────────────────────────────────────────────────────

    async def execute(
        self,
        name: str,
        arguments: dict[str, Any],
        *,
        approval_gate: ApprovalGate | None = None,
    ) -> ToolResult:
        """Dispatch a tool call by name. Unknown tool → a failed ``ToolResult``.

        A tool whose metadata sets ``requires_approval`` is refused unless an
        :class:`ApprovalGate` grants it. This FAILS CLOSED: with no gate
        installed the call is refused, not run.

        That default is deliberate. ``requires_approval`` is declared by
        exactly ``repl`` and ``remote_trigger`` — the two tools that execute
        arbitrary code on the host and reach outward over the network — and
        until this change nothing read the flag at all, so both ran unchecked
        for every volk they were exposed to. Refusing is a visible,
        diagnosable failure; running unapproved is not.

        *approval_gate* overrides the registry-wide gate for one call, so a
        caller can grant approval without mutating shared state.
        """
        tool = self._tools.get(name)
        if tool is None:
            return ToolResult(name, False, "", f"Unknown tool: {name}")

        if tool.metadata.requires_approval:
            gate = approval_gate if approval_gate is not None else self._approval_gate
            if gate is None:
                return ToolResult(
                    name,
                    False,
                    "",
                    f"{name} requires human approval and no approval gate is "
                    f"installed — refused. Install an ApprovalGate on the "
                    f"ToolRegistry, or pass approval_gate= for this call.",
                )
            if not gate.approve(name, dict(arguments)):
                return ToolResult(
                    name, False, "", f"{name} was denied by the approval gate."
                )

        return await tool.execute(dict(arguments))

    # ── dunders ─────────────────────────────────────────────────────

    def __getitem__(self, name: str) -> BaseTool:
        """Return a registered tool by name, raising ``KeyError`` if absent."""
        return self._tools[name]

    def __len__(self) -> int:
        return len(self._tools)

    def __contains__(self, name: object) -> bool:
        return name in self._tools

    def __iter__(self) -> Iterator[BaseTool]:
        return iter(self.list())


# ── Process-wide default registry ────────────────────────────────────

_DEFAULT_REGISTRY = ToolRegistry()


def default_registry() -> ToolRegistry:
    """The process-wide default registry."""
    return _DEFAULT_REGISTRY


def register_tool(tool: BaseTool, *, replace: bool = False) -> None:
    """Register ``tool`` into the default registry."""
    _DEFAULT_REGISTRY.register(tool, replace=replace)