Skip to content

Base

base

Pydantic-based Tool class with validation and schema generation.

This module provides the core Tool class and related base classes for defining tools with input/output validation and auto-generated JSON schemas for LLM function calling.

ToolInput

Bases: BaseModel

Base class for tool input schemas.

Subclass this to define strict input validation for tools. The extra="forbid" config means unknown fields will be rejected.

Example

class MyToolInput(ToolInput): query: str = Field(..., description="Search query") limit: int = Field(default=10, description="Max results")

ToolOutput

Bases: BaseModel

Base class for tool output schemas.

Provides a standardized output format with success status and optional error.

Example

class SearchOutput(ToolOutput): results: List[str] = Field(default_factory=list)

Tool

Tool(function: Optional[Callable] = None, **data)

Bases: BaseModel

Pydantic-based tool with validation and schema generation.

Provides: - Input validation via Pydantic schemas - Auto-generated JSON schemas for LLM function calling - Consistent error handling - Backward compatibility with existing @tool decorator - Support for both sync and async functions

Attributes:

Name Type Description
name str

Unique tool identifier

description str

What the tool does (used in LLM prompts)

function Optional[Callable]

The callable to execute (excluded from serialization)

input_schema Optional[Type[ToolInput]]

Optional Pydantic model for input validation

output_schema Optional[Type[ToolOutput]]

Optional Pydantic model for output wrapping

category str

Tool category for organization

requires_confirmation bool

Whether to prompt user before execution

cacheable bool

Whether results can be cached

cache_ttl int

Cache time-to-live in seconds

Example

From a decorated function

@tool(description="Search the web") async def web_search(query: str) -> str: return f"Results for {query}"

tool_instance = Tool(function=web_search)

With explicit schema

class SearchInput(ToolInput): query: str

tool_instance = Tool( name="web_search", description="Search the web", function=web_search, input_schema=SearchInput )

Initialize a Tool instance.

Can be initialized either with just a function (backward compatible) or with explicit parameters.

Parameters:

Name Type Description Default
function Optional[Callable]

The function to wrap as a tool. If provided as positional arg, enables backward compatibility with old Tool(function) pattern.

None
**data

Additional tool configuration (name, description, etc.)

{}
Source code in reactive_agents/core/tools/base.py
def __init__(self, function: Optional[Callable] = None, **data):
    """Initialize a Tool instance.

    Can be initialized either with just a function (backward compatible)
    or with explicit parameters.

    Args:
        function: The function to wrap as a tool. If provided as positional arg,
                 enables backward compatibility with old Tool(function) pattern.
        **data: Additional tool configuration (name, description, etc.)
    """
    # Handle backward-compatible initialization: Tool(function)
    if function is not None and "function" not in data:
        data["function"] = function

    # Extract metadata from function if not provided
    func = data.get("function")
    if func is not None:
        # Get name from function if not provided
        if not data.get("name"):
            data["name"] = getattr(func, "__name__", "unnamed_tool")

        # Get description from function docstring or decorator if not provided
        if not data.get("description"):
            # First try to get from tool_definition (set by @tool decorator)
            if hasattr(func, "tool_definition"):
                func_def = func.tool_definition
                if isinstance(func_def, dict) and "function" in func_def:
                    data["description"] = func_def["function"].get("description", "")

            # Fallback to docstring
            if not data.get("description"):
                data["description"] = getattr(func, "__doc__", "") or ""

    super().__init__(**data)

tool_definition property

tool_definition: Dict[str, Any]

Generate OpenAI-compatible function schema for LLM tool calling.

Returns a schema in the format expected by OpenAI's function calling API: { "type": "function", "function": { "name": "tool_name", "description": "What the tool does", "parameters": { ... JSON Schema ... } } }

use async

use(params: Dict[str, Any]) -> Any

Execute tool with input validation.

Parameters:

Name Type Description Default
params Dict[str, Any]

Parameters to pass to the tool function

required

Returns:

Type Description
Any

Tool execution result, or ToolOutput with error on failure

