loom.rest.fastapi

FastAPI-specific REST layer for Loom.

Provides the FastAPI binding for compiled REST interfaces: - MsgspecJSONResponse — zero-copy JSON response. - bind_interfaces() — binds compiled routes to FastAPI. - create_fastapi_app() — composition root. - describe_fastapi_app() — application self-description.

class loom.rest.fastapi.MsgspecJSONResponse(*args, **kwargs)[source]

Bases: Response

FastAPI Response subclass that encodes content with msgspec.json.

Drop-in replacement for fastapi.responses.JSONResponse with two advantages:

  • Native msgspec.Struct serialisation — no dict conversion needed.

  • Single encoding pass — content is written directly to bytes with no intermediate JSON string or Pydantic round-trip.

Parameters:

  • content – Any object supported by msgspec.json.encode.

  • status_code – HTTP status code. Defaults to 200.

  • headers – Additional response headers.

  • media_type – Defaults to "application/json".

  • background – Optional Starlette background task.

  • args (Any)

  • kwargs (Any)

Return type:

Any

Example:

return MsgspecJSONResponse(content=my_struct, status_code=201)
render(content)[source]

Encode content to JSON bytes using msgspec.json.encode.

Parameters:

content (object) – Object to serialise.

Returns:

UTF-8 encoded JSON bytes.

Return type:

bytes

loom.rest.fastapi.bind_interfaces(app, compiled_routes, factory, executor, observability_runtime)[source]

Register compiled routes on a FastAPI application.

For each CompiledRoute, creates a dynamic async handler and registers it via app.add_api_route. Path parameters are inferred from the route path template and exposed in the handler signature so FastAPI validates and documents them correctly.

Nested JSON Schema $defs produced by msgspec/pydantic are collected into a shared component registry and returned. The caller is responsible for injecting these into components.schemas of the OpenAPI document.

Parameters:

  • app (fastapi.FastAPI) – FastAPI application instance to register routes on.

  • compiled_routes (Sequence[CompiledRoute]) – Ordered list of fully resolved routes produced by RestInterfaceCompiler.

  • factory (UseCaseFactory) – Use-case factory for constructing instances per request.

  • executor (RuntimeExecutor) – Runtime executor that drives the use-case pipeline.

  • observability_runtime (ObservabilityRuntime) – Shared runtime used to emit request lifecycle events around each handler execution.

Returns:

Mapping of schema name → JSON Schema fragment for all collected $defs that should appear under components.schemas.

Return type:

dict[str, Any]

Example:

compiler = RestInterfaceCompiler(use_case_compiler)
routes = compiler.compile(UserRestInterface)
component_schemas = bind_interfaces(
    app,
    routes,
    factory,
    executor,
    observability_runtime=ObservabilityRuntime.noop(),
)
loom.rest.fastapi.create_app(*config_paths, code_path=None, metrics_registry=None, authenticator=None, resolvers=())[source]

Create a FastAPI application from one or more YAML config files.

Config files are merged left-to-right — later files override earlier ones. Each file may also declare a top-level includes list to pull in additional base files before its own values (resolved by loom.core.config.ConfigContext.from_yaml()).

${secrets:...} and ${ssm:...} placeholders resolve through loom’s built-in AWS resolvers unless resolvers supplies one with the same name.

TraceIdMiddleware is mounted automatically. Structured logging and OTEL come from the top-level observability: section. Prometheus middleware is mounted when observability.prometheus.enabled is true.

Authentication is mechanism-agnostic: configure app.rest.auth.jwt for the built-in stateless JWT mechanism, or pass authenticator for any other. Exactly one of the two may be active.

The optional sql: section wires the SQL subsystem: its connections open inside the app lifespan and two collaborators are registered in the container (null implementations raising an actionable ConfigError when the section is absent). CallerBoundSql derives the query roles from the verified caller and is what a use case serving a request should inject; SqlQueryService takes the roles as an argument and is for system work that has no caller. Connections opting in with sql_endpoint.enabled plus an explicit sql_endpoint.auth mount a POST /sql/{name} endpoint; auth: identity additionally requires a configured authentication mechanism, and a non-empty allowed_roles also requires that mechanism to bind roles to the verified caller identity (for JWT, app.rest.auth.jwt.roles_claim).

