Concepts and lifecycle¶

One application can describe three kinds of realtime routes: WebSocket handlers, HTTP routes that stream Server-Sent Events, and ChannelsPlugin topics. This example has one of each:

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

The generated channels¶

{
  "channels": {
    "/orders/events": {
      "address": "/orders/events",
      "messages": {
        "orders_events_send": {
          "payload": {},
          "contentType": "text/event-stream"
        }
      },
      "bindings": {
        "http": {}
      },
      "x-litestar-source": "sse"
    },
    "/ws/chat": {
      "address": "/ws/chat",
      "messages": {
        "chat_receive": {
          "payload": {
            "type": "string"
          },
          "contentType": "text/plain",
          "x-websocket-mode": "text"
        },
        "chat_send": {
          "payload": {
            "type": "string"
          },
          "contentType": "text/plain",
          "x-websocket-mode": "text"
        }
      },
      "bindings": {
        "ws": {}
      },
      "x-litestar-source": "websocket"
    },
    "orders": {
      "address": "orders",
      "description": "ChannelsPlugin broker topic",
      "messages": {
        "orders_send": {
          "payload": {
            "$ref": "#/components/schemas/Order"
          },
          "contentType": "application/json"
        }
      },
      "bindings": {},
      "x-litestar-source": "channels"
    }
  }
}

Directions¶

AsyncAPI describes the application, not its clients. An operation with action: receive means the application receives messages, as the chat listener does with its data parameter. action: send means the application sends them, as the listener's return value, the event stream and the broker topic do. The console in the documentation UI shows the client's view, so its labels are reversed.

Channels and operations¶

A channel key names the entry in the document's channels map; the address is where messages travel. Discovered routes use their path for both, and a ChannelsPlugin topic uses its name. Each operation in the operations map refers to one channel and to messages of that channel. Its key comes from the handler name and action, such as chat_receive, and is unique without regard to case. AsyncAPI 3 has no operationId field, so the key is the operation's identity.

Discovery sources¶

Each discovered channel names its source in the x-litestar-source extension and carries a matching binding:

• websocket: a WebSocket handler, with a ws binding.
• sse: an HTTP route that returns ServerSentEvent, with an http binding and text/event-stream messages. A stream has no item type, so the payload stays unconstrained until a decorator declares one.
• channels: a ChannelsPlugin route or topic, with an empty bindings object. The orders topic has no route; channel_payloads gives it the Order schema.

Channels declared in AsyncAPIConfig.channels carry no marker. A declared channel replaces a discovered channel with the same key. Two discovered channels with the same key raise ImproperlyConfiguredException and name both routes. Handlers with include_in_schema=False, on the handler or a parent layer, are left out.

Lifecycle¶

Nothing is generated at startup. The first request for the document, a call to get_asyncapi_schema(app) or a CLI export builds it from the finalized routes, with a schema registry of its own and the application's schema plugins and type encoders. The result is cached; clear_cache() discards it, and AsyncAPIConfig(use_cache=False) rebuilds it on every request. get_asyncapi() and get_asyncapi_schema() return copies, and get_asyncapi_json() returns the cached bytes.

An application accepts one AsyncAPIPlugin, and a plugin instance belongs to the first application it serves, even with caching off. A second plugin on the same application raises ImproperlyConfiguredException, as does using one plugin with a second application. Create one plugin per application.

Recap¶

• Operations describe what the application sends and receives, and each refers to one channel.
• WebSocket routes, Server-Sent Event routes and ChannelsPlugin topics are discovered and marked with x-litestar-source.
• Next, WebSocket handlers shows what each kind of handler produces.