FastAPI

forze[fastapi] connects an ExecutionRuntime to a FastAPI app: it runs the runtime from the app's lifespan, binds per-request context (and identity) via middleware, and maps CoreExceptions to HTTP responses. Routes are ordinary FastAPI handlers that resolve the context and run operations — written by hand, or generated from an operation registry.

Install¶

uv add 'forze[fastapi]'

No external service — FastAPI is in-process.

Run the runtime from lifespan¶

The runtime's lifecycle opens and closes every backing client (Postgres, Redis, …). runtime_lifespan holds runtime.scope() open for the app's lifetime — context created and startup run on app startup, shutdown run (and the context reset) on app shutdown, even if the app's lifetime ends with an error:

from fastapi import FastAPI
from forze.application.execution import build_runtime
from forze_fastapi import runtime_lifespan

runtime = build_runtime(...)  # deps modules + lifecycle modules/steps

app = FastAPI(title="Orders API", lifespan=runtime_lifespan(runtime))

build_runtime assembles the runtime in one call — it builds and freezes the deps registry and the lifecycle plan and returns the ExecutionRuntime.

Bind request context¶

Two ASGI middlewares attach the per-request context, both given a factory that returns the current ExecutionContext — runtime.get_context is that factory:

from forze_fastapi.middlewares import (
    InvocationMetadataMiddleware,
    SecurityContextMiddleware,
)

app.add_middleware(InvocationMetadataMiddleware, ctx_dep=runtime.get_context)
app.add_middleware(SecurityContextMiddleware, ctx_dep=runtime.get_context)

InvocationMetadataMiddleware binds correlation/execution metadata and the Idempotency-Key header; SecurityContextMiddleware binds the authenticated identity and tenant.

A request whose credential fails to verify is refused before routing — which is wrong for the handful of paths that exist because the caller has no working credential. Name them exactly:

app.add_middleware(
    SecurityContextMiddleware,
    ctx_dep=runtime.get_context,
    anonymous_paths={"/auth/login", "/auth/refresh"},
)

On those paths an authentication-kind failure — an expired or invalid credential, ambiguous credentials, a tenant mismatch — binds no identity instead of 401ing, so the route authenticates from its body exactly as it would for a request carrying nothing. A valid credential still binds normally, and every other failure kind still returns the error response: a secrets-store outage is a server fault, not a missing credential. This matters most in cookie mode, where a stale access cookie rides every request including the one that would replace it. Paths are exact, never prefixes.

Probe paths want the opposite treatment. Both middlewares resolve the execution context on every HTTP request, so in front of /livez they answer 500 while the runtime scope is not yet open — precisely the window a liveness probe exists to observe. Name the paths neither middleware should run for at all:

from forze.base.logging import DEFAULT_HEALTH_PATHS

app.add_middleware(
    InvocationMetadataMiddleware,
    ctx_dep=runtime.get_context,
    bypass_paths=DEFAULT_HEALTH_PATHS,
)

A bypassed path serves with no identity, no tenant and no invocation envelope bound, and no error shaping — list probe and scrape paths, never anything that reads or writes tenant data. anonymous_paths is the softer tool: it still runs the middleware, and still binds a valid credential. Paths are exact here too, so a route under a bypassed path stays governed — and they are the full mounted path, since middleware runs before routing: list /api/livez, not /livez, for a router mounted under /api.

check_bypass_paths (run automatically by runtime_lifespan) fails the boot on the three ways that goes wrong: a bypassed path serving a generated operation route (it would read and write tenant data with nothing bound), the two middlewares carrying different sets (the one still resolving the context fails the request anyway, so the bypass reads as configured and does nothing), and a non-empty set matching no route at all — the prefix mistake above. A superset is fine and expected: DEFAULT_HEALTH_PATHS names ten paths and most apps serve three. An app that supplies its own lifespan instead of runtime_lifespan has to call check_bypass_paths(app) and check_websocket_allowlist(app) itself — both are exported from forze_fastapi.middlewares, and neither runs on its own.

When an upstream Forze service forwards its remaining time budget as X-Forze-Deadline-Budget, opt in to honoring it with InvocationMetadataMiddleware(..., bind_deadline_from_header=True) — binding is tighten-only, so a forged value can only shorten the sender's own request.

