API Reference

Complete reference for all public classes, methods, and enums in the pipeline engine.

pipeline.context

StepContext

Frozen dataclass passed from step to step. The pipeline engine only reads sample and metadata — domain-specific fields are added by subclassing.

@dataclass(frozen=True)
class StepContext:
    sample: Any = None
    metadata: MappingProxyType = field(
        default_factory=lambda: MappingProxyType({})
    )

Behavior:

• metadata is auto-coerced from dict to MappingProxyType in __post_init__
• Subclasses inherit .replace() — it works on all fields including subclass-defined ones

pipeline.protocol

StepProtocol

Structural protocol that every step (and Pipeline/Branch) must satisfy.

@runtime_checkable
class StepProtocol(Protocol):
    requires: AbstractSet[str]
    provides: AbstractSet[str]

    def __call__(self, ctx: StepContext) -> StepContext: ...

Notes:

• AbstractSet[str] accepts both set and frozenset
• @runtime_checkable enables isinstance(step, StepProtocol) checks

SampleResult

Outcome for one sample after the pipeline has run.

@dataclass
class SampleResult:
    sample: Any
    output: StepContext | None
    error: Exception | None
    failed_at: str | None
    cause: Exception | None = None

Notes:

• Mutable — background threads update it in-place when background steps complete
• For background steps, output/error may be None until wait_for_background() completes

pipeline.pipeline

Pipeline

Ordered sequence of steps. Satisfies StepProtocol — can be nested inside other pipelines.

Constructor

Pipeline(steps: list | None = None, hooks: list[PipelineHook] | None = None)

Validates step ordering and infers contracts at construction time.

Attributes

Methods

then

def then(self, step: object) -> Pipeline

Append a step and return self for chaining. Validates ordering immediately.

Returns: self (for method chaining)

Raises: PipelineOrderError if the step requires a field produced by a later step

branch

def branch(
    self,
    *pipelines: object,
    merge: MergeStrategy | Callable = MergeStrategy.RAISE_ON_CONFLICT,
) -> Pipeline

Append a Branch step and return self for chaining. Shorthand for .then(Branch(*pipelines, merge=merge)).

Returns: self (for method chaining)

run

def run(
    self,
    contexts: Iterable[StepContext],
    workers: int = 1,
    on_sample_done: Callable[[SampleResult], None] | None = None,
    cancel_token: CancellationToken | None = None,
) -> list[SampleResult]

Process contexts through the pipeline (sync entry point).

Returns: list[SampleResult] — one result per input context

Notes: Calls asyncio.run(self.run_async(...)) internally. For background steps, call wait_for_background() after this returns. When cancel_token is provided, also sets cancel_token_var so code inside steps (e.g. LLM clients) can read it.

run_async

async def run_async(
    self,
    contexts: Iterable[StepContext],
    workers: int = 1,
    on_sample_done: Callable[[SampleResult], None] | None = None,
    cancel_token: CancellationToken | None = None,
) -> list[SampleResult]

Async entry point. Use await pipe.run_async(contexts) from coroutine contexts.

Same parameters and return type as run().

__call__

def __call__(self, ctx: StepContext) -> StepContext

Run all steps sequentially on a single context. Used when the pipeline is nested as a step inside another pipeline.

Notes: async_boundary markers are ignored in this mode — all steps run to completion.

wait_for_background

def wait_for_background(self, timeout: float | None = None) -> None

Block until all background tasks complete.

Raises: TimeoutError if timeout elapses before completion

background_stats

def background_stats(self) -> dict[str, int]

Return a snapshot of background task progress. Thread-safe.

Returns: {"active": int, "completed": int}

pipeline.branch

MergeStrategy

Enum of built-in merge strategies for Branch outputs.

class MergeStrategy(Enum):
    RAISE_ON_CONFLICT = "raise_on_conflict"
    LAST_WRITE_WINS = "last_write_wins"
    NAMESPACED = "namespaced"

Branch

Runs multiple pipelines in parallel, then merges their outputs. Satisfies StepProtocol.

Constructor

Branch(
    *pipelines: object,
    merge: MergeStrategy | Callable = MergeStrategy.RAISE_ON_CONFLICT,
)

Raises: ValueError if no pipelines are provided

Attributes

Methods

__call__

def __call__(self, ctx: StepContext) -> StepContext

Sync fan-out via ThreadPoolExecutor. All branches run to completion before any failure is raised.

Raises: BranchError if any branch fails

__call_async__

async def __call_async__(self, ctx: StepContext) -> StepContext

Async fan-out via asyncio.gather. Sync children are wrapped with asyncio.to_thread.

Raises: BranchError if any branch fails

pipeline.protocol — Hooks

PipelineHook

Observation-only protocol fired around each foreground step. Hooks cannot modify context — both methods return None.

@runtime_checkable
class PipelineHook(Protocol):
    def before_step(self, step_name: str, ctx: StepContext) -> None: ...
    def after_step(self, step_name: str, ctx: StepContext) -> None: ...

Notes:

• step_name is type(step).__name__ — hooks know what ran but cannot inspect or mutate the step instance
• Hooks fire for foreground steps only — background steps (after async_boundary) do not trigger hooks
• If a hook raises, the pipeline logs the error and continues — a broken hook never kills the pipeline
• For Branch steps, hooks fire once for "Branch" as a whole, not for inner steps

pipeline.errors

CancellationToken

Thread-safe cancellation signal. Create a fresh token per run() invocation.

class CancellationToken:
    def cancel(self) -> None: ...

    @property
    def is_cancelled(self) -> bool: ...

cancel_token_var

ContextVar set by Pipeline.run_async() so code inside steps (e.g. LLM clients) can read the current cancel token without parameter changes.

cancel_token_var: ContextVar[CancellationToken | None]  # default: None

Notes:

• Set before steps run, reset after run_async() completes
• asyncio.to_thread() copies contextvars automatically — visible in sync steps too
• Read with cancel_token_var.get(None) — returns None when no pipeline is running

PipelineCancelled

class PipelineCancelled(Exception): ...

A cancel_token was triggered. Surfaces in SampleResult.error — never propagated to the caller of run(). Callers check isinstance(result.error, PipelineCancelled) to distinguish cancellation from step failures.

PipelineOrderError

class PipelineOrderError(Exception): ...

A step requires a field that no earlier step provides (but a later step does). Raised at construction time.

PipelineConfigError

class PipelineConfigError(Exception): ...

Invalid pipeline wiring. Raised at construction time. Examples:

• More than one async_boundary = True step in the same pipeline
• An async_boundary = True step inside a Branch child

BranchError

class BranchError(Exception):
    failures: list[BaseException]

One or more branch pipelines failed. All branches always run to completion before this is raised. Raised at runtime.

Step class attributes

Optional attributes a step class can declare to control pipeline behavior: