API reference

Generated from the source docstrings.

Resources

asgimachine.resource.Resource

Base class for graph-lane endpoints. Override only what you care about.

Generic over its context type C: subclass Ctx for typed per-request state and declare it via context_class (and Resource[MyCtx] for the checker). Plain Resource uses base Ctx.

Source code in src/asgimachine/resource.py

class Resource[C: Ctx = Ctx]:
    """Base class for graph-lane endpoints. Override only what you care about.

    Generic over its context type ``C``: subclass ``Ctx`` for typed per-request
    state and declare it via ``context_class`` (and ``Resource[MyCtx]`` for the
    checker). Plain ``Resource`` uses base ``Ctx``.
    """

    # The Ctx (sub)class the core constructs for each request. Declare a subclass
    # alongside ``Resource[MyCtx]`` when a resource needs typed per-request state.
    context_class: ClassVar[type[Ctx]] = Ctx

    async def lifespan(self, ctx: C) -> AsyncGenerator[None]:
        """Per-request setup/teardown, wrapping the whole graph walk.

        Override as a *plain async generator* — acquire in the setup half,
        ``yield`` exactly once, release after. **No decorator**: the core wraps
        this in an async context manager itself, so a forgotten
        ``@asynccontextmanager`` can't bite. (Forgetting the ``yield`` instead is
        a type error, since the declared return is ``AsyncIterator[None]``.)

        The core opens this before B13 and closes it on the way out, guaranteed
        across every exit — a normal response, a halt (404/401/…), a raised
        error, or a client disconnect. The teardown is cancellation-shielded, so
        a disconnect cannot interrupt resource release; and for a *streaming*
        response it is deferred until the body is fully drained, so a connection
        stashed on ``ctx`` stays alive for the life of the stream. An in-flight
        exception is fed into the generator (so ``async with conn.transaction()``
        rolls back), but the lifespan must not *suppress* it::

            async def lifespan(self, ctx: MyCtx) -> AsyncIterator[None]:
                async with self._pool.acquire() as conn:  # released on any exit
                    ctx.conn = conn
                    yield
        """
        yield

    KNOWN_METHODS = frozenset(
        {"GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"},
    )

    # B10 -> 405 + Allow. The set of methods this resource supports. Static shape,
    # not per-request behavior: 405 is a property of the target resource (RFC 9110
    # §15.5.6), while per-principal gating belongs in `forbidden` (403). Declare
    # it on the class (or per-instance in __init__); it is the schema anchor.
    ALLOWED_METHODS: frozenset[str] = frozenset({"GET", "HEAD"})

    async def allowed_methods(self, ctx: C) -> frozenset[str]:
        # The graph reads the method set through this callback, defaulting to the
        # ALLOWED_METHODS declaration. Override only for genuine per-request
        # variation; the schema still documents ALLOWED_METHODS.
        return self.ALLOWED_METHODS

    # --- B13 ---------------------------------------------------------------
    async def service_available(self, ctx: C) -> bool | RetryHint:
        # True = available. False = 503 with no hint. An int (delta-seconds) or a
        # datetime (an HTTP-date) = 503 carrying a ``Retry-After`` header (RFC 9110
        # §10.2.3) — e.g. return 30 during a brief maintenance window. This is
        # service-*wide* backpressure; a per-client quota is within_rate_limit below.
        return True

    # --- B13a --------------------------------------------------------------
    async def within_rate_limit(self, ctx: C) -> bool | RetryHint:
        # True = within limit, proceed. False = 429 (Too Many Requests, RFC 6585)
        # with no hint. An int (delta-seconds) or datetime (an HTTP-date) = 429
        # carrying a ``Retry-After`` header. Runs right after service_available and
        # before method/auth/body checks, so an over-limit request is shed at the
        # cheapest point — ideal for throttling a login against credential-stuffing.
        # Key the limiter on ctx.request (IP, header) — the principal isn't known
        # yet. 429 postdates webmachine v3; this is an additive node, recorded only
        # when it fires. Default True (no limit). Contrast service_available (503,
        # everyone) with this (429, this client).
        return True

    # --- B11 ---------------------------------------------------------------
    async def uri_too_long(self, ctx: C) -> bool:  # -> 414
        # True when the request target is longer than this resource will serve.
        # Default False — most deployments cap URI length at the server/proxy
        # first, so this rarely fires; recorded in the trace only when it does.
        return False

    # --- B8 ----------------------------------------------------------------
    async def is_authorized(self, ctx: C) -> bool | str:
        # True = authorized; False = 401 (no challenge); str = 401 with that
        # WWW-Authenticate challenge value.
        return True

    # --- B7 ----------------------------------------------------------------
    async def forbidden(self, ctx: C) -> bool:
        return False

    # --- B7a (v4, httpdd) --------------------------------------------------
    async def is_legally_restricted(self, ctx: C) -> bool:  # -> 451 (RFC 7725)
        # True when the resource is denied for legal reasons (e.g. a takedown).
        return False

    # --- G7 ----------------------------------------------------------------
    async def resource_exists(self, ctx: C) -> bool:
        return True

    # --- conditional GET (G8-L17 subset) -----------------------------------
    async def generate_etag(self, ctx: C) -> str | None:
        return None

    async def last_modified(self, ctx: C) -> datetime | None:
        return None

    # --- precondition-required (428, RFC 6585) -----------------------------
    async def require_conditional_write(self, ctx: C) -> bool:
        # True demands that a PUT/PATCH/DELETE carry an update precondition
        # (``If-Match`` or ``If-Unmodified-Since``); an unconditional write is a
        # 428, so a client can't blindly overwrite state it hasn't seen — the
        # "lost update" guard. Default False. A *present* precondition still flows
        # through the normal 412 path; 428 fires only when none was sent.
        return False

    # O18 -> 300. When the resource offers several representations and wants the
    # client to choose, return 300 with the list (from PRODUCES). Default: pick
    # one via negotiation and return 200.
    async def multiple_choices(self, ctx: C) -> bool:
        return False

    # --- caching (v3): emitted on cacheable responses (200/304) ------------
    async def expires(self, ctx: C) -> datetime | None:
        # -> Expires header.
        return None

    async def cache_control(self, ctx: C) -> str | None:
        # -> Cache-Control header, e.g. "public, max-age=31536000, immutable"
        # for an archived, immutable feed page.
        return None

    # C3/C4 content negotiation. PRODUCES lists the offered media types in
    # preference order (declared-first wins ties); a codec encodes the single
    # representation built by represent(). Set on the class or per-instance.
    PRODUCES: tuple[str, ...] = ("application/json",)

    # C4a -> serve default instead of 406 (v4, RFC 9110 §12.1): when an Accept
    # can't be satisfied, a resource that declares this True disregards Accept and
    # serves PRODUCES[0] rather than 406. Static shape, read through the thin
    # callback below (§2.7); the schema drops 406 from the surface when it is set.
    IGNORE_UNACCEPTABLE: ClassVar[bool] = False

    async def ignore_unacceptable(self, ctx: C) -> bool:
        return self.IGNORE_UNACCEPTABLE

    async def represent(self, ctx: C) -> Any:
        # The representation value (a domain model / dict / etc.), encoded by the
        # negotiated codec. The typed return is the response model for schema.
        raise NotImplementedError(
            f"{type(self).__name__} offers {self.PRODUCES} but does not "
            "implement represent().",
        )

    async def variances(self, ctx: C) -> Sequence[str]:
        # Extra request-header names this representation varies on, emitted in
        # Vary. The core adds "Accept" automatically when more than one media
        # type is offered (and the matching Accept-* axis for each of
        # LANGUAGES/ENCODINGS offered), so only list additional axes.
        return []

    # --- D/F proactive negotiation (v3) ------------------------------------
    # Language / content-coding, each parallel to PRODUCES: declare the offered
    # values in preference order (first wins ties) and the core negotiates against
    # the matching Accept-* header. Empty (the default) means the axis is not
    # negotiated — the header is ignored, no Vary axis, no 406. When offered, an
    # unsatisfiable Accept-* is a 406 (unless ``ignore_unacceptable`` serves the
    # default instead), and the choice is exposed on ctx for ``represent``.
    #
    # (Charset — webmachine's E nodes — is deliberately absent: RFC 9110 §12.5.2
    # deprecates ``Accept-Charset``. Charset belongs on the Content-Type parameter.)
    #
    # asgimachine negotiates and advertises; it does not compress or transcode
    # (that is the substrate's / your representation's job — "rent Layer 2"). Read
    # ``ctx.chosen_language`` to pick a translation and ``ctx.chosen_encoding`` to
    # produce bytes in that content-coding and set ``Content-Encoding``.

    # D4/D5 -> 406. Offered language tags; matched RFC 4647 lookup-style (a request
    # for ``en-US`` is served by an offered ``en``, and vice versa). Sets
    # Content-Language.
    LANGUAGES: tuple[str, ...] = ()

    async def languages(self, ctx: C) -> Sequence[str]:
        return self.LANGUAGES

    # F6/F7 -> 406. Offered content-codings; ``identity`` is acceptable by default
    # unless the client refuses it (RFC 9110 §12.5.3).
    ENCODINGS: tuple[str, ...] = ()

    async def encodings(self, ctx: C) -> Sequence[str]:
        return self.ENCODINGS

    # --- error bodies (§4 v4, RFC 9457) ------------------------------------
    # Media types offered for *error* bodies (4xx/5xx), negotiated against Accept
    # separately from PRODUCES — the main negotiation may have failed (406) or not
    # run yet (401 before C4). Declare more (e.g. add "text/html") + a codec to
    # serve browsers HTML errors; an unmatched Accept falls back to the first.
    ERROR_PRODUCES: ClassVar[tuple[str, ...]] = ("application/problem+json",)

    async def error_body(self, ctx: C, status: int, media_type: str) -> Any | None:
        # The body for a 4xx/5xx response, encoded as ``media_type`` (the
        # negotiated error representation). Default: an RFC 9457 problem detail.
        # Override to add ``detail``/``instance``/custom members, render per
        # ``media_type``, or return None for an empty body.
        return {"type": "about:blank", "title": _reason(status), "status": status}

    # --- unexpected exceptions ---------------------------------------------
    async def on_exception(self, ctx: C, exc: Exception) -> HttpResponse | None:
        # Catch-all for an *unexpected* exception raised during the walk. Only
        # ``Exception`` reaches here — a client disconnect (``CancelledError``) and
        # other ``BaseException``s always propagate (teardown + re-raise). Runs
        # inside the walk with ``ctx`` in scope, so it is where you report the error
        # and record its id onto ``ctx`` before the response is built.
        #
        # Default: **re-raise**, so the exception propagates to the substrate's
        # outer handler (Starlette's ServerErrorMiddleware, an ASGI error reporter,
        # ...) — behavior is unchanged unless you opt in. Return instead to have the
        # graph own the 500: ``None`` -> the standard negotiated ``problem+json``
        # body (via ``error_body``); an ``HttpResponse`` -> that response; or raise
        # ``HaltResponse(...)`` for full control. Set a default for the whole app at
        # ``build_app(on_exception=...)``; override here for one resource.
        raise exc

    # --- write path (§4 v2) -----------------------------------------------
    # Body-validation nodes. Traversed only for body-bearing methods
    # (POST/PUT/PATCH); each ships a correct default that passes.
    async def malformed_request(self, ctx: C) -> bool:  # B9 -> 400
        return False

    async def valid_content_headers(self, ctx: C) -> bool:  # B6 -> 501
        return True

    # B4 -> 413. The largest request body (bytes) this resource will accept.
    # A declaration, not a callback: the graph rejects a larger declared
    # Content-Length here, and the substrate caps the *actual* read at this value
    # (the backstop for a chunked or lying Content-Length). Raise it for an
    # upload resource; ``ClassVar`` so the checker forbids per-instance mutation.
    MAX_BODY_BYTES: ClassVar[int] = DEFAULT_MAX_BODY_BYTES

    async def valid_entity_length(self, ctx: C) -> bool:  # B4 -> 413
        # Reject a declared Content-Length over the limit. Absent or unparseable
        # Content-Length falls through to the substrate's read cap.
        declared = ctx.request.headers.get("content-length")
        if declared is None:
            return True
        try:
            return int(declared) <= self.MAX_BODY_BYTES
        except ValueError:
            return True

    # The mirror of PRODUCES on the write side: request Content-Types this
    # resource accepts. Empty by default (a read-only resource declares none).
    CONSUMES: ClassVar[tuple[str, ...]] = ()

    async def apply(self, ctx: C, body: Any) -> Any:
        # The write handler for PUT/PATCH/POST-create. The core decodes the
        # request via the negotiated codec and parses it into ``body``'s declared
        # type before calling this — annotate ``body: NoteInput`` (a Pydantic
        # model) and a bad body is a 400 (parse, don't validate). Annotate it
        # loosely (``dict``/``object``) to receive the decoded structure as-is.
        # Its return is the response representation.
        raise NotImplementedError(
            f"{type(self).__name__} accepts writes but does not implement apply().",
        )

    async def known_content_type(self, ctx: C) -> bool:
        # B5 -> 415. Default: if the resource declares CONSUMES, the request's
        # Content-Type must be one of them; otherwise anything is accepted.
        if not self.CONSUMES:
            return True
        media = parse_content_type(ctx.request.headers.get("content-type"))
        return media is not None and media in self.CONSUMES

    async def is_conflict(self, ctx: C) -> bool:
        # O14 -> 409. e.g. a PUT that would violate an invariant.
        return False

    # DELETE (M20/M16)
    async def delete_resource(self, ctx: C) -> bool:
        # Perform the delete; return True once enacted. Resources that allow
        # DELETE must implement this.
        raise NotImplementedError(
            f"{type(self).__name__} allows DELETE but does not implement "
            "delete_resource().",
        )

    async def delete_completed(self, ctx: C) -> bool:
        # M20 -> True = fully done (204); False = accepted for later (202).
        return True

    # POST (N11)
    async def post_is_create(self, ctx: C) -> bool:
        # True -> POST creates a new resource at create_path (201 + Location).
        # False -> POST is an action; process_post handles it.
        return False

    async def create_path(self, ctx: C) -> str:
        raise NotImplementedError(
            f"{type(self).__name__} sets post_is_create but does not implement "
            "create_path().",
        )

    async def process_post(self, ctx: C) -> Any:
        raise NotImplementedError(
            f"{type(self).__name__} handles POST but does not implement "
            "process_post() (or set post_is_create).",
        )

    async def see_other(self, ctx: C) -> str | None:  # N11 -> 303
        # After the POST's side effects run (create_path/apply or process_post),
        # return a URL to redirect the client to with 303 See Other — the
        # POST-Redirect-Get pattern. None (the default) responds 201 (create) /
        # 200 (action) normally. The URL overrides the created/action response.
        return None

    async def accepted(self, ctx: C) -> str | None:  # O20a -> 202
        # After a write handler *enqueues* work it can't finish inside the request
        # budget, return a URL to a status-monitor resource: the graph responds 202
        # Accepted + Location instead of framing a completed 200/201/204 — the
        # asynchronous request-reply pattern (hand off to a background task, let the
        # client poll the monitor). None (the default) frames the result normally.
        # Applies to POST/PUT/PATCH; DELETE has its own 202 via delete_completed.
        # Mutually exclusive with see_other (which, on POST, is checked first).
        return None

    # --- G7-false branch: the resource does not (currently) exist ----------
    async def previously_existed(self, ctx: C) -> bool:  # K7
        # True routes a missing resource to redirect/gone handling.
        return False

    async def moved_permanently(self, ctx: C) -> str | None:  # K5 -> 301
        return None

    async def permanent_redirect(self, ctx: C) -> str | None:  # K5a -> 308 (v4)
        # RFC 7538: a permanent redirect that *preserves the method* (301 does
        # not). Prefer this over moved_permanently for non-idempotent targets.
        return None

    async def moved_temporarily(self, ctx: C) -> str | None:  # L5 -> 307
        return None

    # --- schema (§10) ------------------------------------------------------
    def describe(self) -> ResourceDescription | None:
        # Opt in to OpenAPI generation by returning a ResourceDescription.
        # None (the default) leaves this route out of any generated schema.
        return None