Both middlewares refuse raw websocket scopes (the upgrade handshake is closed with a policy violation): identity, tenancy, and the envelope are resolved for HTTP only, so a raw @app.websocket route would otherwise run with none of them — silently. Framework-attached websocket routes (attach_realtime_ws_route) are allowlisted by exact full mounted path (router prefixes included) with allowed_websocket_paths={"/realtime/ws"} — they resolve identity at connect themselves, and check_websocket_allowlist (run automatically by runtime_lifespan) fails the boot if an allowlisted path doesn't serve exactly one governed route. Only if you deliberately self-manage your own websocket routes, opt out app-wide with allow_raw_websockets=True; you then own identity, tenancy, and error shaping on every websocket route yourself. For governed duplex realtime, use the Socket.IO integration or the WebSocket route below; for server-push, the SSE route below.

Map errors to HTTP¶

register_exception_handlers turns a CoreException into a response — the kind decides the status, the code rides an error-code header, and details are exposed only when the kind's egress policy allows:

from forze_fastapi.exceptions import register_exception_handlers

register_exception_handlers(app)
# raise exc.not_found("...") in a handler → 404 {"detail": "..."}

Readiness probe¶

attach_readiness_route(router, runtime) adds a GET /readyz that reflects the runtime's scope state: 200 while serving, 503 draining once shutdown flips the drain gate, 503 unavailable before the scope exists. Point your load balancer's readiness check at it so routing stops before the drain window starts.

That answers whether the process is willing to take traffic. Pass probes to also ask whether it can reach what it needs — see Readiness.

Routes¶

Routes are ordinary FastAPI handlers. A route resolves the context and runs an operation through the frozen registry (or a facade) — the domain code stays untouched:

from forze_kits.aggregates.document import DocumentFacade

@app.post("/orders")
async def create_order(cmd: CreateOrderCmd) -> ReadOrder:
    facade = DocumentFacade(ctx=runtime.get_context(), registry=registry, namespace=order_spec.default_namespace)
    return await facade.create(cmd)

Generated routes¶

Instead of hand-writing each route, project a frozen operation registry onto a router with attach_document_routes. Request and response schemas come from the operation descriptors, and each route's operationId is the registry operation key verbatim (notes.get) — so the HTTP surface, MCP tool names, and the operation catalog share one identity:

from fastapi import APIRouter
from forze_fastapi.routes import attach_document_routes

router = APIRouter(prefix="/notes", tags=["notes"])

attach_document_routes(
    router,
    registry=registry,  # build_document_registry(spec, dtos).freeze()
    ns=spec.default_namespace,
    ctx_dep=runtime.get_context,
    style="rest",
)

app.include_router(router)

Only operations the registry holds are attached, so a read-only spec yields a read-only router. Sibling helpers project search, storage (including direct and resumable uploads), and authn registries the same way, and apply_openapi_security declares the auth scheme in the generated OpenAPI. The full catalog — rest vs rpc styles, every endpoint each generator produces, the include/path_overrides knobs, and the upload flow — is in FastAPI route generators.

Realtime egress over SSE¶

attach_realtime_sse_route serves the realtime egress plane as an authenticated text/event-stream endpoint — the browser-native transport when a duplex socket is more than you need. On connect it replays the offline mailbox past the device's cursor (a browser-supplied Last-Event-ID beats the stored cursor), then tails live signals from a per-node hub; a POST …/ack endpoint alongside carries the cumulative ack. Frames use the same versioned {id, data} envelope as the Socket.IO gateway — one wire protocol, two transports:

from forze_fastapi.realtime import (
    RealtimeSseHub,
    attach_realtime_sse_route,
    realtime_sse_tail_lifecycle_step,
)

hub = RealtimeSseHub()
attach_realtime_sse_route(
    router,
    ctx_dep=runtime.get_context,
    # the same stores the gateway fills; retention has no default, so the factory
    # names the window (and a sweeper must be registered to back it)
    mailbox_factory=lambda ctx: build_realtime_mailbox(
        ctx, retention=MailboxRetention(max_age=timedelta(days=7)),
    ),
    cursors_factory=build_realtime_cursors,
    hub=hub,
)
# register alongside the app's lifecycle steps: one supervised tail loop per node
step = realtime_sse_tail_lifecycle_step(hub, stream_spec=realtime_stream_spec())

