Orchestrating Parallel Agent Systems

Welcome back! In the previous lessons, you built a fully async agent system that can handle multiple conversations concurrently and execute tools in parallel. Now we're ready to take this parallelization to the next level by building an orchestrator agent that can delegate work to specialized agents.

In this lesson, you'll discover two different approaches for wrapping agents as tools: the synchronous wrapper approach and the asynchronous wrapper approach. You'll understand the trade-offs of each method and learn how to make your Agent class smart enough to handle both sync and async tools intelligently.

Agent orchestration is a pattern in which one coordinator agent manages and delegates work to multiple specialized agents. Think of it like a project manager who receives a complex task and breaks it down into smaller pieces, assigning each piece to a team member with the right expertise. The orchestrator doesn't need to know how to do every task itself; it just needs to understand the problem well enough to delegate effectively.

In our case, we'll build an orchestrator that handles complex math problems by delegating calculations to specialized calculator agents. When you ask the orchestrator to calculate the total cost of purchases from three different stores, it recognizes that each store's calculation is independent and can happen in parallel. Instead of solving all three calculations sequentially, the orchestrator delegates each one to a separate calculator agent call. These agent calls execute concurrently, just like the parallel tool execution you implemented in the previous lesson. The beauty of this pattern is that it combines the strengths of specialization and parallelization, creating a scalable system in which adding more specialized agents or handling more complex problems doesn't require rewriting your core logic.

Before we can create the orchestrator, we need to build the specialized agent that will handle the actual calculations. This calculator agent will be equipped with all the math tools we've been using throughout the course, making it an expert at solving mathematical problems:

import json
from agent import Agent
from functions import sum_numbers, multiply_numbers, subtract_numbers, divide_numbers, power, square_root

# Load tool schemas
with open('schemas.json', 'r') as f:
    tool_schemas = json.load(f)

# Math tools
math_tools = {
    "sum_numbers": sum_numbers,
    "multiply_numbers": multiply_numbers,
    "subtract_numbers": subtract_numbers,
    "divide_numbers": divide_numbers,
    "power": power,
    "square_root": square_root
}

# Create a calculator assistant
calculator_assistant = Agent(
    name="calculator_assistant",
    system_prompt="You are a calculator assistant. You specialize in mathematical calculations and solving equations.",
    tools=math_tools,
    tool_schemas=tool_schemas
)

The calculator assistant is a straightforward agent with a focused purpose. Its system prompt clearly defines its role as a specialist in mathematical calculations and equations. We provide it with all six math tools, which give the calculator everything it needs to handle a wide range of mathematical problems. Notice that we're loading the tool schemas from the schemas.json file, which contains the definitions for all these math tools. The calculator assistant doesn't need to know anything about orchestration or delegation; it just needs to be good at solving the specific math problems it receives, which makes the system easier to understand and maintain.

The first approach for wrapping an agent as a tool is to use a synchronous wrapper. This is the simpler approach and works perfectly well with our current Agent implementation. Here's how to create a synchronous agent tool:

def create_agent_tool_sync(agent, description):
    """
    Create a synchronous tool function for using an agent as a tool.
    This tool will run in a worker thread when called.

    Args:
        agent: The agent to wrap as a tool
        description: Description of what this agent tool does

    Returns:
        tuple: (tool_function, tool_schema)
    """
    def agent_tool_function(message):
        # Call the agent synchronously inside the tool (executed in a worker thread)
        _, response = agent.run_sync([{"role": "user", "content": message}])

        print(f"📊 Agent ({agent.name}) as tool finished working")

        return response

    # Create schema for this agent tool
    tool_schema = {
        "name": f"{agent.name}_agent",
        "description": description,
        "input_schema": {
            "type": "object",
            "properties": {
                "message": {
                    "type": "string",
                    "description": "The message to send to the agent"
                }
            },
            "required": ["message"]
        }
    }

    return agent_tool_function, tool_schema

