Skip to content

Runnable examples

Every application under docs/examples/ is tested: the suite serves its document, and make validate-asyncapi checks it against the official AsyncAPI schemas. Run an example from the repository root after make install with the Litestar CLI, for example uv run litestar --app docs.examples.transports:app run, then open http://localhost:8000/asyncapi/.

Each file starts with a PEP 723 block that lists its dependencies and points at this checkout. Running a file directly only builds the application; the Litestar CLI serves it.

Examples that enable the console serve it at /asyncapi/playground. Its validation is advisory: it can send text that does not match the schema, so connect it only to servers you control. The applications leave authorization and message validation to you.

WebSocket listener

Typed dataclass input and output through a DTO. The application receives the input and sends the returned value.

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "litestar[standard]",
#     "litestar-asyncapi",
# ]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../../.." }
# ///
from dataclasses import dataclass
from typing import TYPE_CHECKING

from litestar import Litestar, websocket_listener
from litestar.dto import DataclassDTO

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, DocsConfig
from litestar_asyncapi.spec import Server

if TYPE_CHECKING:
    from litestar import WebSocket

__all__ = ("ChatMessage", "chat_listener")


@dataclass
class ChatMessage:
    room: str
    text: str


@websocket_listener("/ws/chat", dto=DataclassDTO[ChatMessage], signature_namespace={"ChatMessage": ChatMessage})
async def chat_listener(socket: "WebSocket", data: ChatMessage) -> ChatMessage:
    return ChatMessage(room=data.room, text=f"echo: {data.text}")


config = AsyncAPIConfig(
    title="Websocket Listener",
    docs=DocsConfig(interactive=True),
    servers={"local": Server(host="localhost:8000", protocol="ws")},
)

app = Litestar(route_handlers=[chat_listener], plugins=[AsyncAPIPlugin(config)])

WebSocket stream

An asynchronous generator sends a JSON event every second until the client disconnects.

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "litestar[standard]",
#     "litestar-asyncapi",
# ]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../../.." }
# ///
import asyncio
from dataclasses import dataclass
from typing import TYPE_CHECKING

from litestar import Litestar, websocket_stream

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, DocsConfig
from litestar_asyncapi.spec import Server

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator

__all__ = ("StreamItem", "stream_items")


@dataclass
class StreamItem:
    value: int


@websocket_stream("/ws/stream", signature_namespace={"StreamItem": StreamItem})
async def stream_items() -> "AsyncGenerator[StreamItem, None]":
    i = 0
    while True:
        yield StreamItem(value=i)
        i += 1
        await asyncio.sleep(1)


config = AsyncAPIConfig(
    title="Websocket Stream",
    docs=DocsConfig(interactive=True),
    servers={"local": Server(host="localhost:8000", protocol="ws")},
)

app = Litestar(route_handlers=[stream_items], plugins=[AsyncAPIPlugin(config)])

Server-Sent Events

An HTTP route streams price updates; the decorator declares their payload.

# /// script
# requires-python = ">=3.11"
# dependencies = ["litestar[standard]", "litestar-asyncapi"]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../.." }
# ///
"""Document an HTTP route that streams Server-Sent Events."""

import asyncio
from collections.abc import AsyncGenerator
from dataclasses import dataclass

from litestar import Litestar, get
from litestar.response import ServerSentEvent
from litestar.serialization import encode_json

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, asyncapi_message
from litestar_asyncapi.spec import Server

__all__ = ("Price", "app", "prices")


@dataclass
class Price:
    symbol: str
    amount: float


async def price_updates(symbol: str) -> AsyncGenerator[str, None]:
    for tick in range(3):
        yield encode_json(Price(symbol=symbol, amount=100 + tick)).decode()
        await asyncio.sleep(1)


@asyncapi_message(action="send", payload=Price, name="price", summary="Latest price for a symbol")
@get("/prices/{symbol:str}")
async def prices(symbol: str) -> ServerSentEvent:
    return ServerSentEvent(price_updates(symbol))


