Run parallel Claude agents with asyncio and a message hub

Coordinating a fleet of AI agents that run concurrently, talk to each other, and synthesise their outputs into a single answer is one of the most practically valuable — and least well-documented — patterns in applied ML. The approach covered here uses Python's asyncio event loop and Anthropic's Claude API to run multiple agent instances in parallel, each with its own message inbox, while a shared hub routes communication between them without polling. The result is a lightweight orchestration layer you can drop your own domain tools into without touching the core infrastructure.

Who needs this? Any team running pipelines where a single sequential agent creates unacceptable latency: research aggregation, multi-step code review, parallel document analysis, or any workflow where subtasks are independent enough to run concurrently. The hardware requirement is minimal — a laptop is sufficient — because the concurrency is I/O-bound (waiting on API responses), not CPU-bound. The real cost is API spend: each agent in the team makes its own sequence of messages.create calls, so a three-agent team with ten turns each burns roughly thirty API round-trips.

This guide is adapted from the Anthropic Cookbook's async_multi_agent_orchestration.ipynb, published under the MIT licence. For background on why agent cost efficiency matters in production, see our coverage of Anthropic's Claude Sonnet pricing changes.

---

Prerequisites

• Python 3.11+ — asyncio.wait_for with the timeout keyword and top-level await in Jupyter both require a recent stdlib.
• Anthropic Python SDK — anthropic>=0.27 for async support.
• ANTHROPIC_API_KEY environment variable set. The async client reads it automatically.
• Jupyter or an async-capable REPL — the top-level await calls in the demo cells require a running event loop. In a plain Python script, wrap everything in asyncio.run(...).
• No GPU, no local model, no special hardware.

---

Step 1: Install the SDK

%pip install -qU anthropic

---

Step 2: Imports and client setup

The AsyncAnthropic client is the key difference from single-agent code. It returns awaitables from every API call, which lets asyncio interleave multiple in-flight requests without threads. MODEL is set once and shared by every agent in the run.

import asyncio
import itertools
from collections import Counter, defaultdict

import anthropic

MODEL = "claude-opus-4-8"  # swap for the newest Claude model available to you
client = anthropic.AsyncAnthropic()  # reads ANTHROPIC_API_KEY from env

---

Step 3: Build the message hub

The Hub is the central nervous system of the entire pattern. Each agent gets an inbox (a plain list of dicts) and an asyncio.Event. When agent A calls post, it appends a message to agent B's inbox and sets B's event, waking B if it is blocked on wait_for_message. The drain method atomically empties the inbox and resets the event, preventing any message from being delivered twice.

class Hub:
    def __init__(self):
        self.inbox: dict[str, list[dict]] = defaultdict(list)
        self.event: dict[str, asyncio.Event] = defaultdict(asyncio.Event)
        self.status: dict[str, str] = {}
        self._ids = itertools.count(1)

    def register(self, name: str):
        self.status[name] = "active"
        _ = self.inbox[name], self.event[name]

    def new_name(self, prefix="helper") -> str:
        n = f"{prefix}{next(self._ids)}"
        self.register(n)
        return n

    def post(self, sender: str, recipients: list[str], content: str) -> list[str]:
        delivered = []
        for rid in recipients:
            if rid in self.status:
                self.inbox[rid].append({"from": sender, "content": content})
                self.event[rid].set()
                delivered.append(rid)
        return delivered

    def drain(self, name: str) -> list[dict]:
        msgs, self.inbox[name] = self.inbox[name], []
        self.event[name] = asyncio.Event()
        return msgs

    @staticmethod
    def render(msgs: list[dict]) -> str:
        if not msgs:
            return ""
        body = "\n".join(
            f'<agent-message from="{m["from"]}">\n{m["content"]}\n</agent-message>' for m in msgs
        )
        return f"\n\n[Messages received while you were working:]\n{body}"

render wraps messages in XML-style tags so Claude can parse their structure unambiguously. This matters because the messages are injected into a tool result, not a separate conversation turn.

---

Step 4: Define the messaging tools

Every agent receives these two tool definitions. They are the only mechanism agents have to reach each other — plain text in an assistant turn goes nowhere except back to the model itself.

