orchestrata_ultimate.tools.base

Reference › orchestrata_ultimate › tools › base

This module defines what a tool is in orchestrata: the contract every first-class tool implements (SPEC-062). Nearly every other module in the tools package imports it, as do agents.tool_registry and plugin.registries. tests/tools/test_base.py covers it.

• ToolResult is the uniform outcome of one execution: tool name, success flag, output text and an optional error string.
• Tool is a runtime-checkable protocol with one async execute method, a thin structural interface.
• ToolCategory (file, shell, search, web, task, memory, planning, agent, config, mcp, utility) and ToolPriority (P0 to P3) are string enums that classify a tool.
• ToolMetadata is a frozen description: name, description, category, JSON-Schema parameters, priority, a requires_approval flag, governance article, estimated hours, dependencies, tags and version. to_function_schema renders it as an OpenAI or DeepSeek style function-calling definition, which is how a tool is described to an LLM.
• BaseTool is the abstract base. A subclass supplies the metadata property and an async execute; name, description, category and to_function_schema derive from the metadata.
• FunctionToolAdapter wraps a plain async executor(name, args) function and a ToolMetadata as a BaseTool, so existing function tools join the framework unchanged.

The registry that indexes these objects is in registry.

At a glance

Class diagram

Dependency diagram

API

Classes

ToolResult (dataclass)

Decorators: dataclass · Line 29

Result of a single tool execution.