The live leg reads the realtime stream with a plain (non-group) tail — broadcast semantics, so every node sees every signal, with zero consumer-group lifecycle — and is at-most-once by contract: the mailbox carries the durable guarantee, and the Socket.IO gateway remains its sole writer. Without a hub the endpoint is catch-up-only (the browser's auto-reconnect gives long-poll-style delivery). Topics are subscribed per connection with ?topics=a,b (live-only, like Socket.IO rooms) and are fail-closed: they require an authorize_topics resolver — (ctx, principal, tenant, requested) -> granted — and the connection is refused (realtime_topics_unauthorized) unless every requested topic is granted. Topic membership is the app's authorization decision, exactly as Socket.IO topic rooms are joined by app code; the requested set is also bounded (max_topics, default 32).

On the tenancy ladder's namespace tier — the realtime stream route wired tenant_aware — use realtime_sse_sharded_tail_lifecycle_step(hub, shard=shard) instead: one supervised tail loop per shard tenant, each bound to its tenant, so signals fan out under the stream's trusted identity rather than an untrusted header (the SSE analog of TenantShardedSignalSource). Hand it the same RealtimeShard the publish-side steps use.

Pass the same presence store the Socket.IO side uses (attach_realtime_sse_route(..., presence=...)) and open SSE streams join their principal/topic rooms for the connection's lifetime — so "is this user online" counts an SSE stream and a socket identically. With a TTL-backed store (e.g. RedisRealtimePresence) also register realtime_sse_presence_heartbeat_lifecycle_step(hub, presence) so live streams re-assert within the TTL.

Realtime over raw WebSocket¶

attach_realtime_ws_route is the duplex sibling for clients that cannot run Socket.IO (strict-protocol peers, non-JS embedded clients): the same replay + live egress as SSE — sharing the hub, presence store, and topic authorization — plus a typed ingress. Identity is resolved by an app-supplied resolver from the upgrade request (add the path to the middlewares' allowed_websocket_paths); the ack rides inline ({"type": "realtime.ack", "up_to"}), a rotating token refreshes in place (realtime.reauth, same principal and tenant only), and — given a frozen registry plus RealtimeCommandRoute declarations (the same ones a Socket.IO namespace router registers) — {"type": "cmd", …} frames dispatch through the identical governed operation path as HTTP, error-acked with the shared envelope and bounded by in-flight and frame-size limits. The full framing is normative in the realtime wire protocol; asyncapi_document documents WS commands via its commands= parameter.

Authenticate the connection¶

build_ws_connection_resolver is the shipped resolver, so the ~30 lines every app writes by hand — pick a credential, verify it, carry the expiry, handle the reauth — exist once. The ladder is fixed: a realtime.reauth payload, then the cookie, then Authorization: Bearer, then the query parameter, each source enabled by naming it. The first source that presents a credential is the one used; an invalid one refuses the connection rather than falling through to the next.

from forze_fastapi.realtime import attach_realtime_ws_route, build_ws_connection_resolver

attach_realtime_ws_route(
    router,
    ctx_dep=ctx_dep,
    resolve=build_ws_connection_resolver(
        ctx_dep=ctx_dep,
        authn_spec=spec,                     # verified through the wired authn plane
        cookie_name="forze_access",          # the browser path — pair it with the allowlist below
        origin_allowlist_attested=True,      # your statement that the pairing is there
    ),
    allowed_origins=["https://app.example.com"],  # the entire cross-site perimeter
    mailbox_factory=..., cursors_factory=...,
)

Four things to know before you wire it:

• Cookie mode requires allowed_origins. A browser attaches the cookie to a cross-site upgrade by itself and the handshake has no CORS preflight, so the server-side Origin check is the whole defense. The factory cannot see the attach call's argument, so it asks you to attest it — building cookie mode without origin_allowlist_attested=True is a configuration error.
• The query-parameter source is off by default. Query strings land in access logs, proxy logs and anything that reads a URL. Enable query_param="token" only for clients that can set neither cookie nor header, with short-lived tokens.
• expires_at comes from the verified credential, and the route enforces it continuously — the socket closes once past it, and a realtime.reauth frame swaps in a fresh one without reconnecting.
• The tenant is the resolver's binding, looked up on the authn spec's own route (the shipped tenancy module registers it there), falling back to the issuer's own claim; the X-Tenant-Id header is not honored here, because an upgrade's headers are set by the client being authenticated.

forze_socketio.build_socketio_connection_resolver is the same ladder over a Socket.IO handshake, so a deployment serving both transports authenticates by one set of rules.

What it provides¶

Unlike a backend, FastAPI doesn't implement Forze contracts — it's the edge that runs them. The surface, at a glance:

Notes¶

• No external service — FastAPI runs in-process; the runtime's lifecycle owns the backing clients.
• You write or generate routes. Handlers resolve the context and run operations; the attach_*_routes helpers project a frozen registry, but you still mount the router.
• Identity is extracted, not enforced. Middleware binds the principal; enforcement lives in the engine's authn/authz hooks, and apply_openapi_security only documents it.
• Guard write-granting routes. deactivate, presigned-upload, and multipart-session endpoints ship unguarded or grant write — bind authn/authz before exposing them.