transformers

Raised when a graph run exits early due to a drain request.

This indicates the graph stopped cooperatively at a superstep boundary because RunControl.request_drain() was called (e.g., in response to SIGTERM). The checkpoint is saved and the run can be resumed later.

Raised when a subgraph is interrupted, suppressed by the root graph. Never raised directly, or surfaced to the user.

A protocol event emitted by the streaming infrastructure.

Wraps a raw stream part (values, messages, custom, etc.) in a uniform envelope with a monotonic sequence number assigned by the root StreamMux. Consumers that need a total order across root events should use seq, not params.timestamp (which is wall-clock and not monotonic).

Extension point for custom stream projections.

Transformers observe protocol events flowing through the StreamMux and build typed derived projections (StreamChannels, promises, etc.).

Set _native = True on a transformer to have its projection keys exposed as direct attributes on the run stream (in addition to appearing in run.extensions).

Subclasses must implement init and override at least one of process / aprocess. The finalize / afinalize and fail / afail hooks are optional — the default implementations are no-ops. StreamChannel instances in the projection dict are auto-closed / auto-failed by the mux, so most transformers don't need finalize or fail at all.

Transformers that need async work pick the async lane by:

1. Overriding aprocess (and optionally afinalize / afail), or
2. Calling self.schedule(coro) from inside a sync process, or
3. Setting requires_async = True explicitly.

The mux detects these cases at registration and raises if they're used under sync stream() — they only work under astream().

Use aprocess when the pump must wait for async work before the next transformer sees the event (e.g. PII redaction that mutates event in place). Use schedule() for decoupled async work whose result lands on an independent projection (e.g. async moderation scoring, cost lookup, external tracing).

Async handle for a discovered subgraph (extends AsyncGraphRunStream).

Sync handle for a discovered subgraph (extends GraphRunStream).

Single-consumer drainable queue for streaming events, with optional protocol auto-forwarding.

When constructed with a name, the StreamMux auto-wires every push() to also inject a ProtocolEvent into the main event stream using the channel's name as the method. When constructed without a name, the channel is local-only — items are only visible to in-process consumers that iterate the channel directly.

Items are popped off the front as the consumer advances — there is no retention beyond what's currently queued. A channel accepts exactly one subscriber; a second __iter__ / __aiter__ call raises. Use tee(n) / atee(n) for fan-out.

Starts unbound — neither __iter__ nor __aiter__ is available until the StreamMux calls _bind(is_async). After binding, only the matching iteration protocol works; the other raises TypeError.

Pump wiring (set by the run stream, not by _bind): - _request_more: sync pump callable, returns True if a new event was produced. - _arequest_more: async pump coroutine factory, same contract.

Memory is bounded by caller pace: both sync and async use caller- driven pumps, so each cursor advance produces at most one event.

Lazy-subscribe: push appends to the local buffer only when a subscriber has registered. Auto-forward via _wire_fn always fires regardless of subscription state.

Lifecycle (close / fail) is managed by the mux — transformers don't need to close their channels manually.

Central event dispatcher for the streaming infrastructure.

Owns the main event log and routes events through a transformer pipeline. StreamChannels with a name discovered in transformer projections are auto-wired so that every push() also injects a ProtocolEvent into the main log. StreamChannels without a name are local-only.

Pass is_async=True when the mux will be consumed via async iteration (handler.astream()). All StreamChannel instances discovered during registration are automatically bound to the matching mode.

Capture values events as a drainable stream of state snapshots.

Provides the run.values projection. run.output, run.interrupted and run.interrupts are tracked directly by the run stream and do not depend on this transformer.

Native transformer — projection keys are exposed as direct attributes on the run stream (e.g. run.values).

Only values events at the run's own level are captured; snapshots from deeper subgraphs are left in the main event log but excluded from the projection. "Own level" is defined by scope, which stream_events(version="v3") / astream_events(version="v3") populate from the caller's checkpoint namespace so that a nested stream_events(version="v3") call still sees its own root snapshots.

Capture custom events as a drainable stream of arbitrary payloads.

Nodes emit custom data via get_stream_writer(). This transformer surfaces those events on run.custom as a StreamChannel[Any], preserving payloads in arrival order.

Only events at the run's own scope are captured; custom data from deeper subgraphs is available on the respective subgraph handle's .custom projection.

Native transformer — run.custom is a direct attribute.

Capture updates events as a drainable stream of node outputs.

Surfaces stream_mode="updates" data on run.updates as a StreamChannel[dict[str, Any]]. Each item is a dict mapping a node (or task) name to the update it returned after a step.

