tools_runtime

ToolBackend

Bases: Protocol

Protocol defining what a dynamic tool backend must implement.

Source code in wintermute/ai/tools_runtime.py

@runtime_checkable
class ToolBackend(Protocol):
    """Protocol defining what a dynamic tool backend must implement."""

    async def get_ai_tools(self) -> List[Dict[str, Any]]: ...
    async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str: ...

ToolRegistry

Source code in wintermute/ai/tools_runtime.py

class ToolRegistry:
    def __init__(self, base_path: str = "/opt") -> None:
        self._tools: Dict[str, Tool] = {}
        self._path_mapping: Dict[str, Dict[str, str]] = {}
        # Environment override
        self.base_path = os.getenv("WINTERMUTE_TOOLS_ROOT", base_path)

    def load_tool_configs(self, json_path: str) -> None:
        """Read distributable_db/tools.json and store in _path_mapping."""
        if not os.path.exists(json_path):
            log.warning(f"Tool config file not found: {json_path}")
            return

        try:
            with open(json_path, "r", encoding="utf-8") as f:
                data = json.load(f)
                if isinstance(data, list):
                    for entry in data:
                        if "name" in entry:
                            self._path_mapping[entry["name"]] = entry
                elif isinstance(data, dict):
                    # In case it's a dict-based mapping
                    self._path_mapping = data
        except Exception as e:
            log.warning(f"Failed to load tool configs from {json_path}: {e}")

    def register(self, tool: Tool) -> None:
        # Smart Registration: update description with absolute path if mapped
        description = tool.description
        if tool.name in self._path_mapping:
            entry = self._path_mapping[tool.name]
            if "directory" in entry and "executable" in entry:
                abs_path = os.path.join(
                    self.base_path, entry["directory"], entry["executable"]
                )
                description = f"{description}\n\nAbsolute Path: {abs_path}".strip()

        # Update the tool with the new description if changed
        updated_tool = Tool(
            name=tool.name,
            input_schema=tool.input_schema,
            output_schema=tool.output_schema,
            handler=tool.handler,
            description=description,
        )
        self._tools[updated_tool.name] = updated_tool

    def call(self, name: str, args: JSONObject) -> JSONObject:
        if name not in self._tools:
            raise KeyError(f"Tool {name} not found")
        tool = self._tools[name]
        return tool.handler(args)

    def unregister(self, name: str) -> bool:
        """Remove a single tool from the registry by name.

        Returns ``True`` if a tool was actually removed, ``False`` if it
        was already absent. Used by the cartridge manager when unloading
        a cartridge so its methods stop showing up in the AI tool surface.
        """
        return self._tools.pop(name, None) is not None

    def get_definitions(self) -> List[Dict[str, Any]]:
        """Convert registered tools to OpenAI-compatible function definitions."""
        definitions = []
        for name, tool in self._tools.items():
            definitions.append(
                {
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.input_schema,
                    },
                }
            )
        return definitions

get_definitions()

Convert registered tools to OpenAI-compatible function definitions.

Source code in wintermute/ai/tools_runtime.py

def get_definitions(self) -> List[Dict[str, Any]]:
    """Convert registered tools to OpenAI-compatible function definitions."""
    definitions = []
    for name, tool in self._tools.items():
        definitions.append(
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.input_schema,
                },
            }
        )
    return definitions

load_tool_configs(json_path)

Read distributable_db/tools.json and store in _path_mapping.

Source code in wintermute/ai/tools_runtime.py

def load_tool_configs(self, json_path: str) -> None:
    """Read distributable_db/tools.json and store in _path_mapping."""
    if not os.path.exists(json_path):
        log.warning(f"Tool config file not found: {json_path}")
        return

    try:
        with open(json_path, "r", encoding="utf-8") as f:
            data = json.load(f)
            if isinstance(data, list):
                for entry in data:
                    if "name" in entry:
                        self._path_mapping[entry["name"]] = entry
            elif isinstance(data, dict):
                # In case it's a dict-based mapping
                self._path_mapping = data
    except Exception as e:
        log.warning(f"Failed to load tool configs from {json_path}: {e}")

unregister(name)

Remove a single tool from the registry by name.

Returns True if a tool was actually removed, False if it was already absent. Used by the cartridge manager when unloading a cartridge so its methods stop showing up in the AI tool surface.

Source code in wintermute/ai/tools_runtime.py

def unregister(self, name: str) -> bool:
    """Remove a single tool from the registry by name.

    Returns ``True`` if a tool was actually removed, ``False`` if it
    was already absent. Used by the cartridge manager when unloading
    a cartridge so its methods stop showing up in the AI tool surface.
    """
    return self._tools.pop(name, None) is not None

ToolsRuntime

Orchestrates tool execution across local static tools (ToolRegistry) and dynamic backends (MCP Servers like Surgeon).

Source code in wintermute/ai/tools_runtime.py

