How to Orchestrate Multiple AI Agents: A Coordinator Pattern in Python

This tutorial assumes you've already built (or read) Build an AI Agent From Scratch — we're reusing that single-agent loop as a building block, not reinventing it. If you haven't, Step 1 below rebuilds a short version of it so this tutorial still stands on its own.

Here's the thing nobody tells you before you build your first agent: giving one agent fifteen unrelated tools and a system prompt that tries to cover all of them makes it worse at picking the right tool, not better. The model has to hold every tool's purpose in its head on every single turn, and the more of them there are, the more often it reaches for the wrong one or hedges with an unnecessary call. A handful of narrow agents, each with two or three tools and a system prompt about exactly one job, make sharper decisions than one agent trying to be all of them at once.

Multi-agent isn't a smarter agent. It's context isolation — splitting one overloaded prompt into several focused ones and gluing them together with a coordinator. That's the whole idea this tutorial builds.

What You'll Build

• A reusable run_agent() function implementing the single-agent tool-calling loop — the same primitive every agent in this tutorial is built from
• A Researcher agent with one tool that searches an internal notes database
• A Writer agent with one tool that saves a finished draft
• A Coordinator agent whose only two tools are delegate_to_researcher and delegate_to_writer — each one runs the corresponding sub-agent's full loop to completion and hands the result back up as a tool result
• A parallel version of delegation using asyncio.gather, for when the Coordinator's two sub-tasks don't depend on each other

Step 1: The Single-Agent Loop, as a Reusable Function

Everything below is built on one primitive: send the conversation and the available tools to the model, execute whatever tools it calls, append the results, and repeat until it returns plain text instead of a tool call.

1import json
2from openai import OpenAI
3
4CLIENT_KWARGS = dict(
5    base_url="http://localhost:11434/v1/",  # drop base_url to use the real OpenAI API instead
6    api_key="ollama",  # required by the SDK, ignored by Ollama
7)
8
9def run_agent(system_prompt, tools_schema, tool_impls, user_task, model="qwen3.5:9b", max_iterations=8):
10    # A fresh client per call, not a shared module-level instance — see the note
11    # below Step 5 on why that matters once these calls start running on threads.
12    client = OpenAI(**CLIENT_KWARGS)
13    messages = [
14        {"role": "system", "content": system_prompt},
15        {"role": "user", "content": user_task},
16    ]
17
18    for _ in range(max_iterations):
19        response = client.chat.completions.create(
20            model=model,
21            messages=messages,
22            tools=tools_schema,
23        )
24        message = response.choices[0].message
25        messages.append(message.model_dump(exclude_none=True))
26
27        if not message.tool_calls:
28            return message.content
29
30        for tool_call in message.tool_calls:
31            fn_name = tool_call.function.name
32            fn_args = json.loads(tool_call.function.arguments)
33            result = tool_impls[fn_name](**fn_args)
34            messages.append({
35                "role": "tool",
36                "tool_call_id": tool_call.id,
37                "content": str(result),
38            })
39
40    return "Gave up after max_iterations without a final answer."

Three things matter here and stay true for every agent built on top of this function: the max_iterations cap is not optional — without it, a model that keeps deciding to call another tool will loop forever and burn tokens doing it. The tool result is always appended as a "role": "tool" message tied to the exact tool_call_id it answers, or the model loses track of which result belongs to which call. And tool_impls[fn_name](**fn_args) executes whatever the model asked for with whatever arguments it supplied — this is fine for read-only, narrow tools like the ones below, but it is exactly the line you'd sandbox before giving an agent anything destructive.

Step 2: The Researcher Agent

The Researcher gets exactly one tool: a search over a small in-memory notes database, standing in for a real search API or vector store.

1NOTES_DB = [
2    {"topic": "vector databases", "text": "Vector databases index embeddings for approximate nearest-neighbor search, the retrieval half of most RAG pipelines."},
3    {"topic": "distributed tracing", "text": "Distributed tracing stitches spans from multiple services into one trace using a shared trace ID propagated over HTTP headers."},
4    {"topic": "rbac", "text": "Role-Based Access Control binds permissions to roles, then roles to principals, rather than granting permissions to users directly."},
5]
6
7def search_notes(query: str) -> str:
8    hits = [n["text"] for n in NOTES_DB if query.lower() in n["topic"]]
9    return "\n".join(hits) if hits else "No matching notes found."
10
11RESEARCHER_TOOLS = [
12    {
13        "type": "function",
14        "function": {
15            "name": "search_notes",
16            "description": "Search an internal notes database for background information on a topic.",
17            "parameters": {
18                "type": "object",
19                "properties": {
20                    "query": {"type": "string", "description": "Topic keyword to search for, e.g. 'vector databases'."}
21                },
22                "required": ["query"],
23            },
24        },
25    }
26]
27
28RESEARCHER_IMPLS = {"search_notes": search_notes}
29
30def run_researcher(task: str) -> str:
31    return run_agent(
32        system_prompt=(
33            "You are a research assistant. Use search_notes to gather background "
34            "information before answering. Summarize findings in 3-5 sentences."
35        ),
36        tools_schema=RESEARCHER_TOOLS,
37        tool_impls=RESEARCHER_IMPLS,
38        user_task=task,
39    )

