Serving over HTTP

pyrula.workflows.web serves a workflow app over HTTP: run status, a server-sent events (SSE) stream, and the control endpoints for signals, resumes, updates, and cancellation. It needs the ui extra, which carries FastAPI for the router and the run browser alike; JWT verification is the separate auth extra (see Authentication):

pip install "pyrula-workflows[ui]"

Three pieces fit together:

• WorkflowWorker executes runs. The web layer never runs a workflow itself.
• StoreControl binds the router to your store.
• RunRouter exposes the endpoints; 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 asyncio.gather(task, return_exceptions=True)


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

Submitting is opt-in. Pass the workflows you are willing to start from outside as workflows=[...], and the router adds POST /{name}/runs. Omit it and no such route exists: a read and control surface only.

That list is the allowlist. A workflow that is registered in the process but absent from it is not submittable, and asking for one returns 404 rather than starting work you did not mean to expose.

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 the workflow’s own parameters, validated against its signature: a missing required argument or a wrong type is a 422 before anything is written. A metadata key is pulled out and stored as run metadata rather than passed to the workflow. stream_path is where to tail the run, already correct for wherever you mounted the router.

Retries and duplicate submissions

Send an Idempotency-Key header and a repeat of the same request returns the original run instead of starting a second one:

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

Reusing a key with a different body is a 409: that combination almost always means a client bug rather than a retry. Keys are held for idempotency_ttl_s (24 hours by default).

Shedding load

Set max_queue_depth and a submission arriving when that many runs are already pending is rejected with 429 and a Retry-After header, instead of growing a queue no worker will reach.

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

A whole app in one call

When the FastAPI app exists only to serve runs, create_run_app builds it: a RunRouter over your control plus /healthz and /readyz, ready for uvicorn.

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 toggles the two probes (include_health, include_ready) and their paths; config takes the same RunRouterConfig the router accepts directly.

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; returns {"run_id": ...} when it fires a batch (only with workflows=[...])
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
POST	/runs/approval/runs/{run_id}/resume	Answer an interrupt
POST	/runs/approval/runs/{run_id}/update	Send an update; waits up to timeout (default 30 s, max 300) for a worker, else 504
POST	/runs/approval/runs/{run_id}/query	Run a query handler
DELETE	/runs/approval/stream/{run_id}	Cancel the run

Delivering a signal returns as soon as it is written. The worker’s claim loop picks the run up and continues past the wait; you do not drive that yourself.

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 requires its wait_id - read it off that event - alongside the action. A wait_id that no longer names the run’s open interrupt (already answered, or a newer one since opened) gets a 409, not a silent no-op:

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: a sleeping or signal-waiting run has nothing more to send until it wakes, and holding the connection open through a long sleep would pin a socket for no reason. Every frame carries an id:, so a browser EventSource reconnects and resumes from its Last-Event-ID automatically. A non-browser client should send that header itself rather than re-reading from the top.

Authentication

Endpoints are open unless you configure auth. APIKeyAuth and JWTAuth both plug in through RunRouterConfig:

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

RunRouter(
    control=control,
    config=RunRouterConfig(auth_config=AuthConfig(provider=APIKeyAuth("sk-secret"))),
).attach(app, prefix="/runs")

The router warns when built without a provider. To run open on purpose (behind an authenticating gateway, or on a laptop), pass AuthConfig(provider=None).

The run UI needs its own provider. It reads the store directly rather than going through RunRouter, so auth on the router does not cover it:

mount_run_ui(app, prefix="/ui", control=control, title="Orders", auth=APIKeyAuth("sk-secret"))

Without auth, every run and every event payload in the store is readable by anyone who can reach the port, and mount_run_ui warns. Pass auth=None for a deliberate local-only mount, which is what pyrula dev does.

Two limits:

• auth refuses; it doesn’t log anyone in. Both providers read a Bearer header, which a browser can’t attach to a page load or the timeline’s EventSource, so a person gets a 401. Put an authenticating reverse proxy in front that injects the header, or keep the UI on loopback and tunnel to it.
• Any authenticated caller sees every run; there’s no per-owner filtering. It’s an operator view, not an end-user surface.

Browsing without a worker

StoreControl also backs a read-only mount. Point it at a store your workers already write to and you get the run browser with no execution in that process:

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

This is what pyrula dev --store valkey://... does. See the CLI for the attach-mode workflow.