config = AsyncAPIConfig(title="Price stream", servers={"local": Server(host="localhost:8000", protocol="http")})

app = Litestar(route_handlers=[prices], plugins=[AsyncAPIPlugin(config)])

Discovery sources

A WebSocket listener, a Server-Sent Event route and a typed ChannelsPlugin topic in one application.

# /// script
# requires-python = ">=3.11"
# dependencies = ["litestar[standard]", "litestar-asyncapi"]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../.." }
# ///
"""Discover a WebSocket listener, a Server-Sent Event route and a ChannelsPlugin topic in one app."""

from collections.abc import AsyncGenerator
from dataclasses import dataclass

from litestar import Litestar, get, websocket_listener
from litestar.channels import ChannelsPlugin
from litestar.channels.backends.memory import MemoryChannelsBackend
from litestar.response import ServerSentEvent

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin

__all__ = ("Order", "app", "chat", "order_events")


@dataclass
class Order:
    order_id: str
    total: float


@websocket_listener("/ws/chat")
async def chat(data: str) -> str:
    return data.upper()


async def order_stream(channels: ChannelsPlugin) -> AsyncGenerator[bytes, None]:
    async with channels.start_subscription(["orders"]) as subscriber:
        async for event in subscriber.iter_events():
            yield event


@get("/orders/events")
async def order_events(channels: ChannelsPlugin) -> ServerSentEvent:
    return ServerSentEvent(order_stream(channels))


config = AsyncAPIConfig(title="Order desk", channel_payloads={"orders": Order})

app = Litestar(
    route_handlers=[chat, order_events],
    plugins=[AsyncAPIPlugin(config), ChannelsPlugin(MemoryChannelsBackend(), channels=["orders"])],
)

Raw socket and multiple messages

A raw socket accepts two named message types and returns a JSON result. Decorators supply its contract.

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "litestar[standard]",
#     "litestar-asyncapi",
# ]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../../.." }
# ///
from dataclasses import dataclass
from typing import TYPE_CHECKING

import msgspec
from litestar import Litestar, websocket
from litestar.exceptions import WebSocketDisconnect

from litestar_asyncapi import (
    AsyncAPIConfig,
    AsyncAPIPlugin,
    DocsConfig,
    MessageDefinition,
    asyncapi_message,
    asyncapi_operation,
)
from litestar_asyncapi.spec import Server

if TYPE_CHECKING:
    from litestar import WebSocket

__all__ = ("Incoming", "handler")


@dataclass
class Incoming:
    value: int


@dataclass(slots=True)
class Reset:
    reset: bool


@asyncapi_operation(
    action="receive",
    operation_id="incoming_receive",
    summary="Inbound messages",
    description="Accept either an integer value or a reset command.",
    messages=[
        MessageDefinition(name="Incoming", payload=Incoming, content_type="application/json", examples=[{"value": 1}]),
        MessageDefinition(name="Reset", payload=Reset, content_type="application/json", examples=[{"reset": True}]),
    ],
)
@asyncapi_message(action="send", payload=dict[str, int | str], name="Result", content_type="application/json")
@websocket("/ws/overrides")
async def handler(socket: "WebSocket") -> None:
    """Handle incoming WebSocket messages with decorator overrides."""
    await socket.accept()
    try:
        while True:
            data = await socket.receive_json()
            if "reset" in data:
                msgspec.convert(data, type=Reset)
                value = 0
            else:
                value = msgspec.convert(data, type=Incoming).value
            response = {"received": value, "status": "processed"}
            await socket.send_json(response)
    except WebSocketDisconnect:
        pass


config = AsyncAPIConfig(
    title="Decorator Overrides",
    docs=DocsConfig(interactive=True),
    servers={"local": Server(host="localhost:8000", protocol="ws")},
)

