Phase 8: Production Agent Infrastructure · 60 min · Python · asyncio · Anthropic SDK
The Problem
When an agent calls tools serially — one after another — it wastes time waiting. Each await blocks the next call from starting. For an agent that needs to call 6 tools, this means the user waits for the sum of all call durations. In real-world systems with network I/O, this adds up fast.
Async programming with Python's asyncio module is the performance multiplier that makes agents fast enough for real-time user interactions. Instead of waiting for each tool to return before starting the next, you fire them all at once and collect the results together.
Asyncio fundamentals for tool execution
The core pattern is simple: replace sequential await calls with asyncio.gather().
Serial execution — the slow way:
result1 = await tool_search_web(query)
result2 = await tool_query_database(user_id)
result3 = await tool_call_api(endpoint)
# Total time: sum of all three
Parallel execution — the fast way:
result1, result2, result3 = await asyncio.gather(
tool_search_web(query),
tool_query_database(user_id),
tool_call_api(endpoint)
)
# Total time: max of all three
asyncio.gather() runs all coroutines concurrently — they're all in-flight at the same time. The wall-clock time equals the duration of the slowest call, not the sum. For 3 calls of 2 seconds each, that's 2 seconds instead of 6.
asyncio.as_completed() is the streaming alternative — it yields results as they arrive rather than waiting for all to finish:
tasks = [tool_search_web(q) for q in queries]
async for coro in asyncio.as_completed(tasks):
result = await coro
process_result(result) # handle immediately, don't wait for others
Use gather() when you need all results before proceeding. Use as_completed() when you can start processing partial results — for example, display the first search result to the user while the others are still fetching.
Three API calls each take 2 seconds. Running them in parallel with asyncio.gather() takes:
asyncio.gather() runs all coroutines concurrently. Wall-clock time = max(individual times), not sum(individual times). All 3 API calls are in-flight simultaneously — the async event loop handles all 3 while awaiting each. This is the fundamental parallelism benefit: 3 independent I/O operations in 2 seconds instead of 6.
Timeouts and cancellation
Parallel execution without timeouts is dangerous. A single slow or hung tool call blocks the entire gather(). In production, you need per-tool timeout budgets.
asyncio.wait_for() adds a timeout to any coroutine:
try:
result = await asyncio.wait_for(tool_call(), timeout=5.0)
except asyncio.TimeoutError:
result = {"error": "Tool timed out after 5s"}
Different tools get different timeout budgets based on their expected latency:
- Fast APIs (classification, lookup): 500ms
- Standard APIs (search, database): 5s
- Slow APIs (web scraping, large downloads): 30s
asyncio.shield() protects a task from cancellation when you need it to complete regardless:
# This write to the database must complete even if the overall workflow is cancelled
result = await asyncio.shield(write_to_database(data))
A TimeoutManager tracks the global execution budget across the entire agent run. When the budget is exhausted, it cancels remaining tasks gracefully — returning a partial result rather than running forever.
An async tool call has been running for 30 seconds (expected: 2s). You should:
Runaway async tasks are a production risk. A tool call that should take 2s but runs for 30s is likely hung — network issue, server overload, infinite loop. The asyncio.wait_for timeout should have fired at the configured limit. Log the timeout for debugging, cancel the task cleanly, and return a structured error the model can reason about and potentially recover from.
Dependency-aware parallel execution
Not all tools can run in parallel — some depend on the outputs of others. A TaskGraph models these dependencies and executes with maximum parallelism while respecting ordering constraints.
In this graph, search_web and query_database can run in parallel. summarize_results can only run after both complete. draft_response runs last.
The executor algorithm:
- Find all tasks with no unmet dependencies → run them in parallel
- When tasks complete, check which newly-unblocked tasks can now run
- Continue until all tasks complete or the global timeout fires
Build
Build it
Before diving into the full artifact, this simulation shows the core idea — dependency-aware scheduling — using simple round counters instead of real async I/O, so it runs anywhere.
import itertools
# Each task: (name, duration_rounds, dependencies)
tasks = {
"search_web": (2, []),
"query_database": (1, []),
"summarize": (1, ["search_web", "query_database"]),
"draft_response": (1, ["summarize"]),
}
remaining = {name: dur for name, (dur, _deps) in tasks.items()}
done = set()
round_num = 0
while len(done) < len(tasks):
round_num += 1
# Find tasks whose dependencies are all satisfied and still running
runnable = [
name for name, (dur, deps) in tasks.items()
if name not in done and all(d in done for d in deps)
]
print(f"Round {round_num}: running {runnable}")
for name in runnable:
remaining[name] -= 1
if remaining[name] <= 0:
done.add(name)
print(f" -> {name} complete")
print(f"\nTotal rounds (parallel): {round_num}")
serial_rounds = sum(dur for dur, _ in tasks.values())
print(f"Total rounds (serial): {serial_rounds}")
The code artifact implements a complete AsyncToolExecutor with timeout management, a TaskGraph for dependency-aware execution, and a benchmark comparing serial vs. parallel execution.
Run the artifact:
python lesson_08_01_async_tools.py
Expected output:
=== Serial Execution ===
Serial time: 4.30s
=== Parallel Execution (asyncio.gather) ===
Parallel time: 2.01s
Speedup: 2.1x
=== as_completed() Streaming ===
[802ms] api_call: OK
[1501ms] db_query: OK
[2003ms] web_search: OK
=== TaskGraph: Dependency-Aware Parallel Execution ===
web_search: OK (2001ms)
db_query: OK (1502ms)
summarize: OK (1001ms)
Total graph time: 3.50s (vs 4.5s serial)
The full implementation is in lesson_08_01_async_tools.py (under Code artifacts at the end of this lesson).