pydantic_graph.step
Step-based graph execution components.
This module provides the core abstractions for step-based graph execution,
including step contexts, step functions, and step nodes that bridge between
the declarative BaseNode API and the builder graph.
Bases: Generic[StateT, DepsT, InputT]
Context information passed to step functions during graph execution.
The step context provides access to the current graph state, dependencies, and input data for a step.
The input data for this step.
This must be a property to ensure correct variance behavior
Type: InputT
Bases: Protocol[StateT, DepsT, InputT, OutputT]
Protocol for step functions that can be executed in the graph.
Step functions are async callables that receive a step context and return a result.
def __call__(ctx: StepContext[StateT, DepsT, InputT]) -> Awaitable[OutputT]
Execute the step function with the given context.
Awaitable[OutputT] — An awaitable that resolves to the step’s output
The step context containing state, dependencies, and inputs
Bases: Protocol[StateT, DepsT, InputT, OutputT]
Protocol for stream functions that can be executed in the graph.
Stream functions are async callables that receive a step context and return an async iterator.
def __call__(ctx: StepContext[StateT, DepsT, InputT]) -> AsyncIterator[OutputT]
Execute the stream function with the given context.
AsyncIterator[OutputT] — An async iterator yielding the streamed output
The step context containing state, dependencies, and inputs
Bases: Generic[StateT, DepsT, InputT, OutputT]
A step in the graph execution that wraps a step function.
Steps represent individual units of execution in the graph, encapsulating a step function along with metadata like ID and label.
Unique identifier for this step.
Type: NodeID Default: id
Optional human-readable label for this step.
Type: str | None Default: label
The step function to execute. This needs to be a property for proper variance inference.
Type: StepFunction[StateT, DepsT, InputT, OutputT]
def as_node(inputs: None = None) -> StepNode[StateT, DepsT]
def as_node(inputs: InputT) -> StepNode[StateT, DepsT]
Create a step node with bound inputs.
StepNode[StateT, DepsT] — A StepNode with this step and the bound inputs
inputs : InputT | None Default: None
The input data to bind to this step, or None
Bases: BaseNode[StateT, DepsT, Any]
A BaseNode that represents a builder step with bound inputs.
StepNode lets a BaseNode subclass hand off to a builder
Step by wrapping the step together with the value it should
receive as inputs. It is not meant to be run directly; returning a StepNode from a
BaseNode.run method tells the graph builder which step to invoke next.
The step to execute.
Type: Step[StateT, DepsT, Any, Any]
The inputs bound to this step.
Type: Any
@async
def run(ctx: GraphRunContext[StateT, DepsT]) -> BaseNode[StateT, DepsT, Any] | End[Any]
Attempt to run the step node.
BaseNode[StateT, DepsT, Any] | End[Any] — The result of step execution
The graph execution context
NotImplementedError— Always raised as StepNode is not meant to be run directly
Bases: Step[StateT, DepsT, Any, BaseNode[StateT, DepsT, Any] | End[Any]]
A step that wraps a BaseNode type for execution by the builder graph.
NodeStep lets a BaseNode subclass participate as a step in the
builder graph. It validates that the input is an instance of the expected node type and runs
it with the appropriate graph context.
The BaseNode type this step executes.
Type: type[BaseNode[StateT, DepsT, Any]] Default: get_origin(node_type) or node_type
def __init__(
node_type: type[BaseNode[StateT, DepsT, Any]],
id: NodeID | None = None,
label: str | None = None,
)
Initialize a node step.
The BaseNode class this step will execute
id : NodeID | None Default: None
Optional unique identifier, defaults to the node’s get_node_id()
Optional human-readable label for this step
Type alias for a step function with any type parameters.
Default: StepFunction[Any, Any, Any, Any]