API Reference

Classes

AbstractSaga

source

Abstract base class for orchestration sagas.

Subclasses declare their saga identity (get_id) and terminal condition (is_completed). Steps are registered via add_step and the full sequence is driven by execute.

Type Parameters: T: The saga state / context type (unused by the base class but carried for subclass typing convenience).

Lifecycle: PENDING → RUNNING → COMPLETED (all steps succeed) PENDING → RUNNING → COMPENSATING → FAILED (a step fails; already-completed steps are unwound)

__init__

def __init__(store: SagaStoreProtocol | None = None) -> None

source

get_version

def get_version() -> int

source

Return the current schema version for this saga class.

Returns

Type	Description
int	The integer version declared on the class.

is_compatible_with

def is_compatible_with(stored_version: int) -> bool

source

Check if stored state version is compatible with this class.

Override for custom migration logic (e.g. allow loading V1 in V2).

Parameters

Parameter	Type	Description
stored_version	int	Version number found in the persisted state.

Returns

Type	Description
bool	`True` when the stored version matches the class VERSION.

get_id

def get_id() -> str

source

Return the stable identifier for this saga instance.

Returns

Type	Description
str	A string that uniquely identifies this saga process.

is_completed

def is_completed() -> bool

source

Return whether the saga has reached a terminal state.

Returns

Type	Description
bool	`True` when no further steps remain (success *or* fully compensated failure).

add_step

def add_step(step: SagaStep) -> None

source

Register a step to be executed as part of this saga.

Steps are executed in the order they are added.

Parameters

Parameter	Type	Description
step	SagaStep	The step to append to the execution sequence.

was_compensated

def was_compensated(step_name: str) -> bool

source

Return whether step_name has been successfully compensated.

Parameters

Parameter	Type	Description
step_name	str	The SagaStep.name to check.

Returns

Type	Description
bool	`True` if the step's compensation action ran successfully.

mark_compensated

def mark_compensated(step_name: str) -> None

source

Manually record step_name as compensated.

Useful when compensation is performed externally (e.g., a separate process handles the rollback) and the saga state needs to be updated.

Parameters

Parameter	Type	Description
step_name	str	The SagaStep.name to mark as compensated.

execute

async def execute() -> Result[None, SagaError]

source

Execute all registered steps in registration order.

On first step failure, compensation is triggered for all previously completed steps (in reverse order) and an Err(SagaError(...)) is returned.

Returns

Type	Description
Result[None, SagaError]	`Ok(None) when all steps complete successfully. Err(SagaError)` when any step fails (compensation already attempted before returning).

compensate

async def compensate() -> None

source

Manually trigger compensation for all completed steps.

Useful when the calling code detects a failure condition outside the normal execute flow.

---

AbstractWorkflowNode

source

Abstract base class for all workflow nodes.

All built-in and custom nodes inherit from this class. The only required override is execute.

Parameters

Parameter	Type	Description
name		Unique node identifier within the workflow graph.
node_type		The NodeType label for this node.

__init__

def __init__(
    name: str,
    node_type: NodeType = NodeType.CUSTOM
) -> None

source

name

property name() -> str

source

Unique node identifier.

node_type

property node_type() -> NodeType

source

Type label for this node.

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Execute this node and return a state-update dict.

Parameters

Parameter	Type	Description
state	dict[str, Any]	Current shared workflow state (read-only reference; mutations are applied via the returned dict).

Returns

Type	Description
dict[str, Any]	Dict of key-value pairs to merge into the shared state. Return an empty dict if no state updates are needed.

---

AgentNode

source

Workflow node that executes a Quadkit agent.

Parameters

Parameter	Type	Description
name		Node identifier.
agent		Object implementing AgentProtocol.
executor		Optional AgentExecutorProtocol instance.
input_key		State key to read the agent input from.
output_key		State key where the agent response is stored.

__init__

def __init__(
    name: str,
    *,
    agent: Any,
    executor: Any | None = None,
    input_key: str = 'input',
    output_key: str = 'output'
) -> None

source

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Run the agent with the value of state[input_key].

Parameters

Parameter	Type	Description
state	dict[str, Any]	Current workflow state.

Returns

Type	Description
dict[str, Any]	Dict with {output_key: agent_response_text}.

---

BulkBatchResult

source

Result of processing a single batch.

duration

property duration() -> float

source

Get the duration of this batch.

success_count

property success_count() -> int

source

Get the number of successful items in this batch.

error_count

property error_count() -> int

source

Get the number of failed items in this batch.

is_successful

property is_successful() -> bool

source

Check if the batch completed successfully.

---

BulkItemError

source

Structured error information for a failed batch.

Attributes: batch_id: The ID of the batch that failed. error: Human-readable error message. error_type: The fully-qualified type name of the exception. retry_count: Number of retry attempts made before giving up. timestamp: Monotonic timestamp of the failure.

---

BulkOperation

source

Manages bulk operations with batching, concurrency control, and error handling.

__init__

def __init__(
    config: BulkOperationConfig | None = None,
    processor: Callable[[list[T]], Awaitable[list[R]]] | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None,
    checkpoint_store: CacheBackendProtocol | None = None,
    *,
    on_progress: Callable[[int, int, str], Awaitable[None]] | None = None,
    on_complete: Callable[[int, int, str], Awaitable[None]] | None = None,
    on_error: Callable[[int, int, str], Awaitable[None]] | None = None
) -> None

source

state

property state() -> BulkOperationState

source

Get the current operation state.

metrics

property metrics() -> BulkOperationMetrics

source

Get operation metrics.

add_progress_callback

def add_progress_callback(callback: Callable[[BulkOperationMetrics], Awaitable[None]]) -> None

source

Add a progress tracking callback.

clear_checkpoint

async def clear_checkpoint(operation_id: str) -> None

source

Remove a stored checkpoint, allowing the operation to restart from scratch.

Parameters

Parameter	Type	Description
operation_id	str	Unique identifier for the bulk operation run to clear.

execute

async def execute(
    items: list[T] | AsyncIterable[T],
    processor: Callable[[list[T]], Awaitable[list[R]]] | None = None,
    operation_id: str | None = None
) -> AsyncIterator[BulkBatchResult[T, R]]

source

Execute bulk operation on items.

Parameters

Parameter	Type	Description
items	list[T] | AsyncIterable[T]	Items to process — either a list or an async iterable.
processor	Callable[[list[T]], Awaitable[list[R]]] | None	Optional processor to override the one set at construction.
operation_id	str | None	Stable identifier used for checkpoint/resume. If supplied and a `checkpoint_store` was provided at construction, completed batches from a previous interrupted run are automatically skipped.

cancel

async def cancel() -> None

source

Cancel the bulk operation.

shutdown

async def shutdown() -> None

source

Shut down the bulk operation, cancelling any in-progress execution.

Safe to call even when the operation is not running. Ensures the operation reaches a terminal state so resources can be reclaimed.

---

BulkOperationConfig

source

Configuration for bulk operations and pipeline execution.