Parameters:

  • *config_paths (str) – One or more paths to YAML configuration files.

  • code_path (str | None) – Optional override for app.code_path. Resolved relative to the first config file when not absolute.

  • metrics_registry (CollectorRegistry | None) – Optional Prometheus CollectorRegistry used for PrometheusMiddleware and the scrape endpoint. Defaults to the global registry. Pass a fresh CollectorRegistry() in tests to avoid ValueError: Duplicated timeseries when multiple apps with observability.prometheus.enabled: true are created in the same process.

  • authenticator (Authenticator | None) – Custom authentication mechanism. Mutually exclusive with the app.rest.auth.jwt config section.

  • resolvers (Sequence[ConfigResolver]) – Resolvers for ${name:key} placeholders, registered before the built-in secrets and ssm defaults. A resolver named like a default replaces it.

Returns:

Configured fastapi.FastAPI application, ready to serve.

Raises:

ConfigError – When no config path is given, or when both authenticator and app.rest.auth.jwt are supplied.

Return type:

fastapi.FastAPI

Example — single config:

app = create_app("config/app.yaml")

Example — a mechanism of your own:

app = create_app("config/app.yaml", authenticator=MyMtlsAuthenticator())

Example — base + environment override:

app = create_app("config/base.yaml", "config/production.yaml")

Example — single file using inline includes:

# config/app.yaml
# includes:
#   - base.yaml
#   - secrets.yaml
app = create_app("config/app.yaml")
loom.rest.fastapi.create_fastapi_app(result, routes=None, *, interfaces=None, observability_runtime=None, middleware=(), defaults=None, **fastapi_kwargs)[source]

Create a FastAPI application from a bootstrap result and REST interfaces.

Compiles all RestInterface declarations via RestInterfaceCompiler, binds each compiled route to the FastAPI instance, and returns the ready application.

Compilation is fail-fast: any structural error (missing use-case plan, duplicate route, missing prefix) raises InterfaceCompilationError before the app starts accepting requests.

Parameters:

  • result (BootstrapResult) – Fully initialised BootstrapResult from bootstrap_app().

  • routes (RouteSources | None) – Which interfaces to mount and from which origin — see RouteSources. routes.config compiles after routes.python, deterministically; a (method, path) collision between the two aborts naming both — neither side takes precedence. routes.disabled is applied to routes.python before the merge, so a disabled Python route can be redeclared in routes.config without colliding with itself; an entry matching no Python-declared route aborts startup instead of doing nothing. Required unless interfaces is given instead.

  • interfaces (Sequence[type[RestInterface[Any]]] | None) – Deprecated keyword-only alias for routes=RouteSources(python=interfaces). Kept only so code written before RouteSources existed keeps working when it called create_fastapi_app(result, interfaces=[...]) — every published example did; emits a DeprecationWarning naming routes as the replacement. Mutually exclusive with routes.

  • observability_runtime (ObservabilityRuntime | None) – Shared runtime used to emit lifecycle events around each request.

  • middleware (Sequence[Any]) –

    ASGI middleware classes to register on the application. Added in declaration order (first = outermost wrapper). Accepts any class compatible with FastAPI.add_middleware. Example:

    from loom.rest.middleware import TraceIdMiddleware
from loom.prometheus import PrometheusMiddleware

app = create_fastapi_app(
    result,
    RouteSources(python=[...]),
    observability_runtime=ObservabilityRuntime.noop(),
    middleware=[TraceIdMiddleware, PrometheusMiddleware],
)

  • defaults (RestApiDefaults | None) – Global REST API defaults (pagination mode, profile policy). Falls back to RestApiDefaults when not provided.

  • **fastapi_kwargs (Any) – Additional keyword arguments forwarded to the FastAPI constructor (e.g. title, version, docs_url).

Returns:

Configured fastapi.FastAPI instance ready to serve requests.

Raises:
Return type:

fastapi.FastAPI

Example:

app = create_fastapi_app(
    result,
    RouteSources(python=[UserRestInterface, OrderRestInterface]),
    defaults=RestApiDefaults(pagination_mode=PaginationMode.CURSOR),
    observability_runtime=ObservabilityRuntime.noop(),
    title="My API",
    version="2.0.0",
)
loom.rest.fastapi.describe_fastapi_app(app)[source]

Describe an application built by create_app().

The document always carries the application identity under "app", plus one section per wired pillar — "agents" when an ai: section is configured. Only publishable values appear: no instructions, no model binding, no URL and no credential reference.

Parameters:

app (fastapi.FastAPI) – Application previously built by create_app().

Returns:

JSON-encodable self-description of the application.

Raises:

IntrospectionError – When app was not built by create_app(), or when a pillar contribution cannot be resolved.

Return type:

dict[str, Any]

Example:

app = create_app("config/app.yaml")
describe_fastapi_app(app)["app"]
# {'name': 'billing', 'version': '1.4.0'}