pyflowstep

A lightweight, typed Python library for composing functions into readable, reusable flows — a functional replacement for the fluent-interface (method-chaining) pattern, with flows that can be defined, validated and stored as JSON.

pyflowstep focuses on a functional style:

• steps are plain functions: (subject, *args, **kwargs) -> subject
• flows are immutable values that compose with >>
• step arguments are validated and processed when a flow is built, not halfway through running it
• raw JSON values become typed arguments with a pyargprocessors Process marker next to the parameter
• steps get external objects (a mailer, a database session) through FastAPI-style Depends
• the caller hands per-run values to a flow by name: flow(page, user=user) fills every user: User = Input()
• registry-based registration keeps steps organized and discoverable
• flow definitions can be compiled from dictionaries or JSON
• every registry can describe its flow language as a JSON Schema

It pairs naturally with its siblings pyspecification (predicates) and pyformula (formulas). Its only dependency is pyargprocessors (argument processing), which itself has none.

---

Why use pyflowstep?

A fluent API makes you put every operation on the class and return self from each method:

page.navigate("https://www.google.com").fill("input[name=q]", "pyflowstep").click("#go").wait("#result")

That couples what can be done to one class, you cannot store the chain, reuse half of it, build it from data, or add an operation without editing the class.

With pyflowstep, operations are small functions and the chain is a value:

search = navigate("https://www.google.com") >> fill("input[name=q]", "pyflowstep") >> click("#go")

search(page)                      # run it
search >> wait("#result")         # extend it (a new flow, `search` is unchanged)
login >> search >> logout         # combine flows

…and the same flow can come from JSON:

[
  {"name": "navigate", "args": ["https://www.google.com"]},
  {"name": "fill", "args": ["input[name=q]", "pyflowstep"]},
  {"name": "click", "kwargs": {"selector": "#go"}}
]

---

Installation

pip install pyflowstep

or with uv:

uv add pyflowstep

Requires Python 3.12+.

---

Quick start

from typing import Any

from pyflowstep import step


class Page:
    def navigate(self, url: str) -> None:
        print(f"Navigating to {url}")

    def click(self, selector: str) -> None:
        print(f"Clicking on {selector}")

    def fill(self, selector: str, value: Any) -> None:
        print(f"Filling {selector} with {value}")


@step
def navigate(page: Page, url: str) -> Page:
    page.navigate(url)
    return page


@step
def fill(page: Page, selector: str, value: Any) -> Page:
    page.fill(selector, value)
    return page


@step
def click(page: Page, selector: str) -> Page:
    page.click(selector)
    return page


search = navigate("https://www.google.com") >> fill("input[name=q]", 1111) >> click("#go")

print(search)  # Flow(navigate >> fill >> click)
search(Page())
# Navigating to https://www.google.com
# Filling input[name=q] with 1111
# Clicking on #go

---

Core concepts

Flow

A Flow[T] is an immutable sequence of T -> T actions. Calling it threads the subject through every action in order.

from pyflowstep import Flow, compose

increment = Flow[int](lambda n: n + 1)
double = Flow[int](lambda n: n * 2)

(increment >> double)(3)            # 8
(double >> increment)(3)            # 7
compose(increment, double)(3)       # 8, same as increment >> double
Flow[int]()(3)                      # 3, an empty flow is the identity

• >> accepts flows and plain callables, on either side: flow >> fn, fn >> flow.
• Composition flattens: (a >> b) >> c and a >> (b >> c) have the same three actions.
• Flows support len(), iteration, .actions and a readable repr.

step

@step turns a function whose first positional parameter is the subject into a step factory. Calling the factory with the remaining arguments returns a single-step flow.

from pyflowstep import step


@step
def add(total: int, amount: int) -> int:
    return total + amount


@step
def multiply(total: int, factor: int) -> int:
    return total * factor


pipeline = add(2) >> multiply(10) >> add(amount=1)
pipeline      # Flow(add >> multiply >> add)
pipeline(1)   # 31

step and tap do one thing: turn a function into a flow factory. Naming a step belongs to the registry. Three markers can sit on a parameter: Process to convert the value passed for it, Depends to inject an object nobody passes, and Input to receive a value from whoever runs the flow.

Arguments are bound against the function signature immediately, so mistakes fail fast:

add()           # MissingArgumentError: missing a required argument: 'amount' for step 'add'
add(1, 2)       # TooManyArgumentsError: too many positional arguments for step 'add'
add(1, x=2)     # UnexpectedKeywordArgumentError: got an unexpected keyword argument 'x' ...

tap

For side-effect steps on a mutable subject, @tap ignores the return value and passes the subject on unchanged — no more return page boilerplate:

from pyflowstep import tap


@tap
def click(page: Page, selector: str) -> None:
    page.click(selector)

override

A variant of a step keeps the step's body and parameters but changes a default: a value that used to be an argument now comes from whoever runs the flow, from a provider, or from a fixed default. Write the variant as a stub with the same parameters and decorate it with override(original). The stub's body never runs; the original's does.

from pyflowstep import Depends, Input, override, step


@step
def fill(form: dict, selector: str, *, value: str) -> dict:
    """Type a value into a field."""
    return {**form, selector: value}


@override(fill)
def fill_from_run(form: dict, selector: str, *, value: str = Input()) -> dict: ...


flow = fill_from_run("#date")
flow.inputs                     # frozenset({'value'})
flow({}, value="2026-11-08")    # {'#date': '2026-11-08'}

Any default can change. A parameter can become Depends(...) or get a plain default, and an Input() of the original can become an ordinary argument:

def today() -> str:
    return "2026-11-08"


@override(fill)
def fill_today(form: dict, selector: str, *, value: str = Depends(today)) -> dict: ...