Attributes: batch_size: Number of items per batch. max_concurrency: Maximum number of concurrent operations. timeout: Operation timeout in seconds. retry_attempts: Number of retry attempts. retry_delay: Delay between retries in seconds. enable_progress_tracking: Whether to track progress. circuit_breaker_config: Optional circuit breaker configuration. pipeline_timeout: Default pipeline execution timeout in seconds.

---

BulkOperationMetrics

source

Metrics for bulk operations.

record_start

def record_start(
    total_items: int,
    total_batches: int
) -> None

source

Record the start of the operation.

record_end

def record_end() -> None

source

Record the end of the operation.

record_batch_result

def record_batch_result(result: BulkBatchResult[Any, Any]) -> None

source

Update metrics with result from a processed batch.

record_error

def record_error(error: Exception) -> None

source

Record an unexpected operation-level error.

duration

property duration() -> float | None

source

Get the total duration of the operation.

success_rate

property success_rate() -> float

source

Get the success rate as a percentage.

throughput

property throughput() -> float

source

Get items processed per second.

---

BulkOperationState

source

States for bulk operations.

---

CacheContentCheckpointStore

source

Content-addressed checkpoint store backed by a cache backend.

Entries are JSON-serialized and stored with the checkpoint key string as the cache key. list_by_stage is not supported (cache backends lack prefix scanning).

__init__

def __init__(
    cache: CacheBackendProtocol,
    default_ttl: int | None = 86400
) -> None

source

get

async def get(key: ContentCheckpointKey) -> ContentCheckpointEntry | None

source

set

async def set(
    key: ContentCheckpointKey,
    entry: ContentCheckpointEntry
) -> None

source

evict

async def evict(key: ContentCheckpointKey) -> None

source

list_by_stage

async def list_by_stage(
    stage_id: str,
    tenant_id: str | None = None,
    limit: int = 100
) -> Sequence[ContentCheckpointKey]

source

---

ConditionalStep

source

Pipeline step that executes conditionally based on a predicate.

__init__

def __init__(
    name: str,
    condition: Callable[[PipelineContext], Result[bool, Exception] | Awaitable[Result[bool, Exception]]],
    true_step: PipelineStep,
    false_step: PipelineStep | None = None,
    dependencies: list[str] | None = None,
    timeout: float | None = None
) -> None

source

execute

async def execute(context: PipelineContext) -> Result[Any, Exception]

source

Execute the appropriate step based on condition.

should_skip

async def should_skip(context: PipelineContext) -> Result[bool, Exception]

source

Conditional steps should not be skipped.

---

ContentAddressedSaga

source

Saga where every stage handler is content-addressed.

Subclass and register stages via add_stage(); the framework computes the input hash, checks the checkpoint store, returns cached output, or runs the handler and caches the result.

Parameters

Parameter	Type	Description
saga_id		Stable identifier for this saga instance.
checkpoint_store		Persistence backend for checkpoint entries.
store		Optional saga state store (passed to `AbstractSaga`).
blob_store		Optional blob store for offloading large outputs.
tenant_id		Optional tenant identifier (included in content-address key).
inline_threshold_bytes		Max bytes to store inline; beyond this threshold the output is offloaded to `blob_store`.

__init__

def __init__(
    saga_id: str,
    checkpoint_store: ContentCheckpointStoreProtocol,
    *,
    store: SagaStoreProtocol | None = None,
    blob_store: BlobStoreProtocol | None = None,
    tenant_id: str | None = None,
    inline_threshold_bytes: int = 1048576
) -> None

source

get_id

def get_id() -> str

source

is_completed

def is_completed() -> bool

source

add_stage

def add_stage(stage: ContentAddressedStage) -> None

source

execute

async def execute() -> Result[None, SagaError]

source

run_stage

async def run_stage(
    stage: ContentAddressedStage,
    inputs: dict[str, Any]
) -> Any

source

compensate

async def compensate() -> None

source

---

ContentAddressedStage

source

Descriptor for a single content-addressed saga stage.

Unlike SagaStep (which uses a generic action callable), this stage is designed for content addressing: its inputs, handler version, and config-affecting-output are all hashed into ContentCheckpointKey.

---

ContentCheckpointConfig

source

Configuration for content-addressed checkpointing.

Attributes: enabled: Enable content-addressed checkpointing. inline_threshold_bytes: Max bytes to store inline; beyond this threshold outputs are offloaded to blob store. default_ttl_seconds: Default TTL for cache-backed stores.

---

DatabaseContentCheckpointStore

source

Content-addressed checkpoint store backed by a SQL database.

Creates a workflow_content_checkpoints table with columns key_str (PK), entry_json, stage_handler_version, output_size_bytes, completed_at, and created_at.

__init__

def __init__(
    provider: DatabaseProviderProtocol,
    table_name: str = 'workflow_content_checkpoints'
) -> None

source

get

async def get(key: ContentCheckpointKey) -> ContentCheckpointEntry | None

source

set

async def set(
    key: ContentCheckpointKey,
    entry: ContentCheckpointEntry
) -> None

source

Set a checkpoint entry, using ON CONFLICT DO NOTHING for safety.

If two concurrent sagas race to set the same key, the second insert is silently ignored — content-addressed checkpoints are deterministic, so the output is identical regardless of which saga’s insert wins.

evict

async def evict(key: ContentCheckpointKey) -> None

source

list_by_stage

async def list_by_stage(
    stage_id: str,
    tenant_id: str | None = None,
    limit: int = 100
) -> Sequence[ContentCheckpointKey]

source

---

ExecutionHistory

source

Formats and inspects the execution trace of a completed workflow run.

Parameters

Parameter	Type	Description
result		The completed GraphResult to inspect.

__init__

def __init__(result: GraphResult) -> None

source

as_dict

def as_dict() -> dict[str, Any]

source

Return the full execution trace as a JSON-serialisable dict.

Returns

Type	Description
dict[str, Any]	Dict with keys iterations, duration_ms, terminated_at, and nodes.

as_text

def as_text() -> str

source

Return a multi-line human-readable execution trace.

Returns

Type	Description
str	A string with one line per node showing its outcome.

failed_nodes

def failed_nodes() -> list[NodeResult]

source

Return node results that terminated with an error.

Returns

Type	Description
list[NodeResult]	Filtered list of failed NodeResult entries.

succeeded_nodes

def succeeded_nodes() -> list[NodeResult]

source

Return node results that completed successfully.

Returns

Type	Description
list[NodeResult]	Filtered list of successful NodeResult entries.

total_node_time_ms

def total_node_time_ms() -> float

source

Return cumulative node execution time in milliseconds.

---

FunctionStep

source

Pipeline step that executes a function.

__init__

def __init__(
    name: str,
    func: Callable[[PipelineContext], Result[Any, Exception] | Awaitable[Result[Any, Exception]]],
    dependencies: list[str] | None = None,
    skip_condition: Callable[[PipelineContext], Result[bool, Exception] | Awaitable[Result[bool, Exception]]] | None = None,
    error_handler: Callable[[PipelineContext, Exception], Result[Any, Exception] | Awaitable[Result[Any, Exception]]] | None = None,
    cleanup_func: Callable[[PipelineContext], Result[None, Exception] | Awaitable[Result[None, Exception]]] | None = None,
    timeout: float | None = None
) -> None

