The Problem
An agent needs to call 4 tools: check inventory, get pricing, look up shipping time, and verify payment method. Each takes ~800ms. Done sequentially: 4 × 800ms = 3.2 seconds. Done in parallel: 800ms. The information is completely independent — there's no reason to wait for inventory before starting pricing.
Async tool execution is a free 4x speedup available to every agent that makes multiple independent API calls. It's something most beginners never think about — and something every production AI engineer does automatically.
Sequential vs. parallel tool execution
Most LLM APIs now support returning multiple tool calls in a single model response. When the model calls check_inventory, get_pricing, and get_shipping_time simultaneously, your job is to execute them concurrently.
Sequential execution:
[tool_1 800ms][tool_2 800ms][tool_3 800ms][tool_4 800ms] = 3200ms total
Parallel execution:
[tool_1 800ms]
[tool_2 800ms] All start at the same time
[tool_3 800ms]
[tool_4 800ms]
= 800ms total (+ ~50ms overhead)
In Python, asyncio.gather() runs multiple coroutines concurrently and collects results when all complete:
import asyncio
results = await asyncio.gather(
check_inventory("PROD-001"),
get_pricing("PROD-001"),
get_shipping_time("US", "CA"),
verify_payment("cus_123"),
)
# All 4 ran concurrently — took ~800ms, not 3200ms
Dependency graphs for tool execution
Not all tool calls are independent. Some tools need the output of other tools:
get_user(user_email) → returns user_id
↓
get_orders(user_id=...)
↓
get_order_details(order_ids=[...])
You can't call get_orders before you have user_id from get_user. Build a dependency graph: nodes are tool calls, edges represent data dependencies. Tools at the same "level" of the graph can run in parallel.
Level 0 (no dependencies): get_user(email), get_product_info(sku)
↓ ↓
Level 1 (depends on L0): get_orders(user_id) (independent)
↓
Level 2 (depends on L1): calculate_total(order_ids)
Optimal execution: parallel(get_user, get_product_info) → get_orders → calculate_total
This takes max(get_user_time, get_product_info_time) + get_orders_time + calculate_total_time instead of sum(all_times).
An agent calls 3 tools simultaneously using asyncio.gather(). One tool times out after 5s. The other two complete in 2s. What's the correct behavior?
Use asyncio.gather(return_exceptions=True) or asyncio.wait() with FIRST_EXCEPTION to isolate failures. Return a ToolResult with status="timeout" for the failed tool and complete results for the others. The model can work with partial results — much better than failing the entire response.
Timeout management
A slow tool should not block your entire agent loop. Apply two levels of timeout:
Per-tool timeout: each individual tool call has a maximum execution time (e.g., 5s). Use asyncio.wait_for(tool_coroutine(), timeout=5.0). If the tool exceeds its timeout, you get an asyncio.TimeoutError — catch it and return a structured ToolResult with status="timeout".
Total execution timeout: the entire parallel batch has a wall-clock timeout (e.g., 15s). Use asyncio.wait_for(asyncio.gather(...), timeout=15.0). This ensures the agent loop doesn't hang even if multiple tools are slow.
Partial results policy: when one tool times out in a parallel group, return the results you have:
async def execute_parallel(tool_calls: list) -> list[ToolResult]:
tasks = [asyncio.wait_for(run_tool(tc), timeout=5.0) for tc in tool_calls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
r if not isinstance(r, Exception)
else ToolResult(status=ErrorType.TRANSIENT, error_code="timeout", reason=str(r))
for r in results
]
Tool B requires the output of Tool A. Tool C requires nothing. Which execution strategy is correct?
This is a dependency graph: A→B (B depends on A's output), C is independent. Optimal execution: parallel(A, C) → B. This takes max(A_time, C_time) + B_time instead of A_time + B_time + C_time. Running all three in parallel is wrong because B needs A's result. Running all in sequence wastes the C parallelism opportunity.
Build
Build it
The code below benchmarks sequential vs. parallel execution on a 4-tool agent. Run it to see the actual timing difference.
"""
lesson_02_03_async_tools.py
Parallel tool execution with asyncio.
Run:
pip install openai aiohttp
export OPENAI_API_KEY=...
python lesson_02_03_async_tools.py
"""
import asyncio
import json
import os
import random
import time
from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional
from openai import OpenAI
@dataclass
class ToolResult:
name: str
status: str # "success" | "timeout" | "error"
data: Optional[Any] = None
error: Optional[str] = None
duration_ms: float = 0.0
# Simulate tools with realistic latencies
async def check_inventory(product_id: str) -> ToolResult:
t0 = time.time()
await asyncio.sleep(0.8 + random.uniform(-0.1, 0.1)) # ~800ms
return ToolResult("check_inventory", "success",
{"product_id": product_id, "qty": 42},
duration_ms=(time.time() - t0) * 1000)
async def get_pricing(product_id: str) -> ToolResult:
t0 = time.time()
await asyncio.sleep(0.6 + random.uniform(-0.1, 0.1)) # ~600ms
return ToolResult("get_pricing", "success",
{"product_id": product_id, "price": 49.99},
duration_ms=(time.time() - t0) * 1000)
async def get_shipping_estimate(region: str) -> ToolResult:
t0 = time.time()
await asyncio.sleep(0.9 + random.uniform(-0.1, 0.1)) # ~900ms
return ToolResult("get_shipping_estimate", "success",
{"region": region, "days": 3},
duration_ms=(time.time() - t0) * 1000)
async def verify_payment(customer_id: str) -> ToolResult:
t0 = time.time()
await asyncio.sleep(0.7 + random.uniform(-0.1, 0.1)) # ~700ms
return ToolResult("verify_payment", "success",
{"customer_id": customer_id, "payment_valid": True},
duration_ms=(time.time() - t0) * 1000)
async def run_sequential() -> tuple[list[ToolResult], float]:
"""Run 4 tools one after another."""
t0 = time.time()
results = [
await check_inventory("PROD-001"),
await get_pricing("PROD-001"),
await get_shipping_estimate("US-WEST"),
await verify_payment("cus_123"),
]
return results, (time.time() - t0) * 1000
async def run_parallel() -> tuple[list[ToolResult], float]:
"""Run 4 tools concurrently using asyncio.gather()."""
t0 = time.time()
results = await asyncio.gather(
check_inventory("PROD-001"),
get_pricing("PROD-001"),
get_shipping_estimate("US-WEST"),
verify_payment("cus_123"),
)
return list(results), (time.time() - t0) * 1000
async def run_with_timeout() -> tuple[list[ToolResult], float]:
"""Run tools with per-tool timeout — one will 'timeout' for demo."""
t0 = time.time()
async def slow_tool() -> ToolResult:
await asyncio.sleep(10) # This will time out
return ToolResult("slow_tool", "success", {})
tasks = [
asyncio.wait_for(check_inventory("PROD-001"), timeout=3.0),
asyncio.wait_for(get_pricing("PROD-001"), timeout=3.0),
asyncio.wait_for(slow_tool(), timeout=1.5), # Will timeout
]
raw_results = await asyncio.gather(*tasks, return_exceptions=True)
results = []
for i, r in enumerate(raw_results):
if isinstance(r, asyncio.TimeoutError):
results.append(ToolResult(f"tool_{i}", "timeout", error="Per-tool timeout exceeded"))
elif isinstance(r, Exception):
results.append(ToolResult(f"tool_{i}", "error", error=str(r)))
else:
results.append(r)
return results, (time.time() - t0) * 1000
if __name__ == "__main__":
print("ASYNC TOOL EXECUTION BENCHMARK")
print("=" * 50)
# Sequential benchmark
print("\nRunning sequential (4 tools, one at a time)...")
seq_results, seq_time = asyncio.run(run_sequential())
print(f"Sequential total: {seq_time:.0f}ms")
for r in seq_results:
print(f" {r.name}: {r.duration_ms:.0f}ms")
# Parallel benchmark
print("\nRunning parallel (asyncio.gather)...")
par_results, par_time = asyncio.run(run_parallel())
print(f"Parallel total: {par_time:.0f}ms")
for r in par_results:
print(f" {r.name}: {r.duration_ms:.0f}ms")
speedup = seq_time / par_time
print(f"\nSpeedup: {speedup:.1f}x ({seq_time:.0f}ms → {par_time:.0f}ms)")
# Timeout demo
print("\nRunning with per-tool timeout (one tool will timeout)...")
timeout_results, total = asyncio.run(run_with_timeout())
print(f"Total with timeout: {total:.0f}ms")
for r in timeout_results:
icon = "✓" if r.status == "success" else "✗"
print(f" {icon} {r.name}: {r.status}")
print("Agent loop continues with partial results — no crash.")
Here's a minimal, dependency-free version of the same idea using time.sleep to simulate latency, so you can see the sequential-vs-parallel gap without needing asyncio or real network calls.
import time
def simulate_tool_calls(latencies_ms):
"""Sequential total is just the sum; parallel total is the max."""
sequential_total = sum(latencies_ms)
parallel_total = max(latencies_ms)
return sequential_total, parallel_total
latencies = {
"check_inventory": 800,
"get_pricing": 600,
"get_shipping_estimate": 900,
"verify_payment": 700,
}
seq_total, par_total = simulate_tool_calls(list(latencies.values()))
print("Tool latencies (ms):", latencies)
print(f"Sequential total: {seq_total}ms")
print(f"Parallel total (bounded by slowest tool): {par_total}ms")
print(f"Speedup: {seq_total / par_total:.2f}x")