@override(fill)
def fill_blank(form: dict, selector: str, *, value: str = "") -> dict: ...


fill_today("#date")({})               # {'#date': '2026-11-08'}
fill_blank("#note")({})               # {'#note': ''}
fill_blank("#note", value="hi")({})   # {'#note': 'hi'}


@step
def greet(lines: list, name: str = Input()) -> list:
    return [*lines, f"Hello {name}"]


@override(greet)
def greet_name(lines: list, name: str) -> list: ...


greet_name("Ada")([])   # ['Hello Ada']

The kind of step carries over: an override of a tap ignores the return value too, and an override of an output still stores its value.

@tap
def log(form: dict, message: str) -> None:
    print(message, form)


@override(log)
def log_saved(form: dict, message: str = "saved") -> None: ...


log_saved()({"#a": "1"})   # prints "saved {'#a': '1'}", returns {'#a': '1'}

Several variants in one flow share an input when they keep the parameter name, because the name of an input is the name of its parameter. override does not rename parameters. For separate inputs, write a step whose one-line body calls the original:

(fill_from_run("#from") >> fill_from_run("#to"))({}, value="2026-01-01")
# {'#from': '2026-01-01', '#to': '2026-01-01'}


@step
def fill_to(form: dict, selector: str, *, to: str = Input()) -> dict:
    return fill(selector, value=to)(form)


(fill_from_run("#from") >> fill_to("#to"))({}, value="2026-01-01", to="2026-12-31")
# {'#from': '2026-01-01', '#to': '2026-12-31'}

Markers come from the stub, so Process is kept only when the stub repeats it. A value that arrives through Input is passed as given, never processed. Dropping a Process without injecting the parameter is an error, because the conversion would silently disappear:

@step
def repeat(lines: list, word: str, times: Annotated[int, Process(int)]) -> list:
    return [*lines, *[word] * times]


@override(repeat)
def twice(lines: list, word: str, times: Annotated[int, Process(int)] = 2) -> list: ...


@override(repeat)
def repeat_from_run(lines: list, word: str, times: int = Input()) -> list: ...


twice("hi", "3")([])                 # ['hi', 'hi', 'hi']
repeat_from_run("hi")([], times=2)   # ['hi', 'hi']  (times="2" would reach the body as a string)

Docstring and name. The stub's docstring is used, or the original's when the stub has none. The stub's name is the step name, shown in repr and errors. An override of an override runs the first real body:

fill_blank.__doc__   # 'Type a value into a field.'


@override(fill)
def fill_date(form: dict, selector: str = "#date", *, value: str) -> dict:
    """Type the date of the voucher."""


@override(fill_date)
def fill_today_date(form: dict, selector: str = "#date", *, value: str = Depends(today)) -> dict: ...


fill_date.__doc__       # 'Type the date of the voucher.'
fill_today_date()({})   # {'#date': '2026-11-08'}
fill_from_run()         # MissingArgumentError: missing a required argument: 'selector' for step 'fill_from_run'

In a registry, registry.override(original, name=, description=, hidden=) registers the variant, so JSON can use it:

steps = StepsRegistry[dict]()


@steps.step()
def fill(form: dict, selector: str, *, value: str) -> dict:
    """Type a value into a field."""
    return {**form, selector: value}


@steps.override(steps["fill"], name="fill_date", description="Type the voucher date.")
def fill_voucher_date(form: dict, selector: str = "#date", *, value: str = Input()) -> dict: ...


@steps.override(steps["fill"], hidden=True)
def fill_debug(form: dict, selector: str = "#debug", *, value: str = "on") -> dict: ...


flow = FlowCompiler(steps.steps).compile([{"name": "fill_date"}])
flow                           # Flow(fill_date)
flow({}, value="2026-11-08")   # {'#date': '2026-11-08'}
list(steps.steps)              # ['fill', 'fill_date']

In the JSON schema, fill_date takes only selector (default "#date") and is described as "Type the voucher date.". value is a run input, so it is absent, like every input.

Errors are all raised when the stub is decorated:

override(plain)
# InvalidStepError: 'plain' is not a step factory; override takes a factory made by step, tap, output or a StepsRegistry

@override(fill)
def fill_x(form: dict, field: str, *, value: str = Input()) -> dict: ...
# InvalidStepError: Override 'fill_x' must keep the parameters of 'fill' after the subject, with the same
# names, order and kinds: expected (selector: positional or keyword, value: keyword-only),
# got (field: positional or keyword, value: keyword-only)

@override(repeat)
def thrice(lines: list, word: str, times: int = 3) -> list: ...
# InvalidStepError: Override 'thrice' drops the Process of 'times' from 'repeat'; repeat the
# Annotated[..., Process(...)] on the stub, or inject the parameter

@override(pos)  # def pos(form: dict, selector: str, /) -> dict
def pos_input(form: dict, selector: str = Input(), /) -> dict: ...
# InvalidInputError: Step 'pos_input' cannot take positional-only parameter 'selector' as a run input

@steps.override(steps["fill"], name="fill_date")
def again(form: dict, selector: str, *, value: str = "") -> dict: ...
# StepAlreadyRegisteredError: Step 'fill_date' is already registered.

Why a stub. Python's typing cannot say "the parameters of fill, with value now optional", so the variant spells them out. In return the type checker sees the real signature, (selector: str, *, value: str = Input()) -> Flow[dict], and checks every call of the variant. The parameters must match the original by name, order and kind. Only defaults and annotations may change.

---

Registry

StepsRegistry[T] collects named steps for one subject type. It is the vocabulary of your flow language: the compiler and the JSON schema only know about the steps registered in it.

from pyflowstep import StepsRegistry

page_steps = StepsRegistry[Page]()


@page_steps.tap()
def navigate(page: Page, url: str) -> None:
    """Open a url in the page."""
    page.navigate(url)


