Redesign Your Agent Loop for Async Tool Calling

What changes when the model stops waiting

The standard agent loop is a strict alternation: the model emits a tool call, your application executes it, you hand back the result, the model continues. Every tool call is a barrier. The model's reasoning and your tool's latency are serialised even when they have nothing to do with each other, and the turn takes as long as the sum of its slowest parts. Async tool calling breaks that alternation. The model issues a call, is handed control back immediately, and carries on reasoning — calling other tools, answering the parts of the request that do not depend on the pending result — while your application runs the slow job in the background and delivers the result when it has one.

This is now a first-class API capability rather than a framework trick. OpenAI's developer documentation states plainly that "async tool calling is supported by GPT-6 Astra and later models", and that you "set async: true on a function or custom tool definition to let the model continue working after issuing that call". GPT-6 Astra was released to approved organisations on 3 September 2026, with general availability the following day. Google reached a similar place from a different direction: the Gemini Live API lets a function declaration carry a NON_BLOCKING behaviour, which "ensures the function runs asynchronously while you can continue interacting with the model". Anthropic's documented position as of September 2026 is parallel tool use rather than model-side async — several tool_use blocks per assistant turn, concurrency entirely at your discretion, results returned together in the next user message.

The capability is portable; the parameter names are not. What follows is the pattern — a task registry, a dispatch policy, a dependency gate and a reconciliation rule — with provider specifics shown where they differ. If you implement the pattern, swapping providers costs you an adapter, not a rewrite.

Prerequisites

You should already be running a tool-calling agent in production, with typed tool schemas and structured error returns; if that part is still shaky, fix it first, because async multiplies every schema ambiguity you already have. The groundwork is covered in the guides to designing agent tools with schemas, errors and retries and to debugging agent argument extraction. You should be comfortable with Python's asyncio — create_task, wait, wait_for and cancellation semantics — and you should have per-turn tracing in place before you start, not after.

The blocking loop and what it actually costs

Here is the loop nearly everyone ships first, written against the OpenAI Responses API but structurally identical everywhere. Note the inner for loop: each call is awaited to completion before the next one starts, and none of the results reach the model until all of them exist.

import json

MODEL = "gpt-6-astra"

async def blocking_turn(client, user_input, tools, impls):
    resp = await client.responses.create(model=MODEL, tools=tools, input=user_input)
    while True:
        calls = [i for i in resp.output if i.type == "function_call"]
        if not calls:
            return resp.output_text

        outputs = []
        for call in calls:                       # strictly serial: this is the cost
            result = await impls[call.name](**json.loads(call.arguments))
            outputs.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result),
            })

        resp = await client.responses.create(
            model=MODEL, tools=tools,
            previous_response_id=resp.id, input=outputs,
        )

Two separate wastes hide in that code, and they need separate fixes. The first is intra-batch serialisation: three independent lookups that each take two seconds take six, when they could have taken two. That one is cheap to fix — replace the loop with asyncio.gather — and it is what parallel tool use gives you on every major provider today. The second waste is the one async tool calling exists for: the model itself is idle for the entire duration. If a retrieval takes twenty seconds and the user also asked a question the model could answer from its own context, that answer waits twenty seconds for no reason at all. Fan a CI run or an estate-wide search into the mix and the idle window becomes the dominant term in your p95.

Before changing anything, instrument the loop to record, per turn, total wall-clock time and the time spent inside tool execution while no model tokens were being generated. That second number — tool-blocked time — is your entire addressable win. At ten per cent of turn time, async will not repay the complexity; at sixty per cent, as it often is for retrieval-heavy and CI-driven agents, it will.

Which tools qualify for async, and which never do

The classification is the whole safety argument, so do it explicitly and write it into the tool registry rather than leaving it to a prompt. The test has two parts. First: is the tool idempotent, so that running it twice, or running it and discarding the result, is harmless? Second: is there any point between dispatch and result at which the model could produce a wrong answer because the result is not yet in hand? A tool must pass both to be dispatched asynchronously.

Anthropic's guidance for parallel execution draws the same line in almost the same words — independent, read-only operations are usually safe to run concurrently, while tools with side effects, shared state or ordering requirements are better run sequentially — which is a useful sign that the boundary is a property of the work rather than of any one vendor's API.

