Asynchronous and Streamed Agent Execution Modes

Introduction & Context

Welcome back! In the previous lesson, you learned how to structure your agent’s output using Pydantic models and the output_type parameter. This made your agent’s responses more predictable and easier to use in your applications. You also practiced accessing specific fields from the agent’s output, which is a key skill for building robust systems. Now, let’s take the next step: making your agent-powered applications more responsive and interactive. In many real-world scenarios — such as chatbots, web apps, or tools that need to handle multiple users at once — waiting for an agent to finish its work before doing anything else can make your application feel slow or unresponsive. This is where asynchronous and streamed execution modes come in. These modes allow your program to keep working while the agent is thinking, or even to show results as soon as they are available, creating a smoother and more engaging user experience. In this lesson, you will learn how to run agents asynchronously and how to process their outputs in real time using streaming. By the end, you will be able to build applications that feel fast and interactive, even when working with complex AI agents.

Overview Of Openai Agents SDK Execution Modes

The OpenAI Agents SDK provides three main ways to run agents: synchronous, asynchronous, and streamed execution. You have already used synchronous execution in earlier lessons, where your code waits for the agent to finish before moving on. This is simple and works well for basic scripts, but it can block your program and make it less responsive. Asynchronous execution allows your program to start an agent task and then do other things while waiting for the result. This is especially useful in applications with user interfaces, web servers, or any situation where you do not want to block the main thread. Streamed execution takes this a step further by letting you process the agent’s output as it is being generated, token by token or event by event. This is ideal for real-time applications, such as chatbots that display text as it is typed out. Here is a quick comparison of the three modes: Mode Method Blocking? Real-Time Output? Use Case Example Synchronous Runner.run_sync Yes No Simple scripts, batch jobs Asynchronous Runner.run No No Web servers, UI apps Streamed Runner.run_streamed No Yes Chatbots, live dashboards By using asynchronous and streamed execution, you can make your applications more efficient and user-friendly.

Mode	Method	Blocking?	Real-Time Output?	Use Case Example
Synchronous	`Runner.run_sync`	Yes	No	Simple scripts, batch jobs
Asynchronous	`Runner.run`	No	No	Web servers, UI apps
Streamed	`Runner.run_streamed`	No	Yes	Chatbots, live dashboards

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Streamed Execution Mode

Streamed execution is designed for situations where you want to process the agent’s output as soon as it is available, rather than waiting for the entire response. This is especially useful for real-time applications, such as chatbots or live dashboards, where you want to display information to the user as quickly as possible.

With the OpenAI Agents SDK, you can use the Runner.run_streamed method to start the agent and then iterate over the streaming events as they arrive. Each event can represent a new chunk of text or another type of update from the agent.

Here is an example that builds on the previous code, but now uses streaming to print the agent’s response as it is generated:

Python

import asyncio
from agents import Runner
from openai.types.responses import ResponseTextDeltaEvent

# Define your agent...

async def main():
    # Run the agent in streaming mode
    result = Runner.run_streamed(
        starting_agent=agent,
        input="What are the best travel tips for solo travelers in Japan?"
    )

    # Process streaming events
    async for event in result.stream_events():
        # Check if the event contains a new chunk of text from the agent
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            # Print the new text chunk immediately as it arrives
            print(event.data.delta, end="", flush=True)
            
    # Print the final output after streaming is complete
    print("Final Output:" + result.final_output)

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

import asyncio
from agents import Runner
from openai.types.responses import ResponseTextDeltaEvent

# Define your agent...

async def main():
    # Run the agent in streaming mode
    result = Runner.run_streamed(
        starting_agent=agent,
        input="What are the best travel tips for solo travelers in Japan?"
    )

    # Process streaming events
    async for event in result.stream_events():
        # Check if the event contains a new chunk of text from the agent
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            # Print the new text chunk immediately as it arrives
            print(event.data.delta, end="", flush=True)
            
    # Print the final output after streaming is complete
    print("Final Output:" + result.final_output)

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

Here’s how you can print each token as soon as it is generated using run_streamed in the OpenAI Agents SDK:

Start the agent in streaming mode using Runner.run_streamed. This lets you receive the agent’s response bit by bit, instead of waiting for the whole answer.
Loop through the streaming events with async for. Each event represents a new piece of the agent’s response.
Check if the event contains new text by looking for events of type "raw_response_event" and making sure the event data is a ResponseTextDeltaEvent.
Print the new text immediately by accessing event.data.delta. This will show each token or chunk as soon as it is available, just like watching someone type in a chat. Here’s how the print statement works:
- event.data.delta: Contains the latest chunk or token generated by the agent during streaming.
- end="": Prevents Python from adding a newline after each chunk, so the streamed text appears as a continuous line.
- flush=True: Forces Python to immediately write the output to the console, ensuring each new chunk is displayed to the user as soon as it arrives.

When the agent is finished, you can still access the structured final output as before. Streamed execution is perfect for applications where you want to keep users engaged and informed as soon as new information is available.

Introduction & Context

Overview Of Openai Agents SDK Execution Modes

Join the 1M+ learners on CodeSignal

Asynchronous Execution Mode

Streamed Execution Mode

Real-World Use Cases And Best Practices

Summary & Next Steps