lifespan async

lifespan(ctx)

Per-request setup/teardown, wrapping the whole graph walk.

Override as a plain async generator — acquire in the setup half, yield exactly once, release after. No decorator: the core wraps this in an async context manager itself, so a forgotten @asynccontextmanager can't bite. (Forgetting the yield instead is a type error, since the declared return is AsyncIterator[None].)

The core opens this before B13 and closes it on the way out, guaranteed across every exit — a normal response, a halt (404/401/…), a raised error, or a client disconnect. The teardown is cancellation-shielded, so a disconnect cannot interrupt resource release; and for a streaming response it is deferred until the body is fully drained, so a connection stashed on ctx stays alive for the life of the stream. An in-flight exception is fed into the generator (so async with conn.transaction() rolls back), but the lifespan must not suppress it::

async def lifespan(self, ctx: MyCtx) -> AsyncIterator[None]:
    async with self._pool.acquire() as conn:  # released on any exit
        ctx.conn = conn
        yield

Source code in src/asgimachine/resource.py

async def lifespan(self, ctx: C) -> AsyncGenerator[None]:
    """Per-request setup/teardown, wrapping the whole graph walk.

    Override as a *plain async generator* — acquire in the setup half,
    ``yield`` exactly once, release after. **No decorator**: the core wraps
    this in an async context manager itself, so a forgotten
    ``@asynccontextmanager`` can't bite. (Forgetting the ``yield`` instead is
    a type error, since the declared return is ``AsyncIterator[None]``.)

    The core opens this before B13 and closes it on the way out, guaranteed
    across every exit — a normal response, a halt (404/401/…), a raised
    error, or a client disconnect. The teardown is cancellation-shielded, so
    a disconnect cannot interrupt resource release; and for a *streaming*
    response it is deferred until the body is fully drained, so a connection
    stashed on ``ctx`` stays alive for the life of the stream. An in-flight
    exception is fed into the generator (so ``async with conn.transaction()``
    rolls back), but the lifespan must not *suppress* it::

        async def lifespan(self, ctx: MyCtx) -> AsyncIterator[None]:
            async with self._pool.acquire() as conn:  # released on any exit
                ctx.conn = conn
                yield
    """
    yield

