Streaming Responses in Production with OpenAI Agents SDK: Complete Guide

Streaming Responses in Production with OpenAI Agents SDK: Complete Guide

Streaming responses has become a non-negotiable feature for modern AI applications. Users expect to see tokens appear in real time, not stare at a loading spinner for ten seconds while an agent reasons through a complex task. The OpenAI Agents SDK provides first-class streaming support, but using it correctly in production requires understanding its event model, handling errors gracefully, and integrating it with your existing infrastructure. This guide walks through everything you need to know.

What Is Streaming in the Agents SDK?

The OpenAI Agents SDK is a framework for building agentic workflows — chains of LLM calls, tool invocations, and handoffs between specialized agents. Streaming in this context means receiving incremental updates as the agent works, rather than waiting for the entire run to complete before returning a single response.

Unlike raw chat completions streaming, where you simply receive token-by-token text, the Agents SDK emits a richer stream of events. These events include partial text deltas, tool call starts and completions, agent handoffs, message objects, and the final run result. This granularity lets you build UIs that show not just what the agent is saying, but what it is doing at each step.

Why Streaming Matters in Production

• Perceived latency: First-token latency of 200ms feels dramatically faster than a 5-second full-response wait, even if total time is identical.
• Transparency: Showing tool calls and reasoning steps builds user trust in agentic systems.
• Long-running tasks: Agents that call multiple tools or perform research can take 30+ seconds. Without streaming, users assume the app is broken.
• Cancelability: Streaming lets users interrupt a run early, saving compute costs when they get the answer they need.
• Debuggability: Real-time visibility into agent behavior makes production issues easier to diagnose.

Getting Started: Basic Streaming

The core method for streaming is Runner.run_streamed(). It returns an asynchronous stream of StreamEvent objects. Let's start with a minimal example.

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(
        name="ResearchAssistant",
        instructions="You are a helpful research assistant. Be concise.",
        model="gpt-4o",
    )

    result = Runner.run_streamed(agent, "Explain how transformers work in 3 sentences.")

    async for event in result.stream_events():
        print(event)

asyncio.run(main())

This prints every event the SDK emits. In practice, you will want to filter and handle specific event types. The two most important are ResponseTextDeltaEvent (partial text output) and RunItemStreamEvent (lifecycle events for items like messages and tool calls).

Handling Text Deltas

For most chat-style interfaces, you care primarily about text deltas. Here is how to extract and display them cleanly.

import asyncio
from agents import Agent, Runner
from agents.stream_events import ResponseTextDeltaEvent

async def stream_response(prompt: str):
    agent = Agent(
        name="Writer",
        instructions="You are a concise technical writer.",
        model="gpt-4o",
    )

    result = Runner.run_streamed(agent, prompt)

    async for event in result.stream_events():
        if isinstance(event, ResponseTextDeltaEvent):
            print(event.delta, end="", flush=True)

    print()  # final newline

asyncio.run(stream_response("Write a haiku about distributed systems."))

The flush=True ensures each token appears immediately rather than being buffered. In a web context, you would forward these deltas to the client via Server-Sent Events or WebSockets instead of printing.

Streaming with Tool Calls

Agents become powerful when they use tools. Streaming lets you show users when a tool is being called and when it returns. The RunItemStreamEvent wraps items as they are produced, and each item has a type field you can inspect.

import asyncio
from agents import Agent, Runner, function_tool
from agents.stream_events import ResponseTextDeltaEvent, RunItemStreamEvent
from agents.items import ToolCallItem, ToolCallOutputItem, MessageOutputItem

@function_tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    # Simulated API call
    return f"The weather in {city} is sunny and 72°F."

async def main():
    agent = Agent(
        name="WeatherAgent",
        instructions="You help users with weather queries. Use the get_weather tool.",
        tools=[get_weather],
        model="gpt-4o",
    )

    result = Runner.run_streamed(agent, "What's the weather in Tokyo?")

    async for event in result.stream_events():
        if isinstance(event, ResponseTextDeltaEvent):
            print(event.delta, end="", flush=True)
        elif isinstance(event, RunItemStreamEvent):
            item = event.item
            if isinstance(item, ToolCallItem):
                print("\n[Calling tool...]")
            elif isinstance(item, ToolCallOutputItem):
                print(f"\n[Tool output: {item.output}]")
            elif isinstance(item, MessageOutputItem):
                pass  # text handled by delta events

asyncio.run(main())

Streaming with Agent Handoffs

One of the SDK's signature features is agent handoffs — one agent delegating to another. Streaming surfaces handoff events so your UI can indicate which agent is currently active.

import asyncio
from agents import Agent, Runner, handoff
from agents.stream_events import ResponseTextDeltaEvent, RunItemStreamEvent
from agents.items import HandoffCallItem, HandoffOutputItem

billing_agent = Agent(
    name="BillingAgent",
    instructions="You handle billing questions only.",
    model="gpt-4o",
)

triage_agent = Agent(
    name="TriageAgent",
    instructions="Route users to the right specialist.",
    handoffs=[billing_agent],
    model="gpt-4o",
)

async def main():
    result = Runner.run_streamed(triage_agent, "I was charged twice for my subscription.")

    current_agent = "TriageAgent"
    async for event in result.stream_events():
        if isinstance(event, RunItemStreamEvent):
            if isinstance(event.item, HandoffCallItem):
                print(f"\n[Handing off from {current_agent}...]")
            elif isinstance(event.item, HandoffOutputItem):
                current_agent = event.item.target_agent.name
                print(f"\n[Now speaking with {current_agent}]")
        elif isinstance(event, ResponseTextDeltaEvent):
            print(event.delta, end="", flush=True)

