Tools API Reference¶

Complete API reference for Tool, ToolResult, FunctionTool, the @tool decorator, ToolRegistry, and built-in tools.

`Tool` (Protocol)¶

Tool ¶

Bases: Protocol

Protocol for a tool that can be used by an agent.

Attributes¶

name `property` ¶

name: str

Unique name of the tool.

description `property` ¶

description: str

Description shown to the model.

parameters_schema `property` ¶

parameters_schema: dict[str, Any]

JSON Schema for the tool's parameters.

Methods:¶

execute `async` ¶

execute(ctx: ToolContext, args: dict[str, Any]) -> ToolResult

Execute the tool with the given arguments.

Source code in flux\tools\base.py

async def execute(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult:
    """Execute the tool with the given arguments."""
    ...

@runtime_checkable
class Tool(Protocol):
    """Protocol for a tool that can be used by an agent."""

    @property
    def name(self) -> str: ...

    @property
    def description(self) -> str: ...

    @property
    def parameters_schema(self) -> dict[str, Any]: ...

    async def execute(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult: ...

The Tool protocol defines the interface that all tools must implement. Because it uses @runtime_checkable, you can use isinstance(obj, Tool) to check whether an object satisfies the protocol.

Properties¶

Property	Type	Description
`name`	`str`	Unique identifier for the tool. Used by the model to invoke it.
`description`	`str`	Human-readable description shown to the LLM. Should explain what the tool does and when to use it.
`parameters_schema`	`dict[str, Any]`	JSON Schema object describing the tool's accepted parameters.

Methods¶

`execute`¶

async def execute(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult:

Execute the tool with the given arguments.

Parameter	Type	Description
`ctx`	`ToolContext`	Context for this invocation, providing access to the `RunContext`, tool name, tool call ID, and raw arguments string.
`args`	`dict[str, Any]`	Parsed arguments from the model's tool call.

Returns: ToolResult

Implementing a Custom Tool¶

Any object with the correct properties and an async execute method satisfies the protocol:

from flux.tools.base import Tool, ToolResult
from flux.context import ToolContext

class MyTool:
    @property
    def name(self) -> str:
        return "my_tool"

    @property
    def description(self) -> str:
        return "Does something useful."

    @property
    def parameters_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "The query to process"}
            },
            "required": ["query"],
        }

    async def execute(self, ctx: ToolContext, args: dict) -> ToolResult:
        query = args.get("query", "")
        return ToolResult(output=f"Processed: {query}", success=True)

# Verify protocol compliance
assert isinstance(MyTool(), Tool)  # True

`ToolResult`¶

ToolResult `dataclass` ¶

ToolResult(
    output: str, success: bool = True, metadata: dict[str, Any] = dict()
)

Result of a tool execution.

@dataclass
class ToolResult:
    """Result of a tool execution."""
    output: str
    success: bool = True
    metadata: dict[str, Any] = field(default_factory=dict)

Returned by every tool's execute method. The output string is sent back to the model as a tool result message.

Parameters¶

Parameter	Type	Default	Description
`output`	`str`	required	The result string returned to the model.
`success`	`bool`	`True`	Whether the tool executed successfully.
`metadata`	`dict[str, Any]`	`{}`	Arbitrary metadata (e.g., execution time, source URL).

Usage¶

from flux.tools.base import ToolResult

# Success
result = ToolResult(output="Temperature is 72F")

# Failure
result = ToolResult(
    output="File not found: /tmp/data.csv",
    success=False,
    metadata={"file_path": "/tmp/data.csv"},
)

`FunctionTool`¶

FunctionTool ¶

FunctionTool(
    func: Callable, name: str | None = None, description: str | None = None
)

A tool created from a Python function.

Source code in flux\tools\decorator.py

def __init__(
    self,
    func: Callable,
    name: str | None = None,
    description: str | None = None,
) -> None:
    self._func = func
    self._name = name or func.__name__
    self._description = description or func.__doc__ or f"Tool: {self._name}"
    self._schema = function_to_schema(func)
    self._is_async = asyncio.iscoroutinefunction(func)
    self._accepts_context = _check_context_param(func)

Methods:¶

execute `async` ¶

execute(ctx: ToolContext, args: dict[str, Any]) -> ToolResult

Execute the wrapped function.

Source code in flux\tools\decorator.py