asgimachine.resource.Ctx dataclass

webmachine's ReqData: the framework's per-request state.

Deliberately minimal and domain-agnostic — it holds the request, the trace, negotiation result, and codec registry, plus an untyped extra bag. Domain state (a principal, the loaded entity) is not the framework's business: a resource that wants typed per-request state subclasses Ctx and declares it (see Resource.context_class and PLAN.md §2.7).

Source code in src/asgimachine/resource.py

@dataclass(slots=True)
class Ctx:
    """webmachine's ReqData: the *framework's* per-request state.

    Deliberately minimal and domain-agnostic — it holds the request, the trace,
    negotiation result, and codec registry, plus an untyped ``extra`` bag. Domain
    state (a principal, the loaded entity) is not the framework's business: a
    resource that wants typed per-request state subclasses ``Ctx`` and declares
    it (see ``Resource.context_class`` and PLAN.md §2.7).
    """

    request: HttpRequest
    trace: Trace = field(default_factory=Trace)
    chosen_media_type: str | None = None
    # The proactively-negotiated variant axes (D/F). None when the resource
    # doesn't offer that axis; a representation reads them to serve the right
    # translation / content-coding.
    chosen_language: str | None = None
    chosen_encoding: str | None = None
    allowed_methods: frozenset[str] = field(default_factory=frozenset[str])
    extra: dict[str, Any] = field(default_factory=dict[str, Any])
    # The per-request wide event (§ observability). Callbacks and instrumented code
    # enrich it in place; the core fills owned fields and emits it once at the
    # request boundary. Always present, so writing to it is free; emitted only when
    # an EventSink is configured.
    event: Event = field(default_factory=dict[str, object])
    # Framework config: the media-type -> Codec registry for this request.
    codecs: dict[str, Codec] = field(default_factory=dict[str, "Codec"])

The core walk

asgimachine.core.run async

run(
    resource,
    request,
    *,
    debug=False,
    codecs=None,
    on_exception=None,
    event_sink=None,
)

Walk the graph for one request and return an HttpResponse value object.

In debug mode the ordered node path is attached as the X-Asgimachine-Trace response header on every exit path (§9). codecs is the media-type -> Codec registry (defaults to JSON only). on_exception is the app-level catch-all for an unexpected Exception (a resource may override it); it defaults to re-raising, so the exception propagates to the substrate's outer handler unless a handler opts to own the 500. event_sink, when given, receives one wide event per request (ctx.event), emitted once at the boundary — after lifespan teardown, or at stream-close for a streamed body.

Source code in src/asgimachine/core.py

async def run(
    resource: Resource[Any],
    request: HttpRequest,
    *,
    debug: bool = False,
    codecs: dict[str, Codec] | None = None,
    on_exception: OnException | None = None,
    event_sink: EventSink | None = None,
) -> HttpResponse:
    """Walk the graph for one request and return an HttpResponse value object.

    In ``debug`` mode the ordered node path is attached as the
    ``X-Asgimachine-Trace`` response header on every exit path (§9). ``codecs``
    is the media-type -> Codec registry (defaults to JSON only). ``on_exception``
    is the app-level catch-all for an unexpected ``Exception`` (a resource may
    override it); it defaults to re-raising, so the exception propagates to the
    substrate's outer handler unless a handler opts to own the 500. ``event_sink``,
    when given, receives one wide event per request (``ctx.event``), emitted once at
    the boundary — after lifespan teardown, or at stream-close for a streamed body.
    """

    started = perf_counter()
    ctx = resource.context_class(
        # Copy per request: an injected registry is shared by reference across
        # every request through the endpoint, so ctx must get its own dict (as the
        # default branch already does) — else one request mutating ctx.codecs
        # would corrupt others in flight.
        request=request,
        codecs=dict(codecs) if codecs is not None else dict(DEFAULT_CODECS),
    )
    # The resource's per-request lifespan wraps the entire walk. It is a plain
    # async generator (no @asynccontextmanager on the override — §5); the core
    # owns the wrapping so it can also own guaranteed, cancellation-safe teardown.
    lifespan = asynccontextmanager(resource.lifespan)(ctx)
    await lifespan.__aenter__()
    streaming = False
    torn_down = False
    # The unexpected exception, once handled into a response — fed to teardown so a
    # lifespan transaction still rolls back even though we return a 500, not raise.
    handled_exc: Exception | None = None
    # Bound for the finally (which runs even if the walk raised before assigning).
    final_response: HttpResponse | None = None
    try:
        try:
            response = await _walk(resource, ctx)
        except HaltResponse as halt:
            response = halt.response
            await _apply_error_body(resource, ctx, response)
        except Exception as exc:  # noqa: BLE001 — the graph-owned catch-all
            # Exception only — a disconnect/cancellation is a BaseException and falls
            # through to the teardown-and-reraise arm below. The handler may return a
            # response, or re-raise (propagating to the arm below).
            handled_exc = exc
            response = await _handle_exception(resource, ctx, exc, on_exception)
        final_response = response
        if debug:
            response.headers[TRACE_HEADER] = ctx.trace.header_value
        if response.is_stream:
            # The streamed body outlives this call; hand teardown to the wrapper,
            # which releases when the body drains, errors, or is closed — even if
            # the substrate never starts iterating it (a pre-first-chunk
            # disconnect). Ownership transfers, so this scope must not release. The
            # wide event is emitted at that same close, so late work (a DB merge in
            # the lifespan teardown) and the final duration land in it.
            stream = cast("AsyncIterator[bytes]", response.body)
            streamed = response

            def _on_close(close_exc: BaseException | None) -> None:
                stream_exc = close_exc if isinstance(close_exc, Exception) else None
                _record_and_emit(
                    event_sink, ctx, resource, streamed, stream_exc, started
                )

            response.body = _ClosingStream(stream, lifespan, _on_close)
            streaming = True
        return response
    except BaseException as exc:
        # A propagated error (the handler re-raised) or a cancellation (client
        # disconnect): tear down with the exception in flight so a lifespan
        # transaction rolls back, emit the (statusless) event, then re-raise.
        torn_down = True
        await _teardown(lifespan, exc)
        _record_and_emit(event_sink, ctx, resource, None, exc, started)
        raise
    finally:
        # Non-streaming success/halt/handled-500: release here, exactly once — in a
        # finally so a throw anywhere above can't skip it. A handled exception is
        # fed in so the transaction still rolls back (the contextmanager absorbs the
        # expected re-raise). Then emit, after teardown. The streaming path
        # transferred ownership; the propagate path already released and emitted.
        if not streaming and not torn_down:
            await _teardown(lifespan, handled_exc)
            _record_and_emit(
                event_sink, ctx, resource, final_response, handled_exc, started
            )

asgimachine.http.HttpRequest

Bases: Protocol

The read-only view of a request the core depends on.

Case-insensitive header access is required of headers (per RFC 9110); the Starlette adapter satisfies this. Kept minimal on purpose — the core only reads what the v0 graph needs.

Source code in src/asgimachine/http.py