@page_steps.tap()
def click(page: Page, selector: str) -> None:
    """Click the element matching a CSS selector."""
    page.click(selector)


@page_steps.tap(name="type")
def fill(page: Page, selector: str, value: str) -> None:
    """Type a value into a field."""
    page.fill(selector, value)


@page_steps.tap(hidden=True)
def debug_dump(page: Page) -> None:
    """Python-only helper, never exposed to JSON."""


page_steps["click"]("#go")        # look up a step factory by name
"type" in page_steps              # True
list(page_steps.steps)            # ['navigate', 'click', 'type']  (hidden steps are excluded)

page_steps.register(lambda page: page, name="noop")  # register without a decorator

Method / attribute	Description
step(name=, description=, hidden=)	Decorator for (subject, ...) -> subject functions
tap(name=, description=, hidden=)	Decorator for side-effect functions (return value ignored)
register(fn, *, name=, description=, hidden=, passthrough=)	Register without decorator syntax
override(original, name=, description=, hidden=)	Decorator registering a variant of a step, see override
chain(name=, description=, hidden=)	Decorator registering a function that builds a flow, see Chains
output(name, step_name=, description=, hidden=)	Decorator registering a step whose return value is a run input, see Outputs
steps	Read-only mapping of visible steps
registry[name], name in registry, len(), iteration	Lookup over visible steps

• name is the step's public name: the compiler looks it up, and repr and argument errors show it (Flow(type) above, not Flow(fill)).
• description overrides the docstring, which becomes the step description in the JSON schema.

Chains

A chain is a named step made of other steps. There are four ways to reuse a sequence of steps, and each one adds something to the one before:

from pyflowstep import Flow, chain, step


@step
def fill(form: dict, selector: str, value: str) -> dict:
    return {**form, selector: value}


@step
def click(form: dict, selector: str) -> dict:
    return {**form, "clicked": selector}


# 1. a Flow value: fine when nothing varies
save = fill("#status", "draft") >> click("#save")


# 2. a plain function: takes arguments, but errors are plain TypeErrors
def submit(date_selector: str) -> Flow[dict]:
    return fill(date_selector, "2026-11-08") >> click("#save")


# 3. @chain: arguments are bound and processed like a step's
@chain
def submit_voucher(date_selector: str) -> Flow[dict]:
    """Fill the date and save."""
    return fill(date_selector, "2026-11-08") >> click("#save")


submit_voucher("#date")({})   # {'#date': '2026-11-08', 'clicked': '#save'}
submit()                      # TypeError: submit() missing 1 required positional argument: 'date_selector'
submit_voucher()              # MissingArgumentError: missing a required argument: 'date_selector' for step 'submit_voucher'

4. @registry.chain(name=, description=, hidden=) also registers the chain, so JSON can use it (below).

A chain's factory takes every parameter as a build argument. There is no subject. The flow it returns stays flat, so chains sit next to steps and inside other chains, and repr lists the underlying steps:

@chain
def open_and_submit(date_selector: str) -> Flow[dict]:
    return click("#new") >> submit_voucher(date_selector)


flow = open_and_submit("#date") >> fill("#note", "done")
flow       # Flow(click >> fill >> click >> fill)
flow({})   # {'clicked': '#save', '#date': '2026-11-08', '#note': 'done'}

A chain's inputs are its steps' inputs. They show in flow.inputs and are checked before the first step runs:

@step
def fill_number(form: dict, selector: str, number: str = Input()) -> dict:
    return {**form, selector: number}


@chain
def submit_numbered() -> Flow[dict]:
    return fill_number("#no") >> click("#save")


flow = click("#new") >> submit_numbered()
flow.inputs               # frozenset({'number'})
flow({})                  # MissingInputError: missing run input 'number' for step 'fill_number'; run the flow as flow(subject, number=...)
flow({}, number="V-7")    # {'clicked': '#save', '#no': 'V-7'}

Process works on chain arguments, like on step arguments:

@chain
def fill_rows(selector: str, count: Annotated[int, Process(int)]) -> Flow[dict]:
    return Flow(*(action for i in range(count) for action in fill(f"{selector}{i}", "x")))


fill_rows("#row", "2")({})   # {'#row0': 'x', '#row1': 'x'}

Chains from JSON

steps = StepsRegistry[dict]()
# ... fill and click registered with @steps.step()


@steps.chain(description="Fill the date and save.")
def submit(date_selector: str = "#date") -> Flow[dict]:
    return steps["fill"](date_selector, "2026-11-08") >> steps["click"]("#save")


flow = FlowCompiler(steps.steps).compile([
    {"name": "submit"},
    {"name": "submit", "kwargs": {"date_selector": "#due"}},
])
flow       # Flow(fill >> click >> fill >> click)
flow({})   # {'#date': '2026-11-08', 'clicked': '#save', '#due': '2026-11-08'}

In the JSON schema a chain looks like a step: submit takes date_selector in args or kwargs (a string, default "#date") and is described as "Fill the date and save.".

Errors. Bad arguments raise the usual ArgumentError, with the JSON path when compiled. Input and Depends belong on the steps a chain returns, not on the chain:

FlowCompiler(steps.steps).compile([{"name": "click", "args": ["#new"]}, {"name": "submit", "args": ["#a", "#b"]}])
# TooManyArgumentsError: too many positional arguments for step 'submit'
# at $[1]

@chain
def submit_input(date: str = Input()) -> Flow[dict]: ...
# InvalidStepError: Chain 'submit_input' cannot take 'date' as Input; a chain has no run of its own,
# declare it on the steps the chain returns   (the same for Depends)

@chain
def forgot(selector: str):
    fill(selector, "x")          # no return


forgot("#a")
# InvalidStepError: Chain 'forgot' must return a Flow, got None