async def execute(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult:
    """Execute the wrapped function."""
    try:
        if self._accepts_context:
            result = self._func(ctx, **args) if self._is_async else self._func(ctx, **args)
        else:
            result = self._func(**args) if self._is_async else self._func(**args)

        if asyncio.iscoroutine(result):
            result = await result

        return ToolResult(output=str(result), success=True)
    except Exception as e:
        return ToolResult(output=f"Error: {e}", success=False)

class FunctionTool:
    """A tool created from a Python function."""

    def __init__(
        self,
        func: Callable,
        name: str | None = None,
        description: str | None = None,
    ) -> None:
        self._func = func
        self._name = name or func.__name__
        self._description = description or func.__doc__ or f"Tool: {self._name}"
        self._schema = function_to_schema(func)
        self._is_async = asyncio.iscoroutinefunction(func)
        self._accepts_context = _check_context_param(func)

Wraps a Python function into a Tool-compatible object. Schema generation, sync/async handling, and context injection are all handled automatically.

Parameters¶

Parameter	Type	Default	Description
`func`	`Callable`	required	The function to wrap.
`name`	`str \\| None`	`None`	Tool name. Defaults to `func.__name__`.
`description`	`str \\| None`	`None`	Tool description. Defaults to the function's docstring.

Properties¶

Property	Type	Description
`name`	`str`	The tool's name.
`description`	`str`	The tool's description.
`parameters_schema`	`dict[str, Any]`	Auto-generated JSON Schema from type hints and docstring.

Methods¶

`execute`¶

async def execute(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult:

Executes the wrapped function. Handles:

Async functions: awaited automatically.
Sync functions: called directly (no thread offload).
Context injection: if the function has a parameter annotated with ToolContext or RunContext (named ctx, context, tool_context, or run_context), it is injected automatically.
Exceptions: caught and returned as ToolResult(success=False).

Returns: ToolResult

Usage¶

from flux.tools.decorator import FunctionTool

def add(a: int, b: int) -> int:
    """Add two numbers.

    Args:
        a: First number.
        b: Second number.
    """
    return a + b

tool = FunctionTool(add)
print(tool.name)                # "add"
print(tool.parameters_schema)   # {"type": "object", "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}}, "required": ["a", "b"]}

`@tool` Decorator¶

tool ¶

tool(
    _func: Callable | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
) -> Callable

Decorator to create a Tool from a function.

Can be used with or without parentheses

@tool def my_tool(x: str) -> str: ...

@tool(name="custom_name", description="Custom description") def my_tool(x: str) -> str: ...

Source code in flux\tools\decorator.py

def tool(
    _func: Callable | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
) -> Callable:
    """Decorator to create a Tool from a function.

    Can be used with or without parentheses:
        @tool
        def my_tool(x: str) -> str: ...

        @tool(name="custom_name", description="Custom description")
        def my_tool(x: str) -> str: ...
    """

    def decorator(func: Callable) -> FunctionTool:
        return FunctionTool(func, name=name, description=description)

    if _func is not None:
        # @tool without parentheses
        return decorator(_func)

    # @tool() with parentheses or @tool(name=...)
    return decorator

def tool(
    _func: Callable | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
) -> Callable:

Decorator to create a FunctionTool from a Python function. The most common way to define tools in Flux.

Can be used with or without parentheses:

from flux.tools.decorator import tool

# Without parentheses
@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny in {city}"

# With parentheses
@tool(name="fetch_weather", description="Fetch weather data")
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny in {city}"

Parameters¶

Parameter	Type	Default	Description
`_func`	`Callable \\| None`	`None`	The decorated function (set automatically when used without parentheses).
`name`	`str \\| None`	`None`	Custom tool name.
`description`	`str \\| None`	`None`	Custom description.

Returns: FunctionTool

Context Injection¶

If the function declares a parameter named ctx, context, tool_context, or run_context with a context-related type annotation, ToolContext is injected at call time:

from flux.context import ToolContext

@tool
def log_message(ctx: ToolContext, message: str) -> str:
    """Log a message with context."""
    print(f"Tool: {ctx.tool_name}, Turn: {ctx.run_context.turn_count}")
    return f"Logged: {message}"

Async Support¶

Both sync and async functions are supported:

import httpx

@tool
async def fetch_url(url: str) -> str:
    """Fetch content from a URL."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.text

Schema Generation¶

The @tool decorator automatically generates a JSON Schema from:

Type hints on function parameters.
Google-style docstrings (Args: section) for parameter descriptions.
Parameters named ctx, context, tool_context, or run_context are excluded from the schema.

@tool
def search(query: str, max_results: int = 5) -> str:
    """Search the web.

    Args:
        query: The search query.
        max_results: Maximum number of results to return.
    """
    return f"Results for: {query}"

Generated schema:

{
    "type": "object",
    "properties": {
        "query": {
            "type": "string",
            "description": "The search query."
        },
        "max_results": {
            "type": "integer",
            "description": "Maximum number of results to return."
        }
    },
    "required": ["query"]
}