source

execute

async def execute(context: PipelineContext) -> Result[Any, Exception]

source

Execute the function, honoring optional per-step timeout.

should_skip

async def should_skip(context: PipelineContext) -> Result[bool, Exception]

source

Check skip condition.

on_error

async def on_error(
    context: PipelineContext,
    error: Exception
) -> Result[Any, Exception]

source

Handle errors.

cleanup

async def cleanup(context: PipelineContext) -> Result[None, Exception]

source

Cleanup resources.

---

GateNode

source

Routing node that forces the graph to branch on state values.

Parameters

Parameter	Type	Description
name		Node identifier.
routes		Mapping of target-node-name to a (state) -> bool predicate.

__init__

def __init__(
    name: str,
    *,
    routes: dict[str, Callable[[dict[str, Any]], bool]] | None = None
) -> None

source

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Return empty dict — routing is handled by WorkflowEdge conditions.

Parameters

Parameter	Type	Description
state	dict[str, Any]	Current workflow state (not mutated here).

Returns

Type	Description
dict[str, Any]	Empty dict — gate nodes produce no output.

---

GraphConfig

source

Configuration for the workflow graph engine.

Attributes: enabled: Enable the graph workflow subsystem. max_iterations: Maximum graph traversal steps (anti-infinite-loop guard). node_timeout: Per-node execution timeout in seconds. 0 = no limit. total_timeout: Total workflow timeout in seconds. 0 = no limit. checkpoint_enabled: Persist state checkpoints for resumption after failures. parallel_branches: Execute independent parallel branches via asyncio.gather. max_parallel_branches: Max parallel branches at once. 0 = unlimited.

validate_for_environment

def validate_for_environment(env: Environment | None = None) -> list[ConfigIssue]

source

Check config is safe for the target environment.

Parameters

Parameter	Type	Description
env	Environment | None	Target deployment environment.

Returns

Type	Description
list[ConfigIssue]	List of configuration issues (empty if all OK).

---

GraphResult

source

Aggregate result of a complete graph workflow execution.

Attributes: final_state: The workflow state after all nodes have executed. node_results: Ordered list of per-node results. iterations: Total graph traversal steps taken. duration_ms: Total wall-clock execution time. terminated_at: Name of the terminal node that ended execution.

output

property output() -> Any

source

Return the output key from final_state, or None.

succeeded

property succeeded() -> bool

source

True when all executed nodes succeeded.

---

HumanNode

source

Pause the workflow to collect a human response.

Parameters

Parameter	Type	Description
name		Node identifier (doubles as checkpoint_id).
prompt		Question or instruction to display to the operator. Supports {key} substitution from state.
output_key		State key where the human response will be stored.
resume_key		State key the engine populates with the human answer when WorkflowEngine.resume() is called.

__init__

def __init__(
    name: str,
    *,
    prompt: str = 'Human input required.',
    output_key: str = 'human_response',
    resume_key: str | None = None
) -> None

source

resume_key

property resume_key() -> str

source

State key WorkflowEngine.resume populates with the answer.

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Return human response if resumed, otherwise pause execution.

Parameters

Parameter	Type	Description
state	dict[str, Any]	Current workflow state.

Returns

Type	Description
dict[str, Any]	Dict with {output_key: human_response} when resumed.

Raises

Exception	Description
HumanInputRequiredError	When no prior human response is in state.

---

InMemoryContentCheckpointStore

source

In-memory dict-backed checkpoint store for testing.

__init__

def __init__() -> None

source

get

async def get(key: ContentCheckpointKey) -> ContentCheckpointEntry | None

source

set

async def set(
    key: ContentCheckpointKey,
    entry: ContentCheckpointEntry
) -> None

source

evict

async def evict(key: ContentCheckpointKey) -> None

source

list_by_stage

async def list_by_stage(
    stage_id: str,
    tenant_id: str | None = None,
    limit: int = 100
) -> Sequence[ContentCheckpointKey]

source

---

LLMNode

source

Workflow node that makes a raw LLM call.

Parameters

Parameter	Type	Description
name		Node identifier.
llm		LLMClientProtocol implementation.
prompt_template		Prompt string with optional {key} variables.
system_prompt		Optional system message prepended to the call.
output_key		State key where the completion is stored.
model		Optional model override.
temperature		Sampling temperature override.
max_tokens		Max output token override.

__init__

def __init__(
    name: str,
    *,
    llm: Any,
    prompt_template: str = '{input}',
    system_prompt: str = '',
    output_key: str = 'output',
    model: str | None = None,
    temperature: float | None = None,
    max_tokens: int | None = None
) -> None

source

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Render the prompt from state and call the LLM.

Parameters

Parameter	Type	Description
state	dict[str, Any]	Current workflow state used for prompt substitution.

Returns

Type	Description
dict[str, Any]	Dict with {output_key: completion_text}.

---

NodeResult

source

Result produced by a single workflow graph node execution.

Attributes: node_name: Name of the node that produced this result. output: State update dict returned by the node. duration_ms: Wall-clock duration of the node execution. error: Error message if the node raised; None on success. skipped: True when the node was bypassed by conditional routing.

succeeded

property succeeded() -> bool

source

True when the node completed without error and was not skipped.

---

NodeType

source

Enumeration of built-in workflow node types.

---

ParallelStep

source

Pipeline step that executes multiple steps in parallel.

__init__

def __init__(
    name: str,
    steps: list[PipelineStep],
    dependencies: list[str] | None = None,
    *,
    fail_fast: bool = True,
    timeout: float | None = None
) -> None

source

execute

async def execute(context: PipelineContext) -> Result[list[Any], Exception]

source

Execute all steps in parallel.

---

Pipeline

source

Declarative pipeline for composing business workflows.

__init__

def __init__(
    name: str,
    steps: list[PipelineStep],
    checkpoint_store: CacheBackendProtocol | None = None
) -> None

source

add_step

def add_step(step: PipelineStep) -> None

source

Append a step to this pipeline.

Parameters

Parameter	Type	Description
step	PipelineStep	The PipelineStep to append. Validates pipeline structure after addition.

Raises

Exception	Description
ValueError	If the new step creates duplicate names or broken deps.

to_dot

def to_dot() -> str

source

Return a Graphviz DOT representation of the pipeline execution graph.

Nodes represent pipeline steps; directed edges represent declared dependencies (dep → step). The output can be rendered with dot -Tpng or pasted into an online Graphviz viewer.

Returns

Type	Description
str	A string containing valid Graphviz DOT source for the pipeline.

Example

dot_src = pipeline.to_dot()
# dot -Tsvg -o pipeline.svg <<< "$(python -c '...')"

dot_src = pipeline.to_dot()
# dot -Tsvg -o pipeline.svg <<< "$(python -c '...')"

execute

