Going Async with GPT Agents

Welcome to the first lesson of Parallelizing GPT Agentic Systems! In previous courses, you built a solid foundation by creating an agent class that can handle conversations, use tools, and even hand off control to other specialized agents. You've also seen how the AsyncOpenAI client works. Now, we're going to take your agent system to the next level by converting your agent to handle multiple concurrent conversations using async/await patterns. By the end, you'll have an agent that can juggle several API calls at once without blocking, dramatically improving the efficiency of your workflows.

Let's start by understanding why we need async in the first place. When you make a regular API call to OpenAI's GPT, your program stops and waits for the response. This is called "blocking" behavior. If you need to have three separate conversations with GPT, your program handles them one at a time: start conversation 1, wait for all responses, finish conversation 1, then start conversation 2, and so on.

This sequential approach wastes time. While your program waits for GPT to respond to conversation 1, it could be starting conversation 2 or 3. Network calls and API processing take time, but your CPU sits idle during these waits.

Async programming solves this problem. When you make an async API call, your program can continue doing other work while waiting for the response. Think of it like a restaurant: a synchronous waiter takes one order, goes to the kitchen, waits for the food, delivers it, and only then takes the next order. An async waiter takes multiple orders, sends them all to the kitchen, and delivers each meal as it becomes ready. The kitchen (OpenAI's API) processes multiple requests in parallel, and your program efficiently manages all of them.

For agent systems, this means a single agent can manage multiple conversations simultaneously, or you can run several agents in parallel. This becomes especially powerful when agents need to make multiple tool calls or coordinate with other agents through handoffs. Let's start by making the necessary changes to your agent class.

The first change we'll make is in your agent's initialization. You've seen the AsyncOpenAI client before, and switching to it is straightforward. In your __init__ method, we'll simply replace the regular OpenAI() client with AsyncOpenAI():

from openai import AsyncOpenAI

class Agent:
    def __init__(
        self,
        name,
        system_prompt="You are a helpful assistant.",
        model="gpt-5",
        tools=None,
        tool_schemas=None,
        handoffs=None,
        max_turns=10,
        reasoning_effort="low"
    ):
        self.client = AsyncOpenAI()  # Changed from OpenAI()
        self.name = name
        self.model = model
        # ... rest of initialization

This single-line change tells your agent to use the async version of the OpenAI client. The AsyncOpenAI client has the same interface as the regular client, but its methods return "awaitables" instead of immediate results. This means you'll need to use the await keyword when calling its methods, which we'll cover in the next section.

Now that your agent has an async client, we need to update the run method to work asynchronously. This involves two key changes: adding the async keyword to the method definition and using await when making API calls. Here's how we'll update the run method:

async def run(self, input_messages):  # Changed: Added 'async' keyword
    messages = input_messages.copy()

    turn = 0
    while turn < self.max_turns:
        turn += 1
        all_tools = []
        if self.tool_schemas:
            all_tools.extend(self.tool_schemas)
        if self.handoffs:
            all_tools.append(self.handoff_schema)
        
        response = await self.client.responses.create(  # Changed: Added 'await' keyword
            model=self.model,
            instructions=self.system_prompt,
            input=messages,
            tools=all_tools,
            reasoning={"effort": self.reasoning_effort},
            store=False
        )

        function_calls = [
            item for item in response.output
            if item.type == "function_call"
        ]

        if function_calls:
            function_outputs = []
            for function_call in function_calls:
                if function_call.name == "handoff":
                    handoff_success, handoff_result = await self.call_handoff(function_call, messages)  # Changed: Added 'await' keyword
                    if handoff_success:
                        return handoff_result
                    else:
                        function_outputs.append(handoff_result)
                else:
                    messages.append({
                        "type": "function_call",
                        "name": function_call.name,
                        "arguments": function_call.arguments,
                        "call_id": function_call.call_id
                    })
                    tool_result = self.call_tool(function_call)
                    function_outputs.append(tool_result)
            messages.extend(function_outputs)
        else:
            messages.append({
                "role": "assistant",
                "content": response.output_text
            })
            return messages, response.output_text
            
    raise Exception("Max turns reached")

The async def keyword at the beginning tells the runtime this is an asynchronous function. Inside the method, we use await before self.client.responses.create(). This is where the magic happens. When your code hits the await keyword, it tells the runtime: "This operation will take some time. While we're waiting for the API response, feel free to do other work." Behind the scenes, the event loop can now switch to other tasks. If you have multiple agent conversations running, the event loop can start processing another conversation while this one waits for GPT's response. Notice that we also use await when calling self.call_handoff() because handoffs involve calling another agent's run() method, which is now also async.

When your agent hands off control to another agent, it needs to wait for that agent to complete its work. Since the other agent's run() method is now async, we'll also need to make the call_handoff method async:

async def call_handoff(self, function_call, messages):  # Changed: Added 'async' keyword
    args = json.loads(function_call.arguments or "{}")
    agent_name = args.get("name")
    reason = args.get("reason", "No reason provided")
    
    print(f"🔄 Handoff to: {agent_name}")
    print(f"📝 Reason: {reason}")
    
    try:
        target_agent = next(agent for agent in self.handoffs if agent.name == agent_name)
        return True, await target_agent.run(messages)  # Changed: Added 'await' keyword
    except StopIteration:
        return False, {
            "type": "function_call_output",
            "call_id": function_call.call_id,
            "output": json.dumps({"result": f"Handoff failed: Agent '{agent_name}' not found. Available agents: {[agent.name for agent in self.handoffs]}"})
        }
    except Exception as e:
        return False, {
            "type": "function_call_output",
            "call_id": function_call.call_id,
            "output": json.dumps({"result": f"Handoff failed: Error during handoff to '{agent_name}': {str(e)}"})
        }

The key change here is adding async to the method definition and using await when calling target_agent.run(messages). This allows the handoff to happen asynchronously. If the target agent needs to make multiple API calls or use tools, your original agent doesn't sit idle waiting. The event loop can switch to other work while the handoff completes. Notice that the method returns a tuple (success, result) — the first value indicates whether the handoff succeeded, and the second contains either the result from the target agent or an error message. Now that your agent is fully async, let's see how we can run multiple conversations in parallel.

Before we run multiple conversations concurrently, we need to set up our imports and tools. The key addition here is importing asyncio, which provides the tools for running async code:

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

# Load the schemas from JSON file
with open('schemas.json', 'r') as f:
    tool_schemas = json.load(f)

# Create a dictionary mapping tool names to functions
tools = {
    "sum_numbers": sum_numbers,
    "multiply_numbers": multiply_numbers,
    "subtract_numbers": subtract_numbers,
    "divide_numbers": divide_numbers,
    "power": power,
    "square_root": square_root
}

With our tools ready, we can now define an async main function that creates and runs multiple conversations concurrently:

async def main():
    # Write multiple prompts that we'll process concurrently
    prompts = [
        "Solve this: (2 + 3) * (4*4)",
        "Find the roots of x^2 - 5x + 6 = 0",
    ]

    # Create a single agent that will handle multiple conversations concurrently
    agent = Agent(
        name="math_assistant",
        system_prompt="You are a helpful math assistant. Always use the available tools to perform calculations accurately.",
        tools=tools,
        tool_schemas=tool_schemas,
        max_turns=15
    )

    # Create a list of coroutine tasks, one for each prompt (not started yet)
    tasks = [
        agent.run([{"role": "user", "content": prompt}])
        for prompt in prompts
    ]

    # Run all tasks concurrently and wait for all to complete
    results = await asyncio.gather(*tasks)

    # Display the final response from each conversation
    for idx, (_, result) in enumerate(results, start=1):
        print(f"\n=== run {idx} ===")
        print(result)

We start by creating a list of prompts that we want to process concurrently, then initialize our agent with the necessary tools and schemas. The key part is creating a list of tasks by calling agent.run() for each prompt. Notice that we don't use await here yet — each call returns a coroutine object (a promise of future work) but doesn't start executing immediately. Then we use asyncio.gather(*tasks) with await. The gather() function takes all our tasks and runs them concurrently. It waits for all of them to complete and returns their results in the same order as the input tasks. While one conversation waits for an API response, another conversation can make progress.

Finally, we need an entry point that creates the event loop and runs our async main function:

# Entry point - creates event loop and runs the async main() function
if __name__ == "__main__":
    asyncio.run(main())

The asyncio.run(main()) function creates an event loop, runs our main() function, and handles cleanup when everything completes. This is the standard way to start an async program.

When we run this code, you'll see output that demonstrates the concurrent execution:

🔧 Tool called: sum_numbers({'a': 2, 'b': 3})
🔧 Tool called: multiply_numbers({'a': 4, 'b': 4})
🔧 Tool called: multiply_numbers({'a': 5, 'b': 16})
🔧 Tool called: power({'base': -5, 'exponent': 2})
🔧 Tool called: multiply_numbers({'a': 4, 'b': 1})
🔧 Tool called: multiply_numbers({'a': 1, 'b': 6})
🔧 Tool called: multiply_numbers({'a': 4, 'b': 6})
🔧 Tool called: subtract_numbers({'a': 25, 'b': 24})
🔧 Tool called: square_root({'number': 1})
🔧 Tool called: sum_numbers({'a': 5, 'b': 1})
🔧 Tool called: subtract_numbers({'a': 5, 'b': 1})
🔧 Tool called: divide_numbers({'a': 6, 'b': 2})

=== run 1 ===
80

=== run 2 ===
The roots are x = 2 and x = 3.

Notice how the tool calls from both conversations are interleaved. This shows that both conversations are running concurrently. The agent switches between them as it waits for API responses, making efficient use of time. The first conversation solves the arithmetic problem while the second finds the roots of the quadratic equation, and both complete much faster than if they ran sequentially.

Now that we've seen concurrent execution in action, let's reflect on what we've achieved and what still needs improvement. By converting our agent to use async/await, we've enabled a single agent instance to manage multiple conversations simultaneously. We can now process different user requests in parallel without creating multiple agent instances. In our example above, one agent handles both math problems at the same time, switching between them efficiently while waiting for API responses.

But what would have happened if we tried to run multiple conversations with our original, synchronous agent? Even if we created separate threads or processes, the conversations would effectively execute sequentially. The synchronous OpenAI client blocks while waiting for responses, wasting time that could be spent processing other conversations.

However, there's an important limitation in our current implementation. Notice in the output how tool calls execute one at a time, even when they come from different conversations:

🔧 Tool called: sum_numbers({'a': 2, 'b': 3})
🔧 Tool called: multiply_numbers({'a': 4, 'b': 4})
🔧 Tool called: multiply_numbers({'a': 5, 'b': 16})

These tool calls execute sequentially because our call_tool() method is synchronous. When one conversation needs to run a tool, all other conversations must wait. This becomes a bottleneck when we have many tool calls or when tools involve external API calls. In the next lesson, we'll parallelize tool execution, allowing our agent to make multiple tool calls simultaneously across different conversations.

You've successfully converted your agent system to use async/await patterns by switching to AsyncOpenAI(), adding async and await to the run() and call_handoff() methods, and using asyncio.gather() to run multiple conversations in parallel. This async foundation is crucial for building more advanced parallel agent systems, and in the upcoming practice exercises, you'll implement concurrent agent workflows and explore different patterns for coordinating multiple agents.