WebSocket handlers¶

A WebSocket stream sends every value its generator yields:

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

The generated channel¶

{
  "channels": {
    "/ws/stream": {
      "address": "/ws/stream",
      "messages": {
        "stream_items_send": {
          "payload": {
            "$ref": "#/components/schemas/StreamItem"
          },
          "contentType": "application/json",
          "x-websocket-mode": "text"
        }
      },
      "bindings": {
        "ws": {}
      },
      "x-litestar-source": "websocket"
    }
  }
}

Streams¶

A stream only sends, so its channel has one send message and one send operation. The yielded type, StreamItem, becomes the payload schema. Litestar sends strings as text frames, bytes as binary frames and other values as JSON text, and the document records that frame type in the x-websocket-mode extension, separate from the logical contentType. x-websocket-mode is an extension of this library, not a field of the standard WebSocket binding.

Listeners¶

A listener, as in the installation example, has both directions. The annotation of its data parameter describes the messages the application receives, and its return annotation describes the messages it sends. With a DTO, the schema describes the shape the DTO transfers rather than the model itself.

Raw sockets¶

A handler declared with @websocket reads and writes frames itself, so its parameters say nothing about the messages. Undecorated raw handlers are left out of the document. Describe them with the decorators, or set AsyncAPIConfig(include_raw_websocket_routes=True) to document them with unconstrained payloads and a warning.

Handler metadata¶

A handler's own summary, description and operation_id fill the operation fields that decorators leave empty. With AsyncAPIConfig(use_handler_docstrings=True), the handler's docstring fills the description too. This also applies to operations that a decorator adds, such as the explicit operations of a raw socket.

include_in_schema=False on a handler, router or controller removes its routes from the document. Messages that reach the application through a broker rather than a route are not inferred; declare them as explicit channels.

Server-Sent Events¶

An HTTP route that returns ServerSentEvent streams events to the client, so it becomes a channel with one send operation:

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

{
  "channels": {
    "/prices/{symbol}": {
      "address": "/prices/{symbol}",
      "messages": {
        "price": {
          "name": "price",
          "summary": "Latest price for a symbol",
          "payload": {
            "$ref": "#/components/schemas/Price"
          },
          "contentType": "text/event-stream"
        }
      },
      "parameters": {
        "symbol": {
          "description": "Path parameter: symbol (type: str)"
        }
      },
      "bindings": {
        "http": {}
      },
      "x-litestar-source": "sse"
    }
  }
}

ServerSentEvent does not record what its iterator yields, so the payload comes from the asyncapi_message decorator; without one, the payload is unconstrained. Messages use contentType: text/event-stream, the channel carries an http binding and x-litestar-source: sse, and path parameters become channel parameters.

Only the AsyncAPI decorators describe these channels. The handler's HTTP summary, description and operation_id belong to the OpenAPI document. Declaring a receive operation on a Server-Sent Event route raises ImproperlyConfiguredException.

The channel key is the route path. When a WebSocket route uses the same path, the key becomes {path}:sse; when several HTTP methods stream from one path, each gets {path}:sse:{method}. The address stays the path. AsyncAPIConfig(include_sse_routes=False) turns this discovery off.

Recap¶

• Streams send, listeners receive and send, and raw sockets need explicit metadata.
• HTTP routes that return ServerSentEvent become send-only channels marked x-litestar-source: sse.
• Next, Decorators shows how to declare operations and messages explicitly.