A pipeline is a graph of nodes in JSON. The work happens in stages — Python classes you write. Between them travels one immutable frame of variables, and everything the graph does with that frame is visible in the JSON: branching, retries, error handling, concurrency, nested graphs.

Because the pipeline is data, it can be stored, diffed, generated, validated before it runs — and drawn:

That is the editor: a separate static page that draws and debugs a graph while this core executes it.

Install

pip install stageflow-framework

Python 3.11+.

A pipeline in 30 seconds

import asyncio

from stageflow import BaseStage, Pipeline, Session, register_stage


@register_stage("HelloStage")
class HelloStage(BaseStage):
    """
    description: "Greets whoever the pipeline points at"
    arguments:
      name: string
    outputs:
      greeting: string
    """

    async def run(self):
        name = self.get_arguments().get("name", "world")
        self.set_outputs({"greeting": f"Hello, {name}!"})


pipeline = Pipeline.from_dict({
    "nodes": [
        {"id": "start", "type": "entry",
         "variables": {"user_name": "Alice"}, "next": "hello"},
        {"id": "hello", "type": "stage", "stage": "HelloStage",
         "arguments": {"vars": {"name": "user_name"}},
         "outputs": {"greeting": "greeting"}, "next": "finish"},
        {"id": "finish", "type": "terminal",
         "result": {"status": "ok"}, "artifacts": ["greeting"]},
    ],
})
pipeline.validate()

result = asyncio.run(Session(id="demo", pipeline=pipeline).run())
print(result.result)     # {'status': 'ok'}
print(result.artifacts)  # {'greeting': 'Hello, Alice!'}

The docstring is the stage's specification. validate() checks the graph against it, so a wrong argument name is an error before anything runs.

What the graph can do

Documentation

The tutorial builds one working pipeline step by step, with screenshots from the editor. The reference covers the rest:

Quick start · Node types · Data model · Expressions · Errors · Variable typing · Session control · Step debugging

One page per node type: entry · stage · condition · switch · parallel · try · map · subpipeline · terminal

Sources are in docs/ (*.md English, *.ru.md Russian) and publish themselves on every push to main.

The rest of the project

Development

pip install -e ".[dev]"
python -m unittest discover -s tests

Pipelines can be tested declaratively:

from stageflow.testing import PipelineTestSpec, run_pipeline_test

Releases are tag-driven: the tag has to match project.version in pyproject.toml, and the workflow publishes to PyPI.

License

MIT — see LICENSE.