SEND_MESSAGE = {
    "name": "send_message",
    "description": (
        "Send a message to one or more other agents. It will appear appended to their next tool "
        "result. This is the ONLY way to reach other agents — plain text in your turn goes nowhere."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "recipient_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1},
            "content": {"type": "string"},
        },
        "required": ["recipient_ids", "content"],
    },
}
WAIT_FOR_MESSAGE = {
    "name": "wait_for_message",
    "description": (
        "Block until another agent messages you. Note: messages also arrive automatically appended "
        "to the result of ANY other tool call, so only use this when you have nothing else to do."
    ),
    "input_schema": {"type": "object", "properties": {}},
}
BASE_TOOLS = [SEND_MESSAGE, WAIT_FOR_MESSAGE]

The description of wait_for_message contains a critical hint to the model: messages arrive on any tool result, not only on an explicit wait. This prevents agents from calling wait_for_message unnecessarily and burning a turn.

---

Step 5: Implement the core agent loop

run_agent is the single function that drives any agent, regardless of role. It runs a standard tool-use loop, dispatches the two base tools against the hub, and routes any additional tools through extra_dispatch. The critical implementation detail is on the line marked # ← the key line: inbox messages are appended to the last tool result in the batch, so the model sees them as part of the tool-use response without requiring an extra round-trip.

TRACE: dict[str, list[str]] = defaultdict(list)


def _snip(s, n=60):
    s = str(s).replace("\n", " ")
    return s if len(s) <= n else s[:n] + "…"


async def run_agent(
    hub: Hub,
    name: str,
    system: str,
    first_user_turn: str,
    tools: list = None,
    extra_dispatch=None,
    max_turns: int = 20,
) -> str:
    tools = tools or BASE_TOOLS
    extra_dispatch = extra_dispatch or {}
    messages = [{"role": "user", "content": first_user_turn}]

    try:
        for _ in range(max_turns):
            resp = await client.messages.create(
                model=MODEL,
                max_tokens=2048,
                system=system,
                tools=tools,
                messages=messages,
            )
            messages.append({"role": "assistant", "content": resp.content})

            if resp.stop_reason == "end_turn":
                hub.status[name] = "done"
                return "".join(getattr(b, "text", "") for b in resp.content)
            if resp.stop_reason != "tool_use":
                raise RuntimeError(f"unexpected stop_reason: {resp.stop_reason}")

            results = []
            for block in resp.content:
                if block.type != "tool_use":
                    continue
                TRACE[name].append(block.name)
                if block.name == "send_message":
                    rids = block.input["recipient_ids"]
                    delivered = hub.post(name, rids, block.input["content"])
                    unknown = [r for r in rids if r not in delivered]
                    out = f"delivered to {delivered}" + (f"; unknown: {unknown}" if unknown else "")
                elif block.name == "wait_for_message":
                    hub.status[name] = "idling"
                    try:
                        await asyncio.wait_for(hub.event[name].wait(), timeout=60)
                        out = "woke: new messages"
                    except TimeoutError:
                        out = "woke: 60s timeout"
                    hub.status[name] = "active"
                elif block.name in extra_dispatch:
                    out = await extra_dispatch[block.name](block)
                else:
                    out = f"error: no dispatch for {block.name}"
                print(f"  [{name}] {block.name}({_snip(block.input)}) → {_snip(out)}")
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": out})

            inbox = hub.drain(name)
            for m in inbox:
                print(f"  [{name}] ← received from {m['from']}: {_snip(m['content'])}")
            if results:
                results[-1]["content"] += hub.render(inbox)  # ← the key line
            messages.append({"role": "user", "content": results})

        hub.status[name] = "done"
        return f"[{name} hit max_turns={max_turns}]"
    except Exception:
        hub.status[name] = "crashed"
        raise


def print_trace():
    for agent in sorted(TRACE):
        counts = Counter(TRACE[agent])
        print(f"  {agent}: " + ", ".join(f"{n}×{t}" for t, n in counts.most_common()))
    TRACE.clear()

---

Step 6: Pattern 1 — Fixed N-agent team