@runtime_checkable
class HttpRequest(Protocol):
    """The read-only view of a request the core depends on.

    Case-insensitive header access is required of ``headers`` (per RFC 9110);
    the Starlette adapter satisfies this. Kept minimal on purpose — the core
    only reads what the v0 graph needs.
    """

    @property
    def method(self) -> str: ...

    @property
    def path(self) -> str: ...

    @property
    def headers(self) -> Mapping[str, str]: ...

    @property
    def path_params(self) -> Mapping[str, str]: ...

    @property
    def query_params(self) -> Mapping[str, str]:
        """The parsed query string (``?limit=20`` -> ``{"limit": "20"}``).

        A repeated key (``?tag=a&tag=b``) yields its *last* value through this
        mapping; the substrate's concrete type may offer ``getlist`` for all of
        them. Empty when there is no query string."""
        ...

    @property
    def route(self) -> str | None:
        """The matched route *template* (e.g. ``/notes/{id}``), or None if unrouted.

        The low-cardinality grouping key for the wide event (``http.route``), as
        distinct from ``path`` (the concrete ``/notes/42``)."""
        ...

    async def body(self) -> bytes:
        """The buffered request body.

        May raise :class:`BodyTooLarge` (the substrate bounds the read at the
        resource's limit → 413) or :class:`BodyMalformed` (the bytes read
        disagree with ``Content-Length`` → 400). The graph and command lanes map
        these to their statuses.
        """
        ...

query_params property

query_params

The parsed query string (?limit=20 -> {"limit": "20"}).

A repeated key (?tag=a&tag=b) yields its last value through this mapping; the substrate's concrete type may offer getlist for all of them. Empty when there is no query string.

route property

route

The matched route template (e.g. /notes/{id}), or None if unrouted.

The low-cardinality grouping key for the wide event (http.route), as distinct from path (the concrete /notes/42).

body async

body()

The buffered request body.

