Concurrency in Agent Systems: Parallel Tools, Fan-Out, and Rate Limits

A financial research agent was supposed to fetch ten stock quotes and summarize them. The implementation did exactly that — sequentially. Each get_quote call averaged 350 ms. The agent needed 3.5 seconds before the model could begin its summary. Multiply by three more tool rounds and the full task took 15 seconds. A one-line change to asyncio.gather() brought that to 4 seconds. The latency math had been sitting there the whole time; nobody had bothered to make it visible.

Concurrency is where most agent implementations leave half their performance on the table. This article covers the mechanics of parallel tool execution, the fan-out/join pattern at the orchestrator level, rate-limit-aware scheduling, and how to cancel in-flight work without leaving resource leaks behind. The agent loop fundamentals and tool round-trip mechanics are prerequisites — this article picks up where those leave off.

{ "type": "agent-loop", "scenario": "weather-trip", "title": "The agent loop these tool calls run inside" }

How the API parallel tool call actually works

When a model decides to call tools in one response, both the OpenAI and Anthropic APIs can return multiple tool invocations in the same message. OpenAI represents these as an array of tool_calls; Anthropic returns multiple tool_use content blocks. The protocol is explicit: the caller must execute all of them and return all results before the model continues.

import asyncio
import anthropic

client = anthropic.AsyncAnthropic()

async def execute_tool(tool_name: str, tool_input: dict) -> str:
    # your actual tool dispatch here
    await asyncio.sleep(0.1)  # simulate I/O
    return f"result of {tool_name}"

async def run_agent_turn(messages: list) -> str:
    response = await client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        tools=[...],  # your tool definitions
        messages=messages,
    )

    if response.stop_reason != "tool_use":
        return response.content[0].text

    # collect all tool_use blocks
    tool_calls = [
        block for block in response.content if block.type == "tool_use"
    ]

    # execute ALL of them in parallel
    results = await asyncio.gather(
        *[execute_tool(tc.name, tc.input) for tc in tool_calls],
        return_exceptions=True,
    )

    # build the tool_results turn — one turn for all results
    tool_results = [
        {
            "type": "tool_result",
            "tool_use_id": tc.id,
            "content": str(result) if not isinstance(result, Exception) else f"Error: {result}",
        }
        for tc, result in zip(tool_calls, results)
    ]

    # continue the conversation
    return await run_agent_turn(
        messages + [
            {"role": "assistant", "content": response.content},
            {"role": "user", "content": tool_results},
        ]
    )