app = Litestar(route_handlers=[handler], plugins=[AsyncAPIPlugin(config)])

Payload schemas

Fixed-length and variable-length tuples, an optional field and an explicit 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"))])

Channels and publish endpoint

ChannelsPlugin subscriber routes for news and alerts, a typed news payload and a POST route that publishes to either topic.

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "litestar[standard]",
#     "litestar-asyncapi",
# ]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../../.." }
# ///
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

from litestar import Litestar, post, websocket_listener
from litestar.channels.backends.memory import MemoryChannelsBackend
from litestar.channels.plugin import ChannelsPlugin
from litestar.dto import DataclassDTO

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, DocsConfig
from litestar_asyncapi.spec import Server

if TYPE_CHECKING:
    from litestar import WebSocket

__all__ = ("EchoPayload", "PublishPayload", "echo", "publish_message")


@dataclass
class EchoPayload:
    message: str


@dataclass
class PublishPayload:
    message: str


@websocket_listener("/ws/echo", dto=DataclassDTO[EchoPayload], signature_namespace={"EchoPayload": EchoPayload})
async def echo(socket: "WebSocket[Any, Any, Any]", data: EchoPayload) -> EchoPayload:
    return data


@post("/publish/{channel:str}")
async def publish_message(channel: str, data: PublishPayload, channels: ChannelsPlugin) -> dict[str, str]:
    """Publish a message to a channel.

    Args:
        channel: The channel name to publish to.
        data: The payload containing the message.
        channels: The ChannelsPlugin instance for publishing.

    Returns:
        Status dict with 'published' status and channel name.
    """
    channels.publish({"message": data.message}, channel)
    return {"status": "published", "channel": channel}


channels_plugin = ChannelsPlugin(MemoryChannelsBackend(), channels=["news", "alerts"], create_ws_route_handlers=True)

config = AsyncAPIConfig(
    title="Channels Plugin",
    channel_payloads={"news": PublishPayload},
    docs=DocsConfig(interactive=True),
    servers={"local": Server(host="localhost:8000", protocol="ws")},
)

app = Litestar(route_handlers=[echo, publish_message], plugins=[AsyncAPIPlugin(config), channels_plugin])

HTMX HTML fragments

An HTMX page at / exchanges HTML fragments over a WebSocket. It loads HTMX and Pico CSS from public CDNs. The console does not support its text/HTML messages, so they are documented only.

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "litestar[standard]",
#     "litestar-asyncapi",
# ]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../../.." }
# ///
import json
from html import escape
from typing import TYPE_CHECKING
from urllib.parse import parse_qs

from litestar import Litestar, Response, get, websocket
from litestar.enums import MediaType

from litestar_asyncapi import AsyncAPIPlugin, asyncapi_message, asyncapi_operation
from litestar_asyncapi.spec import OperationAction

if TYPE_CHECKING:
    from litestar import WebSocket

__all__ = ("htmx_socket", "playground")


@asyncapi_operation(
    action=OperationAction.SEND,
    operation_id="htmx_chat",
    summary="HTMX WebSocket chat endpoint",
    description="Receives text messages and responds with HTML fragments for HTMX swap.",
)
@asyncapi_message(
    action="receive", payload=str, name="HtmxMessage", content_type="text/plain", summary="JSON or form-encoded text"
)
@asyncapi_message(
    action="send", payload=str, name="HtmlFragment", content_type="text/html", summary="HTML fragment response"
)
@websocket("/ws/htmx")
async def htmx_socket(socket: "WebSocket") -> None:
    """Handle HTMX WebSocket messages.

    Receives form data or JSON from HTMX and returns HTML fragments for OOB swap.
    """
    await socket.accept()
    while True:
        raw = await socket.receive_text()
        message = _extract_message(raw)
        html = (
            '<div id="messages" hx-swap-oob="beforeend">'
            f"<article><strong>Client:</strong> {escape(message)}</article>"
            "</div>"
        )
        await socket.send_text(html)