The first pattern is a fixed team of three agents created before the run starts: one lead and two named helpers. All three are pre-registered in the hub, so they can address each other by name from the first message.

TEAM_SYSTEM = "You are {name}, one of 3 agents working together (peers: {peers})."

TASKS = {
    "lead": "You are the lead. Introduce yourself to the others. Once everyone has introduced "
    "themselves, finish with a one-sentence summary of the team.",
    "helper1": "You are a backend engineer named Ada. Introduce yourself to the others, then wait "
    "for their replies.",
    "helper2": "You are a designer named Bo. Introduce yourself to the others, then wait for their "
    "replies.",
}


async def run_team() -> str:
    hub = Hub()
    names = list(TASKS)
    for n in names:
        hub.register(n)

    helper_tasks = [
        asyncio.create_task(
            run_agent(
                hub,
                n,
                system=TEAM_SYSTEM.format(name=n, peers=[p for p in names if p != n]),
                first_user_turn=TASKS[n],
            )
        )
        for n in names[1:]
    ]
    try:
        return await run_agent(
            hub,
            "lead",
            system=TEAM_SYSTEM.format(name="lead", peers=names[1:]),
            first_user_turn=TASKS["lead"],
        )
    finally:
        for t in helper_tasks:
            t.cancel()
        await asyncio.gather(*helper_tasks, return_exceptions=True)

answer = await run_team()
print(f"\n[lead final answer]\n{answer}\n\nTool-call summary:")
print_trace()

The helpers are created as background asyncio.Task objects and cancelled in the finally block once the lead finishes. This is the correct cleanup pattern: if a helper is blocked on wait_for_message when the lead exits, cancellation propagates as a CancelledError and the return_exceptions=True in gather ensures the cancellation does not raise in the caller.

---

Step 7: Pattern 2 — Dynamic subagent spawning

The second pattern gives the lead tools to spawn, inspect, and kill helpers at runtime. Helpers don't exist before the run starts — the lead creates them programmatically based on its task.

SLEEP = {
    "name": "sleep",
    "description": "Sleep for the given number of seconds, then return.",
    "input_schema": {
        "type": "object",
        "properties": {"seconds": {"type": "integer", "minimum": 0, "maximum": 10}},
        "required": ["seconds"],
    },
}
SUBAGENT_TOOLS = [
    {
        "name": "create_subagents",
        "description": "Spawn helper subagents. Returns immediately — helpers run concurrently in the "
        "background in parallel with you. Each gets base_instruction, optionally + "
        "per_subagent_instructions[i].",
        "input_schema": {
            "type": "object",
            "properties": {
                "base_instruction": {"type": "string"},
                "per_subagent_instructions": {
                    "type": "array",
                    "items": {"type": "string"},
                    "maxItems": 10,
                },
            },
            "required": ["base_instruction"],
        },
    },
    {
        "name": "get_status",
        "description": "Status of every helper (active / idling / done / crashed).",
        "input_schema": {"type": "object", "properties": {}},
    },
    {
        "name": "kill_subagents",
        "description": "Cancel running helpers you no longer need.",
        "input_schema": {
            "type": "object",
            "properties": {
                "subagent_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1}
            },
            "required": ["subagent_ids"],
        },
    },
]


async def dispatch_sleep(block) -> str:
    s = int(block.input["seconds"])
    await asyncio.sleep(s)
    return f"slept {s}s"