The last three rows are where teams get hurt, and the failure looks exactly like a classic distributed-systems bug rather than an AI bug. If you have not already read it, the companion guide to concurrency bugs in multi-agent systems covers lost updates, duplicate side effects and write skew in detail; async tool calling introduces every one of them into a single agent, which previously only multi-agent systems had to worry about.

The redesigned loop

Three components replace the inner for loop: a registry that owns in-flight jobs and their handles, a drain step that converts finished jobs into tool outputs, and a wait tool the model can call when it genuinely needs a pending result. OpenAI's async flow uses exactly this shape — the model assigns a task_handle to each async call, your application binds that handle to the original call_id and the running job, and a wait_for_tasks function lets the model name the handles it now needs. Handles must stay unique for the whole conversation, including completed tasks and repeated lookups.

import asyncio, json, time, uuid
from dataclasses import dataclass, field

@dataclass
class Task:
    handle: str
    call_id: str
    name: str
    job: asyncio.Task
    started_at: float = field(default_factory=time.monotonic)

class TaskRegistry:
    def __init__(self):
        self._tasks: dict[str, Task] = {}

    def start(self, call_id, name, arguments, impl, handle=None):
        handle = handle or f"{name}-{uuid.uuid4().hex[:8]}"
        if handle in self._tasks:
            raise ValueError(f"duplicate task handle: {handle}")
        job = asyncio.create_task(impl(**arguments), name=handle)
        self._tasks[handle] = Task(handle, call_id, name, job)
        return handle

    def drain_completed(self):
        """Turn every finished job into a function_call_output item."""
        outputs = []
        for handle, task in list(self._tasks.items()):
            if not task.job.done():
                continue
            del self._tasks[handle]
            try:
                payload = {"status": "completed", "result": task.job.result()}
            except asyncio.CancelledError:
                payload = {"status": "cancelled", "handle": handle}
            except Exception as exc:
                payload = {"status": "failed", "handle": handle,
                           "error": f"{type(exc).__name__}: {exc}"}
            payload["elapsed_s"] = round(time.monotonic() - task.started_at, 3)
            outputs.append({
                "type": "function_call_output",
                "call_id": task.call_id,
                "output": json.dumps(payload),
            })
        return outputs

    async def wait_for(self, handles, timeout):
        jobs = [self._tasks[h].job for h in handles if h in self._tasks]
        if jobs:
            await asyncio.wait(jobs, timeout=timeout)
        return self.drain_completed()

    def pending(self):
        return [h for h, t in self._tasks.items() if not t.job.done()]

    async def cancel_all(self):
        jobs = [t.job for t in self._tasks.values() if not t.job.done()]
        for job in jobs:
            job.cancel()
        if jobs:
            await asyncio.gather(*jobs, return_exceptions=True)
        self._tasks.clear()

Two details are load-bearing. asyncio.CancelledError is caught before the general Exception handler because it derives from BaseException and would otherwise escape; a cancelled job must still produce an output, or the model has a call it never gets an answer to. And drain_completed deletes as it drains, so calling it twice in one iteration cannot duplicate an output.

The loop itself then becomes a dispatcher with three branches: background it, wait for it, or run it inline.

ASYNC_TOOLS = {"search_docs", "fetch_pricing", "run_ci"}   # idempotent reads only

async def run_turn(client, user_input, tools, impls, registry, budget_s=120):
    deadline = time.monotonic() + budget_s
    resp = await client.responses.create(model=MODEL, tools=tools, input=user_input)

    try:
        while True:
            tail = []
            for item in resp.output:
                if item.type != "function_call":
                    continue
                args = json.loads(item.arguments)

                if item.name == "wait_for_tasks":
                    remaining = max(0.0, deadline - time.monotonic())
                    tail.extend(await registry.wait_for(args["handles"], remaining))
                    tail.append({
                        "type": "function_call_output",
                        "call_id": item.call_id,
                        "output": json.dumps({"pending": registry.pending()}),
                    })
                elif getattr(item, "async_", False) and item.name in ASYNC_TOOLS:
                    registry.start(item.call_id, item.name, args,
                                   impls[item.name], args.pop("task_handle", None))
                else:
                    result = await impls[item.name](**args)     # sync contract
                    tail.append({
                        "type": "function_call_output",
                        "call_id": item.call_id,
                        "output": json.dumps(result),
                    })

            # Results first, wait status last: the model resumes holding the data.
            next_input = registry.drain_completed() + tail

            if not next_input:
                if not registry.pending():
                    return resp.output_text
                remaining = max(0.0, deadline - time.monotonic())
                if remaining == 0.0:
                    raise TimeoutError("turn budget exhausted with tasks in flight")
                next_input = await registry.wait_for(registry.pending(), remaining)

            resp = await client.responses.create(
                model=MODEL, tools=tools,
                previous_response_id=resp.id, input=next_input,
            )
    finally:
        await registry.cancel_all()