def _extract_message(raw: str) -> str:
    try:
        payload = json.loads(raw)
        if isinstance(payload, dict) and "message" in payload:
            return str(payload["message"])
    except json.JSONDecodeError:
        pass

    parsed = parse_qs(raw)
    if parsed.get("message"):
        return parsed["message"][0]

    return raw


@get("/", sync_to_thread=False)
def playground() -> Response[str]:
    html = """
    <!DOCTYPE html>
    <html lang="en">
      <head>
        <meta charset="utf-8" />
        <meta name="viewport" content="width=device-width, initial-scale=1" />
        <title>HTMX WebSocket Playground</title>
        <link rel="stylesheet" href="https://unpkg.com/@picocss/pico@2/css/pico.min.css" />
        <style>
          body { padding: 2rem; }
          #messages article { margin-bottom: 0.75rem; }
        </style>
      </head>
      <body>
        <main class="container">
          <h1>HTMX WebSocket Playground</h1>
          <p>Uses HTMX WebSocket extension to send messages and receive HTML fragments.</p>
          <p>
            <a href="/asyncapi/" target="_blank" rel="noreferrer">AsyncAPI UI</a> ·
            <a href="/asyncapi/asyncapi.json" target="_blank" rel="noreferrer">AsyncAPI JSON</a>
          </p>

          <div hx-ext="ws" ws-connect="/ws/htmx">
            <form ws-send>
              <label for="message">Message</label>
              <input id="message" name="message" type="text" placeholder="Say hello" required />
              <button type="submit">Send</button>
            </form>
            <section id="messages"></section>
          </div>
        </main>

        <script src="https://unpkg.com/htmx.org@1.9.12"></script>
        <script src="https://unpkg.com/htmx.org@1.9.12/dist/ext/ws.js"></script>
      </body>
    </html>
    """
    return Response(html.strip(), media_type=MediaType.HTML)


app = Litestar(route_handlers=[playground, htmx_socket], plugins=[AsyncAPIPlugin()])

Application error responses

The application answers handled validation cases with structured error messages. Its error action raises an exception, which stays an uncaught server error; the plugin installs no exception handling.

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "litestar[standard]",
#     "litestar-asyncapi",
# ]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../../.." }
# ///
"""Error handling example demonstrating WebSocket error scenarios.

This example shows how WebSocket handlers can gracefully handle various
error conditions and how the AsyncAPI documentation reflects the message schemas.
"""

from dataclasses import dataclass
from typing import TYPE_CHECKING

from litestar import Litestar, websocket_listener
from litestar.dto import DataclassDTO

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, DocsConfig
from litestar_asyncapi.spec import Server

if TYPE_CHECKING:
    from litestar import WebSocket

__all__ = ("ErrorPayload", "ErrorResponse", "error_demo")


@dataclass
class ErrorPayload:
    """Payload for triggering error scenarios."""

    action: str
    """Action to perform: 'echo', 'error', or 'validate'."""
    value: str | None = None
    """Optional value for the action."""


@dataclass
class ErrorResponse:
    """Response from the error demo handler."""

    status: str
    """Status of the operation: 'ok' or 'error'."""
    message: str
    """Descriptive message about the result."""
    original_action: str | None = None
    """The action that was requested."""


_INTENTIONAL_ERROR_MSG = "Intentional error triggered by client"