asyncio.run(main())

Building a Production Streaming Endpoint

For real applications, you need an HTTP endpoint that streams to the browser. Server-Sent Events (SSE) is the simplest protocol for this. Here is a complete FastAPI example.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from agents import Agent, Runner
from agents.stream_events import ResponseTextDeltaEvent, RunItemStreamEvent
from agents.items import ToolCallItem, ToolCallOutputItem
import json

app = FastAPI()

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    model="gpt-4o",
)

class ChatRequest(BaseModel):
    message: str

async def event_generator(message: str):
    result = Runner.run_streamed(agent, message)

    async for event in result.stream_events():
        if isinstance(event, ResponseTextDeltaEvent):
            yield f"data: {json.dumps({'type': 'delta', 'content': event.delta})}\n\n"
        elif isinstance(event, RunItemStreamEvent):
            if isinstance(event.item, ToolCallItem):
                yield f"data: {json.dumps({'type': 'tool_call_start'})}\n\n"
            elif isinstance(event.item, ToolCallOutputItem):
                yield f"data: {json.dumps({'type': 'tool_call_end', 'output': str(event.item.output)})}\n\n"

    yield f"data: {json.dumps({'type': 'done'})}\n\n"

@app.post("/chat")
async def chat(req: ChatRequest):
    return StreamingResponse(
        event_generator(req.message),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # disable nginx buffering
        },
    )

Client-Side Consumption

On the browser side, the EventSource API or a fetch-based reader consumes the SSE stream. Here is a vanilla JavaScript example.

async function streamChat(message) {
    const response = await fetch("/chat", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ message }),
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = "";

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split("\n");
        buffer = lines.pop();

        for (const line of lines) {
            if (!line.startsWith("data: ")) continue;
            const data = JSON.parse(line.slice(6));

            if (data.type === "delta") {
                document.getElementById("output").textContent += data.content;
            } else if (data.type === "tool_call_start") {
                console.log("Tool call started...");
            } else if (data.type === "tool_call_end") {
                console.log("Tool output:", data.output);
            } else if (data.type === "done") {
                console.log("Stream complete.");
            }
        }
    }
}

Error Handling and Resilience

Production streams break. Network connections drop, rate limits trigger, and tool calls fail. You need to handle these cases without leaving users staring at a half-finished response.

import asyncio
from agents import Agent, Runner
from agents.exceptions import AgentsException

async def safe_stream(agent: Agent, prompt: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            result = Runner.run_streamed(agent, prompt)
            async for event in result.stream_events():
                yield event
            return  # success
        except AgentsException as e:
            if attempt == max_retries - 1:
                yield {"type": "error", "message": str(e)}
                return
            await asyncio.sleep(2 ** attempt)
        except Exception as e:
            yield {"type": "error", "message": f"Unexpected error: {e}"}
            return

On the client side, always implement reconnection logic. If the stream drops mid-response, offer the user a retry button rather than silently failing.

Best Practices

• Buffer strategically: Some downstream systems (like TTS) need complete sentences, not raw tokens. Buffer deltas and flush on punctuation if your consumer needs larger chunks.
• Set timeouts: Use asyncio.wait_for or framework-level timeouts to prevent hung streams from consuming server resources indefinitely.
• Log structured events: Persist stream events server-side for debugging and analytics. This is invaluable when users report "the agent said something weird."
• Handle backpressure: If the client reads slowly, your server can accumulate memory. Use bounded queues and drop or slow down if the consumer cannot keep up.
• Disable proxy buffering: Set X-Accel-Buffering: no for nginx and equivalent headers for other proxies, or tokens will arrive in large batches instead of streaming smoothly.
• Validate input before streaming starts: Once a stream begins, validation errors mid-stream are awkward to communicate. Validate request shape, auth, and rate limits before the first event.
• Use streaming-aware guardrails: The Agents SDK supports guardrails that run alongside the agent. For streaming, prefer output guardrails that can evaluate partial content or run at the end without blocking the stream.
• Test with slow networks: Streaming behavior changes dramatically under network constraints. Use tools like Charles Proxy or toxiproxy to simulate latency and packet loss.

Structured Output with Streaming

If your agent produces structured JSON, streaming becomes trickier because partial JSON is not parseable. The recommended pattern is to stream a "thinking" indicator while the agent works, then emit the final structured object once complete.

import asyncio
import json
from pydantic import BaseModel
from agents import Agent, Runner
from agents.stream_events import ResponseTextDeltaEvent

class ResearchSummary(BaseModel):
    topic: str
    key_points: list[str]
    confidence: float

agent = Agent(
    name="Researcher",
    instructions="Produce a structured research summary.",
    output_type=ResearchSummary,
    model="gpt-4o",
)

async def main():
    result = Runner.run_streamed(agent, "Summarize the state of quantum computing.")

    async for event in result.stream_events():
        if isinstance(event, ResponseTextDeltaEvent):
            print(".", end="", flush=True)  # progress indicator

    final = result.final_output
    print(f"\n\n{json.dumps(final.model_dump(), indent=2)}")

asyncio.run(main())

Conclusion

Streaming transforms AI applications from batch-processing tools into interactive experiences. The OpenAI Agents SDK's event-based streaming model gives you fine-grained control over what users see and when they see it — from the first token of a response to tool call progress and agent handoffs. By combining the SDK's native streaming with robust error handling, proper HTTP infrastructure, and thoughtful client-side consumption, you can build production-grade agentic interfaces that feel responsive and trustworthy. Start with the basic run_streamed pattern, layer in event filtering as your UI complexity grows, and always test under real network conditions before shipping to users.