Source code in reactive_agents/core/tools/base.py
async def use(self, params: Dict[str, Any]) -> Any:
    """Execute tool with input validation.

    Args:
        params: Parameters to pass to the tool function

    Returns:
        Tool execution result, or ToolOutput with error on failure
    """
    if self.function is None:
        return ToolOutput(success=False, error="Tool function not defined")

    # Validate input if schema exists
    if self.input_schema:
        try:
            validated = self.input_schema.model_validate(params)
            params = validated.model_dump()
        except Exception as e:
            return ToolOutput(success=False, error=f"Input validation failed: {e}")

    # Execute the function
    try:
        if asyncio.iscoroutinefunction(self.function):
            result = await self.function(**params)
        else:
            # Support sync functions too
            result = self.function(**params)

        # Wrap in ToolOutput if output_schema specified and result isn't already that type
        if self.output_schema and not isinstance(result, self.output_schema):
            return self.output_schema(result=result)

        return result

    except Exception as e:
        error_msg = f"Tool Execution Error: {e}"
        if self.output_schema:
            return self.output_schema(success=False, error=error_msg)
        return error_msg

from_function classmethod

from_function(function: Callable, name: Optional[str] = None, description: Optional[str] = None, input_schema: Optional[Type[ToolInput]] = None, output_schema: Optional[Type[ToolOutput]] = None, **kwargs) -> 'Tool'

Create a Tool from a function.

Factory method that extracts name and description from function metadata if not provided explicitly.

Parameters:

Name Type Description Default
function Callable

The callable to wrap

required
name Optional[str]

Override the tool name (defaults to function.name)

None
description Optional[str]

Override the description (defaults to docstring)

None
input_schema Optional[Type[ToolInput]]

Optional Pydantic model for input validation

None
output_schema Optional[Type[ToolOutput]]

Optional Pydantic model for output wrapping

None
**kwargs

Additional tool configuration

{}

Returns:

Type Description
'Tool'

A configured Tool instance

Example

async def my_tool(query: str) -> str: '''Search for something.''' return f"Results for {query}"

tool = Tool.from_function(my_tool)

name="my_tool", description="Search for something."
Source code in reactive_agents/core/tools/base.py
@classmethod
def from_function(
    cls,
    function: Callable,
    name: Optional[str] = None,
    description: Optional[str] = None,
    input_schema: Optional[Type[ToolInput]] = None,
    output_schema: Optional[Type[ToolOutput]] = None,
    **kwargs
) -> "Tool":
    """Create a Tool from a function.

    Factory method that extracts name and description from function metadata
    if not provided explicitly.

    Args:
        function: The callable to wrap
        name: Override the tool name (defaults to function.__name__)
        description: Override the description (defaults to docstring)
        input_schema: Optional Pydantic model for input validation
        output_schema: Optional Pydantic model for output wrapping
        **kwargs: Additional tool configuration

    Returns:
        A configured Tool instance

    Example:
        async def my_tool(query: str) -> str:
            '''Search for something.'''
            return f"Results for {query}"

        tool = Tool.from_function(my_tool)
        # name="my_tool", description="Search for something."
    """
    # Extract description from decorator or docstring if not provided
    final_description = description
    if not final_description:
        if hasattr(function, "tool_definition"):
            func_def = function.tool_definition
            if isinstance(func_def, dict) and "function" in func_def:
                final_description = func_def["function"].get("description", "")
        if not final_description:
            final_description = getattr(function, "__doc__", "") or ""

    return cls(
        name=name or getattr(function, "__name__", "unnamed_tool"),
        description=final_description,
        function=function,
        input_schema=input_schema,
        output_schema=output_schema,
        **kwargs
    )

__hash__

__hash__() -> int

Make Tool hashable for use in sets.

Source code in reactive_agents/core/tools/base.py
def __hash__(self) -> int:
    """Make Tool hashable for use in sets."""
    return hash(self.name)

__eq__

__eq__(other: object) -> bool

Check equality based on name.

Source code in reactive_agents/core/tools/base.py
def __eq__(self, other: object) -> bool:
    """Check equality based on name."""
    if isinstance(other, Tool):
        return self.name == other.name
    return False