@websocket_listener(
    "/ws/errors",
    dto=DataclassDTO[ErrorPayload],
    return_dto=DataclassDTO[ErrorResponse],
    signature_namespace={"ErrorPayload": ErrorPayload, "ErrorResponse": ErrorResponse},
)
async def error_demo(socket: "WebSocket", data: ErrorPayload) -> ErrorResponse:
    """Handle messages and demonstrate error scenarios.

    Supported actions:
    - 'echo': Echo back the value
    - 'error': Intentionally raise an error
    - 'validate': Validate the value is not empty

    Args:
        socket: The WebSocket connection.
        data: The incoming payload with action and optional value.

    Returns:
        ErrorResponse with status and message.

    Raises:
        ValueError: When action is 'error' to demonstrate error handling.
    """
    if data.action == "echo":
        return ErrorResponse(status="ok", message=f"Echoed: {data.value}", original_action=data.action)
    if data.action == "error":
        raise ValueError(_INTENTIONAL_ERROR_MSG)
    if data.action == "validate":
        if not data.value:
            return ErrorResponse(
                status="error", message="Validation failed: value is required", original_action=data.action
            )
        return ErrorResponse(status="ok", message=f"Validation passed for: {data.value}", original_action=data.action)
    return ErrorResponse(status="error", message=f"Unknown action: {data.action}", original_action=data.action)


config = AsyncAPIConfig(
    title="Error Handling",
    docs=DocsConfig(interactive=True),
    servers={"local": Server(host="localhost:8000", protocol="ws")},
)

app = Litestar(route_handlers=[error_demo], plugins=[AsyncAPIPlugin(config)])

Broker request and reply

Typed broker channels declared without a route or broker client. The request operation names its reply channel and message.

# /// script
# requires-python = ">=3.11"
# dependencies = ["litestar[standard]", "litestar-asyncapi"]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../.." }
# ///
"""A typed broker request/reply contract; declaring it creates no broker client."""

from dataclasses import dataclass

from litestar import Litestar

from litestar_asyncapi import (
    AsyncAPIConfig,
    AsyncAPIPlugin,
    ChannelDefinition,
    MessageDefinition,
    OperationDefinition,
    ReplyDefinition,
)

__all__ = ("app",)


@dataclass
class QuoteRequest:
    sku: str
    quantity: int


@dataclass
class Quote:
    sku: str
    total: float


config = AsyncAPIConfig(
    title="Pricing broker",
    channels=[
        ChannelDefinition(
            key="quote-requests",
            address="pricing.requests",
            operations=[
                OperationDefinition(
                    action="receive",
                    operation_id="handleQuoteRequest",
                    messages=[MessageDefinition(name="QuoteRequest", payload=QuoteRequest)],
                    reply=ReplyDefinition("quotes", messages=["Quote"]),
                )
            ],
        ),
        ChannelDefinition(
            key="quotes",
            address="pricing.quotes",
            operations=[
                OperationDefinition(
                    action="send", operation_id="sendQuote", messages=[MessageDefinition(name="Quote", payload=Quote)]
                )
            ],
        ),
    ],
)
app = Litestar(plugins=[AsyncAPIPlugin(config)])

JWT-protected WebSocket

JWTAuth protects a WebSocket listener. The document inherits the bearer scheme and a wss server from the OpenAPI config, refers to the scheme by name, shares one operation trait between both operations and declares a correlation ID.

# /// script
# requires-python = ">=3.11"
# dependencies = ["litestar[standard,jwt]", "litestar-asyncapi"]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../.." }
# ///
"""A JWT-protected WebSocket whose security scheme and server come from the OpenAPI config."""

from dataclasses import dataclass
from typing import Any

from litestar import Litestar, websocket_listener
from litestar.connection import ASGIConnection
from litestar.openapi import OpenAPIConfig
from litestar.openapi.spec import Server
from litestar.security.jwt import JWTAuth, Token

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, asyncapi_message, asyncapi_operation
from litestar_asyncapi.spec import CorrelationId, OperationTrait, Tag

__all__ = ("app",)


@dataclass
class User:
    name: str


@dataclass
class PlaceOrder:
    order_id: str
    sku: str
    quantity: int


@dataclass
class OrderStatus:
    order_id: str
    status: str


async def retrieve_user(token: Token, connection: ASGIConnection[Any, Any, Any, Any]) -> User:
    return User(name=token.sub)


