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.

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.

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,
    mailbox_factory=build_realtime_mailbox,  # the same stores the gateway fills
    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.

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.