Run, Stream, and Inspect OpenAI Agent Results

An OpenAI SDK run loops through model responses, tool calls, handoffs, and results until it reaches a final output, interruption, error, or enforced limit. Streaming exposes events while that loop continues; it does not make partial text a completed outcome. Applications must distinguish progress, final output, resumable state, and operational diagnostics.

What will you be able to do?

You will select synchronous/asynchronous and streamed execution, enforce turn limits, and map SDK results into stable application outcomes.

What does one run do?

The runner calls the current agent’s model, executes requested tools, follows handoffs, and repeats. Runner.run() is asynchronous; Runner.run_sync() is for synchronous contexts; streamed execution returns an object whose events can feed a user interface while the run is active.

Always confirm exact streaming event types in the current language reference before binding production UI logic to them.

How should an application consume results?

Use a boundary function:

python

from dataclasses import dataclass
from typing import Literal

from agents import Agent, Runner


@dataclass(frozen=True)
class RunOutcome:
    status: Literal["succeeded", "waiting_for_approval", "failed"]
    output: object | None
    reason: str | None


async def execute(agent: Agent, prompt: str) -> RunOutcome:
    try:
        result = await Runner.run(agent, prompt, max_turns=8)
    except Exception as error:
        return RunOutcome("failed", None, type(error).__name__)

    if result.interruptions:
        return RunOutcome("waiting_for_approval", None, "approval_required")
    return RunOutcome("succeeded", result.final_output, None)

Production code should classify known SDK/provider failures more precisely and preserve a trace/correlation reference. The example shows the boundary, not a complete error taxonomy.

Which result surfaces matter?

final_output: typed or text output after completion.
last_agent: the specialist owning the end of the run.
to_input_list(): replay-ready history when the application manages conversation input.
interruptions and to_state(): approval/pause continuation.
lower-level items, raw responses, guardrail results, and usage: audits and debugging.

Choose one continuation strategy; do not accidentally send replayed history while also using server-managed continuation.

How do you design streaming safely?

Publish normalized events such as message.delta, tool.proposed, tool.completed, approval.required, and run.completed. Mark deltas as provisional. Store the authoritative completed result separately. Apply backpressure and disconnect handling; decide whether a client disconnect cancels work or only detaches observation.

Failure injection: Disconnect the UI during a tool call. The worker must follow a documented policy: cancel safely or continue durably. It must not leave the outcome unknowable.

How do you prove it works?

Test final-only, tool-using, turn-limit, interruption, cancellation, provider-error, and consumer-disconnect paths. Assert state/event sequences rather than exact generated wording.

What mistakes should you avoid?

Treating the last text delta as a final answer.
Catching every exception and returning success text.
Resetting turn budgets during continuation.
Mixing continuation strategies and duplicating history.
Sending raw provider events directly to long-lived clients as your public API.

Check your understanding

Why is streamed text provisional?
Which result carries resumable approval state?
What should happen when a streaming client disconnects?

Expert extension

Design a normalized event schema with sequence numbers and write a reducer that reconstructs the run view after reconnect without duplicating events.

Official references

Run, Stream, and Inspect OpenAI Agent Results

What does one run do?

Always confirm exact streaming event types in the current language reference before binding production UI logic to them.

How should an application consume results?

Use a boundary function:

python

from dataclasses import dataclass
from typing import Literal

from agents import Agent, Runner


@dataclass(frozen=True)
class RunOutcome:
    status: Literal["succeeded", "waiting_for_approval", "failed"]
    output: object | None
    reason: str | None


async def execute(agent: Agent, prompt: str) -> RunOutcome:
    try:
        result = await Runner.run(agent, prompt, max_turns=8)
    except Exception as error:
        return RunOutcome("failed", None, type(error).__name__)

    if result.interruptions:
        return RunOutcome("waiting_for_approval", None, "approval_required")
    return RunOutcome("succeeded", result.final_output, None)

Production code should classify known SDK/provider failures more precisely and preserve a trace/correlation reference. The example shows the boundary, not a complete error taxonomy.

Which result surfaces matter?

final_output: typed or text output after completion.

last_agent: the specialist owning the end of the run.

to_input_list(): replay-ready history when the application manages conversation input.

interruptions and to_state(): approval/pause continuation.

lower-level items, raw responses, guardrail results, and usage: audits and debugging.

Choose one continuation strategy; do not accidentally send replayed history while also using server-managed continuation.

How do you design streaming safely?

Failure injection: Disconnect the UI during a tool call. The worker must follow a documented policy: cancel safely or continue durably. It must not leave the outcome unknowable.