Parallel tool execution

When an LLM emits multiple tool calls in a single response, Flux runs them concurrently by default. A model that needs to look up weather for Tokyo, London, and New York issues three tool calls in one turn; Flux dispatches all three at the same time instead of waiting for each to finish before starting the next.

How it works

Inside the agent loop, each turn where the LLM returns tool calls lands in execute_tools() in tool_executor.py. When max_concurrent_tools is None (the default), all tool calls in that turn are dispatched with asyncio.gather:

return list(await asyncio.gather(*[_run_one(c) for c in tool_calls]))

No semaphore, no queue — every tool in the batch starts immediately. If the LLM emits five calls, five tasks start concurrently.

When max_concurrent_tools is set to an integer, Flux creates an asyncio.Semaphore with that value and wraps each call:

sem = asyncio.Semaphore(max_concurrent_tools)

async def _limited(call):
    async with sem:
        return await _run_one(call)

return list(await asyncio.gather(*[_limited(c) for c in tool_calls]))

asyncio.gather still runs — but no more than max_concurrent_tools calls hold the semaphore at once. The rest wait.

Setting the concurrency cap

Pass max_concurrent_tools to agent():

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

@task
async def fetch_weather(city: str) -> str:
    """Get the current weather for a city."""
    # ... call a weather API
    return f"Weather in {city}: 22°C, partly cloudy"

@task
async def fetch_population(city: str) -> str:
    """Get the population of a city."""
    # ... call a data API
    return f"Population of {city}: 9.7 million"

@workflow
async def city_research(ctx: ExecutionContext):
    assistant = await agent(
        "Research cities using your tools. Call multiple tools at once when possible.",
        model="ollama/mistral-small:24b",
        tools=[fetch_weather, fetch_population],
        max_concurrent_tools=4,  # at most 4 tools run at the same time
    )
    return await assistant("Compare Tokyo, London, and New York")

max_concurrent_tools=1 gives fully sequential execution — each tool finishes before the next starts. This is useful when tools share mutable state or when the downstream service cannot handle concurrent requests.

max_concurrent_tools=None (the default) means no cap. All tools in a given turn run at the same time.

Result ordering

asyncio.gather returns results in the order the coroutines were submitted, not in the order they complete. A slow tool call finishing last still ends up at the position corresponding to its place in the LLM’s original response. When Flux feeds results back to the LLM, each result is paired with its tool call by index — the LLM sees results in the same order it issued the calls.

Error propagation

Each tool call runs inside a try/except block. If a tool raises an exception, execute_tools() catches it, logs a warning, and returns {"output": "Error: <message>"} for that slot. The other concurrent tools continue running. All tools in the batch run to completion before the agent loop feeds the collected results back to the LLM — which then sees which calls succeeded and which returned error strings.

The one exception is PauseRequested. If a tool hits a workflow pause (for example, when a requires_approval-gated task fires its TASK_AWAITING_APPROVAL event), PauseRequested is re-raised immediately and propagates out of execute_tools(). The agent loop fires any on_pause hooks and re-raises, suspending the entire execution. Concurrent tools that had already started will complete, but their results are discarded because the execution is pausing.

Putting it together

The example below sets max_concurrent_tools=3 so at most three topics are researched at once, even if the LLM emits more:

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

@task
async def search_topic(topic: str) -> str:
    """Search for information about a topic."""
    await asyncio.sleep(1)  # simulate a network call
    return f"Results for {topic}: ..."

@task
async def get_statistics(subject: str) -> str:
    """Get statistics and data about a subject."""
    await asyncio.sleep(1)
    return f"Statistics for {subject}: ..."

@task
async def check_news(topic: str) -> str:
    """Check recent news about a topic."""
    await asyncio.sleep(1)
    return f"News for {topic}: ..."

@workflow
async def parallel_research(ctx: ExecutionContext):
    assistant = await agent(
        "You are a research assistant. Always call multiple tools at once when possible.",
        model="ollama/mistral-small:24b",
        name="parallel_researcher",
        tools=[search_topic, get_statistics, check_news],
        max_concurrent_tools=3,
        max_tool_calls=10,
    )
    return await assistant("Compare AI, quantum computing, and renewable energy trends")

If the LLM emits three tool calls simultaneously (one per topic), all three start at once. With asyncio.sleep(1) in each tool, the batch completes in roughly one second instead of three.

When to cap concurrency

Leave max_concurrent_tools=None (the default) when tools call independent external services that can handle concurrent requests and have no shared mutable state.

Set an integer cap when the downstream service enforces a rate limit, when tools write to the same resource and order matters, or when you want predictable execution while debugging.

Set max_concurrent_tools=1 for strict sequential execution, which runs tools one at a time in the order the LLM emitted them.