Notice run_researcher is just run_agent with the Researcher's prompt and tools baked in. It's a thin wrapper, not a new mechanism — which is exactly why this pattern scales to as many specialists as you need without the underlying loop getting more complicated.

Step 3: The Writer Agent

Same shape, different job: turn whatever text it's given into a finished draft and save it.

1def save_draft(text: str) -> str:
2    return f"Draft saved ({len(text.split())} words)."
3
4WRITER_TOOLS = [
5    {
6        "type": "function",
7        "function": {
8            "name": "save_draft",
9            "description": "Save a finished draft of the requested writing.",
10            "parameters": {
11                "type": "object",
12                "properties": {
13                    "text": {"type": "string", "description": "The full draft text, ready to save."}
14                },
15                "required": ["text"],
16            },
17        },
18    }
19]
20
21WRITER_IMPLS = {"save_draft": save_draft}
22
23def run_writer(task: str) -> str:
24    return run_agent(
25        system_prompt=(
26            "You are a technical writer. Turn the material you're given into a "
27            "tight, three-paragraph summary for a platform engineering audience, "
28            "then call save_draft with the final text."
29        ),
30        tools_schema=WRITER_TOOLS,
31        tool_impls=WRITER_IMPLS,
32        user_task=task,
33    )

Step 4: The Coordinator

Here's the actual multi-agent part. The Coordinator is also just run_agent — its "tools" just happen to be Python functions that run another agent's entire loop to completion and hand back the final text as the tool result.

1def delegate_to_researcher(task: str) -> str:
2    return run_researcher(task)
3
4def delegate_to_writer(task: str) -> str:
5    return run_writer(task)
6
7COORDINATOR_TOOLS = [
8    {
9        "type": "function",
10        "function": {
11            "name": "delegate_to_researcher",
12            "description": "Hand a research task to the Researcher agent and get back its findings.",
13            "parameters": {
14                "type": "object",
15                "properties": {"task": {"type": "string", "description": "What to research."}},
16                "required": ["task"],
17            },
18        },
19    },
20    {
21        "type": "function",
22        "function": {
23            "name": "delegate_to_writer",
24            "description": "Hand a writing task to the Writer agent and get back a draft.",
25            "parameters": {
26                "type": "object",
27                "properties": {"task": {"type": "string", "description": "What to write, including source material to base it on."}},
28                "required": ["task"],
29            },
30        },
31    },
32]
33
34COORDINATOR_IMPLS = {
35    "delegate_to_researcher": delegate_to_researcher,
36    "delegate_to_writer": delegate_to_writer,
37}
38
39final_answer = run_agent(
40    system_prompt=(
41        "You are a coordinator. Break the user's request into a research step and "
42        "a writing step, delegate each to the right specialist in order, then "
43        "return the writer's final draft as your answer."
44    ),
45    tools_schema=COORDINATOR_TOOLS,
46    tool_impls=COORDINATOR_IMPLS,
47    user_task="Research vector databases and write a short summary for a platform engineering audience.",
48)
49print(final_answer)

Run that, and the trace looks like this: the Coordinator's first response is a delegate_to_researcher tool call with a task like "vector databases". That call runs the entire Researcher loop — its own call to the model, its own search_notes tool call, its own final summary — and only the Researcher's last message comes back to the Coordinator as a tool result. The Coordinator never sees the Researcher's intermediate search_notes call at all. It then calls delegate_to_writer with the research notes as the task, which runs the entire Writer loop the same way, and returns the draft as its own final answer.

Three agents, three separate loops, and from the top level it looks like the Coordinator made two tool calls.

Step 5: Sequential vs Parallel Delegation

The example above is sequential on purpose — writing depends on research finishing first. But when the Coordinator decides two sub-tasks are genuinely independent, running them one after another wastes wall-clock time for no reason. asyncio.gather fixes that without touching the single-agent loop itself — wrap the existing synchronous run_researcher/run_writer calls in asyncio.to_thread so they run concurrently instead of rewriting the OpenAI client calls as async:

1import asyncio
2
3async def delegate_to_researcher_async(task: str) -> str:
4    return await asyncio.to_thread(run_researcher, task)
5
6async def delegate_to_writer_async(task: str) -> str:
7    return await asyncio.to_thread(run_writer, task)
8
9async def run_parallel_example():
10    research_result, draft_result = await asyncio.gather(
11        delegate_to_researcher_async("distributed tracing"),
12        delegate_to_writer_async("Draft a one-paragraph explainer of RBAC for a security audience."),
13    )
14    return research_result, draft_result
15
16research_result, draft_result = asyncio.run(run_parallel_example())

