Serving over HTTP

pyrula.workflows.web serves runs over HTTP: status, an SSE stream, and control endpoints (signal, resume, update, query, cancel, terminate, retry). JWT verification needs the auth extra. pyrula init --fastapi writes an app in this shape.

The router executes nothing. WorkflowWorker runs the workflows, StoreControl binds the router to the store, mount_run_ui adds the run browser.

A complete app

import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI
from pyrula.workflows import ValkeyStore, WorkflowContext, WorkflowWorker, workflow
from pyrula.workflows.web import RunRouter, StoreControl, mount_run_ui


@workflow(name="approval")
async def approval(ctx: WorkflowContext, order: str = "A-1") -> dict:
    await ctx.step("prepare", lambda: {"order": order})
    signal = await ctx.wait_for_signal("approved")
    return await ctx.step("finish", lambda: {"order": order, "by": signal.payload})


store = ValkeyStore(url="redis://localhost:6379")
worker = WorkflowWorker(store=store)
control = StoreControl(store)


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(worker.run())
    try:
        yield
    finally:
        worker.stop()
        await task  # a worker that crashed while serving raises here


app = FastAPI(lifespan=lifespan)
RunRouter(control=control, workflows=[approval]).attach(app, prefix="/runs")
mount_run_ui(app, prefix="/ui", control=control, title="Orders")

Run it with uvicorn myapp:app.

Starting runs

workflows=[...] adds POST /{name}/runs and is an allowlist: any other workflow returns 404. Without it the router is read and control only.

curl -X POST localhost:8000/runs/approval/runs \
  -H 'content-type: application/json' \
  -d '{"order": "A-1042"}'

{
  "run_id": "1f9c…",
  "status": "pending",
  "stream_path": "/runs/approval/stream/1f9c…"
}

The body is validated against the workflow’s signature (422 on a bad argument). A metadata key is stored as run metadata, not passed to the workflow. Success is 202.

Idempotency

An Idempotency-Key header makes a repeat return the original run. Reusing a key with a different body is a 409. Keys are held for idempotency_ttl_s (24 hours by default).

curl -X POST localhost:8000/runs/approval/runs \
  -H 'Idempotency-Key: order-A-1042' \
  -H 'content-type: application/json' \
  -d '{"order": "A-1042"}'

Shedding load

With max_queue_depth set, a submission arriving when that many runs are pending gets a 429 and Retry-After.

RunRouter(
    control=control,
    config=RunRouterConfig(max_queue_depth=1000),
    workflows=[approval],
)

A whole app in one call

create_run_app builds a FastAPI app with a RunRouter plus /healthz and /readyz (503 while draining or when the store ping fails). It takes no workflows=, so it is read and control only.

from pyrula.workflows.web import RunAppOptions, create_run_app

app = create_run_app(
    control=StoreControl(store),
    options=RunAppOptions(health_path="/healthz", ready_path="/readyz"),
)

RunAppOptions(host="0.0.0.0") with no auth raises ConfigurationError, unless allow_unauthenticated=True.

Endpoints

With the router attached at /runs, for a workflow named approval:

Method	Path	Purpose
POST	/runs/approval/runs	Start a run (only with workflows=[...])
POST	/runs/{name}/events	Send a trigger event (only with workflows=[...]); 202 {"run_id": ...}, null until the batch fires. With auth on, each caller has their own batch bucket
GET	/runs/runs?name=approval	List a workflow’s runs, paged (limit 1-200, cursor)
GET	/runs/approval/runs/{run_id}	Run status
GET	/runs/approval/runs/{run_id}/events	Stored events, paged
GET	/runs/approval/stream/{run_id}	Live SSE event stream
GET	/runs/approval/runs/{run_id}/replay	Replay the stored stream
POST	/runs/approval/runs/{run_id}/signal	Deliver a signal: {"name", "payload"}
POST	/runs/approval/runs/{run_id}/resume	Answer an interrupt
POST	/runs/approval/runs/{run_id}/update	{"update_name", "payload", "update_id", "timeout"}; waits up to timeout (default 30 s, max 300), else 504 and the update stays queued
POST	/runs/approval/runs/{run_id}/query	{"query", "args", "kwargs"}
POST	/runs/approval/runs/{run_id}/retry	Retry a failed run (202)
DELETE	/runs/approval/stream/{run_id}	Cancel; optional ?reason= (500 characters)
POST	/runs/approval/runs/{run_id}/terminate	End now; optional {"reason": ...}; 501 if the store cannot terminate (see Cancel and terminate)

