configs.streaming.StreamWriter

class configs.streaming.StreamWriter(source, component, state=None)

Bases: object

Stream writer that provides consistent streaming events with automatic step counting.

Eliminates the need for manual if writer: checks and provides step context for task preparation phases.

Initialize stream writer with component context.

Parameters:

• source (str) – Source type (e.g., “framework”, “als_expert”)

• component (str) – Component name (e.g., “orchestrator”, “python_executor”)

• state (Any | None) – Optional AgentState for extracting execution context

__init__(source, component, state=None)

Initialize stream writer with component context.

Parameters:

• source (str) – Source type (e.g., “framework”, “als_expert”)

• component (str) – Component name (e.g., “orchestrator”, “python_executor”)

• state (Any | None) – Optional AgentState for extracting execution context

Methods

__init__(source, component[, state])	Initialize stream writer with component context.
error(message[, error_data])	Emit an error event
status(message)	Emit a status update event
warning(message)	Emit a warning event

__init__(source, component, state=None)

Initialize stream writer with component context.

Parameters:

• source (str) – Source type (e.g., “framework”, “als_expert”)

• component (str) – Component name (e.g., “orchestrator”, “python_executor”)

• state (Any | None) – Optional AgentState for extracting execution context

status(message)

Emit a status update event

error(message, error_data=None)

Emit an error event

warning(message)

Emit a warning event