This approach creates a regular synchronous function that calls agent.run_sync(). When the orchestrator calls this tool through call_tool(), our current implementation uses asyncio.to_thread() to execute it in a worker thread. Inside that thread, run_sync() creates its own event loop to run the async agent code. This approach works because asyncio.to_thread() is designed for synchronous operations—it expects a regular function, not a coroutine.

The synchronous wrapper has some advantages: it's simple, requires no changes to the Agent class, and works immediately with our existing implementation. However, it has a significant limitation: each agent call runs in its own worker thread with its own event loop, which adds overhead. When you have multiple agent calls running in parallel, you're using multiple threads and multiple event loops, which is less efficient than running everything on a single event loop.

The second approach is to make the agent wrapper truly asynchronous, allowing it to run directly on the event loop alongside other async operations. Here's what an async agent wrapper looks like:

def create_agent_tool_async(agent, description):
    """
    Create an async tool function for using an agent as a tool.
    This tool will run directly on the event loop when called.

    Args:
        agent: The agent to wrap as a tool
        description: Description of what this agent tool does

    Returns:
        tuple: (tool_function, tool_schema)
    """
    async def agent_tool_function(message):
        # Call the agent asynchronously inside the tool (executed on the event loop)
        _, response = await agent.run([{"role": "user", "content": message}])

        print(f"📊 Agent ({agent.name}) as tool finished working")

        return response

    # Create schema for this agent tool
    tool_schema = {
        "name": f"{agent.name}_agent",
        "description": description,
        "input_schema": {
            "type": "object",
            "properties": {
                "message": {
                    "type": "string",
                    "description": "The message to send to the agent"
                }
            },
            "required": ["message"]
        }
    }

    return agent_tool_function, tool_schema

Notice the key differences: we define agent_tool_function with async def and use await agent.run() instead of agent.run_sync(). This creates a coroutine function that runs asynchronously. However, there's a problem: our current call_tool() method uses asyncio.to_thread() to execute all tools, and asyncio.to_thread() can't execute async functions! If we try to use this async wrapper right now, we'll get an error.

The async wrapper approach is more efficient when it works: all agent calls run concurrently on the same event loop, sharing the same asynchronous runtime. No extra threads, no nested event loops—just clean, efficient async execution. But to make it work, we need to teach our Agent class to detect async tools and handle them differently.

To support both sync and async tools, we need to update the call_tool() method to check whether a tool is async and execute it appropriately. Python's inspect module provides the perfect tool for this: inspect.iscoroutinefunction().

Here's how inspect.iscoroutinefunction() works:

import inspect

def sync_function():
    return "sync result"

async def async_function():
    return "async result"

# Check if functions are coroutines
print(inspect.iscoroutinefunction(sync_function))   # False
print(inspect.iscoroutinefunction(async_function))  # True

The function returns True for async functions (defined with async def) and False for regular functions. We can use this to implement intelligent tool execution in our Agent class.

Now let's update our call_tool() method to handle both async and sync tools intelligently:

async def call_tool(self, tool_use):
    tool_name = tool_use.name
    tool_input = tool_use.input or {}
    tool_use_id = tool_use.id

    print(f"🔧 Tool called: {tool_name}({tool_input})")
    
    try:
        tool_fn = self.tools[tool_name]
    except KeyError:
        result = f"Error: Tool {tool_name} not found"
    except Exception as e:
        result = f"Error: {str(e)}"
    else:
        try:
            if inspect.iscoroutinefunction(tool_fn):
                # Tool is async: run it directly on the event loop
                result_value = await tool_fn(**tool_input)
            else:
                # Tool is sync: run it in a worker thread
                result_value = await asyncio.to_thread(tool_fn, **tool_input)
            # Convert result to string for Claude's API
            result = str(result_value)
        except Exception as e:
            result = f"Error: {str(e)}"

    return {
        "type": "tool_result",
        "tool_use_id": tool_use_id,
        "content": result
    }