async def execute(
    initial_context: PipelineContext | None = None,
    fail_fast: bool = True
) -> Result[PipelineContext, Exception]

source

Execute the pipeline.

---

PipelineContext

source

Context passed through pipeline execution, accumulating step results and metadata.

get_step_result

def get_step_result(
    step_name: str,
    default: Any = None
) -> Any

source

Get result from a completed step.

Parameters

Parameter	Type	Description
step_name	str	The name of the step whose result to retrieve.
default	Any	Value to return if the step has no recorded result.

Returns

Type	Description
Any	The recorded step result, or *default* if not found.

set_step_result

def set_step_result(
    step_name: str,
    result: Any
) -> None

source

Set result for a step.

Parameters

Parameter	Type	Description
step_name	str	The name of the step.
result	Any	The result value to record.

add_metadata

def add_metadata(
    key: str,
    value: Any
) -> None

source

Add metadata to the context.

Parameters

Parameter	Type	Description
key	str	Metadata key.
value	Any	Metadata value.

get_metadata

def get_metadata(
    key: str,
    default: Any = None
) -> Any

source

Get metadata from the context.

Parameters

Parameter	Type	Description
key	str	Metadata key to look up.
default	Any	Value to return if the key is absent.

Returns

Type	Description
Any	The metadata value, or *default* if not found.

duration

property duration() -> float

source

Total duration of pipeline execution in seconds.

failed_steps

property failed_steps() -> list[StepExecutionResult]

source

List of failed step results.

successful_steps

property successful_steps() -> list[StepExecutionResult]

source

List of successfully completed step results.

has_failed

def has_failed() -> bool

source

Check if any steps in the pipeline failed.

Returns

Type	Description
bool	`True` if at least one step has a FAILED status.

---

PipelineStep

source

Abstract base class for pipeline steps.

Non-execute helpers are deliberately simple: they return plain values and may raise exceptions. This keeps user implementations lightweight and avoids the noise of wrapping every result in Result. Helpers are responsible for handling their own errors; the pipeline driver will catch and convert them into failed step records if necessary.

Steps may specify an optional timeout (in seconds). When set, the execute invocation will be wrapped with asyncio.wait_for so that a long-running or hung step is interrupted and treated as a failure. The timeout does not apply to should_skip, on_error or cleanup helpers.

__init__

def __init__(
    name: str,
    dependencies: list[str] | None = None,
    timeout: float | None = None,
    resilience: ResiliencePipelineProtocol | None = None
) -> None

source

execute

async def execute(context: PipelineContext) -> Result[Any, Exception]

source

Execute the pipeline step.

Parameters

Parameter	Type	Description
context	PipelineContext	The shared pipeline context carrying inter-step data.

Returns

Type	Description
Result[Any, Exception]	A `Result` wrapping the step's output or a failure.

should_skip

async def should_skip(context: PipelineContext) -> Result[bool, Exception]

source

Determine whether to skip this step.

Parameters

Parameter	Type	Description
context	PipelineContext	The current pipeline context.

Returns

Type	Description
Result[bool, Exception]	`Ok(True) to skip, Ok(False) to execute, or Err(e)` on failure while evaluating the condition.

The default implementation always returns Ok(False).

on_error

async def on_error(
    context: PipelineContext,
    error: Exception
) -> Any

source

Handle errors raised during execute.

Parameters

Parameter	Type	Description
context	PipelineContext	The current pipeline context.
error	Exception	The exception that caused the step to fail.

The default behaviour is to re-raise the error, which causes the step (and pipeline) to fail. Custom handlers may return a value to recover and continue the pipeline.

cleanup

async def cleanup(context: PipelineContext) -> Result[None, Exception]

source

Cleanup resources after step execution.

Parameters

Parameter	Type	Description
context	PipelineContext	The current pipeline context.

Called regardless of success or failure. The default implementation does nothing. Exceptions are logged but not propagated.

---

SagaStep

source

Descriptor for a single saga orchestration step.

Shared across quadkit-workflow and quadkit-events via contracts.

The action callable must return Result[Any, SagaStepError]. The compensation callable is called only when the saga is unwinding and may raise exceptions freely (infrastructure failures in compensation should propagate).

Attributes: name: Unique step identifier within the saga. action: Async callable returning Result[Any, SagaStepError]. compensation: Async callable to undo the step on failure. max_retries: Maximum retry attempts if action returns Err. retry_delay: Seconds to wait between retry attempts. idempotent: Whether the compensation is safe to call multiple times.

---

StepExecutionResult

source

Result of executing a single pipeline step.

---

StepStatus

source

Pipeline step execution status.

---

SubworkflowNode

source

Embed a nested WorkflowEngine as a single node in a parent workflow.

Parameters

Parameter	Type	Description
name		Node identifier.
workflow		A WorkflowEngine instance to nest.
input_key		State key to read as the subworkflow input value.
output_key		State key where the subworkflow result is stored.

__init__

def __init__(
    name: str,
    *,
    workflow: Any,
    input_key: str = 'input',
    output_key: str = 'output'
) -> None

source

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Execute the nested workflow with the current parent state.

Parameters

Parameter	Type	Description
state	dict[str, Any]	Current parent workflow state.

Returns

Type	Description
dict[str, Any]	Dict with {output_key: subworkflow_output} on success or {output_key: "Subworkflow error: ..."} on failure.

---

ToolNode

source

Workflow node that executes a Quadkit tool.

Parameters

Parameter	Type	Description
name		Node identifier.
tool		Object implementing ToolProtocol with async execute().
input_key		State key to read the tool primary input from.
output_key		State key where the tool result is stored.
extra_keys		Additional state keys forwarded to the tool as arguments.

__init__

def __init__(
    name: str,
    *,
    tool: Any,
    input_key: str = 'input',
    output_key: str = 'output',
    extra_keys: list[str] | None = None
) -> None

source

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Call the tool with state-derived arguments.

Parameters

Parameter	Type	Description
state	dict[str, Any]	Current workflow state.

Returns

Type	Description
dict[str, Any]	Dict with {output_key: tool_result}.

---

TransformPipe

source

Composable transformation pipe.

Each step in the pipe receives the output of the previous step and returns the (possibly transformed) value. Steps can be sync or async callables.

The pipe is immutable — calling pipe returns a new TransformPipe instance with the added step.

__init__

def __init__(
    steps: list[Callable[Ellipsis, Any]] | None = None,
    error_handler: Callable[[Exception, Any], Any] | None = None
) -> None

source

pipe

def pipe(step: Callable[[T], U | Awaitable[U]]) -> TransformPipe[U]

source

Add a transformation step to the pipe.

The supplied step should accept the current value (of type T) and return either a new value of type U or an awaitable producing U. The returned pipe has type TransformPipe[U] so that subsequent calls are type safe.

Parameters

Parameter	Type	Description
step	Callable[[T], U | Awaitable[U]]	A callable (sync or async) that takes a value and returns the transformed value.

Returns

Type	Description
TransformPipe[U]	A new `TransformPipe` with the step appended.

tap

def tap(side_effect: Callable[[T], Any]) -> TransformPipe[T]