The return_exceptions=True flag on asyncio.gather is not optional: without it, the first tool failure raises immediately while the other coroutines keep running unobserved — you lose their results and leave orphaned work in flight. (Cancelling siblings on failure is asyncio.TaskGroup behavior, not gather's — more on that in the cancellation section.) With the flag, each result is either a value or an exception object — the model can then decide whether the error is recoverable.

One gotcha on OpenAI: strict: true in Structured Outputs mode is incompatible with parallel_tool_calls. If you enable strict structured outputs, you must set parallel_tool_calls: false explicitly — you give up model-batched parallel calls to get schema enforcement. This trips up more engineers than it should — the feature flags sound independent but they conflict at the schema-enforcement layer.

The fan-out/join pattern

A single model turn returning multiple tool calls is the micro version of the problem. The macro version is when an orchestrator agent spawns multiple subagents or calls multiple external services whose results it will synthesize. This is the fan-out/join pattern.

sequenceDiagram
    participant O as Orchestrator
    participant A as Subagent A
    participant B as Subagent B
    participant C as Subagent C
    participant J as Join / Synthesizer

    O->>A: task slice 1
    O->>B: task slice 2
    O->>C: task slice 3
    Note over A,C: run concurrently
    A-->>J: result 1
    C-->>J: result 3 (fastest)
    B-->>J: result 2
    J->>O: merged result

The split between gather and as_completed matters here. asyncio.gather waits for the slowest task before returning anything — right join semantics. asyncio.as_completed yields each result as it finishes — you can start synthesizing early if partial results are useful.

async def fan_out_gather(tasks: list[Coroutine]) -> list:
    """Wait for all. Good when all results are required before proceeding."""
    return await asyncio.gather(*tasks, return_exceptions=True)

async def fan_out_progressive(tasks: list[Coroutine], handler):
    """Process each result as it arrives. Good for streaming synthesis."""
    pending = [asyncio.ensure_future(t) for t in tasks]
    for coro in asyncio.as_completed(pending):
        result = await coro
        await handler(result)

Which to choose depends on the synthesis step. If the orchestrator model needs all facts before it can reason (entity resolution across multiple databases), gather is correct. If the model can begin drafting while additional data arrives (parallel web searches that inform a report), progressive is better — you cut latency by letting the synthesis overlap with slow fetches.

Bounded fan-out

The model might generate 20 parallel tool calls on a complex task. Launching 20 concurrent HTTP requests without restraint has two failure modes: you exhaust your rate-limit quota in one burst, or you overwhelm a downstream service that has its own concurrency ceiling. Both result in 429s and retries that eat more time than sequential execution would have.

The fix is a bounded worker pool:

async def bounded_fan_out(
    tasks: list[Coroutine],
    max_concurrency: int,
) -> list:
    semaphore = asyncio.Semaphore(max_concurrency)

    async def run_with_limit(task):
        async with semaphore:
            return await task

    return await asyncio.gather(
        *[run_with_limit(t) for t in tasks],
        return_exceptions=True,
    )

The right value for max_concurrency is not a guess — it comes from your provider's rate limit divided by a safety factor. If OpenAI tier 3 gives you 10,000 RPM (≈ 167 RPS), and each tool call hits OpenAI, a semaphore of 40–60 at ~400 ms per call sustains ~100–150 requests per second — 60–90% of the 167 RPS ceiling. Size toward the low end and leave headroom: other parts of your system share the same API key.

Rate-limit-aware scheduling

A semaphore caps concurrency. That is necessary but not sufficient when you also have a token-per-minute ceiling, not just a requests-per-minute one. At 10M TPM with an average of 2,000 tokens per call, you can sustain 5,000 calls per minute — 83 per second. But nothing stops a fan-out-heavy workload from firing 200 calls per second early in the window: at 400,000 tokens per second, the 10M budget is gone in 25 seconds, and every call in the remaining 35 seconds eats a 429.

A token-bucket scheduler solves this by smoothing the burst:

import time
import asyncio
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    capacity: float          # max tokens in bucket
    refill_rate: float       # tokens added per second
    _tokens: float = field(init=False)
    _last_refill: float = field(init=False)

    def __post_init__(self):
        self._tokens = self.capacity
        self._last_refill = time.monotonic()

    def consume(self, amount: float) -> float:
        """Reserve `amount` tokens; return seconds to wait before proceeding.

        Debits immediately, even into a negative balance, so every caller
        pays for its own reservation. Without this, N concurrent callers
        would all compute the same deficit against unchanged state and
        stampede through together after the sleep.
        """
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(self.capacity, self._tokens + elapsed * self.refill_rate)
        self._last_refill = now

        self._tokens -= amount
        if self._tokens >= 0:
            return 0.0
        return -self._tokens / self.refill_rate

# 10M TPM = 166,667 tokens/second
tpm_bucket = TokenBucket(capacity=200_000, refill_rate=166_667)

async def rate_limited_call(tokens_needed: int, call_fn):
    wait = tpm_bucket.consume(tokens_needed)
    if wait > 0:
        await asyncio.sleep(wait)
    return await call_fn()

The debit happens before the sleep: the bucket goes negative and the returned wait is the time for it to refill back to zero. That reservation is what keeps the limiter honest under bursts. And because consume is synchronous, no await can interleave between the balance read and the write — within one event loop it needs no lock. If you make it awaitable, add an asyncio.Lock.

For multi-tenant systems where different users or tasks share a pool, LiteLLM's proxy layer handles this at the gateway level with per-key token budgets — you don't have to rebuild it per service. But for a single agent system you own end to end, an in-process token bucket is simpler and avoids the extra hop.

Retry and backoff under rate limits

429s happen. The right retry strategy is exponential backoff with jitter, bounded by a maximum wait:

import random

from openai import RateLimitError

async def with_retry(fn, max_retries: int = 4, base_delay: float = 1.0):
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except RateLimitError as e:
            if attempt == max_retries:
                raise
            # exponential backoff with full jitter
            delay = min(base_delay * (2 ** attempt), 60.0)
            jitter = random.uniform(0, delay)
            await asyncio.sleep(jitter)
        except Exception:
            raise  # non-rate-limit errors propagate immediately

Full jitter (random between 0 and the computed ceiling) is effective at reducing thundering herd collisions when many parallel tasks hit a 429 simultaneously — avoid fixed-interval backoff, where all tasks wake simultaneously and replicate the original burst exactly.

One detail worth getting right: the 429 response from OpenAI includes a Retry-After header. Parse it and use it as the minimum sleep duration rather than computing your own:

async def with_retry_after(fn, max_retries: int = 4):
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except RateLimitError as e:
            if attempt == max_retries:
                raise
            # prefer the provider's declared retry window
            header = e.response.headers.get("retry-after")
            retry_after = float(header) if header else float(2 ** attempt)
            await asyncio.sleep(retry_after + random.uniform(0, 1))

Cancellation

The most underbuilt part of parallel agent systems is cancellation. When a user presses Stop, when a parent agent times out, or when a downstream dependency fails hard enough to make the entire fan-out pointless, you need to stop in-flight work. Not just "stop reading results" — actually stop the work.

The distinction matters for rate limits. If you abandon a coroutine that has already issued an HTTP request, the request continues executing on the server side. It counts against your quota. It might write to a database. It might send an email. Stopping at the result-reader level is a measurement lie.

import httpx
import asyncio

async def cancellable_tool_call(
    client: httpx.AsyncClient,
    cancel_event: asyncio.Event,
    endpoint: str,
    payload: dict,
) -> dict | None:
    # if already cancelled before we start, bail immediately
    if cancel_event.is_set():
        return None

    # race between the actual request and the cancel signal
    request_task = asyncio.create_task(
        client.post(endpoint, json=payload)
    )
    cancel_task = asyncio.create_task(cancel_event.wait())

    done, pending = await asyncio.wait(
        [request_task, cancel_task],
        return_when=asyncio.FIRST_COMPLETED,
    )

    # cancel the loser
    for task in pending:
        task.cancel()
        try:
            await task
        except (asyncio.CancelledError, Exception):
            pass

    if cancel_task in done:
        return None  # cancelled

    response = request_task.result()
    return response.json()

This pattern races the HTTP call against a shared asyncio.Event. When the cancel fires, the HTTP task is cancelled at the asyncio level — which closes the underlying connection, preventing the server from receiving the full request body if it hasn't yet, and preventing your client from keeping the socket open. Pass the same cancel_event to all branches of a fan-out to get coordinated shutdown:

async def fan_out_with_cancel(
    tasks_and_payloads: list[tuple],
    cancel_event: asyncio.Event,
) -> list:
    client = httpx.AsyncClient()
    async with client:
        results = await asyncio.gather(
            *[
                cancellable_tool_call(client, cancel_event, ep, pl)
                for ep, pl in tasks_and_payloads
            ],
            return_exceptions=True,
        )
    return [r for r in results if r is not None]

flowchart TD
    PARENT["Parent agent\nor user interrupt"] -->|"cancel_event.set()"| CE["asyncio.Event\n(shared)"]
    CE --> T1["Tool call A\n(racing against event)"]
    CE --> T2["Tool call B\n(racing against event)"]
    CE --> T3["Tool call C\n(racing against event)"]
    T1 -->|"cancelled"| CONN1["HTTP conn closed\n(no quota consumed)"]
    T2 -->|"result arrived first"| RESULT2["Result kept"]
    T3 -->|"cancelled"| CONN3["HTTP conn closed"]
    RESULT2 --> JOIN["Partial join"]
    style CE fill:#00e5ff,color:#0a0a0f
    style CONN1 fill:#15803d,color:#fff
    style CONN3 fill:#15803d,color:#fff
    style RESULT2 fill:#0e7490,color:#fff

TaskGroup vs gather

Python 3.11's asyncio.TaskGroup is the structured-concurrency primitive that does what many engineers wrongly assume gather does: when one task raises, the group cancels every sibling, waits for the cancellations to finish, and re-raises the failures as an ExceptionGroup. Nothing escapes the async with scope still running.

async def fan_out_all_or_nothing(payloads: list[dict]) -> list[dict]:
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(p)) for p in payloads]
    # reaching this line means every task succeeded
    return [t.result() for t in tasks]