Only events at the run's own scope are captured; updates from deeper subgraphs are available on the respective subgraph handle's .updates projection.

Native transformer — run.updates is a direct attribute.

Capture messages events as ChatModelStream objects.

The messages projection yields one ChatModelStream (or AsyncChatModelStream) per LLM call. Consumers iterate run.messages to get stream handles, then use each handle's typed projections (.text, .reasoning, .tool_calls, .usage, .output) for per-message content.

Two input shapes are handled (via params["data"] = (payload, metadata) from StreamMessagesHandler):

1. Protocol event (dict with "event" key) — emitted by stream_events(version="v3") / astream_events(version="v3") via the on_stream_event callback. Routed to an existing ChatModelStream by metadata["run_id"]. A message-start event creates a new stream; message-finish closes it.
2. Whole AIMessage — emitted from on_chain_end when a node returns a finalized message. Replayed as a synthetic protocol event lifecycle via message_to_events, then the already-complete stream is pushed to the log.

V1 AIMessageChunk tuples (from on_llm_new_token) are not streamed into this projection: chat models that want to populate run.messages with content-block streaming must use stream_events(version="v3") / astream_events(version="v3"). Models called via the legacy stream() method still surface their final AIMessage via on_chain_end when a node returns it as state.

Only events at the run's own level are projected; tokens from deeper subgraphs are left in the main event log but excluded from .messages. "Own level" is defined by scope, which stream_events(version="v3") / astream_events(version="v3") populate from the caller's checkpoint namespace so that a stream_events(version="v3") call inside a node still sees its own root chat model streams on .messages. Consumers that need subgraph tokens should iterate the raw event stream or register a custom transformer.

Native transformer — the messages projection is exposed as a direct attribute on the run stream.

Payload of a lifecycle event surfaced on the lifecycle channel.

Auto-forwarded as lifecycle protocol events (no custom: prefix because LifecycleTransformer is a native transformer) so remote SDK clients receive the same data in-process consumers see via run.lifecycle.

Surface subgraph lifecycle as lifecycle protocol events.

Pushes LifecyclePayload to a StreamChannel named lifecycle. The channel is auto-forwarded by the mux so payloads land in the main event log under method = "lifecycle" (native transformer — no custom: prefix) — visible to remote SDK clients over the wire and to in-process consumers via run.lifecycle.

Tracks subgraphs at every depth strictly below the transformer's scope, so a graph → subgraph → subgraph chain produces lifecycle events for both nested levels in a flat stream.

Native transformer — projection key lifecycle is exposed as run.lifecycle.

Discover subgraph invocations as in-process navigation handles.

Per discovered direct-child subgraph, builds a SubgraphRunStream (or AsyncSubgraphRunStream) wrapping a child mini-mux scoped to the subgraph's namespace. Consumers iterate run.subgraphs to receive handles, then drill into handle.values / handle.messages / handle.subgraphs (recursive grandchildren) / handle.lifecycle.

Each mini-mux owns its own scope and uses its own SubgraphTransformer to discover its direct children, so grandchildren live on the child handle — never on the root's subgraphs log. Forwarding events into the matching child mini-mux is what keeps the child's projections populated.

Native transformer — subgraphs is exposed as run.subgraphs.

Capture checkpoint events as a drainable stream.

Surfaces stream_mode="checkpoints" data on run.checkpoints as a StreamChannel[dict[str, Any]]. Each item is in the same format as returned by get_state().

Checkpoint events are only emitted when a checkpointer is configured on the graph. When no checkpointer is present, the projection exists but receives no events.

Only events at the run's own scope are captured; checkpoint data from deeper subgraphs is available on the respective subgraph handle's .checkpoints projection.

Native transformer — run.checkpoints is a direct attribute.

Capture debug events as a drainable stream.

Surfaces stream_mode="debug" data on run.debug as a StreamChannel[dict[str, Any]]. Each item is a debug event with step-level detail (checkpoint snapshots, task payloads, and task results wrapped with step number and timestamp).

Only events at the run's own scope are captured; debug data from deeper subgraphs is available on the respective subgraph handle's .debug projection.

Native transformer — run.debug is a direct attribute.

Capture raw task events as a drainable stream.

Surfaces stream_mode="tasks" data on run.tasks as a StreamChannel[dict[str, Any]]. Each item is a task payload (start or result).

LifecycleTransformer and SubgraphTransformer also consume tasks events for subgraph discovery and lifecycle tracking. This transformer captures the raw payloads independently for consumers who need task-level detail.

Only events at the run's own scope are captured; task data from deeper subgraphs is available on the respective subgraph handle's .tasks projection.

Native transformer — run.tasks is a direct attribute.