source

Add a side-effect step that observes but does not transform the value.

The side-effect callable receives the current value. Its return value is ignored and the original value is forwarded unchanged.

Parameters

Parameter	Type	Description
side_effect	Callable[[T], Any]	A callable (sync or async) for observation/logging.

Returns

Type	Description
TransformPipe[T]	A new `TransformPipe` with the tap step appended.

pipe_if

def pipe_if(
    predicate: Callable[[T], bool],
    step: Callable[[T], U | Awaitable[U]]
) -> TransformPipe[U | T]

source

Add a conditional transformation step.

The step is only applied when predicate(value) returns True. Otherwise the value passes through unchanged.

Parameters

Parameter	Type	Description
predicate	Callable[[T], bool]	A callable that decides whether to apply *step*.
step	Callable[[T], U | Awaitable[U]]	The transformation to apply when the predicate holds.

Returns

Type	Description
TransformPipe[U | T]	A new `TransformPipe` with the conditional step appended.

catch

def catch(handler: Callable[[Exception, T], Any]) -> TransformPipe[T]

source

Set a raw exception recovery handler for the pipe.

When any step raises and a handler is set, the handler receives (exception, last_value) and its return value becomes the pipe result, swallowing the exception.

Warning catch() operates outside the Result pattern — it intercepts raw exceptions rather than returning Err(…). Prefer designing pipe steps to return Result[T, E] and handling errors at the call-site via result.match() or result.and_then() (async) / result.and_then_sync() when possible. Use catch() only for coarse-grained last-resort recovery (e.g. logging and substituting a sentinel value) where propagating exceptions would crash the surrounding pipeline.

Parameters

Parameter	Type	Description
handler	Callable[[Exception, T], Any]	A callable `(exception, value) -> recovery_value`.

Returns

Type	Description
TransformPipe[T]	A new `TransformPipe` with the error handler configured.

execute

async def execute(
    value: T,
    **kwargs: Any
) -> T

source

Execute the pipe by passing the value through all steps.

Parameters

Parameter	Type	Description
value	T	The initial input value. **kwargs: Additional keyword arguments forwarded to each step.

Returns

Type	Description
T	The final transformed value.

---

WorkflowBuilder

source

Fluent builder for WorkflowEngine.

Parameters

Parameter	Type	Description
name		Workflow identifier used in logs and checkpoints.
config		Default runtime configuration.

__init__

def __init__(
    name: str,
    config: GraphConfig | None = None,
    version: int = 1
) -> None

source

add_node

def add_node(
    name: str,
    *,
    node: AbstractWorkflowNode | None = None,
    agent: Any | None = None,
    tool: Any | None = None,
    llm: Any | None = None,
    prompt: str = '',
    human_prompt: str = '',
    subworkflow: Any | None = None,
    executor: Any | None = None,
    output_key: str = 'output'
) -> WorkflowBuilder

source

Register a node in the workflow graph.

Parameters

Parameter	Type	Description
name	str	Unique node identifier.
node	AbstractWorkflowNode | None	Pre-built AbstractWorkflowNode instance.
agent	Any | None	AgentProtocol wrapped in AgentNode.
tool	Any | None	ToolProtocol wrapped in ToolNode.
llm	Any | None	LLMClientProtocol wrapped in LLMNode.
prompt	str	Prompt template for llm nodes.
human_prompt	str	Prompt shown to the human for human nodes.
subworkflow	Any | None	Nested workflow for SubworkflowNode.
executor	Any | None	Agent executor required for agent nodes.
output_key	str	State key to store the node primary output under.

Returns

Type	Description
WorkflowBuilder	self for chaining.

Raises

Exception	Description
ValueError	If name is already registered or no node type is given.

add_gate

def add_gate(
    name: str,
    routes: dict[str, Callable[[dict[str, Any]], bool]]
) -> WorkflowBuilder

source

Add a GateNode with conditional outgoing edges.

Parameters

Parameter	Type	Description
name	str	Gate node identifier.
routes	dict[str, Callable[[dict[str, Any]], bool]]	Mapping of target-node-name to condition callable.

Returns

Type	Description
WorkflowBuilder	self for chaining.

add_edge

def add_edge(
    source: str,
    target: str,
    *,
    condition: Callable[[dict[str, Any]], bool] | None = None,
    label: str = ''
) -> WorkflowBuilder

source

Add a directed edge from source to target.

Parameters

Parameter	Type	Description
source	str	Source node name.
target	str	Target node name.
condition	Callable[[dict[str, Any]], bool] | None	Optional guard callable (state: dict) -> bool.
label	str	Optional human-readable label.

Returns

Type	Description
WorkflowBuilder	self for chaining.

set_entry

def set_entry(node: str) -> WorkflowBuilder

source

Set node as the workflow entry point.

Parameters

Parameter	Type	Description
node	str	Node name to start execution from.

Returns

Type	Description
WorkflowBuilder	self for chaining.

set_terminal

def set_terminal(
    node: str,
    *,
    condition: Callable[[dict[str, Any]], bool] | None = None
) -> WorkflowBuilder

source

Mark node as a terminal node.

Parameters

Parameter	Type	Description
node	str	Node name to mark as terminal.
condition	Callable[[dict[str, Any]], bool] | None	Optional guard callable.

Returns

Type	Description
WorkflowBuilder	self for chaining.

configure

def configure(config: GraphConfig) -> WorkflowBuilder

source

Replace the default GraphConfig.

Parameters

Parameter	Type	Description
config	GraphConfig	New configuration.

Returns

Type	Description
WorkflowBuilder	self for chaining.

build

def build() -> WorkflowEngine

source

Validate and build the WorkflowEngine.

Returns

Type	Description
WorkflowEngine	Ready-to-execute workflow engine.

Raises

Exception	Description
GraphValidationError	If entry node is not set or graph has structural errors.

---

WorkflowCheckpoint

source

In-memory checkpoint store for workflow states.

Parameters

Parameter	Type	Description
max_size		Maximum number of checkpoints to keep. Oldest entries are evicted once the limit is reached. 0 means unlimited.

__init__

def __init__(max_size: int = 512) -> None

source

save

def save(
    checkpoint_id: str,
    state: WorkflowState
) -> None

source

Persist a snapshot of state under checkpoint_id.

Parameters

Parameter	Type	Description
checkpoint_id	str	Unique identifier for this checkpoint.
state	WorkflowState	Workflow state to snapshot.

load

def load(checkpoint_id: str) -> dict[str, object] | None

source

Retrieve the state snapshot for checkpoint_id.

Parameters

Parameter	Type	Description
checkpoint_id	str	Identifier to look up.

Returns

Type	Description
dict[str, object] | None	The raw state dict if found, otherwise None.

delete

def delete(checkpoint_id: str) -> bool

source

Remove a checkpoint entry.

Parameters

Parameter	Type	Description
checkpoint_id	str	Identifier to remove.

Returns

Type	Description
bool	True if an entry was removed, False if not found.

---

WorkflowCompletedEvent

source