May raise :class:BodyTooLarge (the substrate bounds the read at the resource's limit → 413) or :class:BodyMalformed (the bytes read disagree with Content-Length → 400). The graph and command lanes map these to their statuses.

Source code in src/asgimachine/http.py

async def body(self) -> bytes:
    """The buffered request body.

    May raise :class:`BodyTooLarge` (the substrate bounds the read at the
    resource's limit → 413) or :class:`BodyMalformed` (the bytes read
    disagree with ``Content-Length`` → 400). The graph and command lanes map
    these to their statuses.
    """
    ...

asgimachine.http.HttpResponse dataclass

What the core walk returns; the substrate turns it into a wire response.

Source code in src/asgimachine/http.py

@dataclass(slots=True)
class HttpResponse:
    """What the core walk returns; the substrate turns it into a wire response."""

    status: int
    headers: dict[str, str] = field(default_factory=dict[str, str])
    body: Body = b""

    def __post_init__(self) -> None:
        # Poka-yoke: a malformed header value is un-constructable. Every response
        # from both lanes flows through here, so no header reaches the wire without
        # conforming to the field-value grammar (guards Location/ETag/… built from
        # client-influenced data — e.g. a CRLF-bearing path param).
        for name, value in self.headers.items():
            try:
                _FIELD_VALUE.parse_all(value)
            except ParseError:
                raise MalformedHeader(name) from None

    @property
    def is_stream(self) -> bool:
        return not isinstance(self.body, (bytes, bytearray))

asgimachine.http.Status

Bases: IntEnum

HTTP status codes the v0 graph can select. Extended in later phases.

Source code in src/asgimachine/http.py

class Status(IntEnum):
    """HTTP status codes the v0 graph can select. Extended in later phases."""

    OK = 200
    CREATED = 201
    ACCEPTED = 202
    NO_CONTENT = 204
    MULTIPLE_CHOICES = 300
    MOVED_PERMANENTLY = 301
    SEE_OTHER = 303  # N11: redirect after a POST's side effects (PRG)
    TEMPORARY_REDIRECT = 307
    PERMANENT_REDIRECT = 308  # RFC 7538: like 301 but method-preserving
    NOT_MODIFIED = 304
    BAD_REQUEST = 400
    UNAUTHORIZED = 401
    FORBIDDEN = 403
    NOT_FOUND = 404
    METHOD_NOT_ALLOWED = 405
    NOT_ACCEPTABLE = 406
    CONFLICT = 409
    GONE = 410
    PRECONDITION_FAILED = 412
    REQUEST_ENTITY_TOO_LARGE = 413
    URI_TOO_LONG = 414  # B11: request target too long
    UNSUPPORTED_MEDIA_TYPE = 415
    PRECONDITION_REQUIRED = 428  # RFC 6585: a write must be conditional
    TOO_MANY_REQUESTS = 429  # RFC 6585: B13a rate limit exceeded
    UNAVAILABLE_FOR_LEGAL_REASONS = 451  # RFC 7725
    INTERNAL_SERVER_ERROR = 500
    NOT_IMPLEMENTED = 501
    SERVICE_UNAVAILABLE = 503

asgimachine.http.HaltResponse

Bases: Exception

Short-circuit the walk with an explicit response (PLAN.md §6).

Any callback or node may raise this to terminate the graph immediately.

Source code in src/asgimachine/http.py

class HaltResponse(Exception):
    """Short-circuit the walk with an explicit response (PLAN.md §6).

    Any callback or node may raise this to terminate the graph immediately.
    """

    def __init__(self, response: HttpResponse) -> None:
        self.response = response
        super().__init__(f"halt {response.status}")

Commands

asgimachine.command.Command

Base class for the plain-handler lane. Override :meth:handle.

Source code in src/asgimachine/command.py

class Command:
    """Base class for the plain-handler lane. Override :meth:`handle`."""

    async def handle(self, request: HttpRequest) -> HttpResponse:
        raise NotImplementedError

asgimachine.command.json_response

json_response(value, *, status=Status.OK, headers=None)

Build a JSON :class:HttpResponse — the ergonomic return for a command.

Source code in src/asgimachine/command.py

def json_response(
    value: object,
    *,
    status: Status | int = Status.OK,
    headers: dict[str, str] | None = None,
) -> HttpResponse:
    """Build a JSON :class:`HttpResponse` — the ergonomic return for a command."""

    return HttpResponse(
        status=int(status),
        headers={"Content-Type": "application/json", **(headers or {})},
        body=serialize(value),
    )

Authorization header parsing

asgimachine.auth.parse_authorization

parse_authorization(header)

Split an Authorization value into (scheme, credentials).

The scheme must be a valid token (a non-token is rejected) and is lowercased (it is case-insensitive, RFC 9110 §11.1). The credentials are everything after the first run of whitespace, verbatim (a scheme-only header yields "") — their grammar is scheme-specific, so they are not validated here; that is the job of the scheme helpers below. Returns None for a missing/blank header or a malformed scheme.

Source code in src/asgimachine/auth.py

def parse_authorization(header: str | None) -> tuple[str, str] | None:
    """Split an ``Authorization`` value into ``(scheme, credentials)``.

    The scheme must be a valid ``token`` (a non-token is rejected) and is lowercased
    (it is case-insensitive, RFC 9110 §11.1). The credentials are everything after
    the first run of whitespace, verbatim (a scheme-only header yields ``""``) —
    their grammar is scheme-specific, so they are *not* validated here; that is the
    job of the scheme helpers below. Returns ``None`` for a missing/blank header or a
    malformed scheme.
    """

    if header is None:
        return None
    header = header.strip()
    if not header:
        return None
    parts = header.split(None, 1)
    scheme = parts[0]
    if not _matches(_TOKEN, scheme):
        return None
    credentials = parts[1].strip() if len(parts) > 1 else ""
    return (scheme.lower(), credentials)

asgimachine.auth.bearer_token

bearer_token(header)

The token from an Authorization: Bearer <token> header, or None when the header is absent, uses a different scheme, or the token is not a valid token68 (RFC 6750).

Source code in src/asgimachine/auth.py

def bearer_token(header: str | None) -> str | None:
    """The token from an ``Authorization: Bearer <token>`` header, or ``None`` when
    the header is absent, uses a different scheme, or the token is not a valid
    ``token68`` (RFC 6750)."""

    parsed = parse_authorization(header)
    if parsed is None or parsed[0] != "bearer":
        return None
    token = parsed[1]
    return token if _matches(_TOKEN68, token) else None

asgimachine.auth.basic_credentials

basic_credentials(header)

The (user_id, password) from an Authorization: Basic … header (RFC 7617), or None when the header is absent, uses a different scheme, or is malformed. The password may contain colons; the user id may not (the split is on the first colon).

Source code in src/asgimachine/auth.py

def basic_credentials(header: str | None) -> tuple[str, str] | None:
    """The ``(user_id, password)`` from an ``Authorization: Basic …`` header (RFC
    7617), or ``None`` when the header is absent, uses a different scheme, or is
    malformed. The password may contain colons; the user id may not (the split is on
    the first colon)."""

    parsed = parse_authorization(header)
    if parsed is None or parsed[0] != "basic":
        return None
    try:
        decoded = base64.b64decode(parsed[1], validate=True).decode("utf-8")
    except (ValueError, UnicodeDecodeError):  # bad base64 / not UTF-8
        return None
    user_id, sep, password = decoded.partition(":")
    if not sep:  # RFC 7617 requires the colon
        return None
    return (user_id, password)

Authorization policy

asgimachine.policy.RuleEngine

Ordered Allow/Deny rule engine; first matching rule wins (§7).

Generic over the resource's context type C so rules see typed ctx. default decides when no rule fires (deny-by-default is safe).

Source code in src/asgimachine/policy.py

class RuleEngine[C: Ctx = Ctx]:
    """Ordered Allow/Deny rule engine; first matching rule wins (§7).

    Generic over the resource's context type ``C`` so rules see typed
    ``ctx``. ``default`` decides when no rule fires (deny-by-default is safe).
    """

    __slots__ = ("_default", "_rules")

    def __init__(
        self, rules: Sequence[NamedRule[C]], *, default: Effect = Effect.DENY
    ) -> None:
        self._rules = list(rules)
        self._default = default

    async def evaluate(self, ctx: C) -> Decision:
        for rule in self._rules:
            effect = await rule.check(ctx)
            if effect is not None:
                ctx.trace.record(f"policy:{rule.name}", effect.value)
                return Decision(allowed=effect is Effect.ALLOW, reason=rule.name)
        ctx.trace.record("policy:default", self._default.value)
        return Decision(allowed=self._default is Effect.ALLOW, reason="default")

asgimachine.policy.NamedRule dataclass

Source code in src/asgimachine/policy.py

@dataclass(frozen=True, slots=True)
class NamedRule[C: Ctx]:
    name: str
    check: Callable[[C], Awaitable[Effect | None]]

asgimachine.policy.Effect

Bases: Enum

Source code in src/asgimachine/policy.py

class Effect(Enum):
    ALLOW = "allow"
    DENY = "deny"

asgimachine.policy.Decision dataclass

The outcome of a policy evaluation; reason names the deciding rule.

Source code in src/asgimachine/policy.py

@dataclass(frozen=True, slots=True)
class Decision:
    """The outcome of a policy evaluation; ``reason`` names the deciding rule."""

    allowed: bool
    reason: str

Codecs

asgimachine.codec.Codec

Bases: Protocol

Encodes a representation value to bytes and decodes a body to structure.

Source code in src/asgimachine/codec.py

@runtime_checkable
class Codec(Protocol):
    """Encodes a representation value to bytes and decodes a body to structure."""

    def encode(self, value: object) -> bytes: ...

    def decode(self, raw: bytes) -> object: ...

asgimachine.codec.JsonCodec

application/json codec. Encode reuses :func:http.serialize (Pydantic model_dump_json or json.dumps); decode is a syntactic parse.

Source code in src/asgimachine/codec.py

class JsonCodec:
    """``application/json`` codec. Encode reuses :func:`http.serialize` (Pydantic
    ``model_dump_json`` or ``json.dumps``); decode is a syntactic parse."""

    def encode(self, value: object) -> bytes:
        return serialize(value)

    def decode(self, raw: bytes) -> object:
        return json.loads(raw)

Schema generation

asgimachine.schema.generate_openapi

generate_openapi(
    *,
    title,
    version,
    routes,
    security_schemes=None,
    security=None,
)

Emit an OpenAPI 3.1 document for the resources that declare a describe().

security_schemes populates components.securitySchemes; security is the document-level default requirement (scheme names), overridable per operation via Operation.security.

Source code in src/asgimachine/schema.py

def generate_openapi(
    *,
    title: str,
    version: str,
    routes: Sequence[tuple[str, Resource[Any]]],
    security_schemes: Mapping[str, dict[str, Any]] | None = None,
    security: Sequence[str] | None = None,
) -> dict[str, Any]:
    """Emit an OpenAPI 3.1 document for the resources that declare a ``describe()``.

    ``security_schemes`` populates ``components.securitySchemes``; ``security`` is
    the document-level default requirement (scheme names), overridable per
    operation via ``Operation.security``.
    """

    paths: dict[str, Any] = {}
    schemas: dict[str, Any] = {}  # models hoisted here, deduped, referenced by $ref
    for path, resource in routes:
        description = resource.describe()
        if description is None:
            continue
        params = _parameters(path)
        declared_ops: dict[str, Operation | None] = {
            "get": description.get,
            "post": description.post,
            "put": description.put,
            "patch": description.patch,
            "delete": description.delete,
        }
        path_item: dict[str, Any] = {}
        for method in _DOCUMENTABLE:
            if method.upper() not in resource.ALLOWED_METHODS:
                continue
            path_item[method] = _operation(
                method, declared_ops[method], resource, params, schemas
            )
        if path_item:
            paths[path] = path_item

    document: dict[str, Any] = {
        "openapi": "3.1.0",
        "info": {"title": title, "version": version},
        "paths": paths,
    }
    if security is not None:
        document["security"] = [{name: []} for name in security]
    components: dict[str, Any] = {}
    if schemas:
        components["schemas"] = schemas
    if security_schemes is not None:
        components["securitySchemes"] = dict(security_schemes)
    if components:
        document["components"] = components
    return document

asgimachine.schema.Operation dataclass

One HTTP method's declared surface. Errors are auto-derived; declare the success bodies. security: None inherits the document default, [] marks the operation public, ["name"] requires those schemes.

Source code in src/asgimachine/schema.py

@dataclass(frozen=True, slots=True)
class Operation:
    """One HTTP method's declared surface. Errors are auto-derived; declare the
    success bodies. ``security``: ``None`` inherits the document default, ``[]``
    marks the operation public, ``["name"]`` requires those schemes."""

    summary: str | None = None
    request: Model | None = None
    responses: Mapping[int, Model | None] | None = None
    security: Sequence[str] | None = None

asgimachine.schema.ResourceDescription dataclass

A resource's per-method declarations (bodies + summaries). Methods absent here but present in ALLOWED_METHODS are still documented (auto-errors only).

Source code in src/asgimachine/schema.py

@dataclass(frozen=True, slots=True)
class ResourceDescription:
    """A resource's per-method declarations (bodies + summaries). Methods absent
    here but present in ``ALLOWED_METHODS`` are still documented (auto-errors
    only)."""

    get: Operation | None = None
    post: Operation | None = None
    put: Operation | None = None
    patch: Operation | None = None
    delete: Operation | None = None

Decision trace

asgimachine.trace.Trace dataclass

Source code in src/asgimachine/trace.py

@dataclass(slots=True)
class Trace:
    entries: list[TraceEntry] = field(default_factory=list[TraceEntry])

    def record(self, node: str, outcome: object) -> None:
        self.entries.append(TraceEntry(node, outcome))

    @property
    def nodes(self) -> list[str]:
        return [e.node for e in self.entries]

    @property
    def header_value(self) -> str:
        """The ordered node path as a single header-safe token list."""

        return ",".join(self.nodes)

    def __str__(self) -> str:
        return " -> ".join(str(e) for e in self.entries)

header_value property

header_value

The ordered node path as a single header-safe token list.

asgimachine.trace.TraceEntry dataclass

Source code in src/asgimachine/trace.py

@dataclass(slots=True)
class TraceEntry:
    node: str
    outcome: object

    def __str__(self) -> str:
        return f"{self.node}={self.outcome!r}"

Wide events

asgimachine.event.EventSink

Bases: Protocol

Where a completed wide event goes. One synchronous method — sync so it is safe to call from the request's exit path even under cancellation (a disconnect); hand off to an async queue inside emit if you must.

Source code in src/asgimachine/event.py

@runtime_checkable
class EventSink(Protocol):
    """Where a completed wide event goes. One synchronous method — sync so it is
    safe to call from the request's exit path even under cancellation (a
    disconnect); hand off to an async queue inside ``emit`` if you must."""

    def emit(self, event: Mapping[str, object]) -> None: ...

asgimachine.event.LoggingEventSink

Reference :class:EventSink over the stdlib logging module.

The full event rides on the log record as record.event (via extra), so a structured/JSON handler can render every field; the formatted message is a terse METHOD path -> status summary for plain handlers. Swap in a structlog or OpenTelemetry sink at the composition root when you want those backends.

Source code in src/asgimachine/event.py

class LoggingEventSink:
    """Reference :class:`EventSink` over the stdlib ``logging`` module.

    The full event rides on the log record as ``record.event`` (via ``extra``), so
    a structured/JSON handler can render every field; the formatted message is a
    terse ``METHOD path -> status`` summary for plain handlers. Swap in a structlog
    or OpenTelemetry sink at the composition root when you want those backends.
    """

    __slots__ = ("_level", "_logger")

    def __init__(
        self, logger_name: str = "asgimachine.event", level: int = logging.INFO
    ) -> None:
        self._logger = logging.getLogger(logger_name)
        self._level = level

    def emit(self, event: Mapping[str, object]) -> None:
        self._logger.log(
            self._level,
            "%s %s -> %s",
            event.get("http.request.method", "-"),
            event.get("url.path", "-"),
            event.get("http.response.status_code", "-"),
            extra={"event": dict(event)},
        )

Streaming

asgimachine.streaming.sse_event

sse_event(
    data,
    *,
    event=None,
    event_id=None,
    retry=None,
    comment=None,
)

Format one Server-Sent Events frame as bytes.

data is emitted verbatim if it is a str, otherwise JSON-encoded. Multi-line data becomes multiple data: lines, per the SSE spec.

Source code in src/asgimachine/streaming.py

def sse_event(
    data: object,
    *,
    event: str | None = None,
    event_id: str | None = None,
    retry: int | None = None,
    comment: str | None = None,
) -> bytes:
    """Format one Server-Sent Events frame as bytes.

    ``data`` is emitted verbatim if it is a ``str``, otherwise JSON-encoded.
    Multi-line data becomes multiple ``data:`` lines, per the SSE spec.
    """

    lines: list[str] = []
    if comment is not None:
        lines.append(f": {comment}")
    if event is not None:
        lines.append(f"event: {event}")
    if event_id is not None:
        lines.append(f"id: {event_id}")
    if retry is not None:
        lines.append(f"retry: {retry}")
    text = data if isinstance(data, str) else json.dumps(data)
    lines.extend(f"data: {line}" for line in text.split("\n"))
    return ("\n".join(lines) + "\n\n").encode()

asgimachine.streaming.sse_error

sse_error(data, *, event='error')

Format an SSE frame carrying an error (the post-commit failure channel).

Source code in src/asgimachine/streaming.py

def sse_error(data: object, *, event: str = "error") -> bytes:
    """Format an SSE frame carrying an error (the post-commit failure channel)."""

    return sse_event(data, event=event)

asgimachine.streaming.guard_sse async

guard_sse(source, *, format_error=None)

Yield from source; on a post-commit exception, emit an SSE error frame.

Wrap a producer's event generator with this so a failure after the response has committed surfaces as an event: error frame instead of tearing down the connection. format_error maps the exception to the error payload.

Source code in src/asgimachine/streaming.py

async def guard_sse(
    source: AsyncIterator[bytes],
    *,
    format_error: Callable[[Exception], object] | None = None,
) -> AsyncIterator[bytes]:
    """Yield from ``source``; on a post-commit exception, emit an SSE error frame.

    Wrap a producer's event generator with this so a failure after the response
    has committed surfaces as an ``event: error`` frame instead of tearing down
    the connection. ``format_error`` maps the exception to the error payload.
    """

    try:
        async for chunk in source:
            yield chunk
    except Exception as exc:  # noqa: BLE001 — post-commit: app failures become a frame
        # Only Exception: cancellation (CancelledError/GeneratorExit on client
        # disconnect) is a BaseException and must propagate so the stream stops.
        payload = format_error(exc) if format_error is not None else "internal error"
        yield sse_error(payload)

Starlette substrate

asgimachine.substrate.starlette.build_app

build_app(
    routes,
    *,
    debug=False,
    middleware=None,
    on_exception=None,
    event_sink=None,
)

Assemble the composition root into an ASGI application.

middleware is passed straight to Starlette, so cross-cutting concerns are rented rather than baked into the graph (PLAN.md §2.1). CORS in particular is its own decision machine — mount Starlette's CORSMiddleware here and true preflights are answered before a request ever reaches the graph, while actual responses get their Access-Control-* headers on the way out::

from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

build_app(
    routes,
    middleware=[Middleware(CORSMiddleware, allow_origins=["https://app.example"])],
)

on_exception is the app-wide catch-all for an unexpected Exception raised during a resource's walk (a resource may override its own). It defaults to re-raising — so a bug propagates to Starlette's ServerErrorMiddleware (or an ASGI error reporter) as before — but a handler may report the error, enrich the request context, and return to have the graph own a negotiated 500 instead.

event_sink, when given, receives one wide event per request (ctx.event) — the canonical-log-line seam. None by default (nothing is emitted); :class:asgimachine.event.LoggingEventSink is the reference sink.

Source code in src/asgimachine/substrate/starlette.py

def build_app(
    routes: Sequence[BaseRoute],
    *,
    debug: bool = False,
    middleware: Sequence[Middleware] | None = None,
    on_exception: OnException | None = None,
    event_sink: EventSink | None = None,
) -> Starlette:
    """Assemble the composition root into an ASGI application.

    ``middleware`` is passed straight to Starlette, so cross-cutting concerns are
    rented rather than baked into the graph (PLAN.md §2.1). CORS in particular is
    its own decision machine — mount Starlette's ``CORSMiddleware`` here and true
    preflights are answered before a request ever reaches the graph, while actual
    responses get their ``Access-Control-*`` headers on the way out::

        from starlette.middleware import Middleware
        from starlette.middleware.cors import CORSMiddleware

        build_app(
            routes,
            middleware=[Middleware(CORSMiddleware, allow_origins=["https://app.example"])],
        )

    ``on_exception`` is the app-wide catch-all for an unexpected ``Exception`` raised
    during a resource's walk (a resource may override its own). It defaults to
    re-raising — so a bug propagates to Starlette's ``ServerErrorMiddleware`` (or an
    ASGI error reporter) as before — but a handler may report the error, enrich the
    request context, and return to have the graph own a negotiated 500 instead.

    ``event_sink``, when given, receives one wide event per request (``ctx.event``)
    — the canonical-log-line seam. None by default (nothing is emitted);
    :class:`asgimachine.event.LoggingEventSink` is the reference sink.
    """

    app = Starlette(debug=debug, routes=routes, middleware=middleware)
    # Carried on the app so each endpoint can read them from the ASGI scope at
    # request time (the same way it reads ``debug``), not close over every route.
    app.state.on_exception = on_exception
    app.state.event_sink = event_sink
    return app

asgimachine.substrate.starlette.resource_route

resource_route(path, resource, *, codecs=None)

Build a Starlette Route that runs resource through the graph.

codecs injects a media-type -> Codec registry (default: JSON only).

Source code in src/asgimachine/substrate/starlette.py

def resource_route(
    path: str, resource: Resource[Any], *, codecs: dict[str, Codec] | None = None
) -> Route:
    """Build a Starlette ``Route`` that runs ``resource`` through the graph.

    ``codecs`` injects a media-type -> Codec registry (default: JSON only).
    """

    return Route(
        path,
        _ResourceEndpoint(resource, codecs, route=path),
        name=type(resource).__name__,
    )

asgimachine.substrate.starlette.command_route

command_route(
    path,
    command,
    *,
    methods=("POST",),
    max_body_bytes=DEFAULT_MAX_BODY_BYTES,
)

Build a Route for a command (plain-handler lane, §2.5).

Unlike a resource, a command does not walk the graph, so the router owns method restriction here (405 for an unlisted method) — that's fine for a command-shaped endpoint. max_body_bytes bounds the request body (the graph lane gets this from the resource's MAX_BODY_BYTES; a command has no resource, so it is set here) — an over-cap body is 413, a Content-Length mismatch 400.