The rule of thumb: use TaskGroup when the fan-out is all-or-nothing — one failure invalidates the batch, so cancelling the siblings saves quota and time. Use gather(return_exceptions=True) when partial results are useful — which for agent tool execution is most of the time, because the model reasons over partial failure better than your orchestrator can. And a TaskGroup wrapped around cancellable_tool_call branches combines both mechanisms: the shared event handles external cancellation, the group guarantees nothing outlives the scope.

Worked numbers: what parallelism actually costs you

Scenario: research agent fetching 8 data sources per query.
Each external API call: 400 ms avg, 900 ms p95.
Provider rate limit: 600 RPM (10 RPS).

Sequential execution:
  Latency: 8 × 400 ms = 3,200 ms (avg)
  Throughput: 10 RPS / 8 calls per query = 1.25 queries/second max

Parallel execution (all 8 at once):
  Latency: max(8 × Uniform[200, 900]) ≈ 800 ms avg
  Instantaneous rate-limit hit: 8 calls at t=0 → 8 RPM consumed in one tick

With semaphore(4) — 4 concurrent calls:
  Latency: two waves of max(4 × Uniform[200, 900]) ≈ 2 × 760 ms ≈ 1,500 ms avg
  Rate-limit pressure: max 4 in-flight at any moment
  Throughput: still capped by 10 RPS ÷ 8 calls/query = 1.25 queries/second
  (concurrency moves latency, not the rate-limit ceiling)

