Installation¶

Install the package into the environment of your Litestar application:

uv add litestar-asyncapi

pip install litestar-asyncapi

pdm add litestar-asyncapi

poetry add litestar-asyncapi

It supports Python 3.11 to 3.15 and Litestar 2.24 or later. The documentation UI ships inside the wheel, so the application needs neither Node nor a CDN.

A first application¶

This application has one WebSocket listener and registers AsyncAPIPlugin:

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

Save it as app.py and start it with the Litestar CLI:

litestar --app app:app run

From a checkout of this repository, run uv run litestar --app docs.examples.websocket_listener.app:app run instead.

The generated document¶

Open http://localhost:8000/asyncapi/ for the documentation UI, or download the document from http://localhost:8000/asyncapi/asyncapi.json. Its channels section describes the listener:

{
  "channels": {
    "/ws/chat": {
      "address": "/ws/chat",
      "messages": {
        "chat_listener_receive": {
          "payload": {
            "$ref": "#/components/schemas/ListenerhandlerChatMessageRequestBody"
          },
          "contentType": "application/json",
          "x-websocket-mode": "text"
        },
        "chat_listener_send": {
          "payload": {
            "$ref": "#/components/schemas/ListenerhandlerChatMessageResponseBody"
          },
          "contentType": "application/json",
          "x-websocket-mode": "text"
        }
      },
      "bindings": {
        "ws": {}
      },
      "x-litestar-source": "websocket"
    }
  }
}

How it works¶

The plugin reads the application's routes the first time a document is requested. The listener's data parameter describes the messages the application receives, and its return annotation describes the messages it sends. Both refer to schemas that Litestar generates for the DTO's incoming and outgoing shapes, which the document keeps under components.schemas.

The channel key and its address are both the route path. The ws binding and the x-litestar-source: websocket extension mark the channel as a native WebSocket route, so tools can tell it apart from ChannelsPlugin routes and Server-Sent Event streams.

DocsConfig controls the documentation routes under /asyncapi. The JSON document is served as application/vnd.asyncapi+json. DocsConfig(yaml=True) adds /asyncapi/asyncapi.yaml and /asyncapi/asyncapi.yml. The example also sets interactive=True, which adds a WebSocket console at /asyncapi/playground; it is off by default.

The console connects only to servers that the document declares, which is why the example adds a ws server for localhost:8000. The plugin never turns the request's host name into a server entry. With AsyncAPIConfig(inherit_openapi_servers=True), absolute server URLs from the OpenAPI config become ws or wss servers instead; see Security and servers.

Documents use AsyncAPI 3.1 by default. Set AsyncAPIConfig(spec_version="3.0.0") to generate AsyncAPI 3.0 instead.

Recap¶

• Register one AsyncAPIPlugin and the application serves its AsyncAPI document and UI under /asyncapi.
• Listener annotations become receive and send messages with generated schemas.
• Next, Concepts and lifecycle explains channels, operations and the discovery sources.