Tools and structured output

Pass Flux @task functions as tools for the agent to call, and use Pydantic models to get typed JSON output instead of plain strings.

Two parameters on agent() control what the LLM can do beyond producing text: tools lets the agent call Flux tasks during its loop, and response_format makes it return a Pydantic model instance instead of a string. You can use either independently or together, with one important constraint covered below.

Declaring tools

Pass a list of Flux @task functions to the tools parameter. The agent calls them autonomously during the agentic loop — each invocation is a fully durable Flux task execution, recorded in the event log.

from flux import task, workflow, ExecutionContext
from flux.tasks.ai import agent

@task
async def add(a: int, b: int) -> int:
    """Add two numbers and return the result."""
    return a + b

@task
async def multiply(a: int, b: int) -> int:
    """Multiply two numbers and return the result."""
    return a * b

@workflow
async def math_agent(ctx: ExecutionContext):
    assistant = await agent(
        "You are a math assistant. Use the available tools to compute answers precisely.",
        model="anthropic/claude-sonnet-4-20250514",
        tools=[add, multiply],
    )
    return await assistant(ctx.input["question"])

Run it:

export ANTHROPIC_API_KEY="sk-ant-..."
flux workflow run math_agent '{"question": "What is (12 + 7) * 3?"}'

The agent decides when and how many times to call each tool. It can call add first, then pass the result to multiply, or call them in a different order — the LLM figures out the sequence. By default it caps at 10 tool-call iterations (max_tool_calls=10); once that limit is reached, the agent produces a final answer from whatever it has gathered.

How Flux generates tool schemas

Flux reads each tool’s function signature, docstring, and type hints to build the JSON schema it sends to the provider. The build_tool_schemas function in flux.tasks.ai.tool_executor does this automatically — you do not write schemas by hand.

Given this task:

@task
async def add(a: int, b: int) -> int:
    """Add two numbers and return the result."""
    return a + b

Flux generates a neutral schema:

{
  "name": "add",
  "description": "Add two numbers and return the result.",
  "parameters": {
    "type": "object",
    "properties": {
      "a": {"type": "integer"},
      "b": {"type": "integer"}
    },
    "required": ["a", "b"]
  }
}

Flux then translates this neutral schema into the wire format each provider expects — Anthropic uses input_schema, OpenAI and Ollama use parameters inside a function wrapper, and Gemini uses FunctionDeclaration. You declare tools the same way regardless of which provider is configured. Switching from anthropic/... to openai/... or ollama/... requires no changes to your tool definitions.

Write docstrings. The first line of the docstring becomes the tool’s description field sent to the model. Vague or missing descriptions reduce tool-call accuracy across all providers.

Structured output with Pydantic

Pass a Pydantic BaseModel subclass to response_format to get typed output instead of a string:

from pydantic import BaseModel
from flux import workflow, ExecutionContext
from flux.tasks.ai import agent

class WeatherForecast(BaseModel):
    city: str
    temperature_celsius: float
    conditions: str
    precipitation_chance: float  # 0.0 to 1.0
    recommendation: str

@workflow
async def forecast_workflow(ctx: ExecutionContext):
    forecaster = await agent(
        "You are a weather analyst. Return a structured forecast based on the city provided. "
        "Use realistic estimates. precipitation_chance must be between 0.0 and 1.0.",
        model="anthropic/claude-sonnet-4-20250514",
        response_format=WeatherForecast,
    )
    result: WeatherForecast = await forecaster(ctx.input["city"])
    return {
        "city": result.city,
        "temperature": result.temperature_celsius,
        "conditions": result.conditions,
        "rain_chance": result.precipitation_chance,
        "tip": result.recommendation,
    }

The return type changes from str to an instance of WeatherForecast. Streaming is automatically disabled when response_format is set — stream=True (the default) is silently overridden to False.

For Anthropic, structured output is enforced at the API level. The Anthropic API has no response_format parameter, so Flux registers a synthetic tool (respond_with_structured_output) whose input_schema is the Pydantic model and forces a call to it with tool_choice. The model must call the tool, and its arguments are schema-valid by construction. OpenAI and Gemini use their native JSON-schema modes. In all three cases the agent loop validates the result with model_validate_json() before returning.

Ollama: tools and response_format are mutually exclusive

On Anthropic and OpenAI, tools and response_format can be combined — the agent calls tools during the loop and returns a structured model at the end.

Combining tools and structured output (Anthropic)

This example uses both. The agent calls a lookup tool and returns a structured result:

from pydantic import BaseModel
from flux import task, workflow, ExecutionContext
from flux.tasks.ai import agent

class ConversionResult(BaseModel):
    original_value: float
    original_unit: str
    converted_value: float
    target_unit: str
    formula: str

@task
async def convert_celsius_to_fahrenheit(celsius: float) -> float:
    """Convert a temperature from Celsius to Fahrenheit."""
    return (celsius * 9 / 5) + 32

@task
async def convert_km_to_miles(km: float) -> float:
    """Convert a distance from kilometres to miles."""
    return km * 0.621371

@workflow
async def unit_converter(ctx: ExecutionContext):
    converter = await agent(
        "You are a unit conversion assistant. Use tools to compute the exact converted value, "
        "then return a structured result explaining what was converted and how.",
        model="anthropic/claude-sonnet-4-20250514",
        tools=[convert_celsius_to_fahrenheit, convert_km_to_miles],
        response_format=ConversionResult,
    )
    result: ConversionResult = await converter(ctx.input["request"])
    return result.model_dump()
flux workflow run unit_converter '{"request": "Convert 100 km to miles"}'

The agent calls convert_km_to_miles(100) to get the precise value, then formats it into a ConversionResult.

Tool call limits

Raise max_tool_calls when a task requires many tool invocations, such as a research agent that reads multiple documents. Reduce it to prevent runaway loops on simple agents.

assistant = await agent(
    "You are a research assistant.",
    model="anthropic/claude-sonnet-4-20250514",
    tools=[search_web, fetch_page],
    max_tool_calls=25,
)

When the limit is reached, Flux forces the agent to produce a final answer from whatever tool results it has collected so far. The workflow does not raise an exception.

Tool replay safety

Each tool call inside the agent loop is a Flux task execution. On workflow crash and replay, completed tool calls are not re-executed — their recorded outputs are returned from the event log. Design tool functions to be idempotent where possible. See Agents in workflows for how the event log preserves the full agent state, including tool results, across restarts.

Next steps