Upload mythos/tool.py

mythos/tool.py

@@ -0,0 +1,102 @@
+"""Tool abstraction for Mythos agents — MCP-native design."""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+from typing import Any, Awaitable, Callable, Optional
+
+from pydantic import BaseModel, Field
+
+
+class ToolSpec(BaseModel):
+    """Specification for a tool, compatible with MCP and function-calling APIs."""
+
+    name: str
+    description: str
+    parameters: dict[str, Any] = Field(default_factory=dict)
+    required: list[str] = Field(default_factory=list)
+    returns: Optional[str] = None
+
+
+class Tool(BaseModel):
+    """A callable tool that agents can use."""
+
+    spec: ToolSpec
+    _func: Optional[Callable[..., Any]] = None
+
+    class Config:
+        arbitrary_types_allowed = True
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        self._func = data.get("_func")
+
+    async def run(self, **kwargs) -> Any:
+        """Execute the tool with given arguments."""
+        if self._func is None:
+            raise RuntimeError(f"Tool '{self.spec.name}' has no implementation")
+        if asyncio.iscoroutinefunction(self._func):
+            return await self._func(**kwargs)
+        return self._func(**kwargs)
+
+    @classmethod
+    def from_function(
+        cls,
+        func: Callable[..., Any],
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+    ) -> Tool:
+        """Create a Tool from a Python function."""
+        sig = inspect.signature(func)
+        params = {}
+        required = []
+        for param_name, param in sig.parameters.items():
+            param_type = param.annotation if param.annotation is not inspect.Parameter.empty else "string"
+            params[param_name] = {"type": str(param_type)}
+            if param.default is inspect.Parameter.empty:
+                required.append(param_name)
+
+        spec = ToolSpec(
+            name=name or func.__name__,
+            description=description or (func.__doc__ or ""),
+            parameters=params,
+            required=required,
+        )
+        return cls(spec=spec, _func=func)
+
+
+class ToolRegistry(BaseModel):
+    """Registry of tools available to agents."""
+
+    tools: dict[str, Tool] = Field(default_factory=dict)
+
+    def register(self, tool: Tool) -> None:
+        """Register a tool."""
+        self.tools[tool.spec.name] = tool
+
+    def register_function(
+        self,
+        func: Callable[..., Any],
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+    ) -> Tool:
+        """Register a function as a tool."""
+        tool = Tool.from_function(func, name, description)
+        self.register(tool)
+        return tool
+
+    def get(self, name: str) -> Optional[Tool]:
+        """Get a tool by name."""
+        return self.tools.get(name)
+
+    def list_tools(self) -> list[ToolSpec]:
+        """List all tool specifications."""
+        return [t.spec for t in self.tools.values()]
+
+    async def call(self, name: str, **kwargs) -> Any:
+        """Call a tool by name."""
+        tool = self.get(name)
+        if tool is None:
+            raise KeyError(f"Tool '{name}' not found")
+        return await tool.run(**kwargs)