Contracts¶
from litestar_asyncapi import (
ChannelDefinition,
MessageDefinition,
OperationDefinition,
ReplyDefinition,
asyncapi_message,
asyncapi_operation,
)
The decorators refine what discovery infers from a route handler. The definition types describe the same contracts explicitly: decorators take MessageDefinition values, and AsyncAPIConfig.channels takes ChannelDefinition values. Payload and header types stay Python types until the document is built.
Decorators¶
litestar_asyncapi.asyncapi_operation(*, action: OperationAction | str, operation_id: str | None = None, title: str | None = None, summary: str | None = None, description: str | None = None, traits: list[str | OperationTrait | Reference] | None = None, messages: list[MessageDefinition] | None = None, tags: list[Tag | Reference] | None = None, security: list[str | SecurityScheme | Reference] | None = None, bindings: dict[str, Any] | Reference | None = None, reply: ReplyDefinition | Reply | Reference | None = None) -> Callable[[T], T]
¶
Declare the operation for one action on a WebSocket or Server-Sent Event route handler.
Keyword arguments match the fields of OperationDefinition; None keeps the discovered value.
Decorating the same handler again for the same action updates the declaration and appends messages.
Returns:
| Type | Description |
|---|---|
Callable[[T], T]
|
A decorator that mutates the Litestar route handler metadata in-place. |
litestar_asyncapi.asyncapi_message(*, action: OperationAction | str, payload: object | None = None, name: str | None = None, title: str | None = None, summary: str | None = None, description: str | None = None, examples: list[object] | None = None, headers: object | None = None, content_type: str | None = None, traits: list[str | MessageTrait | Reference] | None = None, correlation_id: CorrelationId | Reference | None = None, bindings: dict[str, Any] | Reference | None = None, tags: list[Tag | Reference] | None = None, extensions: dict[str, Any] | None = None) -> Callable[[T], T]
¶
Add a message to the operation for one action on a WebSocket or Server-Sent Event route handler.
Keyword arguments match the fields of MessageDefinition. A message without a name is
rejected when the operation already has messages.
Returns:
| Type | Description |
|---|---|
Callable[[T], T]
|
A decorator that mutates the Litestar route handler metadata in-place. |
Definitions¶
litestar_asyncapi.ChannelDefinition
¶
An explicit channel keyed independently from its transport address.
key: str
¶
Key of the channel in the document's channels map.
address: str
¶
Transport address of the channel, such as a WebSocket path or broker topic.
operations: Sequence[OperationDefinition] = ()
¶
Operations on the channel.
parameters: dict[str, Parameter | Reference] | None = None
¶
Parameters for expressions in address.
servers: list[str | Reference] | None = None
¶
Names of, or references to, the servers that host the channel; all servers when omitted.
bindings: dict[str, Any] | Reference | None = None
¶
Protocol-specific channel bindings.
description: str | None = None
¶
Description of the channel; CommonMark is allowed.
extensions: dict[str, Any] = field(default_factory=dict)
¶
Specification extensions; every key must start with x-.
litestar_asyncapi.OperationDefinition
¶
An application-perspective operation and its named message choices.
action: OperationAction | str
¶
Direction from the application's perspective: send or receive.
messages: list[MessageDefinition] | None = None
¶
Messages the operation carries; None keeps the inferred messages.
operation_id: str | None = None
¶
Key of the operation in the document; derived from the channel key when omitted.
title: str | None = None
¶
Human-friendly operation title.
summary: str | None = None
¶
Short summary of the operation.
description: str | None = None
¶
Longer description of the operation; CommonMark is allowed.
traits: list[str | OperationTrait | Reference] | None = None
¶
Operation traits, given by name of a configured trait, as an object or as a reference.
tags: list[Tag | Reference] | None = None
¶
Tags for logical grouping.
security: list[str | SecurityScheme | Reference] | None = None
¶
Security schemes that apply to the operation, given by name of a scheme in components, as an object or as a reference.
bindings: dict[str, Any] | Reference | None = None
¶
Protocol-specific operation bindings.
reply: ReplyDefinition | Reply | Reference | None = None
¶
Reply that the operation expects, for request and reply patterns.
litestar_asyncapi.MessageDefinition
¶
A message contract retaining Python payload and header types until assembly.
payload: object | None = None
¶
Payload type, converted to a JSON Schema when the document is built.
A spec Schema, Reference, MultiFormatSchema, schema dict or bool is used as given.
name: str | None = None
¶
Machine-friendly message name.
title: str | None = None
¶
Human-friendly message title.
summary: str | None = None
¶
Short summary of the message.
description: str | None = None
¶
Longer description of the message; CommonMark is allowed.
examples: list[object] | None = None
¶
Example payloads; an empty list suppresses generated examples.
headers: object | None = None
¶
Header type or schema, converted like payload.
correlation_id: CorrelationId | Reference | None = None
¶
Location of the correlation id in the message.
content_type: str | None = None
¶
Media type of the payload; inferred from the payload schema when omitted.
traits: list[str | MessageTrait | Reference] | None = None
¶
Message traits, given by name of a configured trait, as an object or as a reference.
bindings: dict[str, Any] | Reference | None = None
¶
Protocol-specific message bindings.
tags: list[Tag | Reference] | None = None
¶
Tags for logical grouping.
extensions: dict[str, Any] = field(default_factory=dict)
¶
Specification extensions; every key must start with x-.
litestar_asyncapi.ReplyDefinition
¶
A reply named by its channel key and message names, resolved when the document is built.
channel: str
¶
Key of the channel the reply is sent on.
messages: list[str] | None = None
¶
Names of messages on the reply channel that may be sent as the reply; any of them when omitted.
address: ReplyAddress | Reference | None = None
¶
Where the reply is sent when it is not the channel's fixed address.
Transport markers¶
Every discovered channel carries an x-litestar-source extension that names the transport it came from. Channels declared in AsyncAPIConfig.channels carry none.
litestar_asyncapi.asyncapi.datastructures.SOURCE_EXTENSION = 'x-litestar-source'
¶
Channel extension naming the transport of a discovered channel.
litestar_asyncapi.asyncapi.datastructures.SOURCE_MARKERS: dict[DiscoverySource, str] = {DiscoverySource.WEBSOCKET: 'websocket', DiscoverySource.CHANNELS_PLUGIN: 'channels', DiscoverySource.SSE: 'sse'}
¶
x-litestar-source values for discovered channels; explicit config channels carry none.