Tool (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 47

Thin structural interface for a single callable tool.

ToolCategory (enum)

Bases: StrEnum · Line 72

Tool category taxonomy (SPEC-062 Tool Definition Schema).

ToolPriority (enum)

Bases: StrEnum · Line 88

Roadmap priority band for a tool (SPEC-062).

ToolMetadata (dataclass)

Decorators: dataclass(frozen=True) · Line 101

Describes a tool: identity, category, LLM parameter schema, governance.

BaseTool (abstract)

Bases: ABC · Line 136

Abstract base every registrable tool implements.

FunctionToolAdapter

Bases: BaseTool · Line 174

Wrap a plain async executor(name, args) -> ToolResult as a BaseTool.

Source

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

"""Unified tool framework — ``BaseTool`` + ``ToolMetadata``.

The canonical contract every first-class tool implements so it can be
registered, discovered, described to an LLM (OpenAI/DeepSeek function schema),
and dispatched uniformly through :class:`~orchestrata_ultimate.tools.registry.ToolRegistry`.

Before this module the codebase had no tool abstraction: tools were ad-hoc
module-level function-call dicts collected in ``agents/tool_registry.py`` and
dispatched by a hardcoded ``if/elif`` chain. This module introduces the shared
base the greenfield tool suite (SPEC-063…072) builds on, and the metadata index
that makes tool search (SPEC-072) possible.

Spec: SPEC-062 (CheetahClaws Tools Gap) — Tool Definition Schema + category
taxonomy; enables SPEC-072 (ToolSearchTool).
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any, Protocol, runtime_checkable

# ── Result ───────────────────────────────────────────────────────────


@dataclass
class ToolResult:
    """Result of a single tool execution.

    Canonical home for the result type (previously defined in
    ``agents/tool_registry.py``, which now re-imports it) so both the legacy
    executor and the unified registry speak one type.
    """

    tool_name: str
    success: bool
    output: str
    error: str | None = None


# ── Thin structural interface ─────────────────────────────────────────


@runtime_checkable
class Tool(Protocol):
    """Thin structural interface for a single callable tool.

    Composition-over-inheritance boundary (see
    ``plans/build-factory-architecture.md``): a concrete tool implements this
    Protocol — the ONE ABC/Protocol boundary — and COMPOSES its behaviour
    (executor, validator, tracer, registry lookups, ...) via constructor DI
    rather than a deep inheritance chain. ``gen-skeleton``'s ``<<Tool>>``
    emitter targets exactly this Protocol (``ReplTool ..|> Tool`` ->
    ``class ReplTool(Tool):``).

    Distinct from :class:`BaseTool`, the pre-existing ABC below: tools that
    need its shared ``name``/``description``/``category`` properties may
    still subclass ``BaseTool`` directly; new composition-first tools target
    ``Tool``.
    """

    async def execute(self, args: Any) -> ToolResult:
        """Run the tool with its typed arguments and return a ToolResult."""
        ...


# ── Taxonomy ─────────────────────────────────────────────────────────


class ToolCategory(StrEnum):
    """Tool category taxonomy (SPEC-062 Tool Definition Schema)."""

    FILE = "file"
    SHELL = "shell"
    SEARCH = "search"
    WEB = "web"
    TASK = "task"
    MEMORY = "memory"
    PLANNING = "planning"
    AGENT = "agent"
    CONFIG = "config"
    MCP = "mcp"
    UTILITY = "utility"


class ToolPriority(StrEnum):
    """Roadmap priority band for a tool (SPEC-062)."""

    P0 = "P0"
    P1 = "P1"
    P2 = "P2"
    P3 = "P3"


# ── Metadata ─────────────────────────────────────────────────────────


@dataclass(frozen=True)
class ToolMetadata:
    """Describes a tool: identity, category, LLM parameter schema, governance.

    Mirrors the SPEC-062 Tool Definition Schema. ``parameters`` is a JSON Schema
    object (the same shape the OpenAI/DeepSeek function-calling API expects).
    """

    name: str
    description: str
    category: ToolCategory
    parameters: dict[str, Any] = field(default_factory=dict)
    priority: ToolPriority = ToolPriority.P2
    requires_approval: bool = False
    governance_article: str | None = None
    estimated_hours: int | None = None
    dependencies: tuple[str, ...] = ()
    tags: tuple[str, ...] = ()
    version: str = "1.0.0"

    def to_function_schema(self) -> dict[str, Any]:
        """Render as an OpenAI/DeepSeek function-calling tool definition."""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters
                or {"type": "object", "properties": {}, "required": []},
            },
        }


# ── Base tool ────────────────────────────────────────────────────────


class BaseTool(ABC):
    """Abstract base every registrable tool implements.

    A tool exposes immutable :class:`ToolMetadata` and an async ``execute`` that
    takes a JSON-decoded ``arguments`` mapping (the LLM's tool-call arguments)
    and returns a :class:`ToolResult`. Uniform ``(arguments) -> ToolResult``
    dispatch is what lets the registry route any tool without special-casing.
    """

    @property
    @abstractmethod
    def metadata(self) -> ToolMetadata:
        """The tool's metadata (identity, category, parameter schema)."""

    @abstractmethod
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Run the tool with the given arguments."""

    @property
    def name(self) -> str:
        return self.metadata.name

    @property
    def description(self) -> str:
        return self.metadata.description

    @property
    def category(self) -> ToolCategory:
        return self.metadata.category

    def to_function_schema(self) -> dict[str, Any]:
        """Render this tool as an LLM function-calling definition."""
        return self.metadata.to_function_schema()


# ── Adapter for the legacy function-tools ────────────────────────────


class FunctionToolAdapter(BaseTool):
    """Wrap a plain async ``executor(name, args) -> ToolResult`` as a BaseTool.

    Bridges the pre-existing function-tools (``agents/tool_registry.py``'s
    ``ToolExecutor``) into the unified registry without rewriting them, so the
    live tools become discoverable/searchable while the invocation path is
    unchanged. The executor is any callable with the legacy executor signature.
    """

    def __init__(
        self,
        metadata: ToolMetadata,
        executor: Callable[[str, dict[str, Any]], Awaitable[ToolResult]],
    ) -> None:
        self._metadata = metadata
        self._executor = executor

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        return await self._executor(self._metadata.name, arguments)