Schema generation¶
Payload types are ordinary Python types. This stream sends a dataclass with a fixed-length tuple, a variable-length tuple and an optional field, and declares one example:
# /// script
# requires-python = ">=3.11"
# dependencies = ["litestar[standard]", "litestar-asyncapi"]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../.." }
# ///
"""Generate Draft07 payload schemas with tuple bounds and an explicit example."""
import asyncio
from collections.abc import AsyncGenerator
from dataclasses import dataclass
from litestar import Litestar, websocket_stream
from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, asyncapi_message
__all__ = ("Reading", "app", "readings")
@dataclass
class Reading:
sensor: str
position: tuple[float, float]
tags: tuple[str, ...]
note: str | None = None
EXAMPLE = Reading(sensor="roof-1", position=(52.52, 13.40), tags=("outdoor",))
@asyncapi_message(action="send", payload=Reading, name="reading", examples=[EXAMPLE])
@websocket_stream("/ws/readings")
async def readings() -> AsyncGenerator[Reading, None]:
while True:
yield EXAMPLE
await asyncio.sleep(1)
app = Litestar(route_handlers=[readings], plugins=[AsyncAPIPlugin(AsyncAPIConfig(title="Sensor readings"))])
The generated schema¶
{
"components": {
"schemas": {
"Reading": {
"properties": {
"sensor": {
"type": "string"
},
"position": {
"items": [
{
"type": "number"
},
{
"type": "number"
}
],
"type": "array",
"maxItems": 2,
"minItems": 2
},
"tags": {
"items": {
"type": "string"
},
"type": "array"
},
"note": {
"oneOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
}
},
"type": "object",
"required": [
"position",
"sensor",
"tags"
],
"title": "Reading"
}
}
}
}
The message refers to that schema and carries the example:
{
"channels": {
"/ws/readings": {
"address": "/ws/readings",
"messages": {
"reading": {
"name": "reading",
"payload": {
"$ref": "#/components/schemas/Reading"
},
"contentType": "application/json",
"examples": [
{
"payload": {
"sensor": "roof-1",
"position": [
52.52,
13.4
],
"tags": [
"outdoor"
],
"note": null
}
}
],
"x-websocket-mode": "text"
}
},
"bindings": {
"ws": {}
},
"x-litestar-source": "websocket"
}
}
}
How it works¶
Schemas come from Litestar's own schema generation, the same code that builds OpenAPI schemas, so every model type the application supports works here: dataclasses, msgspec structs, Pydantic models, attrs classes and TypedDicts, provided their Litestar integrations are installed. Each document build uses a fresh schema registry and the application's schema plugins, even when OpenAPI is disabled, and named models end up under components.schemas.
AsyncAPI 3 uses JSON Schema Draft07 for inline schemas, so the generated schema is converted to Draft07. A schema construct with no Draft07 equivalent, such as a dynamic reference, raises an error that names its location.
Optional fields keep their null alternative and their default. Model aliases, msgspec array-like structs, recursive models and DTO transfer shapes produce the shape that goes over the wire.
Tuple and schema dialect¶
A fixed-length tuple such as tuple[float, float] becomes a Draft07 positional items array with minItems and maxItems equal to its length; tuple[()] has maxItems: 0. A variable-length tuple such as tuple[str, ...], or a bare tuple, becomes an unbounded array. The document never contains prefixItems.
To write a schema by hand, pass a Schema from litestar_asyncapi.spec, a schema dictionary or a MultiFormatSchema as the payload; these are used as given. The official AsyncAPI schema rejects a bare boolean payload, so write MultiFormatSchema("application/schema+json;version=draft-07", False) for one.
Examples¶
A message's examples come from the first of these that is present:
- examples passed to the decorator or
MessageDefinition, as in this example; - examples declared on the model or its schema;
- an example generated by Litestar's example factory, when
AsyncAPIConfig(create_examples=True).
Explicit examples pass through the application's type encoders, so they look like real messages. An empty list means no examples, and null values are kept. Generated examples are random; prefer explicit ones for documentation that should not change between builds, and for tuples, which the console cannot fill in by itself. Litestar's example factory can need Pydantic: if create_examples=True raises MissingDependencyException, install litestar[pydantic]. When a type has constraints the factory cannot honor, the message gets no generated example and a warning is raised.
Custom types¶
A type that Litestar cannot describe needs a schema plugin. Register the plugin and a type encoder on the application, as you would for OpenAPI; this also works with openapi_config=None:
# /// script
# requires-python = ">=3.11"
# dependencies = ["litestar[standard]", "litestar-asyncapi"]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../.." }
# ///
"""Use a native Litestar schema plugin even when OpenAPI routes are disabled."""
from typing import Any
from litestar import Litestar
from litestar.openapi.spec import OpenAPIType, Schema
from litestar.plugins import OpenAPISchemaPlugin
from litestar.typing import FieldDefinition
from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, ChannelDefinition, MessageDefinition, OperationDefinition
__all__ = ("Ticket", "TicketSchemaPlugin", "app")
class Ticket:
__slots__ = ("value",)
def __init__(self, value: str) -> None:
self.value = value
class TicketSchemaPlugin(OpenAPISchemaPlugin):
__slots__ = ()
def is_plugin_supported_field(self, field_definition: FieldDefinition) -> bool:
return field_definition.annotation is Ticket
def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: Any) -> Schema:
return Schema(type=OpenAPIType.STRING, pattern="^T-[0-9]+$")
config = AsyncAPIConfig(
channels=[
ChannelDefinition(
key="tickets",
address="tickets",
operations=[
OperationDefinition(
action="send",
messages=[MessageDefinition(name="Ticket", payload=Ticket, examples=[Ticket("T-42")])],
)
],
)
]
)
app = Litestar(
openapi_config=None,
type_encoders={Ticket: lambda ticket: ticket.value},
plugins=[TicketSchemaPlugin(), AsyncAPIPlugin(config)],
)
The plugin describes the type in documents only. It adds no decoding or validation at runtime.
Recap¶
- Payload schemas come from Litestar's schema generation and are exported as Draft07.
- Fixed-length tuples get positional
itemswith exact bounds; explicit examples take precedence over generated ones. - Next, Exporting documents writes the document without starting a server.