Agent Orchestration Patterns

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 how to wrap agents as asynchronous tools, allowing one agent to call another agent just as it would call any other tool. You'll learn how to make your Agent class smart enough to detect whether a tool is async or sync and handle each type appropriately, enabling efficient parallel orchestration across multiple specialized agents.

Agent orchestration is a pattern in which one coordinator agent manages and delegates work to multiple specialized agents. Think of it as 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. The Gemini Agent class automatically handles schema mapping internally: it can accept schemas in either Claude format (with input_schema) or Gemini format (with parameters) and passes standard lowercase JSON Schema types to the Google Gen AI SDK. This means you can use the same schema files across different implementations without modification. 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.

To use an agent as a tool, we need to wrap it in a function that can be called like any other tool. With Gemini, we create an asynchronous wrapper that allows the agent to run directly on the event loop alongside other async operations. Here's what an async agent wrapper looks like:

def create_agent_tool(agent, description):
    """
    Create a tool function and schema for using an agent as a tool.

    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
        _, response = await agent.run([{"role": "user", "content": message}])
        print(f"📊 Agent ({agent.name}) as tool finished working")
        return response

    tool_schema = {
        "name": f"{agent.name}_agent",
        "description": description,
        "parameters": {
            "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 characteristics: we define agent_tool_function with async def and use await agent.run() to call the agent asynchronously. This creates a coroutine function that runs directly on the event loop. The schema uses Gemini's parameters format instead of input_schema, defining a simple interface that accepts a message string to send to the agent.

This async wrapper approach is efficient: 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. When the orchestrator makes multiple agent tool calls, they'll all execute in parallel on the event loop, just like the parallel tool execution you implemented in the previous lesson.

To support both sync tools (like our math functions) and async tools (like our agent wrapper), we need 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, allowing it to handle both regular tools and agent-as-tool wrappers seamlessly.

Now let's look at how the _call_tool() method in our Gemini Agent class handles both async and sync tools intelligently:

async def _call_tool(self, name, args):
    import inspect
    print(f"🔧 Tool called: {name}({args})")
    try:
        fn = self.tools[name]
    except KeyError:
        result = f"Error: Function {name} not found"
        print(f"❌ Error: {result}")
        return self._function_response_part(name, result)
    try:
        if inspect.iscoroutinefunction(fn):
            # If the tool is async (e.g., agent-as-tool), run it directly
            result_value = await fn(**args)
        else:
            # If the tool is sync, run it in a thread
            result_value = await asyncio.to_thread(fn, **args)
        result = str(result_value)
        print(f"✅ Result: {result}")
        return self._function_response_part(name, result)
    except Exception as e:
        result = f"Error: {str(e)}"
        print(f"❌ Error: {result}")
        return self._function_response_part(name, result)

The key addition is the if inspect.iscoroutinefunction(fn) check. For async tools (like our agent wrapper), we call them directly with await fn(**args), which runs them on the event loop. For sync tools (like our math functions), we use asyncio.to_thread(fn, **args) 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.

Notice that the method returns results using _function_response_part(), which formats the response in Gemini's expected function response format. This helper method wraps the result in the proper structure that Gemini's API expects when receiving tool execution results.

Let's create our orchestrator using the async agent wrapper to enable efficient parallel delegation:

# Create agent tool for calculator
calculator_tool_function, calculator_tool_schema = create_agent_tool(
    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() 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.

The orchestrator receives both the tool function (which it will call) and the tool schema (which tells Gemini's API what the tool does and what parameters it accepts). This separation allows Gemini to understand when to call the tool while our code handles the actual execution.

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

async def main():
    # ... (agent creation code from above)
    
    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)

if __name__ == "__main__":
    asyncio.run(main())

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': 'Compute 15 * 12.50 + 8.99'})
🔧 Tool called: calculator_assistant_agent({'message': 'Compute 23 * 9.75 + 12.50'})
🔧 Tool called: calculator_assistant_agent({'message': 'Compute 8 * 24.99 + 6.25'})
🔧 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': 224.25, 'b': 12.5})
📊 Agent (calculator_assistant) as tool finished working
🔧 Tool called: sum_numbers({'a': 199.92, 'b': 6.25})
🔧 Tool called: sum_numbers({'a': 187.5, 'b': 8.99})
📊 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.

This is the power of combining async agent wrappers with intelligent sync/async tool detection: the orchestrator and calculator agents run on the event loop, while the math functions safely execute in worker threads, all happening in parallel for maximum efficiency.

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

=== Final Response ===

Here are the totals per store and the grand total:

- Store A: 15 × $12.50 + $8.99 = $196.49
- Store B: 23 × $9.75 + $12.50 = $236.75
- Store C: 8 × $24.99 + $6.25 = $206.17

Grand total: $196.49 + $236.75 + $206.17 = $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 Gemini 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 how to wrap agents as asynchronous tools, allowing one agent to call another agent just like any other tool. You updated your understanding of how the Agent class intelligently detects and handles both sync and async tools using inspect.iscoroutinefunction(). You now understand how async agent wrappers run directly on the event loop for maximum efficiency, while sync tools safely execute in worker threads, all coordinated seamlessly by the intelligent tool execution logic.

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 Gemini agent systems in Python, and you'll have your certificate to prove it. Let's finish strong—you're almost there! 🚀