The ordering on the next_input line matters and is easy to get wrong. OpenAI's guidance is to return each newly completed result on its original call_id first, then return status on the wait call's own call_id, so that the model resumes already holding the results rather than being told they exist. Putting the wait status first produces a model that acknowledges the data and then asks for it again.

The finally block is not decoration. Without it, an exception anywhere in the turn leaves background jobs running against a conversation that no longer exists, and on a busy service those orphans are what eventually exhausts your connection pool.

Dependencies, stale state and reconciliation

Async only pays if genuinely independent work overlaps. The moment tool B needs tool A's output, you are back to a barrier — and the question is who enforces it. The wrong answer is the prompt. Instructions like "wait for the search before summarising" are advisory, and a model under time pressure will violate them at some rate you cannot bound.

The right answer has two halves. First, express the dependency in the schema: make B's parameters require a field that only A's output can supply, so a call to B without it fails validation at the boundary and returns a structured error naming the missing handle. The model then has a mechanical reason to call wait_for_tasks rather than a stylistic one. Second, for anything that reads state which a concurrent job might change, carry a version snapshot and check it before you act.

class VersionLedger:
    """Monotonic version per resource the agent reads or writes."""
    def __init__(self):
        self._versions: dict[str, int] = {}

    def snapshot(self, keys):
        return {k: self._versions.get(k, 0) for k in keys}

    def bump(self, key):
        self._versions[key] = self._versions.get(key, 0) + 1

    def is_stale(self, snapshot):
        return any(self._versions.get(k, 0) != v for k, v in snapshot.items())


async def guarded_write(ledger, snapshot, key, impl, **kwargs):
    """Compare-and-set: refuse a write built on state that moved underneath it."""
    if ledger.is_stale(snapshot):
        return {"status": "stale_state",
                "message": f"{key} changed since it was read; re-read and retry",
                "keys": sorted(snapshot)}
    result = await impl(**kwargs)
    ledger.bump(key)
    return {"status": "completed", "result": result}

That is ordinary optimistic concurrency control, correct here for the same reason it is correct in a database: the model cannot know that the row it read four seconds ago is still the row it is writing to. Returning stale_state as a structured result rather than raising is deliberate — it gives the model something actionable, which is to re-read and retry.

Reconciliation is the mirror-image problem: a result lands after the model has already moved on. Three policies cover nearly every case, and each tool should declare one at registration. Drop the result and count it as wasted work — correct for a speculative prefetch that did not end up mattering. Defer it into the next turn as ordinary context — correct for a conversational agent where the user is still present. Invalidate the answer and re-run the affected step — the only safe choice when the answer asserted something the late result contradicts. Gemini's Live API makes this choice explicit in the protocol: the function response carries a scheduling field of INTERRUPT, WHEN_IDLE or SILENT, which map almost exactly onto invalidate, defer and drop.

Failure modes, cancellation and partial results

An async loop has failure modes a blocking loop cannot have, because a blocking loop has nowhere to leak. Design against each of these explicitly before you ship, not after the first incident.

A per-tool timeout and a per-turn budget are different controls and you need both: the first stops one pathological dependency consuming the turn, the second stops a well-behaved set of tools collectively doing so. Wrap each implementation so a timeout becomes a structured result, not an exception escaping the registry.

async def bounded(name, coro, timeout_s):
    """Never let a tool raise into the loop; always return something the model can read."""
    try:
        return {"status": "completed", "result": await asyncio.wait_for(coro, timeout_s)}
    except asyncio.TimeoutError:
        return {"status": "timeout", "tool": name, "after_s": timeout_s}
    except asyncio.CancelledError:
        raise                                   # cancellation must propagate
    except Exception as exc:
        return {"status": "failed", "tool": name,
                "error": f"{type(exc).__name__}: {exc}"}

Re-raising CancelledError rather than swallowing it is the detail that separates a loop which shuts down cleanly from one that hangs on exit. If you intercept cancellation and return a value, the surrounding gather in cancel_all waits for a task that has decided not to stop. Anything that must run during shutdown — releasing a lease, writing an audit row — belongs inside asyncio.shield, not inside a broad except clause.

