Streaming

Agent streaming types and utilities.

Normalized streaming event interface across providers.

class agent.stream.StreamEvent(**data)

Bases: BaseModel

A normalized streaming event.

Parameters:

• data (Any)

• type (Literal['text_delta', 'tool_call_start', 'tool_call_delta', 'tool_result', 'message_start', 'message_end', 'usage', 'error'])

• text (str | None)

• tool_call (ToolCall | None)

• tool_call_delta (dict[str, Any] | None)

• tool_result (str | None)

• usage (Usage | None)

• error (str | None)

• raw (Any)

type: Literal['text_delta', 'tool_call_start', 'tool_call_delta', 'tool_result', 'message_start', 'message_end', 'usage', 'error']

text: str | None

tool_call: ToolCall | None

tool_call_delta: dict[str, Any] | None

tool_result: str | None

usage: Usage | None

error: str | None

raw: Any

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod text_delta(text, raw=None)

Create a text delta event.

Parameters:

• text (str)

• raw (Any)

Return type:

StreamEvent

classmethod tool_call_start(tool_call, raw=None)

Create a tool call start event.

Parameters:

• tool_call (ToolCall)

• raw (Any)

Return type:

StreamEvent

classmethod tool_call_delta_event(tool_call_id, delta, raw=None)

Create a tool call delta event.

Parameters:

• tool_call_id (str)

• delta (dict[str, Any])

• raw (Any)

Return type:

StreamEvent

classmethod tool_result_event(tool_call_id, result, raw=None)

Create a tool result event.

Parameters:

• tool_call_id (str)

• result (str)

• raw (Any)

Return type:

StreamEvent

classmethod message_start_event(raw=None)

Create a message start event.

Parameters:

raw (Any)

Return type:

StreamEvent

classmethod message_end(usage=None, raw=None)

Create a message end event.

Parameters:

• usage (Usage | None)

• raw (Any)

Return type:

StreamEvent

classmethod usage_event(usage, raw=None)

Create a usage event.

Parameters:

• usage (Usage)

• raw (Any)

Return type:

StreamEvent

classmethod error_event(error, raw=None)

Create an error event.

Parameters:

• error (str)

• raw (Any)

Return type:

StreamEvent

class agent.stream.StreamResponse(_events, provider='', model='')

Bases: object

A streaming response that yields events and accumulates the final response.

Parameters:

• _events (Iterator[StreamEvent])

• provider (str)

• model (str)

property text: str

Get accumulated text (available after iteration).

property tool_calls: list[ToolCall]

Get accumulated tool calls (available after iteration).

property usage: Usage | None

Get usage information (available after iteration).

collect()

Consume all events and return self with accumulated state.

Return type:

StreamResponse

class agent.stream.AsyncStreamResponse(events, provider='', model='')

Bases: object

An async streaming response that yields events and accumulates the final response.

Parameters:

• events (Any)

• provider (str)

• model (str)

property text: str

Get accumulated text (available after iteration).

property tool_calls: list[ToolCall]

Get accumulated tool calls (available after iteration).

property usage: Usage | None

Get usage information (available after iteration).

async collect()

Consume all events and return self with accumulated state.

Return type:

AsyncStreamResponse