@steps.chain(name="click")
def again() -> Flow[dict]: ...
# StepAlreadyRegisteredError: Step 'click' is already registered.

A chain or a step that calls flows? Use a chain when the sequence is fixed once its arguments are known: it stays flat, its inputs are checked up front, and JSON sees its steps in repr and errors. Write a step whose body runs flows, (a() >> b())(page), only when the body has to decide at run time what to do, such as looping until a page shows a result. Such a step is one opaque step to the flow around it.

---

Processing arguments

Flows often come from JSON or a web form, where every value arrives as a string, number, boolean, list, object or null, while your steps want dates, decimals, enums, clean text or loaded data. Mark the parameter with the Process(fn) marker of pyargprocessors inside Annotated, and fn is applied to the value that is passed for it:

from typing import Annotated

from pyargprocessors import Process


@page_steps.tap()
def wait(page: Page, selector: str, timeout: Annotated[float, Process(float)] = 5.0) -> None:
    page.wait(selector, timeout)


wait("#result", "10")   # timeout is 10.0

{"name": "wait", "args": ["#result", "10"]}

The marker sits next to the parameter it changes, and it works with the plain @step and @tap decorators too; no registry is required. pyargprocessors is installed together with pyflowstep.

Name it once, reuse it

A type alias gives a processed type a name, so many steps can share it:

from decimal import Decimal

type Money = Annotated[Decimal, Process(Decimal)]
type Text = Annotated[str, Process(str.strip), Process(str.lower)]   # several processors run left to right


@catalog_steps.step()
def price_between(products: Catalog, low: Money, high: Money) -> Catalog: ...


@catalog_steps.step()
def search(products: Catalog, text: Text, min_stars: Annotated[float, Process(float)] = 0) -> Catalog: ...

Loading data from a value in the JSON

A processor can be any function of one value, so the JSON can send a path and the step can receive what was loaded from it:

def load_names(path: str) -> frozenset[str]:
    return frozenset(Path(path).read_text().splitlines())


type Names = Annotated[frozenset[str], Process(load_names)]


@catalog_steps.step()
def exclude(products: Catalog, names: Names) -> Catalog:
    return tuple(product for product in products if product.name not in names)

{"name": "exclude", "args": ["discontinued.txt"]}

The file is read once, when the flow is built. Every run of that flow reuses the loaded names. For an object that must be fresh on every run, or that the JSON must not choose at all, use a dependency instead.

*args and **kwargs

On *args the processor applies to each item, and on **kwargs to each value:

@barista.step()
def top_with(drink: Drink, *toppings: Text) -> Drink: ...


@catalog_steps.step()
def where(products: Catalog, *, in_stock: Annotated[bool, Process(to_bool)] = False, **fields: Text) -> Catalog: ...


where(in_stock="yes", brand=" SONIC ")   # in_stock=True, brand="sonic"

Lists and nulls

A JSON list or a null is one value, so Process(int) would be applied to the whole of it. The each and optional helpers of pyargprocessors build the processor you need:

from pyargprocessors import each, optional

type Prices = Annotated[list[Decimal], Process(each(Decimal))]   # every item of a list
type Limit = Annotated[int | None, Process(optional(int))]       # null is kept as None

Which values are processed, and when

• Only values that are actually passed, positionally or by keyword. Default values are never processed.
• Processing runs once per factory call, while the flow is being built, not every time the flow runs.
• Arguments are checked against the step's signature first, so a missing or extra argument is reported as an ArgumentError before any processor runs.

flow = add_syrup("  Vanilla ", "2")   # processed here: flavor="vanilla", pumps=2
flow(drink)                           # the step runs with the processed values
flow(another_drink)                   # nothing is processed again

Errors

A processor that fails on a value raises ProcessArgumentError, chained to the original exception. It has the name of the argument (error.argument), the value as it was passed (error.value) and the original exception (error.cause). Inside a compiled flow it also carries the JSON path of the step, so a bad value is found before anything runs:

pyargprocessors.exceptions.ProcessArgumentError: Argument 'limit' with value 'two' failed to process, invalid literal for int() with base 10: 'two'
at $[1]

A marker that cannot work raises InvalidProcessorError when the step is created:

Problem	Example
The processor is not callable	Process("int")
The marker is used as a default value	amount: int = Process(int), write Annotated[int, Process(int)]
The marker is on the subject (the first parameter)	def step(page: Annotated[Page, Process(...)], ...)
One parameter has both Process and Depends or Input	nothing is passed for it, so there is nothing to process

Both exceptions come from pyargprocessors. They are TypeErrors and derive from PyargprocessorsError, not from PyflowstepError:

from pyargprocessors import ProcessArgumentError

try:
    flow = compiler.compile(definition)
except ProcessArgumentError as error:    # a value in the definition is wrong
    ...
except PyflowstepError as error:         # the definition itself is wrong
    ...

In the JSON schema

A processed parameter is described by what the JSON must send. The schema takes it from the type hint of the processor's own parameter (load_names(path: str) means string). If the processor has no type hint, as with int, Decimal or an enum class, the schema falls back to the parameter's type.

Parameter	Schema
names: Annotated[frozenset[str], Process(load_names)]	{"type": "string"}
limit: Annotated[int, Process(int)]	{"type": "integer"}
kind: Annotated[Milk, Process(Milk)]	{"enum": ["whole", "oat", "almond"], "type": "string"}

Upgrading

Argument processing moved to pyargprocessors, and Parse became its Process:

Before	Now
from pyflowstep import Parse	from pyargprocessors import Process
Annotated[float, Parse(float)]	Annotated[float, Process(float)]
pyflowstep.ParseArgumentError	pyargprocessors.ProcessArgumentError, no longer a PyflowstepError
pyflowstep.InvalidParserError	pyargprocessors.InvalidProcessorError, no longer a PyflowstepError
... failed to parse, ... in the message	... failed to process, ...