jwt_auth = JWTAuth[User](
    retrieve_user_handler=retrieve_user,
    token_secret="development-secret-replace-in-production",
    exclude=["/asyncapi", "/schema"],
)
by_order_id = CorrelationId(location="$message.payload#/order_id")


@asyncapi_operation(action="receive", security=["BearerToken"], traits=["orders"])
@asyncapi_operation(action="send", security=["BearerToken"], traits=["orders"])
@asyncapi_message(action="receive", name="PlaceOrder", payload=PlaceOrder, correlation_id=by_order_id)
@asyncapi_message(action="send", name="OrderStatus", payload=OrderStatus, correlation_id=by_order_id)
@websocket_listener("/orders")
async def orders(data: PlaceOrder) -> OrderStatus:
    return OrderStatus(order_id=data.order_id, status="accepted")


app = Litestar(
    route_handlers=[orders],
    on_app_init=[jwt_auth.on_app_init],
    openapi_config=OpenAPIConfig(
        title="Orders", version="1.0.0", servers=[Server(url="https://orders.example.com", description="Production")]
    ),
    plugins=[
        AsyncAPIPlugin(
            AsyncAPIConfig(
                inherit_openapi_servers=True, operation_traits={"orders": OperationTrait(tags=[Tag(name="orders")])}
            )
        )
    ],
)

Native schema plugin

A Litestar schema plugin and type encoder describe a custom Ticket value with OpenAPI disabled.

# /// 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)],
)

Mounted and guarded docs

Router nesting mounts the documentation at /service/asyncapi. Requests need the header Authorization: demo-token; replace this demonstration guard in a real application.

# /// script
# requires-python = ">=3.11"
# dependencies = ["litestar[standard]", "litestar-asyncapi"]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../.." }
# ///
"""Mount guarded documentation with native Litestar routing."""

from litestar import Litestar, Router
from litestar.connection import ASGIConnection
from litestar.exceptions import NotAuthorizedException
from litestar.handlers import BaseRouteHandler

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, DocsConfig

__all__ = ("app", "docs_guard")


def docs_guard(connection: ASGIConnection, handler: BaseRouteHandler) -> None:
    if connection.headers.get("authorization") != "demo-token":
        raise NotAuthorizedException


plugin = AsyncAPIPlugin(AsyncAPIConfig(docs=DocsConfig(enabled=False, guards=[docs_guard])))
app = Litestar([Router("/service", route_handlers=[plugin.create_docs_router()])], plugins=[plugin])

Custom renderers

Scalar at /events, JSON and YAML at custom paths, and the React console as a separate page.

# /// script
# requires-python = ">=3.11"
# dependencies = ["litestar[standard]", "litestar-asyncapi"]
# [tool.uv.sources]
# litestar-asyncapi = { path = "../.." }
# ///
"""Configure native renderers explicitly for AsyncAPI paths and media types."""

from typing import cast

from litestar import Litestar
from litestar.enums import MediaType

from litestar_asyncapi import AsyncAPIConfig, AsyncAPIPlugin, DocsConfig
from litestar_asyncapi.plugins import AsyncAPIUIRenderPlugin, JsonRenderPlugin, YamlRenderPlugin

__all__ = ("app",)

app = Litestar(
    plugins=[
        AsyncAPIPlugin(
            AsyncAPIConfig(
                docs=DocsConfig(
                    path="/events",
                    interactive=True,
                    render_plugins=[
                        AsyncAPIUIRenderPlugin(renderer="scalar"),
                        JsonRenderPlugin(
                            path="/contract.json", media_type=cast("MediaType", "application/vnd.asyncapi+json")
                        ),
                        YamlRenderPlugin(
                            path="/contract.yaml", media_type=cast("MediaType", "application/vnd.asyncapi+yaml")
                        ),
                    ],
                )
            )
        )
    ]
)

Headless export

Any example can be exported without starting a server, for example uv run litestar --app docs.examples.contracts:app asyncapi export. See Exporting documents.