Recommendation: semaphore(4) at this limit, targeting 70% RPM utilization.
At 600 RPM ceiling: 600 × 0.7 = 420 "slots" per minute available.
With 8 calls/query: 420 / 8 = 52 queries/minute sustainable.
Headroom: 600 – 420 = 180 RPM reserved for retries and other traffic.

The right concurrency number is an arithmetic result, not a tuning knob you adjust until things stop breaking. Compute it from the rate limit, your per-query call count, and a headroom factor (0.7 is a reasonable starting point). Instrument it — track your actual RPM against the ceiling in your observability layer — and adjust when your traffic profile changes.

{ "type": "error-compound", "accuracy": 0.85, "steps": 10, "title": "Why sequential failures hurt agent completion rates" }

What breaks

Unbounded fan-out on model hallucination

The model returns 30 tool_use blocks. Your agent launches 30 concurrent requests. You hit the rate limit instantly, 28 calls return 429s, retries pile up, total latency is worse than sequential. The root cause: the model misunderstood its task and decided to be thorough in exactly the wrong way. The fix is a hard cap in the orchestrator layer:

MAX_PARALLEL_TOOLS = 8  # never lift this above your rate-limit math

if len(tool_calls) > MAX_PARALLEL_TOOLS:
    # either truncate or group into sequential batches
    tool_calls = tool_calls[:MAX_PARALLEL_TOOLS]

This is not a hack — it is a contract: the orchestrator is responsible for honoring rate limits, not the model.

Write tool races

Two parallel update_record calls with the last-write-wins semantics corrupt data. The model returning them in the same response does not mean they are safe to run concurrently. Any tool that mutates shared state needs to be identified at schema definition time and routed through a serialized queue:

READ_TOOLS = {"get_user", "search_documents", "get_order_status"}
WRITE_TOOLS = {"update_record", "send_email", "create_ticket"}

async def dispatch(tool_calls: list):
    reads = [tc for tc in tool_calls if tc.name in READ_TOOLS]
    writes = [tc for tc in tool_calls if tc.name in WRITE_TOOLS]

    # reads in parallel, writes in sequence
    read_results = await asyncio.gather(*[execute(tc) for tc in reads])
    write_results = []
    for tc in writes:
        write_results.append(await execute(tc))

    return read_results + write_results

Context rot in long fan-outs

A subagent that runs for 20 tool turns accumulates a 60,000-token context. Fan out 5 of these and your orchestrator is synthesizing from 300,000 tokens of accumulated agent history, most of it tool result text that is no longer relevant. This is the same context rot problem covered in context window management for agents, amplified by the number of subagents.

The fix is to return structured summaries, not raw transcripts. Each subagent's final output should be a compressed fact object — what it found, what it tried, what failed — not its full message history.

Partial join failures

asyncio.gather(return_exceptions=True) means some tasks return values and others return exception objects. If your merge code assumes all results are valid, an exception object in the list will crash the synthesizer in a way that looks like a synthesis bug rather than a tool failure.