async def run_spawn_lead() -> str:
    hub = Hub()
    hub.register("lead")
    helpers: dict[str, asyncio.Task] = {}

    async def _create(block):
        base = block.input["base_instruction"]
        per = block.input.get("per_subagent_instructions") or [""]
        spawned = []
        for suffix in per:
            h = hub.new_name()
            helpers[h] = asyncio.create_task(
                run_agent(
                    hub,
                    h,
                    system=f"You are {h}, a helper.",
                    first_user_turn=f"{base}\n\n{suffix}".strip(),
                    tools=[SLEEP, *BASE_TOOLS],
                    extra_dispatch={"sleep": dispatch_sleep},
                )
            )
            spawned.append(h)
        return f"spawned: {', '.join(spawned)}"

    async def _status(block):
        return "\n".join(f"{n}: {s}" for n, s in hub.status.items() if n != "lead") or "(none)"

    async def _kill(block):
        killed_ids, to_await = [], []
        for sid in block.input["subagent_ids"]:
            if sid in helpers and not helpers[sid].done():
                helpers[sid].cancel()
                hub.status[sid] = "done"
                to_await.append(helpers.pop(sid))
                killed_ids.append(sid)
        await asyncio.gather(*to_await, return_exceptions=True)
        return f"cancelled: {', '.join(killed_ids)}" if killed_ids else "no matching active helpers"

    try:
        return await run_agent(
            hub,
            "lead",
            system="You are the lead.",
            first_user_turn=(
                "Spawn three helper agents. Instruct each to sleep a different number of seconds "
                "(1, 2, 3), report back to you 'done, slept Ns', then wait for your further "
                "instructions. After that, check that they're running, collect all three reports, "
                "then dismiss all three helpers. Finish with a one-line summary."
            ),
            tools=[*SUBAGENT_TOOLS, *BASE_TOOLS],
            extra_dispatch={
                "create_subagents": _create,
                "get_status": _status,
                "kill_subagents": _kill,
            },
        )
    finally:
        for t in helpers.values():
            t.cancel()
        await asyncio.gather(*helpers.values(), return_exceptions=True)

answer = await run_spawn_lead()
print(f"\n[lead final answer]\n{answer}\n\nTool-call summary:")
print_trace()

---

Pattern comparison

---

What to watch out for

The max_turns ceiling is a hard stop, not a graceful one. When an agent hits max_turns=20, run_agent returns a bracketed error string rather than a real answer. The lead will receive this as a message and may silently continue. In production, treat this string as a signal to retry or escalate, not as a valid subtask result.

wait_for_message times out after 60 seconds. If a helper is waiting for a message that never arrives — because the sender crashed, addressed the wrong ID, or the lead exited early — the helper will wake after 60 seconds with "woke: 60s timeout" and likely spin on another wait. You will accumulate idle tasks burning nothing but memory, unless your finally block cancels them explicitly. Always do so.

Message delivery is fire-and-forget. hub.post returns a list of delivered IDs but does not retry or queue for offline agents. If an agent is not yet registered when a message arrives, it is silently dropped. In the fixed-team pattern this is fine because all agents register before any task starts. In dynamic spawning, the lead must not send to a helper before _create returns the spawned IDs.

Context window growth is real. Every turn appends to messages. A long-running agent with many tool calls and large inbox payloads will eventually approach max_tokens or the model's context limit. For production use, implement a sliding window or summarisation step in run_agent rather than accumulating the full history.

Concurrency is I/O-bound, but API rate limits apply per-account. Ten agents making simultaneous messages.create calls will hit your requests-per-minute limit faster than ten sequential calls. Add exponential backoff around the client.messages.create call, or use a semaphore to cap concurrency. The source code does neither; it is a pedagogical baseline, not a rate-limit-safe implementation.

Agent IDs are ephemeral. hub.new_name() uses a process-level counter. Across runs, IDs reset to helper1. If you log or persist agent outputs across multiple run_spawn_lead() invocations in the same process, IDs will collide in your logs. Namespace them by run UUID if observability matters.

The model may hallucinate agent names. Claude occasionally sends a message to an agent ID it invents rather than one it was told about. The post method will return an empty delivered list and include the unknown IDs in the unknown field of the tool result. Prompt engineering helps — be explicit about available peer IDs — but monitor for non-empty unknown lists in production.

---

Where to go next

Drop your own domain tools into extra_dispatch — a web search function, a SQL query executor, a code sandbox — and replace the toy TASKS dict with real subtask specifications. The Hub, run_agent, and the two runner functions above don't change; all your domain logic lives in tool handlers.

For deeper context on how agent memory scales across team-level interactions, see our coverage of TencentDB's team-level memory hub, which addresses exactly the state-sharing challenges that emerge when these patterns graduate from demos to persistent services. The full tool definition reference is at Anthropic's tool use guide, and the notebook this guide is based on lives at the Anthropic Cookbook repository (MIT licence).