Distributed Python pipelines from a decorator

Pipelines are just Python

Mark units of work as steps and compose them into workflows with a decorator; chain them by awaiting one step from another, and express branching, loops and fan-out/fan-in in real code. No YAML DAGs, JSON state machines or DSL to learn, and the same definitions run async or sync.

@app.workflow
async def pipeline(x: int, services=None):
    doubled = await double(x, services=services)
    return await add_ten(doubled, services=services)

Publish a streaming HTTP API

Stack an endpoint decorator on any workflow and the framework generates an OpenAPI schema and typed request/response models from your signature, turning the app into a standard ASGI service. Callers get validated JSON, a job id to poll, and live progress over server-sent events or WebSocket.

@app.endpoint(path="/run", method="POST")
@app.workflow
async def pipeline(x: int, services=None): ...

Warm services & pooled connectors

A service initializes once per worker and stays warm across thousands of tasks, reusing pooled connections and built-in connectors for object storage, databases and popular SaaS tools — expensive setup runs once, not per call.

@app.step(sandboxed=True)   # no filesystem, no direct network
async def run_user_code(x: int, services=None):
    return await services["Database"].query(f"SELECT {x}")

Sandboxed steps for untrusted code

Flip one flag and a step runs inside an isolated WebAssembly sandbox with no host filesystem and no direct network; it reaches your infrastructure only through the specific service methods you approve, so credentials never enter the sandbox. User code is also validated at decoration time and re-verified server-side before it runs.

@method(timeout=30, retries=2, max_concurrency=10,
        rate_limit=10, rate_window=1, rate_key="vendor:default")
async def fetch(self, symbol: str) -> dict: ...

Durable, resumable execution

Every operation is durably recorded, so a worker that restarts mid-task is idempotently re-dispatched and the pipeline finishes instead of failing; abandoned workers are reclaimed on a heartbeat rather than lost.

Full-jitter retries, no storms

Built-in retries use exponential backoff with full jitter, so a downstream blip doesn't become a synchronized retry stampede. Infinite-retry modes are bounded by an attempt cap so a wedged call can't spin forever.

@app.step(gpu="any", gpu_memory="24GB")   # route by memory floor, not a fixed SKU
async def inference(model, services=None): ...

@app.workflow(schedule=Cron("0 9 * * 1-5"))   # server-side, weekdays 09:00
async def daily_digest(sources, services=None): ...

Health-aware regional failover

A stuck unit is detected, cancelled in its old region and re-submitted to the least-loaded healthy region — with an idempotency key appended per attempt so a resubmit can't double-execute. Retries are bounded and non-retryable errors fail fast.

Crash-loop quarantine & respawn

Dead workers are respawned automatically; a worker that keeps crashing is rate-limited and quarantined, so a single poison task can't burn the fleet down in a restart loop.

Fleet-wide rate limits

Cap invocations per sliding window across the whole worker fleet, not per machine — enforced regionally and correct under active-active replication, so you never overshoot a vendor's quota. Denied callers pace themselves instead of parking a worker.

@method(rate_limit=10, rate_window=1, rate_key="vendor:api")
async def fetch(self, symbol: str) -> dict:
    return await self._client.get(symbol)

Distributed concurrency limits

Per-method max-concurrency is enforced across all workers via leased in-flight slots; over-capacity callers wait for a slot rather than being dropped, and a crashed holder's lease expires so capacity is never leaked to zero.

Deadlock-safe deep pipelines

Recursive and nested pipelines raise the warm-worker floor automatically so a deep call chain can't hold-and-wait deadlock, backed by a checked slots ≥ concurrent-roots × depth invariant.

Queue-depth autoscaling & scale-to-zero

Worker pools size themselves from live queue depth, growth rate and the detected bottleneck, and idle per-tenant pools scale to zero after a cooldown — reactivating the moment work arrives, with hysteresis to avoid flapping.

GPUs by type, memory & count

Request accelerators per step by type, a minimum-memory floor, or a GPU count for distributed training; sync GPU work is routed to isolated workers so it can't stall the async event loop.

@app.step(gpu="any", gpu_memory="80GB")
async def infer(model, services=None):
    ...

Micro-batching for inference

Coalesce many individual step calls into one batch — flushing on batch size or a time window, whichever comes first — ideal for GPU inference, with optional padding to a fixed batch size.

Cron & interval scheduling

Schedule any workflow server-side with a standard, timezone-aware cron expression or a fixed interval with jitter — no external scheduler, and exactly one instance fires each occurrence.

@app.workflow(schedule=Cron("0 9 * * 1-5"))   # weekdays 09:00
async def daily_digest(services=None):
    ...

Ephemeral agent workspaces

Give a workflow an isolated, per-tenant workspace to run untrusted or AI-generated code, then throw it away — created, injected as a service, and destroyed around your workflow body by a single decorator.

@app.workflow(workspace={"repo_url": "…", "auto_destroy": True})
async def agent(task, services=None):
    ws = services["workspace"]
    ...

Isolation tier & resource limits

Each workspace gets its own isolation — a userspace-kernel sandbox by default, or an optional microVM tier that gives it a separate kernel — with CPU, memory, PID and disk limits applied per workspace.

Locked-down workspace egress

On a self-hosted cluster, a workspace's outbound traffic is forced through a per-workspace proxy with a domain allow-list, and the network is severed at the install→run boundary so build-time access can't leak into execution.

Thread-persistent agent workspaces

Keep a workspace alive across a conversation: follow-up messages reuse the same sandbox with its cloned repo and state intact, torn down only on a terminal outcome or timeout — built for coding agents.