Streaming

Helpers for consuming a flow as a stream of events and adapting stream chunks to SSE or WebSocket transports.

The StreamChunk payload type itself is documented under Messages & types.

streaming

Utilities for streaming chunk integrations.

stream_flow async

stream_flow(flow: PenguiFlow, parent_msg: Message, *, to: Node | Sequence[Node] | None = None, timeout: float | None = None, include_final: bool = False) -> AsyncIterator[Any]

Yield streamed payloads from a running flow.

This helper emits each StreamChunk produced downstream. When include_final is True the first non-chunk payload encountered after a terminal chunk is also yielded before the generator stops. The caller is responsible for stopping the flow when finished.

Parameters:

Name	Type	Description	Default
flow	PenguiFlow	The running PenguiFlow instance to emit into and fetch results from.	required
parent_msg	Message	The message to emit as the starting point of the run.	required
to	Node | Sequence[Node] | None	Optional node or sequence of nodes to route parent_msg to; forwarded to flow.emit.	None
timeout	float | None	Optional per-fetch timeout in seconds; raises on expiry via asyncio.wait_for.	None
include_final	bool	When True, also yield the first non-chunk payload observed after a terminal (done=True) chunk.	False

Yields:

Type	Description
AsyncIterator[Any]	Each StreamChunk produced by the flow, and optionally a trailing non-chunk
AsyncIterator[Any]	payload when include_final is True.

Raises:

Type	Description
TimeoutError	If timeout is set and a fetch does not complete in time.

emit_stream_events async

emit_stream_events(source: AsyncIterable[Any], ctx: Context, parent_msg: Message, *, adapter: EventAdapter | None = None, to: Node | Sequence[Node] | None = None, final_meta: dict[str, Any] | None = None) -> None

Bridge an async iterable of provider events into StreamChunk emissions.

Parameters:

Name	Type	Description	Default
source	AsyncIterable[Any]	An async iterable of provider-specific event objects.	required
ctx	Context	The Context used to emit chunks via ctx.emit_chunk.	required
parent_msg	Message	The parent message associated with the emitted chunks.	required
adapter	EventAdapter | None	Optional callable converting a provider event into a (text, done, meta) tuple; defaults to (str(event), False, {}).	None
to	Node | Sequence[Node] | None	Optional node or sequence of nodes to route emitted chunks to.	None
final_meta	dict[str, Any] | None	Metadata to attach to a synthesized terminal chunk if the source completes without ever emitting a chunk with done=True.	None

Returns:

Type	Description
None	None. Chunks are emitted as a side effect via ctx.emit_chunk.

chunk_to_ws_json

chunk_to_ws_json(chunk: StreamChunk, *, extra: dict[str, Any] | None = None) -> str

Serialize a StreamChunk as a JSON WebSocket payload.

Parameters:

Name	Type	Description	Default
chunk	StreamChunk	The stream chunk to serialize.	required
extra	dict[str, Any] | None	Additional key/value pairs to merge into the JSON payload.	None

Returns:

Type	Description
str	A JSON-encoded string with stream_id, seq, text, done, and
str	meta fields, plus any entries from extra.

format_sse_event

format_sse_event(chunk: StreamChunk, *, event_name: str | None = None, retry_ms: int | None = None) -> str

Render a StreamChunk as an SSE event string.

Parameters:

Name	Type	Description	Default
chunk	StreamChunk	The stream chunk to render.	required
event_name	str | None	Optional SSE event: name; defaults to "done" when chunk.done is True, otherwise "chunk".	None
retry_ms	int | None	Optional SSE retry: value in milliseconds.	None

Returns:

Type	Description
str	A newline-terminated SSE event string (including the trailing blank line).