For jobs that genuinely outlive a single turn, do not try to hold them in the registry at all. Hand them to a durable runner and give the model a handle to poll; the architecture for that is set out in the guide to background agents and durable execution. The registry in this article is for work measured in seconds, not hours.

Keeping an async loop observable and replayable

A blocking loop has a free property nobody notices until it is gone: its trace is a list. Turn, call, result, turn. You read it top to bottom and reconstruct what happened. An async loop's trace is a directed graph with overlapping intervals, and the naive version is unreadable — spans that start in one turn and finish three turns later, nested under whichever parent was open at dispatch.

The fix is to stop relying on span nesting for causality and record it explicitly. The OpenTelemetry GenAI semantic conventions give you the vocabulary: an execute_tool span carrying gen_ai.tool.name and gen_ai.tool.call.id, which ties the span back to the call the model emitted. Those conventions were still marked in development as of mid-2026, so pin your instrumentation version. Four rules make an async trace reconstructable:

• Parent the tool span to the dispatching turn, not to the draining turn. The span that started the work is the one that caused it, even though it closed long before the work finished.
• Add a span link from the draining turn back to the tool span. Links, not parent-child edges, are the right primitive for a result that arrives out of order.
• Record dispatch time, completion time and delivery time as three separate timestamps. The gap between completion and delivery is pure scheduling overhead and it is invisible if you only log duration.
• Write an append-only decision log per turn — handle dispatched, handle waited on, handle drained, policy applied — keyed by conversation and turn index. This is what you replay, not the spans.

Determinism needs one more thing. Because the interleaving depends on real timing, you cannot reproduce a bug by re-running the same inputs. Make the loop replayable by recording every tool result alongside its handle and offering a replay mode that serves results from that log in the recorded delivery order, ignoring real latency entirely. That turns a heisenbug into a fixture, which is the only way to write a regression test for an ordering defect. The broader instrumentation setup is covered in the guide to instrumenting agents with OpenTelemetry and tail-sampled traces.

Measuring the win, and what to do next

Be clear about what async buys, because it is narrower than the enthusiasm suggests. It buys wall-clock time. It does not buy tokens — it usually costs a few more, because each continuation re-reads the conversation and the wait call is itself a tool call. Any claim that async reduces your bill is a category error until someone shows you both axes.

Neither OpenAI's async documentation nor the Gemini Live API documentation publishes latency figures for the feature, and there is no credible public benchmark across providers as of September 2026. Measure it yourself, on your own workload:

1. Fix the workload. Take fifty to a hundred real production turns, replayed against recorded tool latencies so the comparison is not contaminated by a dependency having a bad afternoon.
2. Record four numbers per turn on the blocking loop: total wall-clock, tool-blocked time, input tokens, output tokens.
3. Run the same set through the async loop with an identical tool set and identical prompts. Change one variable.
4. Compare p50 and p95 separately. Async usually moves p95 far more than p50, because the win scales with the slowest tool in the turn, and p95 is what your users actually complain about.
5. Divide, do not subtract. Report tool-blocked time as a share of turn time before and after. An absolute saving of eight seconds means nothing without the denominator.
6. Track the discard rate. Count async dispatches whose results were dropped or cancelled. A high discard rate means you are paying real tool cost for speculative work, and it is the number that tells you when to tighten the dispatch policy.

The honest summary for most teams is this. If your agent is dominated by a single slow dependency — legal document retrieval, a claims system in a UK insurer, a KYC check against an Indian registry, a CI pipeline on a large monorepo — async tool calling is a structural improvement to perceived latency and worth the engineering. If your tools are uniformly fast, it adds moving parts and concurrency bugs in exchange for tens of milliseconds, and you should spend the same week on caching or routing instead; the arithmetic for those is in the guides to prompt caching and model routing and to the Batch API for asynchronous jobs, which move cost rather than latency.

Start narrow. Pick one tool — your slowest idempotent read — mark it async, build the registry, run the measurement above, and only then widen the set. A loop that backgrounds one tool correctly beats one that backgrounds eight and cannot say which result belongs to which call.

Sources

• OpenAI — Async tool calling guide
• Anthropic — Parallel tool use
• Google — Tool use with the Gemini Live API
• Pydantic AI — Advanced tool features
• GPT-6 Astra release timeline