The key addition is the if inspect.iscoroutinefunction(tool_fn) check. For async tools (like our async agent wrapper), we call them directly with await tool_fn(**tool_input), which runs them on the event loop. For sync tools (like our math functions or sync agent wrapper), we use asyncio.to_thread(tool_fn, **tool_input) to run them in worker threads. This gives us the best of both worlds: efficient event loop execution for async tools and safe thread-based execution for blocking sync tools.

Now that our Agent class can handle both sync and async tools, let's understand when to use each approach:

Synchronous Agent Wrapper (create_agent_tool_sync)

• ✅ Simpler to understand—just calls run_sync()
• ✅ Works with any Agent implementation
• ✅ Good for single agent calls or when thread overhead isn't a concern
• ❌ Each agent runs in its own thread with its own event loop
• ❌ More overhead when running many agents in parallel

Asynchronous Agent Wrapper (create_agent_tool_async)

• ✅ More efficient—all agents share the same event loop
• ✅ Better resource usage for large-scale parallel orchestration
• ✅ Can handle hundreds of concurrent agent operations
• ❌ Requires Agent class to support async tool detection
• ❌ Slightly more complex to understand

For production systems that need to scale, the async approach is superior. But both approaches work correctly and achieve parallel execution—they just use different mechanisms to get there. The sync approach uses multiple threads, while the async approach uses a single event loop with concurrent tasks.

Let's create our orchestrator using the async agent wrapper approach to demonstrate the most efficient pattern:

# Create agent tool for calculator (using async approach)
calculator_tool_function, calculator_tool_schema = create_agent_tool_async(
    calculator_assistant,
    "Call the calculator assistant to solve a single mathematical problem or equation."
)

# Create the orchestrator assistant
orchestrator = Agent(
    name="orchestrator",
    system_prompt=(
        "You are a math orchestration coordinator. When given complex problems with multiple independent "
        "subproblems, you MUST break them down and delegate each subproblem to a separate calculator agent call. "
        "Issue multiple calculator_assistant_agent tool calls IN PARALLEL within the same turn for independent "
        "calculations. Each agent call should handle ONE specific calculation or equation. "
        "After all agents return results, synthesize them into a clear, complete final answer."
    ),
    tools={calculator_tool_schema["name"]: calculator_tool_function},
    tool_schemas=[calculator_tool_schema]
)

We use create_agent_tool_async() to create an async wrapper for the calculator agent. The orchestrator's system prompt instructs it to break down complex problems and make multiple tool calls in parallel. When the orchestrator makes multiple calculator_assistant_agent calls, they'll all execute concurrently on the event loop—this is the most efficient way to coordinate multiple agents.

Note: If you used create_agent_tool_sync() instead, the orchestrator would still work and execute agent calls in parallel, but each agent would run in its own thread. Both approaches achieve parallelization; the async approach is just more efficient at scale.

Let's see the orchestrator in action with a complex problem that requires multiple independent calculations:

# Run the orchestrator on a problem that requires breaking into parallel subproblems
messages = [
    {
        'role': 'user',
        'content': (
            'I need to calculate the total cost of three different purchases:\n'
            '- Store A: 15 items at $12.50 each, plus $8.99 shipping\n'
            '- Store B: 23 items at $9.75 each, plus $12.50 shipping\n'
            '- Store C: 8 items at $24.99 each, plus $6.25 shipping\n'
            'What is the grand total across all three stores?'
        )
    }
]
result_messages, response = await orchestrator.run(messages)

print('\n=== Final Response ===\n')
print(response)

We create a message asking for the total cost across three stores, each with different items, prices, and shipping costs. The orchestrator will recognize that each store's calculation is independent and delegate them to separate calculator agents running in parallel. Because we used the async wrapper approach, all three calculator agents will run concurrently on the event loop. Let's examine the output to see this efficient parallel execution in action.

When we run the orchestrator, the output shows parallel delegation in action:

🔧 Tool called: calculator_assistant_agent({'message': 'Calculate the total cost for Store A: 15 items at $12.50 each, plus $8.99 shipping'})
🔧 Tool called: calculator_assistant_agent({'message': 'Calculate the total cost for Store B: 23 items at $9.75 each, plus $12.50 shipping'})
🔧 Tool called: calculator_assistant_agent({'message': 'Calculate the total cost for Store C: 8 items at $24.99 each, plus $6.25 shipping'})
🔧 Tool called: multiply_numbers({'a': 8, 'b': 24.99})
🔧 Tool called: multiply_numbers({'a': 23, 'b': 9.75})
🔧 Tool called: multiply_numbers({'a': 15, 'b': 12.5})
🔧 Tool called: sum_numbers({'a': 187.5, 'b': 8.99})
🔧 Tool called: sum_numbers({'a': 199.92, 'b': 6.25})
🔧 Tool called: sum_numbers({'a': 224.25, 'b': 12.5})
📊 Agent (calculator_assistant) as tool finished working
📊 Agent (calculator_assistant) as tool finished working
📊 Agent (calculator_assistant) as tool finished working

Notice how the orchestrator makes three calculator_assistant_agent tool calls right at the start. Because we used async wrappers, all three execute concurrently on the event loop—no thread overhead! Each calculator agent then uses its math tools (which are sync functions) that execute in worker threads via asyncio.to_thread(). You can see the interleaved tool calls from all three calculator agents: they're all calling multiply_numbers and sum_numbers at roughly the same time, demonstrating true parallel execution at both the orchestrator and calculator levels. The "Agent finished working" messages appear together, confirming that all three calculator agents completed their work concurrently.

If we had used create_agent_tool_sync() instead, we'd see the same output pattern, but under the hood each agent would be running in its own thread. Both approaches work—the async approach is just more efficient for large-scale systems.

After the three calculator agents return their results, the orchestrator synthesizes them into a clear, complete answer:

=== Final Response ===

## Total Cost Breakdown

Here's the complete breakdown of your purchases across all three stores:

**Store A:** $196.49
- 15 items × $12.50 = $187.50
- Shipping: $8.99

**Store B:** $236.75
- 23 items × $9.75 = $224.25  
- Shipping: $12.50

**Store C:** $206.17
- 8 items × $24.99 = $199.92
- Shipping: $6.25

**Grand Total: $639.41**

The orchestrator successfully broke down the problem, delegated the independent calculations to run in parallel, and then combined the results into a well-formatted final answer. This demonstrates the power of the orchestrator pattern: complex problems get solved efficiently through parallel delegation to specialized agents, while maintaining a clean separation of concerns between coordination and execution.

You've just completed the final lesson of this course, and what an incredible journey it's been! You started by building your first Claude agent, then you learned to give it tools, made it fully asynchronous, enabled parallel tool execution, and now you've built a complete orchestrator system that can coordinate multiple specialized agents running in parallel.

In this lesson, you learned two approaches for wrapping agents as tools: the synchronous wrapper that runs agents in worker threads, and the asynchronous wrapper that runs agents directly on the event loop. You updated your Agent class to intelligently detect and handle both sync and async tools using inspect.iscoroutinefunction(). You now understand the trade-offs: sync wrappers are simpler but less efficient at scale, while async wrappers offer superior performance for large-scale parallel orchestration.

This orchestrator pattern you just implemented represents the cutting edge of agentic system design. You now understand how to break down complex problems, delegate work to specialized agents, handle both sync and async tools intelligently, and leverage Python's async capabilities to execute multiple agent calls concurrently. These are the exact skills that top AI engineering teams use to build production systems that scale.

All that's left now is to complete the practice exercises and claim your certificate! The exercises will challenge you to apply everything you've learned—from basic agent setup to advanced parallel orchestration with intelligent async/sync tool handling. Once you finish them, you'll have demonstrated mastery of building sophisticated Claude agent systems in Python, and you'll have your certificate to prove it. Let's finish strong—you're almost there! 🚀