Coming from 0.1, the processors= option of the registry is gone as well. Move each entry onto its parameter:

0.1	Now
processors={"timeout": float}	timeout: Annotated[float, Process(float)]
processors=normalize (every argument)	annotate each parameter, usually with a shared alias such as Text
processors={"pumps": int, ...: normalize}	pumps: Annotated[int, Process(int)], the others Text
... covering **kwargs	**fields: Text
InvalidProcessorsError	removed; a key can no longer be misspelled

See examples/processing.py for every form working together on data from a web form.

---

Dependencies

Some steps need objects that cannot be written in JSON: a mailer, a database session, an API client. Mark the parameter with Depends(provider) and it stops being an argument. Nobody passes it, neither Python nor JSON; provider is called to produce it when the flow runs.

from pyflowstep import Depends, StepsRegistry

steps = StepsRegistry[Order]()


def get_mailer() -> Mailer:
    return Mailer("smtp.example.com")


@steps.tap()
def send_email(order: Order, template: str, mailer: Mailer = Depends(get_mailer)) -> None:
    mailer.send(order.customer, template)


send_email("receipt")   # only `template` is an argument

[{"name": "send_email", "args": ["receipt"]}]

It works the same with the plain @step and @tap decorators; no registry is required.

Always the default value

Depends(...) is written as the default value of the parameter, in steps and in providers. To name a dependency once and reuse it in many steps, keep the marker in a constant:

# inline: quick, for a one-off
def send_email(order: Order, template: str, mailer: Mailer = Depends(get_mailer)) -> None: ...


# a constant: name the dependency once, reuse it
MAILER = Depends(get_mailer)

def send_email(order: Order, template: str, mailer: Mailer = MAILER) -> None: ...
def send_invoice(order: Order, mailer: Mailer = MAILER) -> None: ...

It is never written inside Annotated. Only a default value tells a type checker that the parameter is not passed, so send_email("receipt") type-checks; Annotated[Mailer, Depends(get_mailer)] raises InvalidDependencyError when the step is created.

One object per flow run

A flow run is one scope, like one request in a web framework. A provider is called at most once per run, and every step of that run receives the same object. The next run starts fresh.

A provider written as a generator is cleaned up when the run ends. If a step fails, the error is raised inside the provider at its yield, so it can roll back:

from collections.abc import Iterator


def get_session() -> Iterator[Session]:
    session = Session()
    try:
        yield session          # shared by every step of this run
    except Exception:
        session.rollback()     # a step failed
        raise
    else:
        session.commit()       # the whole flow succeeded
    finally:
        session.close()


SESSION = Depends(get_session)


@steps.tap()
def save(order: Order, session: Session = SESSION) -> None:
    session.add(order)


@steps.tap()
def audit(order: Order, action: str, session: Session = SESSION) -> None:
    session.add(AuditRow(order.id, action))   # the same session `save` used


flow = save() >> audit("fulfilled")

flow(order_1)   # session A: opened, used twice, committed, closed
flow(order_2)   # session B

Cleanups run in reverse order, and a provider cannot swallow a step's error: it is always re-raised. Flows nested inside a flow, or called from inside a step, join the run that is already open.

Sub-dependencies

A provider's own parameters can use Depends too:

def get_settings() -> Settings:
    return Settings()


def get_mailer(settings: Settings = Depends(get_settings)) -> Mailer:
    return Mailer(settings.smtp_host)

Any other provider parameter must have a default. A provider can be any callable: a function, a class, a functools.partial, or an object with __call__.

Replacing a dependency in tests

from pyflowstep import override_dependencies

with override_dependencies({get_mailer: FakeMailer}):
    flow(order)          # every step that asked for get_mailer gets a FakeMailer

flow(order)              # the real mailer again

Keys are the original providers and values the providers to call instead. Nested blocks add up, and everything is restored when the block exits, even after an error.

Rules

• Invisible to JSON. A dependency is absent from the JSON schema, cannot be processed, and passing one raises UnexpectedKeywordArgumentError (or TooManyArgumentsError) when the flow is built.
• Declarations are checked early, when the step is created, with InvalidDependencyError: a provider that is not callable, is circular, or has a required parameter that is not a dependency; a dependency on the subject or on a positional-only parameter; Depends written inside Annotated.
• Providers run late, when the flow runs. An error inside a provider surfaces then, not at compile time.
• A dependency is for an object the JSON must not choose. When the JSON should pick one by name ("via": "email"), use Process with a function that turns the name into the object.

Using ruff? Its B008 rule flags function calls in argument defaults. Tell it Depends is a marker:

[tool.ruff.lint.flake8-bugbear]
extend-immutable-calls = ["pyflowstep.Depends"]

See examples/dependencies.py for a complete flow with a mailer, a per-run session and a test override.

Upgrading from 0.4

Annotated[T, Depends(provider)] is gone, for the same reason Input[T] went in 0.4: a type checker reported send_email("receipt") as missing its mailer argument. Move the marker to the default value, in steps and in providers:

0.4	0.5
mailer: Annotated[Mailer, Depends(get_mailer)]	mailer: Mailer = Depends(get_mailer)
type MailerDep = Annotated[Mailer, Depends(get_mailer)]	MAILER = Depends(get_mailer)
mailer: MailerDep	mailer: Mailer = MAILER

---

Run inputs

Some values exist only where the flow is run: the logged-in user, the record being processed, credentials read from a prompt. No provider can build them and JSON must not hold them. Give the parameter the default Input() and the caller supplies it, by name, when it runs the flow:

from pyflowstep import Input, StepsRegistry

steps = StepsRegistry[Page]()


