tiphys.tools.base

  1"""
  2Tool definitions and base classes.
  3
  4Provides the foundation for defining agent tools with Pydantic schemas.
  5"""
  6
  7from collections.abc import Awaitable, Callable
  8from functools import wraps
  9from typing import Any, TypeVar
 10
 11from pydantic import BaseModel, Field, create_model
 12
 13# Type variable for handler functions
 14T = TypeVar("T", bound=Callable[..., Awaitable[str]])
 15
 16
 17class ToolDefinition(BaseModel):
 18    """Definition of an agent tool."""
 19
 20    name: str
 21    description: str
 22    args_schema: type[BaseModel]
 23    handler: Callable[..., Awaitable[str]]
 24    requires_permission: bool = False
 25    optional: bool = False
 26
 27    class Config:
 28        arbitrary_types_allowed = True
 29
 30
 31class ToolRegistry:
 32    """Registry for agent tools."""
 33
 34    def __init__(self) -> None:
 35        """Initialize the tool registry."""
 36        self._tools: dict[str, ToolDefinition] = {}
 37
 38    def register(self, tool: ToolDefinition) -> None:
 39        """Register a tool."""
 40        self._tools[tool.name] = tool
 41
 42    def unregister(self, name: str) -> bool:
 43        """Unregister a tool. Returns True if removed."""
 44        if name in self._tools:
 45            del self._tools[name]
 46            return True
 47        return False
 48
 49    def get(self, name: str) -> ToolDefinition | None:
 50        """Get a tool by name."""
 51        return self._tools.get(name)
 52
 53    def get_all(self) -> dict[str, ToolDefinition]:
 54        """Get all registered tools."""
 55        return self._tools.copy()
 56
 57    def list_names(self) -> list[str]:
 58        """List all tool names."""
 59        return list(self._tools.keys())
 60
 61    def clear(self) -> int:
 62        """Clear all tools. Returns count cleared."""
 63        count = len(self._tools)
 64        self._tools.clear()
 65        return count
 66
 67
 68def tool(
 69    name: str,
 70    description: str,
 71    requires_permission: bool = False,
 72) -> Callable[[T], T]:
 73    """
 74    Decorator to define an agent tool.
 75
 76    Automatically extracts args schema from type hints.
 77
 78    Usage:
 79        @tool(name="read_file", description="Read a file")
 80        async def read_file(path: str) -> str:
 81            ...
 82    """
 83
 84    def decorator(func: T) -> T:
 85        # Extract type hints for arguments
 86        import inspect
 87        from typing import get_type_hints
 88
 89        hints = get_type_hints(func)
 90        hints.pop("return", None)
 91
 92        # Build Pydantic model from hints
 93        fields: dict[str, Any] = {}
 94        sig = inspect.signature(func)
 95
 96        for param_name, param in sig.parameters.items():
 97            if param_name in ("context", "self"):
 98                continue
 99
100            param_type = hints.get(param_name, str)
101
102            # Get description from docstring if available
103            param_desc = ""
104
105            # Check if parameter has default
106            if param.default is inspect.Parameter.empty:
107                fields[param_name] = (param_type, Field(..., description=param_desc))
108            else:
109                fields[param_name] = (
110                    param_type,
111                    Field(default=param.default, description=param_desc),
112                )
113
114        args_schema = create_model(f"{name.title().replace('_', '')}Args", **fields)
115
116        # Create tool definition
117        tool_def = ToolDefinition(
118            name=name,
119            description=description,
120            args_schema=args_schema,
121            handler=func,
122            requires_permission=requires_permission,
123        )
124
125        # Attach to function
126        func._tool_definition = tool_def  # type: ignore[attr-defined]
127
128        @wraps(func)
129        async def wrapper(*args: Any, **kwargs: Any) -> str:
130            return await func(*args, **kwargs)
131
132        wrapper._tool_definition = tool_def  # type: ignore[attr-defined]
133        return wrapper  # type: ignore[return-value]
134
135    return decorator
136
137
138def get_tool_definition(func: Callable[..., Any]) -> ToolDefinition | None:
139    """Get the tool definition from a decorated function."""
140    return getattr(func, "_tool_definition", None)
141
142
143# Global tool registry (deprecated - use AppContext)
144_tool_registry: ToolRegistry | None = None
145
146
147def get_tool_registry() -> ToolRegistry:
148    """
149    Get the global tool registry.
150
151    .. deprecated::
152        Use AppContext.tool_registry instead. This function will be removed
153        in a future version.
154
155    Returns:
156        The global ToolRegistry instance.
157    """
158    import warnings
159
160    from tiphys.context import get_context, has_context
161
162    # If AppContext is available, use it (preferred path)
163    if has_context():
164        warnings.warn(
165            "get_tool_registry() is deprecated. Use AppContext.tool_registry instead.",
166            DeprecationWarning,
167            stacklevel=2,
168        )
169        return get_context().tool_registry
170
171    # Fallback to global singleton for backwards compatibility
172    global _tool_registry
173    if _tool_registry is None:
174        _tool_registry = ToolRegistry()
175    return _tool_registry