Single-threaded agent orchestration hits a wall when task counts reach the hundreds. A sequential loop that dispatches agents one at a time, waiting for each to finish before starting the next, turns a 10-minute parallel workflow into an hours-long serial queue. Python’s asyncio provides the tools to coordinate hundreds of concurrent agent tasks on a single thread: but using those tools correctly requires understanding how concurrency, cancellation, rate limiting, and failure isolation interact in an async agent pipeline.
Scatter-Gather with asyncio.gather() and TaskGroup
Scatter-gather is the most common async coordination pattern: fan out a task to many agents simultaneously, then collect and merge all results. The naive implementation, asyncio.gather(*[agent(task) for task in tasks]): breaks under two conditions: slow agents hang the entire gather, and exceptions in one agent cancel nothing else (or cancel everything, depending on return_exceptions).
import asyncio
from dataclasses import dataclass
@dataclass
class AgentResult:
agent_id: str
result: dict | None
error: str | None
duration_ms: float
async def scatter_gather(tasks: list[dict], timeout_seconds: float = 30.0) -> list[AgentResult]:
"""
WHY asyncio.wait_for per task rather than a single timeout on gather:
a single outer timeout cancels ALL tasks if any one task takes too long.
Per-task timeouts let fast agents complete and return results while
slow or hung agents are cancelled individually: maximizing partial success.
"""
async def bounded_agent_call(task: dict) -> AgentResult:
start = asyncio.get_event_loop().time()
try:
result = await asyncio.wait_for(
call_agent(task),
timeout=timeout_seconds,
)
duration = (asyncio.get_event_loop().time() - start) * 1000
return AgentResult(task["agent_id"], result, None, duration)
except asyncio.TimeoutError:
return AgentResult(task["agent_id"], None, "TimeoutError", timeout_seconds * 1000)
except Exception as exc:
duration = (asyncio.get_event_loop().time() - start) * 1000
return AgentResult(task["agent_id"], None, str(exc), duration)
# WHY return_exceptions=True: without this, the first exception from any
# agent immediately cancels all remaining agents. We want to collect
# results from successful agents even when some fail.
results = await asyncio.gather(*[bounded_agent_call(t) for t in tasks], return_exceptions=True)
return [r for r in results if isinstance(r, AgentResult)]flowchart TD
O["Orchestrator"] -->|"asyncio.gather()"| A1["Agent 1\n(wait_for 30s)"]
O -->|"asyncio.gather()"| A2["Agent 2\n(wait_for 30s)"]
O -->|"asyncio.gather()"| A3["Agent 3\n(wait_for 30s)"]
O -->|"asyncio.gather()"| A4["Agent N\n(wait_for 30s)"]
A1 -->|"result"| R["Fan-in:\nmerge results"]
A2 -->|"result"| R
A3 -->|"TimeoutError\n(cancelled)"| R
A4 -->|"result"| R
R --> F["Aggregated Output\n(partial success OK)"]
style O fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style R fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style F fill:#f0fdf9,stroke:#0D9488,color:#0F172A
style A3 fill:#fff7ed,stroke:#f59e0b,color:#0F172ASaga Pattern for Distributed Agent Transactions
When an agent workflow spans multiple independent services, each with its own state, you cannot use database transactions to guarantee consistency. The saga pattern manages this: each step is a local operation with a pre-registered compensating action. On failure, the orchestrator runs compensations in reverse to restore consistent state.
from typing import Callable, Awaitable
@dataclass
class SagaStep:
name: str
action: Callable[..., Awaitable[dict]]
compensate: Callable[..., Awaitable[None]]
class SagaOrchestrator:
def __init__(self, steps: list[SagaStep]):
self.steps = steps
# WHY track completed steps separately from the step list:
# on failure, we iterate completed_steps in reverse to compensate
# only what was actually executed: not steps we never reached.
self.completed: list[tuple[SagaStep, dict]] = []
async def execute(self, context: dict) -> dict:
for step in self.steps:
try:
result = await step.action(context)
context.update(result)
self.completed.append((step, result))
except Exception as exc:
await self._compensate(exc)
raise
return context
async def _compensate(self, original_error: Exception) -> None:
"""
WHY reverse order compensation: steps build on each other.
Compensating in reverse ensures we undo the most recent
(and possibly dependent) operations before undoing earlier ones.
"""
for step, result in reversed(self.completed):
try:
await step.compensate(result)
except Exception as comp_error:
# Log but continue: best-effort compensation.
# Alert ops: this saga needs manual intervention.
logger.error(f"Compensation failed for {step.name}: {comp_error}")
# Usage: loan application workflow
loan_saga = SagaOrchestrator([
SagaStep(
name="create_credit_file",
action=credit_agent.create_file,
compensate=credit_agent.delete_file,
),
SagaStep(
name="reserve_funds",
action=funds_agent.reserve,
compensate=funds_agent.release,
),
SagaStep(
name="send_notification",
action=notify_agent.send,
compensate=notify_agent.send_cancellation,
),
])Semaphores and Circuit Breakers
Two patterns protect your async pipeline from external service limits and cascading failures: semaphores bound concurrency, circuit breakers stop calling degraded services.
flowchart TD
T["500 Agent Tasks"] --> S["asyncio.Semaphore(100)\n(max 100 concurrent)"]
S -->|"100 slots available"| LLM["LLM API\n(rate limit: 100 RPS)"]
S -->|"400 tasks waiting\n(non-blocking await)"| W["Event Loop\n(handles other work)"]
LLM -->|"response"| CB["Circuit Breaker\n(tracks error rate)"]
CB -->|"error rate < 50%\n(CLOSED)"| R["Return result"]
CB -->|"error rate >= 50%\n(OPEN)"| F["Fail fast\n(no LLM call made)"]
style S fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style CB fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style R fill:#f0fdf9,stroke:#0D9488,color:#0F172A
style F fill:#fff7ed,stroke:#f59e0b,color:#0F172Aimport asyncio
from collections import deque
from datetime import datetime, UTC
# Shared semaphore: all agent tasks in the process compete for these slots
llm_semaphore = asyncio.Semaphore(100)
class AsyncCircuitBreaker:
def __init__(self, failure_threshold: float = 0.5, window_seconds: int = 60):
self.failure_threshold = failure_threshold
self.window_seconds = window_seconds
self._calls: deque[tuple[datetime, bool]] = deque()
self.open = False
def _prune(self):
cutoff = datetime.now(UTC).timestamp() - self.window_seconds
while self._calls and self._calls[0][0].timestamp() < cutoff:
self._calls.popleft()
def _error_rate(self) -> float:
if not self._calls:
return 0.0
failures = sum(1 for _, success in self._calls if not success)
return failures / len(self._calls)
async def call(self, coro):
if self.open:
raise RuntimeError("Circuit breaker OPEN, refusing call to degraded service")
try:
# WHY acquire semaphore inside circuit breaker check:
# if the circuit is open, we never acquire the semaphore slot,
# keeping it available for tasks that CAN succeed (e.g., cached
# responses, fallback services) rather than wasting capacity
# on calls we already know will fail.
async with llm_semaphore:
result = await coro
self._calls.append((datetime.now(UTC), True))
return result
except Exception:
self._calls.append((datetime.now(UTC), False))
self._prune()
if self._error_rate() >= self.failure_threshold:
self.open = True
logger.warning("Circuit breaker opened, error rate exceeded threshold")
raise
circuit_breaker = AsyncCircuitBreaker(failure_threshold=0.5, window_seconds=60)The patterns in this lesson: scatter-gather with per-task timeouts, saga orchestration for distributed consistency, semaphores for rate limiting, and circuit breakers for failure isolation, form a complete toolkit for coordinating large agent populations without blocking, losing work, or cascading failures through your system. Together with the durable messaging and contract design from the previous lessons, these patterns give you the foundation for an agent network that can scale to real enterprise workloads.