asyncio.to_thread runs each blocking run_agent call in a worker thread, and asyncio.gather waits for both to finish — the two full agent loops overlap instead of running back to back. The same technique applies one level up, too: when a model's response contains multiple tool_calls in a single turn (it can request several at once), the for tool_call in message.tool_calls loop in run_agent executes them one at a time by default. Swapping that loop for asyncio.gather over asyncio.to_thread-wrapped calls parallelizes same-turn tool calls the same way.

This is also exactly why Step 1's run_agent creates its own OpenAI(**CLIENT_KWARGS) instance on every call instead of sharing one module-level client: the SDK's underlying HTTP client keeps a connection pool that isn't designed to be hammered from multiple threads at once, and mixing threads with async code on top of a shared client is a documented way to end up with a hung request instead of a slow one. A fresh, cheap client per call sidesteps the whole problem — it doesn't open a connection until the first request goes out, so there's no real cost to not sharing it.

Two coordination patterns are hiding in what you just built, worth naming explicitly:

Message-passing (what this tutorial builds) — the Coordinator only ever sees a sub-agent's final answer, never its intermediate steps. Clean isolation, easy to reason about, but the Coordinator can't intervene mid-task if a sub-agent starts going the wrong direction.

Shared blackboard — instead of returning a value, sub-agents read and write directly to a shared state object (a dict, a database row, a scratch file) that every agent in the system can see. More visibility and less duplicated context passed around, but now you're managing concurrent writes and the coordination logic is implicit in whatever touches the blackboard, which is harder to trace than an explicit call graph.

Common Issues

Token cost multiplies with every layer. A flat single agent answering a question directly makes one round of model calls. The Coordinator pattern above makes a Coordinator call, which triggers a full Researcher loop (its own multiple model calls), then a full Writer loop (more model calls) — three to five times the tokens of the flat version isn't unusual once you count every layer's system prompt, tool schemas, and intermediate turns. That's a fine trade when specialization genuinely improves output quality. It's a bad trade when it doesn't.

Runaway recursive delegation. If a sub-agent is itself handed a delegate_to_x tool, nothing in the code above stops delegation from nesting indefinitely — Coordinator delegates to A, A delegates to B, B delegates back to something that delegates to A again. max_iterations only caps a single agent's own loop; it does nothing to cap the depth of the whole tree. If you allow nested delegation, track and enforce a maximum depth explicitly, separate from any single agent's iteration limit.

No global budget. Each run_agent call independently caps itself at max_iterations, but three agents each hitting their own cap can still take far longer and cost far more than you expect, because nothing caps the total across the whole run. For anything beyond a demo, track a shared token or time budget across the entire Coordinator call and abort the whole tree when it's exhausted, not just each agent individually.

Reaching for multi-agent before you need it. If the "specialists" end up sharing most of the same context anyway, you've paid the delegation tax without the isolation benefit. Try a single agent with a slightly longer, clearer system prompt and a few more tools first. Split into multiple agents only once that single agent is visibly struggling to pick the right tool.

Frequently Asked Questions

When is multi-agent actually worth the added complexity, versus a single agent with more tools?

When the tools genuinely belong to different domains with different context — a Researcher tool needs search results and citations in its context window, a Writer tool needs style guidance and doesn't need to see raw search results at all. If every tool would be equally at home in one shared prompt, that's a sign one well-designed agent is simpler and cheaper than several coordinated ones.

How deep should delegation be allowed to nest?

As shallow as the task allows — one level (Coordinator → specialist) covers most real use cases. Two levels is rarely necessary and should be a deliberate decision with an explicit depth counter passed down and checked at every delegation, not something that's allowed to happen by accident because a sub-agent happens to have a delegate_to_x tool too.

How do sub-agents share context or state safely?

The pattern in this tutorial passes state explicitly, as a string, through the task argument and the returned result — nothing is shared implicitly. That's the safe default: if you move to a shared blackboard for efficiency, you take on the same concurrency problems any shared mutable state has (locking, stale reads, partial writes) and should treat it with the same care you'd give a shared database, not a convenience shortcut.

Is this the same thing frameworks like LangGraph or CrewAI do under the hood?

Conceptually, yes. Strip away the graph definitions, the built-in state schemas, and the visualization tooling, and what's left is the same idea: agents as callable units, a coordinator (or graph) deciding which one runs next, and results passed between them as messages or shared state. Building it by hand once is what makes the framework's abstractions legible instead of magic the next time you reach for one.

Official References

• OpenAI function calling — the tool schema and tool_calls response format this tutorial builds on
• Python asyncio — gather, to_thread, and the rest of the concurrency primitives used in Step 5
• LangGraph docs — further reading on graph-based multi-agent orchestration once you've outgrown a hand-rolled Coordinator