class ToolsRuntime:
    """
    Orchestrates tool execution across local static tools (ToolRegistry)
    and dynamic backends (MCP Servers like Surgeon).
    """

    def __init__(self) -> None:
        self.dynamic_backends: List[ToolBackend] = []

    def register_backend(self, backend: ToolBackend) -> None:
        """Register a dynamic backend (e.g., SurgeonBackend)."""
        self.dynamic_backends.append(backend)

    async def get_all_tools(self) -> List[Dict[str, Any]]:
        """Combine static local tools with dynamic backend tools."""
        # 1. Start with local tools from the global registry
        all_tools = tools.get_definitions()

        # 2. Fetch tools from dynamic backends
        for backend in self.dynamic_backends:
            try:
                backend_tools = await backend.get_ai_tools()
                all_tools.extend(backend_tools)
            except Exception as e:
                log.error(f"Failed to fetch tools from backend {backend}: {e}")

        return all_tools

    async def run_tool(self, name: str, args: Dict[str, Any]) -> str:
        """
        Execute a tool by name. Checks dynamic backends first, then local registry.
        """
        log.info(f"AI requested execution of tool: {name}")

        # 1. Check Dynamic Backends (MCP)
        for backend in self.dynamic_backends:
            try:
                # Optimization: We assume backend names are unique or check quickly
                # Ideally, we'd cache the map of tool_name -> backend
                known_tools = await backend.get_ai_tools()
                if any(t["function"]["name"] == name for t in known_tools):
                    return await backend.execute_tool(name, args)
            except Exception as e:
                log.error(f"Error checking backend {backend}: {e}")

        # 2. Fallback to Local Tools
        return await self._run_local_tool(name, args)

    async def _run_local_tool(self, name: str, args: Dict[str, Any]) -> str:
        """
        Executes a tool from the local ToolRegistry.
        """
        try:
            result = tools.call(name, args)
            return str(result)
        except KeyError:
            error_msg = (
                f"Tool '{name}' not found in local registry or connected backends."
            )
            log.warning(error_msg)
            return error_msg
        except Exception as e:
            error_msg = f"Error executing local tool '{name}': {str(e)}"
            log.error(error_msg)
            return error_msg

get_all_tools() async

Combine static local tools with dynamic backend tools.

Source code in wintermute/ai/tools_runtime.py

async def get_all_tools(self) -> List[Dict[str, Any]]:
    """Combine static local tools with dynamic backend tools."""
    # 1. Start with local tools from the global registry
    all_tools = tools.get_definitions()

    # 2. Fetch tools from dynamic backends
    for backend in self.dynamic_backends:
        try:
            backend_tools = await backend.get_ai_tools()
            all_tools.extend(backend_tools)
        except Exception as e:
            log.error(f"Failed to fetch tools from backend {backend}: {e}")

    return all_tools

register_backend(backend)

Register a dynamic backend (e.g., SurgeonBackend).

Source code in wintermute/ai/tools_runtime.py

def register_backend(self, backend: ToolBackend) -> None:
    """Register a dynamic backend (e.g., SurgeonBackend)."""
    self.dynamic_backends.append(backend)

run_tool(name, args) async

Execute a tool by name. Checks dynamic backends first, then local registry.

Source code in wintermute/ai/tools_runtime.py

async def run_tool(self, name: str, args: Dict[str, Any]) -> str:
    """
    Execute a tool by name. Checks dynamic backends first, then local registry.
    """
    log.info(f"AI requested execution of tool: {name}")

    # 1. Check Dynamic Backends (MCP)
    for backend in self.dynamic_backends:
        try:
            # Optimization: We assume backend names are unique or check quickly
            # Ideally, we'd cache the map of tool_name -> backend
            known_tools = await backend.get_ai_tools()
            if any(t["function"]["name"] == name for t in known_tools):
                return await backend.execute_tool(name, args)
        except Exception as e:
            log.error(f"Error checking backend {backend}: {e}")

    # 2. Fallback to Local Tools
    return await self._run_local_tool(name, args)

unregister_tools(names)

Remove the given tool names from the global :data:tools registry.

Counterpart to :func:wintermute.ai.utils.tool_factory.register_tools + :meth:ToolRegistry.register. Used by :class:wintermute.cartridges.manager.CartridgeManager on unload so the AI no longer sees cartridge methods that have been removed from memory.

Parameters:

Name	Type	Description	Default
names	Iterable[str]	Iterable of tool names to drop. Names that are already absent are ignored silently.	required

Returns:

Type	Description
int	The number of tools actually removed.

Source code in wintermute/ai/tools_runtime.py

def unregister_tools(names: Iterable[str]) -> int:
    """Remove the given tool names from the global :data:`tools` registry.

    Counterpart to :func:`wintermute.ai.utils.tool_factory.register_tools` +
    :meth:`ToolRegistry.register`. Used by
    :class:`wintermute.cartridges.manager.CartridgeManager` on unload so the
    AI no longer sees cartridge methods that have been removed from memory.

    Args:
        names: Iterable of tool names to drop. Names that are already
            absent are ignored silently.

    Returns:
        The number of tools actually removed.
    """
    removed = 0
    for name in names:
        if tools.unregister(name):
            removed += 1
    return removed