gemma-spectrum/archon-poc/runtime/tool_registry.py

#!/usr/bin/env python3
"""
Tool Registry - Layer 2 Component

Defines and manages tools available to the Claw runtime.
Tools are executed locally, not sent to the intelligence layer.
"""

import re
import json
import subprocess
from typing import Dict, Any, List, Callable, Optional
from dataclasses import dataclass
from datetime import datetime


@dataclass
class Tool:
    name: str
    description: str
    parameters: Dict[str, Any]
    handler: Callable


class ToolRegistry:
    """
    Registry of available tools for the Claw runtime.
    
    Tools are pattern-matched from user messages and executed
    before routing to the intelligence layer.
    """
    
    def __init__(self):
        self.tools: Dict[str, Tool] = {}
        self._register_builtin_tools()
    
    def _register_builtin_tools(self):
        """Register built-in tools."""
        self.register(Tool(
            name="time",
            description="Get current time",
            parameters={},
            handler=self._get_time
        ))
        
        self.register(Tool(
            name="status",
            description="Get system status",
            parameters={},
            handler=self._get_status
        ))
        
        self.register(Tool(
            name="echo",
            description="Echo a message back",
            parameters={"message": "string"},
            handler=self._echo
        ))
        
        self.register(Tool(
            name="ollama_list",
            description="List available Ollama models",
            parameters={},
            handler=self._ollama_list
        ))
    
    def register(self, tool: Tool):
        """Register a new tool."""
        self.tools[tool.name] = tool
    
    def parse_tool_call(self, message: str) -> Optional[Dict[str, Any]]:
        """
        Parse a message for tool invocation.
        
        Patterns:
        - /tool_name
        - /tool_name param=value
        - @tool_name
        """
        # Match /tool_name or @tool_name patterns
        match = re.match(r'^[/@](\w+)(?:\s+(.+))?$', message.strip())
        if not match:
            return None
        
        tool_name = match.group(1)
        args_str = match.group(2) or ""
        
        if tool_name not in self.tools:
            return None
        
        # Parse parameters
        params = {}
        if args_str:
            # Simple key=value parsing
            for pair in args_str.split():
                if '=' in pair:
                    key, value = pair.split('=', 1)
                    params[key] = value
                else:
                    # Positional argument as 'message'
                    params["message"] = args_str
        
        return {
            "name": tool_name,
            "parameters": params
        }
    
    def execute(self, tool_call: Dict[str, Any]) -> str:
        """Execute a tool call."""
        tool_name = tool_call["name"]
        params = tool_call.get("parameters", {})
        
        if tool_name not in self.tools:
            return f"Error: Unknown tool '{tool_name}'"
        
        tool = self.tools[tool_name]
        
        try:
            result = tool.handler(**params)
            return f"[Tool: {tool_name}]\n{result}"
        except Exception as e:
            return f"[Tool Error: {tool_name}]\n{str(e)}"
    
    def list_tools(self) -> List[str]:
        """List all available tools."""
        return [
            f"{name}: {tool.description}"
            for name, tool in self.tools.items()
        ]
    
    # --- Tool Handlers ---
    
    def _get_time(self) -> str:
        """Get current time."""
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    def _get_status(self) -> str:
        """Get system status."""
        return json.dumps({
            "runtime": "claw",
            "version": "1.0.0-poc",
            "status": "operational",
            "tools_available": len(self.tools)
        }, indent=2)
    
    def _echo(self, message: str = "") -> str:
        """Echo a message."""
        return message
    
    def _ollama_list(self) -> str:
        """List Ollama models."""
        try:
            result = subprocess.run(
                ["curl", "-s", "http://localhost:11434/api/tags"],
                capture_output=True,
                text=True
            )
            data = json.loads(result.stdout)
            models = [m["name"] for m in data.get("models", [])]
            return f"Available models:\n" + "\n".join(f"  - {m}" for m in models)
        except Exception as e:
            return f"Error listing models: {e}"


def main():
    """CLI for testing tool registry."""
    registry = ToolRegistry()
    
    print("Available tools:")
    for tool_info in registry.list_tools():
        print(f"  {tool_info}")
    
    print("\nTest parsing:")
    test_messages = [
        "/time",
        "/status",
        "/echo Hello world",
        "/ollama_list",
        "Regular message without tool"
    ]
    
    for msg in test_messages:
        parsed = registry.parse_tool_call(msg)
        if parsed:
            result = registry.execute(parsed)
            print(f"\n> {msg}")
            print(result)
        else:
            print(f"\n> {msg}")
            print("(No tool call detected)")


if __name__ == "__main__":
    main()