@steps.tap()
def login(page: Page, url: str, credentials: Credentials = Input()) -> None:
    page.navigate(url)
    page.fill("#user", credentials.username)
    page.fill("#password", credentials.password)


@steps.tap()
def fill_year(page: Page, selector: str, voucher: Voucher = Input()) -> None:
    page.fill(selector, voucher.year)


flow = login("https://example.com/login") >> fill_year("#year")   # inputs are not arguments

flow(page, credentials=credentials, voucher=voucher)              # they are given to the run

[
  {"name": "login", "args": ["https://example.com/login"]},
  {"name": "fill_year", "args": ["#year"]}
]

The name of the input is the name of the parameter, and every step that declares it receives the same value.

Input() is written as the default value, never inside Annotated, so that type checkers see the parameter as optional and accept fill_year("#year").

Checked before anything runs

A flow knows the inputs its steps require, and checks them before the first step runs. A forgotten input never fails halfway through:

flow.inputs   # frozenset({'credentials', 'voucher'})

flow(page, credentials=credentials)
# MissingInputError: missing run input 'voucher' for step 'fill_year'; run the flow as flow(subject, voucher=...)

Typed calls

A flow is typed as flow(obj, **inputs: Any), because Python's typing cannot merge the inputs of several steps into one signature. To have a type checker check the inputs, write the signature once as a stub and decorate it with flow.signature:

@flow.signature
def log_in_and_fill(page: Page, /, *, credentials: Credentials, voucher: Voucher) -> Page: ...

log_in_and_fill(page, credentials=credentials)   # type checker: missing argument "voucher"

The subject is positional-only and every input keyword-only. Required inputs have no default; optional ones need one (= ... is fine), but the step's own default is what gets used. The stub is checked against the steps when it is decorated, and InvalidInputError names any input that is missing, unknown, or required when it should be optional (or the reverse). Types are not compared, only names. mypy reports a ... body as empty-body; write raise NotImplementedError instead if that bothers you.

Rules

• Invisible to JSON, like a dependency: an input is absent from the JSON schema, cannot be processed, and passing one while building the flow raises UnexpectedKeywordArgumentError (or TooManyArgumentsError).
• Input(default=...) makes the input optional: with note: str = Input(default=""), "" is used when the caller passes no note. Optional inputs are not listed in flow.inputs.
• Extra inputs are ignored, so one caller can run different flows, each using the inputs it needs.
• Nested flows see the inputs of the run they join. A flow called from inside a step can be given inputs of its own, inner(page, voucher=other); they are laid over the outer ones for that call only.
• Declarations are checked early, when the step is created, with InvalidInputError: an input on the subject, on a positional-only parameter, or written inside Annotated. A provider cannot take an input.

Using ruff? Add Input next to Depends for its B008 rule:

[tool.ruff.lint.flake8-bugbear]
extend-immutable-calls = ["pyflowstep.Depends", "pyflowstep.Input"]

Upgrading from 0.3

Input[T] is gone, because a type checker reported fill_year("#year") as missing its voucher argument. Move the marker to the default value:

0.3	0.4
voucher: Input[Voucher]	voucher: Voucher = Input()
note: Input[str] = ""	note: str = Input(default="")

Outputs

Sometimes a later step needs a value that an earlier step finds: a number read from the page, an id returned by a save. Decorate the earlier step with output(name). Its return value becomes the run input name, and the subject passes through unchanged, like a tap. The later step reads it with an ordinary Input(), so it stays reusable with or without a producer:

from pyflowstep import Input, output, step


@output("number")
def read_number(form: dict) -> str:
    return form["#no"]


@step
def fill(form: dict, selector: str, number: str = Input()) -> dict:
    return {**form, selector: number}


flow = read_number() >> fill("#target")
flow({"#no": "V-7"})   # {'#no': 'V-7', '#target': 'V-7'}
flow.inputs            # frozenset(): the caller no longer gives `number`

One output can be read by several steps, a flow can have several outputs, and an optional Input(default=...) takes the produced value once there is one:

@step
def note(form: dict, number: str = Input()) -> dict:
    return {**form, "note": f"copied {number}"}


@output("year")
def read_year(form: dict) -> str:
    return form["#date"][:4]


@step
def fill_ref(form: dict, number: str = Input(), year: str = Input()) -> dict:
    return {**form, "#ref": f"{year}/{number}"}


@output("user")
def read_user(form: dict) -> str:
    return form["#owner"]


@step
def stamp(form: dict, user: str = Input(default="system")) -> dict:
    return {**form, "#by": user}


(read_number() >> fill("#target") >> note())({"#no": "V-7"})
# {'#no': 'V-7', '#target': 'V-7', 'note': 'copied V-7'}
(read_number() >> read_year() >> fill_ref())({"#no": "V-7", "#date": "2026-11-08"})
# {'#no': 'V-7', '#date': '2026-11-08', '#ref': '2026/V-7'}
(stamp() >> read_user() >> stamp())({"#owner": "ada"})
# {'#owner': 'ada', '#by': 'ada'}   (the first stamp wrote 'system', the second 'ada')

A producer works anywhere in the flow, also inside a chain or from JSON. registry.output(name, step_name=, description=, hidden=) registers one. Its first argument is the output name, so the step name is the keyword step_name:

@chain
def copy_number(selector: str) -> Flow[dict]:
    return read_number() >> fill(selector)


(copy_number("#target") >> note())({"#no": "V-7"})
# {'#no': 'V-7', '#target': 'V-7', 'note': 'copied V-7'}

steps = StepsRegistry[dict]()


@steps.output("number", step_name="read_number")
def read_no(form: dict) -> str:
    """Read the voucher number."""
    return form["#no"]


# ... fill registered with @steps.step()