Emitted when a workflow instance reaches a terminal success state.

Attributes: workflow_id: Unique identifier of the workflow instance. workflow_name: Human-readable name of the workflow definition.

---

WorkflowCompletedHook

source

Payload fired when a workflow instance reaches a terminal state.

Attributes: workflow_id: Unique identifier of the workflow instance. workflow_type: Type or name of the workflow definition. succeeded: True if the workflow completed successfully.

---

WorkflowDefinition

source

Immutable, versioned descriptor for a workflow.

Parameters

Parameter	Type	Description
name		Unique workflow name.
steps		Ordered list of step descriptors (opaque; interpreted by the executor). Defaults to an empty list for graph-based workflows where nodes/edges are the canonical step representation.
version		Definition schema version. Increment this whenever the definition structure changes incompatibly. Defaults to `1`.

---

WorkflowEdge

source

A directed edge in the workflow graph.

Edges connect a source node to a target node. An optional condition callable gates the edge: if provided, the edge is traversed only when condition(state) is truthy. Edges without a condition are always traversed (unconditional/always edges).

For parallel branches, list multiple edges from the same source nodes; WorkflowEngine will collect all edges whose conditions are satisfied and execute them concurrently when GraphConfig.parallel_branches is True.

Attributes: source: Name of the originating node. target: Name of the destination node. condition: Optional callable (state: dict) -> bool. When None the edge is unconditional. label: Optional human-readable label for documentation/debugging. is_parallel: Hint that this edge should be executed concurrently with sibling edges from the same source. The engine enables parallelism automatically when multiple conditions are satisfied; this flag provides an explicit override.

is_active

def is_active(state: dict[str, Any]) -> bool

source

Return True if this edge should be traversed given state.

Parameters

Parameter	Type	Description
state	dict[str, Any]	Current workflow state dict.

Returns

Type	Description
bool	`True` for unconditional edges or when the condition is met.

---

WorkflowEngine

source

Async directed-graph workflow executor.

Construct via WorkflowBuilder.

Parameters

Parameter	Type	Description
name		Display name for logging and checkpointing.
nodes		Dict mapping node name to node instance.
edges		List of all edges in the graph.
entry_node		Name of the starting node.
terminal_conditions		Mapping of terminal node name to optional condition callable (state) -> bool. If None the node is always terminal.
config		Runtime configuration.

__init__

def __init__(
    name: str,
    nodes: dict[str, AbstractWorkflowNode],
    edges: list[WorkflowEdge],
    entry_node: str,
    terminal_conditions: dict[str, Any],
    config: GraphConfig | None = None,
    version: int = 1
) -> None

source

name

property name() -> str

source

Workflow display name used in logs and checkpoints.

version

property version() -> int

source

Definition schema version — incremented when the graph structure changes incompatibly.

execute

async def execute(
    input: str,
    *,
    config: GraphConfig | None = None,
    state: dict[str, Any] | None = None
) -> Result[GraphResult, GraphExecutionError]

source

Execute the workflow from the entry node.

Parameters

Parameter	Type	Description
input	str	User input string injected as state["input"].
config	GraphConfig | None	Optional per-call config override.
state	dict[str, Any] | None	Optional pre-populated initial state.

Returns

Type	Description
Result[GraphResult, GraphExecutionError]	Ok(GraphResult) or Err(GraphExecutionError) on failure.

resume

async def resume(
    checkpoint_state: dict[str, Any],
    human_response: str,
    *,
    config: GraphConfig | None = None,
    response_key: str | None = None
) -> Result[GraphResult, GraphExecutionError]

source

Resume a workflow paused at a human-in-the-loop node.

Execution continues from the paused node (_paused_at in the checkpoint) rather than the entry, so already-completed steps are not re-executed.

Parameters

Parameter	Type	Description
checkpoint_state	dict[str, Any]	State dict snapshot from the checkpoint.
human_response	str	Human operator input.
config	GraphConfig | None	Optional per-call config override.
response_key	str | None	State key to store the response under. When omitted, the paused node's `resume_key is used (the contract HumanNode exposes), defaulting to "human_response"`.

Returns

Type	Description
Result[GraphResult, GraphExecutionError]	Ok(GraphResult) or Err(GraphExecutionError) on failure.

validate

def validate() -> None

source

Validate the workflow graph structure.

Raises

Exception	Description
GraphValidationError	If the graph has structural problems.

---

WorkflowFailedEvent

source

Emitted when a workflow instance terminates due to an unrecoverable error.

Attributes: workflow_id: Unique identifier of the workflow instance. workflow_name: Human-readable name of the workflow definition. error: Human-readable description of the failure reason.

---

WorkflowInstance

source

A single running (or paused) execution of a workflow definition.

Parameters

Parameter	Type	Description
instance_id		Unique identifier for this execution.
workflow_name		Name of the workflow definition this instance belongs to.
definition_version		The WorkflowDefinition.version recorded at instance-creation time. Used to detect incompatible definition changes on resume.

create

def create(
    cls,
    definition: WorkflowDefinition
) -> WorkflowInstance

source

Create a new instance stamped with definition’s version.

Parameters

Parameter	Type	Description
definition	WorkflowDefinition	The workflow definition being started.

Returns

Type	Description
WorkflowInstance	A new WorkflowInstance whose definition_version matches `definition.version`.

validate_definition_version

def validate_definition_version(definition: WorkflowDefinition) -> None

source

Raise if definition’s version differs from the one this instance was started under.

Call this before executing any step when resuming a non-new instance to prevent silently running stale or incompatible steps.

Parameters

Parameter	Type	Description
definition	WorkflowDefinition	The workflow definition being used to resume this instance.

Raises

Exception	Description
WorkflowVersionMismatchError	If `definition.version` differs from the version recorded at instance creation.

---

WorkflowModule

source

Pipeline orchestration, bulk operations, saga state machines, and graph engine.

Call configure to configure the workflow subsystem.

Usage (defaults)

@module(imports=[WorkflowModule.configure()])
class AppModule(Module):
    pass

@module(imports=[WorkflowModule.configure()])
class AppModule(Module):
    pass

Usage (configured)

from quadkit.workflow.config import BulkOperationConfig

@module(
    imports=[WorkflowModule.configure(BulkOperationConfig(batch_size=200))]
)
class AppModule(Module):
    pass

from quadkit.workflow.config import BulkOperationConfig

@module(
    imports=[WorkflowModule.configure(BulkOperationConfig(batch_size=200))]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    config: BulkOperationConfig | None = None,
    saga_store: SagaStoreProtocol | None = None,
    state_machine: StateMachineProtocol | None = None,
    content_checkpoint_store: ContentCheckpointStoreProtocol | None = None,
    content_checkpoint_config: ContentCheckpointConfig | None = None
) -> DynamicModule

source

Create a WorkflowModule with explicit configuration.

Parameters