Source code in src/asgimachine/substrate/starlette.py

def command_route(
    path: str,
    command: Command,
    *,
    methods: Sequence[str] = ("POST",),
    max_body_bytes: int = DEFAULT_MAX_BODY_BYTES,
) -> Route:
    """Build a ``Route`` for a command (plain-handler lane, §2.5).

    Unlike a resource, a command does not walk the graph, so the router owns
    method restriction here (405 for an unlisted method) — that's fine for a
    command-shaped endpoint. ``max_body_bytes`` bounds the request body (the graph
    lane gets this from the resource's ``MAX_BODY_BYTES``; a command has no
    resource, so it is set here) — an over-cap body is 413, a Content-Length
    mismatch 400.
    """

    async def endpoint(request: Request) -> Response:
        started = perf_counter()
        wrapped = _StarletteRequest(request, max_body_bytes)
        sink = getattr(getattr(request, "app", None), "state", None)
        sink = getattr(sink, "event_sink", None)
        status: int | None = None
        exc: BaseException | None = None
        try:
            response = _to_starlette(await command.handle(wrapped))
            status = response.status_code
            return response
        except BodyTooLarge:
            status = int(Status.REQUEST_ENTITY_TOO_LARGE)
            return Response(status_code=status)
        except BodyMalformed:
            status = int(Status.BAD_REQUEST)
            return Response(status_code=status)
        except BaseException as raised:  # emit, then propagate (no graph 500 here)
            exc = raised
            raise
        finally:
            _emit_command_event(sink, command, request, path, status, exc, started)

    endpoint.__name__ = type(command).__name__
    return Route(path, endpoint, methods=list(methods))