Tools

The tools module provides production-safe tool decorators and toolset base classes for organizing AI agent capabilities.

safe_tool

fastroai.tools.safe_tool(timeout=DEFAULT_TOOL_TIMEOUT, max_retries=DEFAULT_TOOL_MAX_RETRIES, on_timeout=None, on_error=None)

Decorator that adds timeout, retry, and error handling to AI tools.

When a tool times out or raises an exception, instead of crashing the conversation, this decorator returns an error message that the AI can use to respond gracefully.

Parameters:

Name	Type	Description	Default
timeout	float	Maximum seconds per attempt. Default: 30.	DEFAULT_TOOL_TIMEOUT
max_retries	int	Maximum retry attempts. Default: 3.	DEFAULT_TOOL_MAX_RETRIES
on_timeout	str | None	Custom message returned on timeout. Default: "Tool timed out after {max_retries} attempts"	None
on_error	str | None	Custom message returned on error. Use {error} placeholder for error details. Default: "Tool failed: {error}"	None

Returns:

Type	Description
Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R | str]]]	Decorated async function with safety features.

Examples:

@safe_tool(timeout=10, max_retries=2)
async def web_search(query: str) -> str:
    '''Search the web for information.'''
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com?q={query}")
        return response.text

# If the API is slow or down:
# - Waits max 10 seconds per attempt
# - Retries up to 2 times with exponential backoff
# - Returns error message on final failure
# - AI sees: "Tool timed out after 2 attempts"

# With custom messages:
@safe_tool(
    timeout=30,
    on_timeout="Search is taking too long. Try a simpler query.",
    on_error="Search unavailable: {error}",
)
async def search(query: str) -> str:
    ...

SafeToolset

fastroai.tools.SafeToolset

Base class for toolsets containing only safe tools.

Safe tools are those that: - Don't access external networks (or have timeout protection) - Don't modify system state - Have bounded execution time - Return graceful error messages instead of raising exceptions

Use this as a base class to mark toolsets as production-safe.

Examples:

@safe_tool(timeout=5)
async def calculator(expression: str) -> str:
    '''Evaluate a math expression.'''
    try:
        # Safe: no network, no state modification
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@safe_tool(timeout=1)
async def get_time() -> str:
    '''Get current time.'''
    from datetime import datetime
    return datetime.now().isoformat()

class UtilityToolset(SafeToolset):
    def __init__(self):
        super().__init__(
            tools=[calculator, get_time],
            name="utilities",
        )

FunctionToolsetBase

fastroai.tools.FunctionToolsetBase

Base class for organized tool sets.

Extends PydanticAI's FunctionToolset with a name for identification and organization purposes.

Examples:

from fastroai.tools import safe_tool, FunctionToolsetBase

@safe_tool(timeout=30)
async def web_search(query: str) -> str:
    '''Search the web.'''
    ...

@safe_tool(timeout=10)
async def get_weather(location: str) -> str:
    '''Get weather for location.'''
    ...

class WebToolset(FunctionToolsetBase):
    def __init__(self):
        super().__init__(
            tools=[web_search, get_weather],
            name="web",
        )

# Use with FastroAgent
agent = FastroAgent(toolsets=[WebToolset()])

__init__(tools, name=None)

Initialize toolset with tools and optional name.

Parameters:

Name	Type	Description	Default
tools	list[Callable[..., Any]]	List of tool functions to include.	required
name	str | None	Name for this toolset. Defaults to class name.	None

---

← Pipelines Usage →