Parameter	Type	Description
config	BulkOperationConfig | None	BulkOperationConfig or `None` for framework defaults.
saga_store	SagaStoreProtocol | None	Optional durable SagaStoreProtocol implementation. Defaults to in-memory.
state_machine	StateMachineProtocol | None	Optional application-configured StateMachineProtocol to register for DI consumers.
content_checkpoint_store	ContentCheckpointStoreProtocol | None	Optional ContentCheckpointStoreProtocol implementation for content-addressed saga checkpoints.
content_checkpoint_config	ContentCheckpointConfig | None	Optional ContentCheckpointConfig governing inline-output threshold and TTL. When omitted, framework defaults are used (1 MiB inline, 24 h TTL).

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return an in-memory WorkflowModule for unit testing.

Uses in-memory saga store and default workflow configuration. No external state backends are required.

Returns

Type	Description
DynamicModule	A DynamicModule with in-memory workflow state.

---

WorkflowProvider

source

DI provider for the workflow subsystem.

Registers pipeline, bulk-operation, and saga infrastructure into the container. All registrations use singleton scope so that the same configured objects are shared across the application.

Attributes: config: BulkOperationConfig controlling bulk and pipeline defaults. saga_store: Optional SagaStoreProtocol implementation for durable saga state. When omitted, sagas operate in-memory only (suitable for development and single-process deployments).

__init__

def __init__(
    config: BulkOperationConfig | None = None,
    saga_store: SagaStoreProtocol | None = None,
    state_machine: StateMachineProtocol | None = None,
    db_provider: DatabaseProviderProtocol | None = None,
    state_table: str = 'workflow_state_transitions',
    content_checkpoint_store: ContentCheckpointStoreProtocol | None = None,
    content_checkpoint_config: ContentCheckpointConfig | None = None
) -> None

source

Initialise the WorkflowProvider.

Parameters

Parameter	Type	Description
config	BulkOperationConfig | None	`BulkOperationConfig for bulk and pipeline defaults. Defaults to BulkOperationConfig()` (framework defaults).
saga_store	SagaStoreProtocol | None	Optional durable store for saga state. When `None`, sagas run in-memory only.
state_machine	StateMachineProtocol | None	Optional pre-configured StateMachine instance to register under StateMachineProtocol. When `None`, no state machine singleton is registered.
db_provider	DatabaseProviderProtocol | None	Optional database provider for state transition persistence.
state_table	str	SQL table name used by `DatabaseStatePersistence. Defaults to "workflow_state_transitions"`.
content_checkpoint_store	ContentCheckpointStoreProtocol | None	Optional ContentCheckpointStoreProtocol implementation for content-addressed saga checkpoints.

from_config

def from_config(
    cls,
    config: BulkOperationConfig,
    **context: Any
) -> Self

source

Create provider from config object.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register workflow services into the DI container.

Bindings registered:

• WorkflowProvider — self-reference for inspection.
• BulkOperationConfig — pipeline/bulk defaults.
• Pipeline / PipelineProtocol — default pipeline factory.
• SagaStoreProtocol — optional durable saga store.

Parameters

Parameter	Type	Description
container	ContainerRegistrarProtocol	DI registrar provided by the framework.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Perform post-registration initialisation.

Resolves BulkOperationConfig and logs the active configuration so that operators can confirm the expected settings on startup.

Parameters

Parameter	Type	Description
container	ContainerResolverProtocol	DI resolver provided by the framework.

shutdown

async def shutdown() -> None

source

Shut down the workflow provider and release any held resources.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check provider health.

The state-persistence component (when a database provider is configured) is probed via is_connected(); an unreachable backend reports the provider as unhealthy. In-memory components (pipeline, saga store, state machine) report their configured type.

Parameters

Parameter	Type	Description
timeout	float	Maximum seconds to wait for health check response.

Returns

Type	Description
HealthCheckResult	HealthCheckResult with status and component details.

---

WorkflowRunner

source

Execute a WorkflowEngine with retry and optional checkpointing.

Parameters

Parameter	Type	Description
engine		The WorkflowEngine to run.
max_retries		Maximum attempts after the first failure.
retry_delay		Seconds to wait between attempts.
checkpoint		Optional WorkflowCheckpoint instance.

__init__

def __init__(
    engine: WorkflowEngine,
    *,
    max_retries: int = 0,
    retry_delay: float = 1.0,
    checkpoint: Any | None = None
) -> None

source

run

async def run(
    input: str,
    *,
    state: dict[str, Any] | WorkflowState | None = None
) -> Result[GraphResult, GraphExecutionError]

source

Execute the workflow, retrying on transient failures.

Parameters

Parameter	Type	Description
input	str	Workflow input string.
state	dict[str, Any] | WorkflowState | None	Optional pre-populated WorkflowState.

Returns

Type	Description
Result[GraphResult, GraphExecutionError]	Ok(GraphResult) on success or Err(GraphExecutionError) on permanent failure.

Raises

Exception	Description
HumanInputRequiredError	When a HumanNode pauses execution.

resume

async def resume(
    human_response: str,
    *,
    state: WorkflowState,
    response_key: str | None = None
) -> Result[GraphResult, GraphExecutionError]

source

Resume a paused HITL workflow by injecting the human response.

Delegates to WorkflowEngine.resume so the definition-version guard applies and execution continues from the paused node (not the entry). The caller’s state is not mutated.

Parameters

Parameter	Type	Description
human_response	str	The operator response to inject into state.
state	WorkflowState	The WorkflowState from the paused point.
response_key	str | None	Optional explicit state key for the response. When omitted, the paused node's `resume_key` decides.

Returns

Type	Description
Result[GraphResult, GraphExecutionError]	Ok(GraphResult) on success or Err(GraphExecutionError).

---

WorkflowStartedEvent

source

Emitted when a workflow instance begins execution.

Attributes: workflow_id: Unique identifier of the workflow instance. workflow_name: Human-readable name of the workflow definition.

---

WorkflowStartedHook

source

Payload fired when a workflow instance begins execution.

Attributes: workflow_id: Unique identifier of the workflow instance. workflow_type: Type or name of the workflow definition.

---

WorkflowState

source

Shared, mutable state container for a workflow execution.

All nodes read from and write to the same WorkflowState instance. Each node’s execute() method receives the state as a plain dict[str, Any] and returns a partial update dict that is merged back via merge.

Keys set by the engine:

• "input" — the original user input string
• "output" — the most recent node output (updated each step)
• "_iteration" — current traversal step count
• "_history" — ordered list of (node_name, output_dict) tuples

Example

from quadkit.logging import get_logger

logger = get_logger(__name__)
state = WorkflowState(input="Summarise this document")
state.merge({"summary": "...", "output": "..."})
logger.info("workflow_output_ready")

from quadkit.logging import get_logger

logger = get_logger(__name__)
state = WorkflowState(input="Summarise this document")
state.merge({"summary": "...", "output": "..."})
logger.info("workflow_output_ready")

__init__

def __init__(
    input: str = '',
    *,
    initial: dict[str, Any] | None = None
) -> None

source

get

def get(
    key: str,
    default: Any = None
) -> Any

source

Return the value for key, or default if absent.

as_dict

def as_dict() -> dict[str, Any]

source

Return a shallow copy of the underlying state dict.

