A lightweight, generic engine for chaining and routing tasks between AI agents — or any callable steps at all.
stagechain orchestrates arbitrary Python callables ("stages") into pipelines with linear
execution, dynamic triage/routing, conditional branching (skip, retry, terminate), automatic
retries with backoff, and a structured, inspectable execution trace. It has no opinion about what
a stage actually does: it can call Claude, OpenAI, a local model, a database, or just run plain
Python logic — stagechain only orchestrates.
This is a standalone, general-purpose open-source library, not tied to any specific product or company. It was built to demonstrate agent-orchestration and pipeline-design patterns as a reusable piece of infrastructure, not as a wrapper around one particular application.
- Why this exists
- Installation
- Quickstart: a linear pipeline
- The triage/router pattern
- Retries
- Observability: the trace
- Design principles
- API overview
- Tech stack
- Development
Most "orchestration" in agent projects is a handful of sequential function calls with a try/
except bolted on. That's fine until you need:
- Routing, not just chaining — a triage step that decides which of several next stages should run, based on the input.
- Branching — a stage that skips ahead, retries an earlier stage, or ends the run early.
- Resilience — a stage that calls a flaky API and needs configurable retries with backoff.
- Observability — a clear answer to "what actually happened during this run, in what order, with what inputs/outputs/timings, and what failed" — without stitching together ad hoc print statements.
stagechain is the smallest library that gives you all four, while staying out of the way of
everything else. It is not a workflow platform: no scheduling, no distributed execution, no
UI, no required database. It's an embeddable engine you import into a script or service.
pip install stagechainRequires Python 3.10+. Zero required third-party dependencies — the core engine is built on the standard library alone.
A pipeline is a sequence of Stages. Each stage is any callable that takes the previous stage's
output and returns the next stage's input.
flowchart LR
In(["input"]) --> S1["Stage\nclean_input"]
S1 --> S2["Stage\nsummarize"]
S2 --> S3["Stage\nformat_output"]
S3 --> Out(["output"])
from stagechain import Pipeline, Stage
def clean_input(text: str) -> str:
return text.strip()
def mock_summarize(text: str) -> str:
words = text.split()
return " ".join(words[:5]) + ("..." if len(words) > 5 else "")
def format_output(summary: str) -> str:
return f"Summary: {summary!r}"
pipeline = Pipeline([
Stage(clean_input, name="clean_input"),
Stage(mock_summarize, name="summarize"),
Stage(format_output, name="format_output"),
])
result = pipeline.run(" stagechain makes it easy to chain agent steps together ")
print(result.output) # Summary: 'stagechain makes it easy to...'
print(result.success) # True
print(result.trace.pretty()) # full step-by-step execution reportRun it yourself with zero configuration: python examples/linear_pipeline.py.
In a real pipeline, any stage could just as easily call an LLM:
def summarize_with_llm(text: str) -> str:
response = my_llm_client.messages.create(..., messages=[{"role": "user", "content": text}])
return response.contentstagechain doesn't know or care — it only cares that the stage is a callable that takes one
input and returns one output.
The core differentiator versus a plain function-chaining helper: a Router stage inspects the
data (and optionally shared context) and decides which stage runs next, instead of always
advancing in a fixed order.
flowchart LR
In(["ticket"]) --> Intake["Stage\nintake"]
Intake --> Triage{"Router\ntriage"}
Triage -->|billing| Billing["Stage\nhandle_billing"]
Triage -->|technical| Technical["Stage\nhandle_technical"]
Triage -->|general| General["Stage\nhandle_general"]
Billing --> Stop1(["Stop"])
Technical --> Stop2(["Stop"])
General --> Stop3(["Stop"])
from stagechain import Pipeline, Stop
def intake(ticket: str) -> str:
return ticket.strip()
def triage(ticket: str) -> str:
lowered = ticket.lower()
if "refund" in lowered or "invoice" in lowered:
return "billing"
if "error" in lowered or "crash" in lowered:
return "technical"
return "general"
def handle_billing(ticket: str) -> Stop:
return Stop(f"[billing team] Reviewing: {ticket!r}")
def handle_technical(ticket: str) -> Stop:
return Stop(f"[technical team] Investigating: {ticket!r}")
def handle_general(ticket: str) -> Stop:
return Stop(f"[general support] Logged: {ticket!r}")
pipeline = Pipeline(name="support-ticket-router")
pipeline.add_stage(intake, name="intake")
pipeline.add_router(triage, name="triage", routes={
"billing": "handle_billing",
"technical": "handle_technical",
"general": "handle_general",
})
pipeline.add_stage(handle_billing, name="handle_billing")
pipeline.add_stage(handle_technical, name="handle_technical")
pipeline.add_stage(handle_general, name="handle_general")
result = pipeline.run("I was charged twice for my last invoice, please refund the duplicate.")
print(result.output)
# [billing team] Reviewing: 'I was charged twice for my last invoice, please refund the duplicate.'Each handler returns Stop(...) because these are mutually exclusive terminal branches — without
it, execution would fall through into the next handler in the list, the same way a linear pipeline
falls through by default. Stop and Goto (skip ahead, or jump back to retry a previous stage)
are the two control-flow primitives any stage can return; Router is a thin, validated
convenience layer on top of Goto purpose-built for the "pick one of N routes" case.
Run it yourself: python examples/support_ticket_router.py.
Any stage can retry itself automatically on failure:
Stage(call_flaky_llm_api, name="summarize", max_retries=3, backoff_base=0.5, backoff_factor=2.0)This retries up to 3 additional times (4 attempts total) on any exception, waiting 0.5s, 1s, then
2s between attempts. Narrow retry_on=(RateLimitError, TimeoutError) to avoid retrying bugs that
will never succeed on a second try.
sequenceDiagram
participant P as Pipeline
participant S as Stage(fn)
P->>S: attempt 1
S--)P: raises TimeoutError
Note over P: sleep(backoff_base)
P->>S: attempt 2
S--)P: raises TimeoutError
Note over P: sleep(backoff_base * factor)
P->>S: attempt 3
S-->>P: returns value
Note over P: trace records attempts=3
Every pipeline.run(...) returns a PipelineResult with a .trace — a structured, ordered record
of every stage that ran, independent of whatever the host application does with the standard
logging module.
Trace(run_id=ff7d22a6671f, success=True, total=0.02ms, steps=3)
[1] intake (stage) [OK] 0.00ms
in: 'I was charged twice for my last invoice, please refund the duplicate.'
out: 'I was charged twice for my last invoice, please refund the duplicate.'
[2] triage (router) [OK] 0.00ms -> goto:handle_billing
in: 'I was charged twice for my last invoice, please refund the duplicate.'
out: 'I was charged twice for my last invoice, please refund the duplicate.'
[3] handle_billing (stage) [OK] 0.00ms -> stop
in: 'I was charged twice for my last invoice, please refund the duplicate.'
out: "[billing team] Reviewing charges related to: 'I was charged twice for my last invoice, please refund the duplicate.'"
trace.to_json() / trace.write(path) export the same data as JSON for logging pipelines or
debugging a failed run after the fact.
Observability is not optional. Every run returns a structured trace: stage name, a truncated repr of input/output, duration, attempt count, and success/failure — not just a log line you have to grep for.
Stages are just callables. stagechain never imports an LLM SDK and never will. A stage can
be def f(data): ... or def f(data, context): ... (the pipeline calls it either way based on its
signature) — there is no base class to subclass, no decorator required. This keeps the library
equally useful for real agent calls, data processing steps, and tests using plain mock functions.
Explicit control flow, no magic. A stage advances to the next stage in order by default. To do
anything else, it returns Goto("stage_name") to jump anywhere in the pipeline (forward to skip
ahead, or backward to retry/redo an earlier stage) or Stop(output) to end the run immediately.
Router is built on the same primitive — it's not a separate execution model, just a validated
way to map a routing decision to a Goto.
Minimal dependencies, by design. The core engine has zero required third-party dependencies.
Retry/backoff logic is hand-rolled (see src/stagechain/retry.py) rather than pulled in from a
package like tenacity, because the actual need — N retries, exponential backoff, optional
jitter, retry only on matching exception types — is small enough to fully own and test in-repo.
Not a workflow platform. No scheduling, no distributed execution, no persistence requirement,
no UI. If you need DAG scheduling across machines, cron-like triggers, or a visual pipeline
builder, reach for Airflow, Prefect, or similar. stagechain is for the much more common case: a
single process that needs to run a sequence of steps with routing, retries, and a trace, and
doesn't want a platform to do it.
| Concept | What it does |
|---|---|
Stage(fn, name=..., max_retries=..., ...) |
Wraps one callable as a pipeline step. |
Router(fn, name=..., routes={...}) |
Wraps a callable that decides the next stage's name. |
Pipeline([...]) / .add() / .add_stage() / .add_router() |
Builds the ordered set of stages/routers. |
pipeline.run(input_data) |
Executes the pipeline, returns a PipelineResult. |
Goto(target, data=...) |
Returned from a stage to jump to another named stage. |
Stop(output=...) |
Returned from a stage to end the run immediately. |
PipelineResult.output / .success / .error / .trace |
The run's outcome. |
Trace.pretty() / .to_dict() / .to_json() / .write(path) |
Inspect or export the execution trace. |
PipelineContext.shared |
Free-form scratch dict passed to any stage that accepts (data, context). |
Every public class and function has a docstring; see the source under src/stagechain/ for full
details, or generate API docs from the docstrings with your tool of choice (e.g. pdoc, sphinx).
agent-orchestration · pipeline-design · triage-routing · retry-with-backoff · observability · tracing · developer-tooling · open-source-library
git clone https://github.com/divyaanshkumar24/stagechain.git
cd stagechain
pip install -e ".[dev]"
pytest --cov=stagechain --cov-report=term-missingSee CONTRIBUTING.md for more.
MIT — see LICENSE.