flow = FlowCompiler(steps.steps).compile([
    {"name": "read_number"},
    {"name": "fill", "args": ["#target"]},
])
flow({"#no": "V-7"})   # {'#no': 'V-7', '#target': 'V-7'}

Order is checked before anything runs. A step that reads number before the producer still needs it from the caller, so a wrong order fails before the first step:

flow = save() >> fill("#target") >> read_number()
flow.inputs            # frozenset({'number'})
flow({"#no": "V-7"})   # MissingInputError: missing run input 'number' for step 'fill'; run the flow as flow(subject, number=...)
                       # and `save` never ran

Flow.signature follows the same rule, so a produced name must not be declared:

@(read_number() >> fill("#target")).signature
def copy(form: dict, /) -> dict: ...


@(read_number() >> fill("#target")).signature
def copy2(form: dict, /, *, number: str) -> dict: ...
# InvalidInputError: Signature 'copy2' does not match the flow: unknown input 'number'

Replacing values. A later output of the same name replaces the earlier one, and a produced value replaces one the caller passed, both only for the steps after the producer. A step that needs its own output name reads it before it writes it, so the caller must give it:

(read_number() >> fill("#a") >> set_no("V-8") >> read_number() >> fill("#b"))({"#no": "V-7"})
# {'#no': 'V-8', '#a': 'V-7', '#b': 'V-8'}

(fill("#a") >> read_number() >> fill("#b"))({"#no": "V-7"}, number="given")
# {'#no': 'V-7', '#a': 'given', '#b': 'V-7'}


@output("number")
def next_number(form: dict, number: str = Input()) -> str:
    return str(int(number) + 1)


(next_number() >> fill("#no")).inputs   # frozenset({'number'})
(next_number() >> fill("#no"))({}, number="41")   # {'#no': '42'}

Scope. An output lives in its run. It is gone when the run ends, so the next run starts without it. A nested flow called with inputs of its own gets an overlay, so what it outputs ends with that call:

inner = read_number() >> fill("#inner")


@step
def run_inner(form: dict) -> dict:
    return inner({**form, "#no": "INNER"}, mode="nested")


(read_number() >> run_inner() >> fill("#outer"))({"#no": "V-7"})
# {'#no': 'INNER', '#inner': 'INNER', '#outer': 'V-7'}

An output step is a step, so it can take its own Input and Depends. A producer that returns None produces None:

@output("number")
def make_number(form: dict, year: str = Input(), prefix: str = Depends(get_prefix)) -> str:
    return f"{prefix}-{year}-{len(form)}"


(make_number() >> fill("#no"))({}, year="2026")   # {'#no': 'INV-2026-0'}


@output("number")
def maybe_number(form: dict) -> str | None:
    return form.get("#no")


@step
def fill_or_skip(form: dict, number: str | None = Input()) -> dict:
    return form if number is None else {**form, "#target": number}


(maybe_number() >> fill_or_skip())({})   # {}

The output name must be usable as a parameter name, and it is checked when the step is decorated:

output("voucher-number")
# InvalidInputError: Invalid output name: 'voucher-number', it names the run input that later steps read,
# so it must be a valid parameter name: an identifier that is not a keyword
output("class")   # the same error: a keyword

Input, Depends or an output?

The object…	Use
is written in the flow definition (a selector, a limit)	a plain argument, with Process if needed
can be built by a function, the same way for every caller (a mailer, a session)	Depends(provider)
is only known to whoever runs the flow (a user, a record, credentials)	Input()
is found by an earlier step of the same run (a number read from the page, a saved id)	output(name) on that step, Input() on the readers

---

Compiling flows from JSON

FlowCompiler turns a flow definition — a list of step dictionaries — into a Flow.

from pyflowstep import FlowCompiler

compiler = FlowCompiler(page_steps.steps)

login = compiler.compile(
    [
        {"name": "navigate", "args": ["https://example.com/login"]},
        {"name": "type", "args": ["#email", "ada@example.com"]},
        {"name": "click", "kwargs": {"selector": "button[type=submit]"}},
    ],
)

login(Page())

compile takes already-parsed data, so a database row or an API payload works too.

Flow files

compile_file picks a parser by the file suffix. JSON and TOML are built in:

login = compiler.compile_file("login.json")  # a list of steps, as above
login = compiler.compile_file("login.toml")

A TOML flow lists its steps as [[steps]] tables:

[[steps]]
name = "navigate"
args = ["https://example.com/login"]

[[steps]]
name = "click"
kwargs = { selector = "button[type=submit]" }

Add any other format by registering a parser that turns the file text into a list of steps:

import yaml

from pyflowstep import register_flow_format

register_flow_format(".yaml", yaml.safe_load)
register_flow_format(".yml", yaml.safe_load)

An unregistered suffix raises UnsupportedFlowFormatError; a file that does not parse raises InvalidFlowDefinitionError. Errors from a file carry an extra in <path> note. load_flow_definition(path) returns the parsed data without compiling it.

Each step dictionary has this shape — only name is required:

{"name": "fill", "args": ["#email"], "kwargs": {"value": "ada@example.com"}}

Everything is validated while compiling, before any step runs. Errors carry a note with their JSON path:

pyargprocessors.exceptions.ProcessArgumentError: Argument 'url' with value 'http://insecure.example.com' failed to process, only https urls are allowed
at $[1]

Problem	Exception
Not a list / malformed step dict / bad file	InvalidFlowDefinitionError
Unknown or hidden step name	StepDoesNotExistError (lists the available steps)
Missing / extra / duplicated arguments	MissingArgumentError, TooManyArgumentsError, ...
A processor rejects a value	ProcessArgumentError (from pyargprocessors)

A compiled flow is an ordinary Flow, so it composes with Python steps: login >> click("#profile").

---

JSON Schema

Describe your flow language so a UI, a validator, or an LLM can produce valid flows:

import json

from pyflowstep import get_flow_json_schema, get_step_json_schema

print(json.dumps(get_flow_json_schema(page_steps.steps), indent=2))

Output, trimmed to the click step:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "array",
  "items": {
    "oneOf": [
      {
        "type": "object",
        "properties": {
          "name": {"const": "click"},
          "args": {"type": "array", "prefixItems": [{"type": "string"}], "items": false},
          "kwargs": {
            "type": "object",
            "properties": {"selector": {"type": "string"}},
            "required": [],
            "additionalProperties": false
          }
        },
        "required": ["name"],
        "additionalProperties": false,
        "description": "Click the element matching a CSS selector."
      }
    ]
  }
}

Supported annotations: str, int, float, bool, None, Decimal, datetime, date, time, UUID, list/tuple/set/frozenset, dict, Literal, Enum, unions, Annotated, TypedDict, and type aliases. Anything else maps to {} (any value). JSON-compatible default values are included as default.

---

Examples

Runnable examples live in examples/:

• examples/browser.py — the Page automation above: tap steps, an https-only processor, reusable sub-flows and a JSON login scenario.

  uv run python -m examples.browser
  

• examples/coffee.py — a coffee shop where every step returns a new immutable Drink. The menu is JSON, orders are customized with Python steps, and a hidden staff-only discount step is invisible to the menu.

  uv run python -m examples.coffee
  

  order("latte", size("L"), add_syrup("vanilla")).describe()
  # 'L espresso, 1 shot(s), 200ml whole milk, 1x vanilla syrup -> $3.65'
  

• examples/processing.py — every way to use Process, side by side. Shop filters arrive from a web form as raw strings ("4", "yes", " SONIC ", a file path) and are turned into typed, clean arguments, including a set of names loaded from that path:

  uv run python -m examples.processing
  

  type Text = Annotated[str, Process(str.strip), Process(str.lower)]
  type Names = Annotated[frozenset[str], Process(load_names)]
  
  @catalog_steps.step()
  def where(products: Catalog, *, in_stock: Annotated[bool, Process(to_bool)] = False, **fields: Text) -> Catalog: ...
  
  @catalog_steps.step()
  def exclude(products: Catalog, names: Names) -> Catalog: ...
  

• examples/dependencies.py — steps that need objects JSON cannot describe. A fulfilment flow stored as JSON uses a mailer (with a sub-dependency) and a database session that is opened once per run, shared by two steps, then committed or rolled back. Also shows swapping the mailer for a fake in a test:

  uv run python -m examples.dependencies
  

  @fulfilment_steps.tap()
  def send_email(order: Order, template: str, mailer: Mailer = Depends(get_mailer)) -> None: ...
  
  @fulfilment_steps.tap()
  def save(order: Order, session: Session = SESSION) -> None: ...
  

Working with pyspecification and pyformula

Predicates and formulas are just callables, so they plug straight into steps:

from dataclasses import dataclass, replace
from decimal import Decimal

from pyformula import variable
from pyspecification import object_rule
from pyflowstep import step


@dataclass(frozen=True)
class Invoice:
    subtotal: Decimal
    tax: Decimal = Decimal(0)
    notes: tuple[str, ...] = ()


@object_rule()
def is_large(invoice: Invoice, limit: Decimal) -> bool:
    return invoice.subtotal >= limit


@variable()
def subtotal(invoice: Invoice) -> Decimal:
    return invoice.subtotal


@step
def note_if(invoice: Invoice, rule, note: str) -> Invoice:
    return replace(invoice, notes=(*invoice.notes, note)) if rule(invoice) else invoice


@step
def apply_tax(invoice: Invoice, formula) -> Invoice:
    return replace(invoice, tax=Decimal(str(formula(invoice))))


checkout = note_if(is_large(Decimal(100)), "needs approval") >> apply_tax(round(subtotal * 0.15, 2))

---

API reference

Name	Kind	Description
Flow[T]	class	Immutable, callable sequence of actions; composes with >>
compose(*actions)	function	Combine actions and flows into one flat flow
step / tap	decorator	Turn a function into a step factory
override(original)	decorator	A variant of a step with other defaults, see override
chain	decorator	A named step factory made of other steps, see Chains
output(name)	decorator	A step whose return value becomes the run input name, see Outputs
StepsRegistry[T]	class	Named collection of steps
pyargprocessors.Process(fn)	marker	Apply fn to the value passed for a parameter, see Processing arguments
Depends(provider)	marker	Inject a parameter by calling provider, see Dependencies
override_dependencies(mapping)	context manager	Replace providers inside a with block, for tests
Input(default=...)	marker	Receive a parameter from the caller of the flow, flow(subject, name=value), see Run inputs
Flow.inputs	property	The names of the inputs a flow requires
Flow.signature(stub)	decorator	Give a flow a typed call signature, checked against its inputs, see Typed calls
FlowCompiler[T](steps)	class	compile(definition) turns parsed step dicts into a flow
validate_step_dict(item, path)	function	Validate one step dictionary
get_json_schema(annotation)	function	JSON Schema of a type annotation
get_step_json_schema(name, step)	function	JSON Schema of one step dictionary
get_flow_json_schema(steps)	function	JSON Schema of a whole flow definition

All exceptions derive from PyflowstepError; argument errors, InvalidDependencyError, InvalidInputError and MissingInputError also derive from TypeError, definition errors from ValueError, and StepDoesNotExistError from LookupError. The exceptions of argument processing, ProcessArgumentError and InvalidProcessorError, belong to pyargprocessors: they are TypeErrors too, but derive from PyargprocessorsError.

---

Development

uv sync
uv run pytest --cov --cov-report term-missing

The test suite also runs every docstring example (--doctest-modules).

License

GPL-3.0