merge

def merge(update: dict[str, Any]) -> None

source

Merge update into state, overwriting existing keys.

Internal keys prefixed with _ are skipped from external updates to prevent accidental manipulation of engine metadata.

record

def record(
    node_name: str,
    output: dict[str, Any]
) -> None

source

Append a node execution record to the history list.

Parameters

Parameter	Type	Description
node_name	str	Name of the node that produced *output*.
output	dict[str, Any]	State update dict returned by the node.

iteration

property iteration() -> int

source

Current graph traversal step count.

history

property history() -> list[tuple[str, dict[str, Any]]]

source

Ordered list of (node_name, output_dict) tuples.

---

WorkflowStateTransitionedHook

source

Payload fired when a workflow instance transitions between states.

Attributes: workflow_id: Unique identifier of the workflow instance. from_state: State name being exited. to_state: State name being entered.

---

Functions

conditional

conditional

def conditional(
    name: str,
    condition: Callable[[PipelineContext], Awaitable[Result[bool, Exception]]],
    true_step: PipelineStep,
    false_step: PipelineStep | None = None,
    dependencies: list[str] | None = None,
    timeout: float | None = None
) -> ConditionalStep

source

Create a conditional pipeline step.

---

parallel

parallel

def parallel(
    name: str,
    steps: list[PipelineStep],
    dependencies: list[str] | None = None,
    *,
    fail_fast: bool = True,
    timeout: float | None = None
) -> ParallelStep

source

Create a parallel execution step.

---

pipeline_step

pipeline_step

def pipeline_step(dependencies: list[str] | None = None) -> Callable[[Callable[[PipelineContext], Awaitable[Result[Any, Exception]]]], FunctionStep]

source

Decorator to convert a function into a pipeline step.

---

saga_step

saga_step

def saga_step(
    name: str | None = None,
    *,
    compensation: Callable[Ellipsis, Any] | None = None
) -> Callable[[F], F]

source

Mark an async function as a saga step with optional compensation.

A saga step participates in distributed transaction management. If the step fails and a compensation handler is provided, it will be invoked during rollback.

Parameters

Parameter	Type	Description
name	str | None	Optional step name override. Defaults to the function name.
compensation	Callable[Ellipsis, Any] | None	Async callable to invoke for rollback on failure.

Returns

Type	Description
Callable[[F], F]	Decorator that attaches saga metadata to the function.

Example

---

step

step

def step(
    name: str,
    func: Callable[[PipelineContext], Awaitable[Result[Any, Exception]]],
    dependencies: list[str] | None = None,
    timeout: float | None = None
) -> FunctionStep

source

Create a function-based pipeline step.

timeout is an optional per-step timeout in seconds; it is forwarded to FunctionStep.

---

workflow

workflow

def workflow(
    name: str | None = None,
    *,
    timeout: float | None = None,
    retries: int = 0
) -> Callable[[F], F]

source

Mark an async function as a workflow definition.

Attaches workflow metadata used by the execution engine for registration, timeout enforcement, and retry policy.

Parameters

Parameter	Type	Description
name	str | None	Optional workflow name override. Defaults to the function name.
timeout	float | None	Maximum execution time in seconds. None means no limit.
retries	int	Number of retry attempts on failure.

Returns

Type	Description
Callable[[F], F]	Decorator that attaches workflow metadata to the function.

Example

---

Exceptions

BulkOperationCancelledError

source

Raised when a bulk operation is cancelled.

__init__

def __init__(message: str = 'Bulk operation cancelled') -> None

source

---

BulkOperationError

source

Base exception for bulk operation errors.

__init__

def __init__(message: str = 'Bulk operation error') -> None

source

---

BulkOperationTimeoutError

source

Raised when a bulk operation times out.

__init__

def __init__(message: str = 'Bulk operation timed out') -> None

source

---

CycleDetectedError

source

Graph engine exceeded max_iterations (likely cycle).

Parameters

Parameter	Type	Description
iterations		Number of iterations completed before the limit.
node		Name of the node that was about to execute again.

__init__

def __init__(
    iterations: int,
    *,
    node: str | None = None
) -> None

source

---

GraphExecutionError

source

Base exception for graph workflow engine operations.

Parameters

Parameter	Type	Description
message		Human-readable error description.
node		Optional name of the node where the error occurred.

__init__

def __init__(
    message: str,
    *,
    node: str | None = None
) -> None

source

---

GraphTimeoutError

source

Graph workflow execution exceeded total timeout.

Parameters

Parameter	Type	Description
timeout		The configured timeout value in seconds.

__init__

def __init__(timeout: float) -> None

source

---

GraphValidationError

source

Graph structure validation failed.

---

HumanInputRequiredError

source

Raised by HumanNode to pause execution awaiting human input.

Parameters

Parameter	Type	Description
prompt		Text/question to display to the human operator.
node		Node that triggered the pause.
checkpoint_id		Identifier for the stored checkpoint.

__init__

def __init__(
    prompt: str,
    *,
    node: str | None = None,
    checkpoint_id: str | None = None
) -> None

source

---

NodeExecutionError

source

A graph node's execute() method failed.

Parameters

Parameter	Type	Description
message		Human-readable error description.
node		Name of the failing node.
cause		Optional underlying exception.

__init__

def __init__(
    message: str,
    *,
    node: str | None = None,
    cause: BaseException | None = None
) -> None

source

---

PipelineExecutionError

source

Error raised during pipeline execution.

__init__

def __init__(
    step_name: str,
    error: Exception,
    details: dict[str, Any] | None = None,
    **kwargs: Any
) -> None

source

---

PipelineStepError

source

Error raised by a pipeline step.

__init__

def __init__(
    message: str = 'Pipeline step error',
    **kwargs: Any
) -> None

source

---

WorkflowCompensationError

source

Raised when workflow compensation/rollback fails.

---

WorkflowError

source

Base exception for workflow orchestration errors.

__init__

def __init__(
    message: str = 'Workflow error',
    **kwargs: Any
) -> None

source

---

WorkflowNotFoundError

source

Raised when a workflow definition cannot be found.

__init__

def __init__(workflow_id: str) -> None

source

---

WorkflowStateError

source

Raised when a workflow is in an invalid state for the requested operation.

---

WorkflowStepError

source

Raised when a workflow step fails to execute.

__init__

def __init__(
    step_name: str,
    reason: str
) -> None

source

---

WorkflowTimeoutError

source

Raised when a workflow or step exceeds its timeout.

---

WorkflowVersionMismatchError

source

Raised when resuming a workflow instance with a mismatched definition version.

Resuming an in-flight instance with an incompatible definition version would silently execute stale steps or skip newly-added ones. This error forces callers to either migrate the instance or restart it under the new definition.

Parameters

Parameter	Type	Description
workflow_name		Name of the workflow definition.
expected_version		Version stored in the running instance (set at creation time).
actual_version		Version of the definition currently being used to resume the instance.

__init__

def __init__(
    workflow_name: str,
    expected_version: int,
    actual_version: int
) -> None

source

---