archipellabs-runtime

An async runtime for distributing work over Redis — Playwright sessions, API calls, SSE streams, timed waits — without losing control of how many run at once.

These tasks spend most of their time waiting (on a network call, a browser, a timer), so you want many of them going at once. Two easy approaches both break:

• A new task per event: a spike starts thousands at once and falls over — too many browsers, too many open connections.
• One at a time: safe, but a single slow call holds up everything behind it. All that waiting happens in sequence instead of together.

A service is the middle ground: a fixed budget of workers (max_slots, say 20) sharing the work. Up to 20 run together — enough to overlap the waiting — and never more. Each service has its own budget, so a slow batch of browser sessions cannot hog the workers your API calls need.

Services talk to each other with three verbs and nothing else. No registry, no service discovery, no configuration tying them together: a producer in one process and a consumer in another agree because they share a string.

Distribution name archipellabs-runtime; it imports as runtime.

Install

pip install archipellabs-runtime   # requires Python 3.12+ and a Redis server

Quickstart

import os
from runtime import App, Service

pricing = Service("pricing", max_slots=20)


@pricing.action("pricing.quote")  # one executant, returns a value
async def quote(ctx, params):
    return {"total": round(19.99 * params["qty"], 2)}


@pricing.event("order.placed")  # every subscriber gets a copy
async def note_order(ctx, params):
    print(f"order {params['id']}")


@pricing.every("500ms")  # runs on a schedule
async def tick(ctx):
    total = await ctx.call("pricing.quote", qty=3)
    print(f"quote: {total}")


app = App(redis=os.environ.get("REDIS_URL", "redis://localhost:6379/0"))
app.include(pricing)
app.start()  # blocking; logs the topology first

The three verbs

Every message can carry a ttl — a deadline that travels with it, and that any call or dispatch the handler makes inherits and can only narrow — and a delay. Failures cross the wire as typed errors and are re-raised in the caller. Actions can declare a pydantic model for their params and get validation before the handler runs.

await ctx.call("pricing.quote", ttl="2s", qty=3)
await ctx.dispatch("report.build", delay="30s", month="2026-07")
await ctx.emit("order.placed", id="o1")

Prior art

The semantics are lifted from Moleculer — actions, call, a propagated context, typed errors, distributed timeouts. What is not taken is its protocol: no registry, no heartbeats, no discovery, because Redis Streams already do that. dispatch is the one addition, where Moleculer folds decoupled work into a balanced emit.

doc/index.md has the argument; doc/limits.md has what it costs.

Shared resources

A service's lifespan opens what its handlers share — a browser, a DB pool, an API client — once at boot, and closes it on shutdown.

from contextlib import asynccontextmanager


@asynccontextmanager
async def store(config):
    db = await connect(config["dsn"])
    try:
        yield {"db": db}  # → ctx.resources["db"]
    finally:
        await db.close()


warehouse = Service("warehouse", max_slots=8, lifespan=store)


@warehouse.action("order.fulfil")
async def fulfil(ctx, params):
    await ctx.resources["db"].fulfil(params["id"])


app.include(warehouse, config={"dsn": "postgres://…"})

Documentation

Full documentation is in doc/:

Runnable examples in examples/:

• minimal — one service, one action, one producer
• rpc — call, typed errors, params validation, a ttl expiry
• events — fan-out to independent subscribers
• jobs — dispatch + task_status, a delay, and a runtime switch
• orders — two processes, lifespans on both sides
• poisson — a service that consumes and produces
• playwright — a heavy, isolated cost profile

Upgrading from 0.2? The API is replaced — see CHANGELOG.md.

Deployment

The same App runs in one process, split by role, or scaled to N replicas with no code change. App(namespace=...) isolates environments sharing one Redis; include(enabled=False) leaves a service out of a process entirely, and runtime switches pause one that is mounted.

Development

uv sync --extra dev
uv run python -m pytest                             # auto-detects a backend
RUNTIME_TEST_BACKEND=fake uv run python -m pytest   # CI runs both legs
RUNTIME_TEST_BACKEND=real uv run python -m pytest   # needs a Redis on :6379
uv run python -m pytest --cov=runtime --cov-report=term-missing
uv run mypy && uv run ruff check .

License

MIT — see LICENSE.