"""Tool system for the agent framework.""" from abc import ABC, abstractmethod from typing import Dict, Any, Callable import inspect from .models import ExecutionContext from .utils import function_to_input_schema, format_tool_definition class BaseTool(ABC): """Abstract base class for all tools.""" def __init__( self, name: str = None, description: str = None, tool_definition: Dict[str, Any] = None, # Confirmation support requires_confirmation: bool = False, confirmation_message_template: str = None ): self.name = name or self.__class__.__name__ self.description = description or self.__doc__ or "" self._tool_definition = tool_definition self.requires_confirmation = requires_confirmation self.confirmation_message_template = confirmation_message_template or ( "The agent wants to execute '{name}' with arguments: {arguments}. " "Do you approve?" ) @property def tool_definition(self) -> Dict[str, Any] | None: return self._tool_definition @abstractmethod async def execute(self, context: ExecutionContext, **kwargs) -> Any: pass async def __call__(self, context: ExecutionContext, **kwargs) -> Any: return await self.execute(context, **kwargs) def get_confirmation_message(self, arguments: dict[str, Any]) -> str: """Generate a confirmation message for this tool call.""" return self.confirmation_message_template.format( name=self.name, arguments=arguments ) class FunctionTool(BaseTool): """Wraps a Python function as a BaseTool.""" def __init__( self, func: Callable, name: str = None, description: str = None, tool_definition: Dict[str, Any] = None, requires_confirmation: bool = False, confirmation_message_template: str = None ): self.func = func self.needs_context = 'context' in inspect.signature(func).parameters self.name = name or func.__name__ self.description = description or (func.__doc__ or "").strip() tool_definition = tool_definition or self._generate_definition() super().__init__( name=self.name, description=self.description, tool_definition=tool_definition, requires_confirmation=requires_confirmation, confirmation_message_template=confirmation_message_template ) async def execute(self, context: ExecutionContext = None, **kwargs) -> Any: """Execute the wrapped function. Context is only required if the wrapped function has a 'context' parameter. """ if self.needs_context: if context is None: raise ValueError( f"Tool '{self.name}' requires a context parameter. " f"Please provide an ExecutionContext instance." ) result = self.func(context=context, **kwargs) else: result = self.func(**kwargs) # Handle both sync and async functions if inspect.iscoroutine(result): return await result return result def _generate_definition(self) -> Dict[str, Any]: """Generate tool definition from function signature.""" parameters = function_to_input_schema(self.func) return format_tool_definition(self.name, self.description, parameters) def tool( func: Callable = None, *, name: str = None, description: str = None, tool_definition: Dict[str, Any] = None, requires_confirmation: bool = False, confirmation_message: str = None ): """Decorator to convert a function into a FunctionTool. Usage: @tool def my_function(x: int) -> int: return x * 2 # Or with parameters: @tool(name="custom_name", description="Custom description") def my_function(x: int) -> int: return x * 2 # With confirmation: @tool(requires_confirmation=True, confirmation_message="Delete file?") def delete_file(filename: str) -> str: ... """ from typing import Union def decorator(f: Callable) -> FunctionTool: return FunctionTool( func=f, name=name, description=description, tool_definition=tool_definition, requires_confirmation=requires_confirmation, confirmation_message_template=confirmation_message ) if func is not None: return decorator(func) return decorator