Metadata-Version: 2.5
Name: stagechain
Version: 0.1.1
Summary: A lightweight, generic engine for chaining and routing tasks between AI agents (or any callable steps), with built-in triage/routing, retries, and structured tracing.
Project-URL: Homepage, https://github.com/divyaanshkumar24/stagechain
Project-URL: Repository, https://github.com/divyaanshkumar24/stagechain
Project-URL: Issues, https://github.com/divyaanshkumar24/stagechain/issues
Project-URL: Changelog, https://github.com/divyaanshkumar24/stagechain/blob/main/CHANGELOG.md
Author: Divyaansh Gupta
License: MIT
License-File: LICENSE
Keywords: agent,llm,orchestration,pipeline,retry,routing,tracing,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: pytest-cov>=4.0; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Description-Content-Type: text/markdown

<p align="center">
  <img src="assets/banner.svg" alt="stagechain — chain agent steps, triage & route, retry with backoff, trace every run" width="100%">
</p>

<p align="center">
  <a href="https://github.com/divyaanshkumar24/stagechain/actions/workflows/ci.yml"><img src="https://github.com/divyaanshkumar24/stagechain/actions/workflows/ci.yml/badge.svg" alt="CI"></a>
  <a href="https://pypi.org/project/stagechain/"><img src="https://img.shields.io/pypi/v/stagechain.svg" alt="PyPI"></a>
  <img src="https://img.shields.io/badge/python-3.10%2B-blue" alt="Python 3.10+">
  <img src="https://img.shields.io/badge/coverage-98%25-brightgreen" alt="Coverage 98%">
  <img src="https://img.shields.io/badge/types-py.typed-blue" alt="Typed">
  <a href="LICENSE"><img src="https://img.shields.io/badge/license-MIT-blue.svg" alt="License: MIT"></a>
  <img src="https://img.shields.io/badge/dependencies-zero-success" alt="Zero required dependencies">
  <img src="https://img.shields.io/badge/PRs-welcome-brightgreen.svg" alt="PRs Welcome">
</p>

<p align="center"><b>A lightweight, generic engine for chaining and routing tasks between AI agents — or any callable steps at all.</b></p>

---

`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.

## Contents

- [Why this exists](#why-this-exists)
- [Installation](#installation)
- [Quickstart: a linear pipeline](#quickstart-a-linear-pipeline)
- [The triage/router pattern](#the-triagerouter-pattern)
- [Retries](#retries)
- [Observability: the trace](#observability-the-trace)
- [Design principles](#design-principles)
- [API overview](#api-overview)
- [Tech stack](#tech-stack)
- [Development](#development)

## Why this exists

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.

## Installation

```bash
pip install stagechain
```

Requires Python 3.10+. Zero required third-party dependencies — the core engine is built on the
standard library alone.

## Quickstart: a linear pipeline

A pipeline is a sequence of `Stage`s. Each stage is any callable that takes the previous stage's
output and returns the next stage's input.

```mermaid
flowchart LR
    In(["input"]) --> S1["Stage\nclean_input"]
    S1 --> S2["Stage\nsummarize"]
    S2 --> S3["Stage\nformat_output"]
    S3 --> Out(["output"])
```

```python
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 report
```

Run it yourself with zero configuration: `python examples/linear_pipeline.py`.

In a real pipeline, any stage could just as easily call an LLM:

```python
def summarize_with_llm(text: str) -> str:
    response = my_llm_client.messages.create(..., messages=[{"role": "user", "content": text}])
    return response.content
```

`stagechain` doesn't know or care — it only cares that the stage is a callable that takes one
input and returns one output.

## The triage/router pattern

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.

```mermaid
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"])
```

```python
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`.

## Retries

Any stage can retry itself automatically on failure:

```python
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.

```mermaid
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
```

## Observability: the trace

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.

## Design principles

**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.

## API overview

| 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`).

## Tech stack

<p>
  <img src="https://img.shields.io/badge/Python-3.10%2B-3776AB?logo=python&logoColor=white" alt="Python">
  <img src="https://img.shields.io/badge/pytest-tested-0A9EDC?logo=pytest&logoColor=white" alt="pytest">
  <img src="https://img.shields.io/badge/GitHub%20Actions-CI%2FCD-2088FF?logo=githubactions&logoColor=white" alt="GitHub Actions">
  <img src="https://img.shields.io/badge/packaging-hatchling-orange" alt="Hatchling">
  <img src="https://img.shields.io/badge/style-PEP%208-informational" alt="PEP 8">
  <img src="https://img.shields.io/badge/stdlib--only-core%20engine-success" alt="stdlib only">
</p>

**agent-orchestration · pipeline-design · triage-routing · retry-with-backoff · observability · tracing · developer-tooling · open-source-library**

## Development

```bash
git clone https://github.com/divyaanshkumar24/stagechain.git
cd stagechain
pip install -e ".[dev]"
pytest --cov=stagechain --cov-report=term-missing
```

See [CONTRIBUTING.md](CONTRIBUTING.md) for more.

## License

MIT — see [LICENSE](LICENSE).
