Implementation:Openai Openai agents python Stream Events

Overview

The stream events module defines the typed event classes that are yielded by RunResultStreaming.stream_events() during a streamed agent run. These classes form the StreamEvent union type used to deliver real-time updates about agent activity.

Class Definitions

RawResponsesStreamEvent

@dataclass
class RawResponsesStreamEvent:
    data: TResponseStreamEvent
    type: Literal["raw_response_event"] = "raw_response_event"

Fields:

Field	Type	Description
data	TResponseStreamEvent	The raw stream event from the LLM provider. Contains text deltas, tool call argument chunks, and other model-level streaming data.
type	Literal["raw_response_event"]	Discriminator field. Always "raw_response_event".

RunItemStreamEvent

@dataclass
class RunItemStreamEvent:
    name: Literal["message_output_created", "handoff_requested", "handoff_occured",
                   "tool_called", "tool_output", "reasoning_item_created",
                   "mcp_approval_requested", "mcp_approval_response", "mcp_list_tools"]
    item: RunItem
    type: Literal["run_item_stream_event"] = "run_item_stream_event"

Fields:

Field	Type	Description
name	Literal[...]	Identifies the specific kind of run item. See the table below for all possible values.
item	RunItem	The full RunItem object containing details about the action.
type	Literal["run_item_stream_event"]	Discriminator field. Always "run_item_stream_event".

Possible name values:

Name	Description
message_output_created	A complete message output has been produced by the agent.
handoff_requested	The current agent has requested a handoff to another agent.
handoff_occured	A handoff between agents has been completed.
tool_called	A tool invocation has been initiated.
tool_output	A tool has returned its result.
reasoning_item_created	A reasoning step has been produced by the model.
mcp_approval_requested	An MCP tool requires user approval before execution.
mcp_approval_response	A response to an MCP approval request.
mcp_list_tools	MCP tools have been listed.

AgentUpdatedStreamEvent

@dataclass
class AgentUpdatedStreamEvent:
    new_agent: Agent[Any]
    type: Literal["agent_updated_stream_event"] = "agent_updated_stream_event"

Fields:

Field	Type	Description
new_agent	Agent[Any]	The agent that has taken over processing after a handoff.
type	Literal["agent_updated_stream_event"]	Discriminator field. Always "agent_updated_stream_event".

StreamEvent Union

StreamEvent = Union[RawResponsesStreamEvent, RunItemStreamEvent, AgentUpdatedStreamEvent]

This union type represents any event that can be yielded by stream_events(). The type field on each variant serves as a discriminator for runtime dispatch and static type narrowing.

Source

• File: src/agents/stream_events.py, lines 1-63
• Import: from agents.stream_events import StreamEvent, RawResponsesStreamEvent, RunItemStreamEvent, AgentUpdatedStreamEvent

Example

from agents import Agent, Runner

agent = Agent(name="assistant", instructions="You are helpful.", tools=[my_tool])
result = Runner.run_streamed(agent, "Search for Python tutorials")

async for event in result.stream_events():
    if event.type == "raw_response_event":
        print(f"Raw: {type(event.data).__name__}")
    elif event.type == "run_item_stream_event":
        print(f"Item: {event.name} -> {type(event.item).__name__}")
    elif event.type == "agent_updated_stream_event":
        print(f"Agent changed to: {event.new_agent.name}")

print(f"Final: {result.final_output}")

Explanation

1. An agent is created with a tool attached.
2. Runner.run_streamed() initiates the streamed run.
3. The async for loop iterates over all events. Each event is dispatched by its type field:
   • raw_response_event -- Prints the class name of the raw data for inspection.
   • run_item_stream_event -- Prints both the event name (e.g., tool_called) and the type of the associated RunItem.
   • agent_updated_stream_event -- Prints the name of the new active agent.
4. After all events are consumed, the final output is printed.

Usage Patterns

Text-Only Streaming

For UIs that only need to display streaming text:

async for event in result.stream_events():
    if event.type == "raw_response_event":
        if hasattr(event.data, 'delta') and hasattr(event.data.delta, 'text'):
            display_text(event.data.delta.text)

Action Logging

For logging semantic actions without raw token data:

async for event in result.stream_events():
    if event.type == "run_item_stream_event":
        log_action(event.name, event.item)

See Also

• Stream Event Consumption Principle -- The theory behind consuming stream events.
• Runner.run_streamed Implementation -- The method that initiates streamed runs.
• RunResultStreaming Implementation -- The result object that provides stream_events().