A signal to a finished run, or one with a cancel requested, gets a 409.

curl -X POST localhost:8000/runs/approval/runs/$RUN_ID/signal \
  -H 'content-type: application/json' \
  -d '{"name": "approved", "payload": {"by": "mike"}}'

/resume answers a run:interrupted wait and needs the wait_id from that event; a stale one gets a 409.

curl -X POST localhost:8000/runs/approval/runs/$RUN_ID/resume \
  -H 'content-type: application/json' \
  -d '{"action": "approve", "wait_id": "b6f2...b0"}'

Streaming and reconnects

The SSE stream stops at a durable pause (sleeping, signal-waiting). Every frame has an id:, so a browser EventSource resumes from Last-Event-ID; other clients send it themselves.

RunRouterConfig	Default
sse_max_streams	256	Further opens get 503 with Retry-After
sse_max_streams_per_owner	16	With auth on
sse_max_duration_s	3600	The stream ends cleanly and the client resumes
sse_keepalive_interval	15 s

Authentication

Endpoints are open unless auth is configured. APIKeyAuth and JWTAuth plug in through RunRouterConfig:

from pyrula.contracts.web import AuthConfig
from pyrula.workflows.web import APIKeyAuth, RunRouter, RunRouterConfig

api_keys = APIKeyAuth({"billing": "sk-billing-secret", "reports": "sk-reports-secret"})
RunRouter(
    control=control,
    config=RunRouterConfig(auth_config=AuthConfig(provider=api_keys)),
).attach(app, prefix="/runs")

APIKeyAuth takes a mapping of caller name to key and authenticates a request as the name, never the key. The name is the run’s owner: billing can’t list or read a run reports submitted. Rotate by replacing a key’s value under the same name. APIKeyAuth("sk-secret") is shorthand for {"api-key": "sk-secret"}. Names are non-empty without : or \x1f, and each needs its own key. Load keys from the environment or a secret store.

Built without a provider, the router warns, then logs one ERROR on the first request from a non-loopback address. It never refuses. To run open on purpose: AuthConfig(provider=None).

With auth on, a run belongs to the caller who submitted it. principal_to_owner (default str) maps the principal to an owner string, and authorize_run(principal, owner) decides access (default: equal). Someone else’s run, or one with no owner, answers 404, and listing returns only your runs.

The run UI reads the store directly, so router auth doesn’t cover it. Give it its own provider:

mount_run_ui(app, prefix="/ui", control=control, title="Orders", auth=api_keys)

Without auth, anyone who can reach the port reads every run and payload, and mount_run_ui warns. auth=None is the deliberate local-only mount (what pyrula dev does).

• Both providers read a Bearer header, which a browser can’t attach to a page load or EventSource. Put an authenticating reverse proxy in front or keep the UI on loopback.
• The UI has no per-owner filtering: any authenticated caller sees every run, including run:init params and step:* inputs and results.
• Failure text is withheld and tracebacks never ship. expose_error_details=True shows redacted, capped failure text; redaction_hook runs your own filter last. show_traceback=True adds a redacted 8 KB traceback and is refused unless auth=None.
• max_streams (256) and max_duration_s (3600) bound open streams; the browser reconnects after the lifetime ends.

Browsing without a worker

Point StoreControl at a store your workers write to for a read-only run browser:

mount_run_ui(app, prefix="/ui", control=StoreControl(store), title="Runs")

This is what pyrula dev --attach --store valkey://... does; see the CLI.