results = await asyncio.gather(*tasks, return_exceptions=True)

successes = []
failures = []
for i, result in enumerate(results):
    if isinstance(result, Exception):
        failures.append({"task": i, "error": str(result)})
    else:
        successes.append(result)

# surface failures to the model explicitly
if failures:
    # include failure info in the tool_results turn
    # so the model can decide whether to retry or proceed without that data
    pass

The model should know which subtasks failed. Hiding failures and proceeding with partial data produces confident wrong answers — worse than a visible error.

Rate limit quota leakage from orphaned requests

A parent task times out after 5 seconds. The children launched at t=0 are still running at t=5. Without cancellation at the transport layer, they run to completion, consuming quota. At scale — 100 concurrent agent sessions with 8 parallel tool calls each — unmanaged orphan requests can consume 30–40% of your rate limit budget on work that will never be used. This is the cancellation problem described above; it is not hypothetical.

Multi-agent fan-out

When the fan-out units are subagents rather than single tool calls, the pattern is the same but the stakes are higher. Each subagent runs its own agent loop, accumulates tokens across multiple turns, and can itself spawn parallel tool calls. The cost amplification is real: a single orchestrator turn that fans out to 5 subagents, each running 10 tool turns, generates 50 LLM calls and potentially hundreds of tool calls. Unbudgeted, this is how a single user request generates a $50 API bill.

The production controls are:

• Token budget per subagent, enforced by passing max_tokens to each spawned context and tracking usage.
• Iteration cap per subagent, hard-coded, not left to the model's self-assessment.
• Total budget for the orchestrator turn, which the orchestrator checks before spawning each additional subagent.

LangGraph expresses this as a graph with a shared RemainingBudget state object that every node reads and decrements. When the budget hits zero, all pending node executions are cancelled. The multi-agent orchestration patterns article covers the topology choices; the concurrency concern here is orthogonal — it applies regardless of which orchestration pattern you choose.

flowchart TD
    ORCH["Orchestrator\ntoken_budget=50k"] -->|"spawn (15k budget)"| SA1["Subagent A\n~3 turns, 12k used"]
    ORCH -->|"spawn (15k budget)"| SA2["Subagent B\n~4 turns, 14k used"]
    ORCH -->|"spawn (15k budget)"| SA3["Subagent C\n~2 turns, 8k used"]
    SA1 -->|"result + 12k used"| BUDGET["Budget tracker\n34k consumed / 50k"]
    SA2 -->|"result + 14k used"| BUDGET
    SA3 -->|"result + 8k used"| BUDGET
    BUDGET -->|"16k remaining"| SYNTH["Synthesis turn\n~5k tokens"]
    style ORCH fill:#a855f7,color:#fff
    style BUDGET fill:#00e5ff,color:#0a0a0f
    style SYNTH fill:#0e7490,color:#fff

The decision in practice

The choice of how much parallelism to run is not a one-time architectural decision — it is a per-workflow configuration that should match the structure of the task.

Run everything in parallel when: all tools are read-only, tasks are independent, and you have verified the combined token burst fits inside your rate-limit window. Standard web research agents fall here.

Run in bounded batches when: you have mixed read/write tools, or you are fanning out to more subagents than your rate limit can absorb. Size the batches from the rate-limit arithmetic above.

Run sequentially when: each step's output is the next step's input (no independence), or you have write tools with shared mutable state that you cannot make idempotent. Sequential is not a performance failure — it is the correct model for dependent pipelines.

Serialize writes, parallelize reads within each turn when: you have a mix and can identify the tool types at definition time. This is the most common real-world case and the read/write dispatch pattern above handles it directly.

The tool-use round-trip mechanics article shows you the raw protocol; reliable tool-using systems covers the surrounding production patterns. What this article adds is the scheduling layer that sits between them: the rate-limit math, the semaphore sizing, the cancellation hygiene, and the bounded fan-out that stops a hallucinating model from blowing your entire daily quota in a single turn.

One final number worth keeping in mind: at 85% per-step accuracy, a 10-step agent succeeds 20% of the time. That compounding applies to model decisions, not to how fast your tools run — so be precise about which win is which. Batching multiple tool calls into a single model turn reduces the number of decisions the model makes, and that is where the completion-rate math improves. Running those batched calls concurrently is the latency win layered on top.