Decorators¶

A raw WebSocket handler reads and writes frames itself, so the decorators describe its messages. This one accepts two kinds of incoming message and sends a result:

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

The generated operations¶

{
  "operations": {
    "incoming_receive": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/~1ws~1overrides"
      },
      "summary": "Inbound messages",
      "description": "Accept either an integer value or a reset command.",
      "messages": [
        {
          "$ref": "#/channels/~1ws~1overrides/messages/Reset"
        },
        {
          "$ref": "#/channels/~1ws~1overrides/messages/Incoming"
        }
      ]
    },
    "ws_overrides_send": {
      "action": "send",
      "channel": {
        "$ref": "#/channels/~1ws~1overrides"
      },
      "messages": [
        {
          "$ref": "#/channels/~1ws~1overrides/messages/Result"
        }
      ]
    }
  }
}

How it works¶

Place the AsyncAPI decorators above the Litestar route decorator. They read and update the route handler that the Litestar decorator creates, not the plain function. Every decorator names an action from the application's point of view: receive for messages the application accepts, send for messages it emits.

asyncapi_operation declares the whole operation for one action. Its messages list offers several named choices, here Incoming and Reset, each a MessageDefinition with its own payload type, content type and examples. operation_id="incoming_receive" sets the operation's key in the document; without it, the key comes from the handler or the channel, as for ws_overrides_send.

asyncapi_message adds one message to the operation for its action, creating the operation when needed. The handler above uses it for the Result message it sends. When an operation already has messages, every further message needs a name, and messages from both decorators are combined by name in either stacking order. A repeated name raises ValueError.

Explicit values win over inference, and an explicit empty list is a value: messages=[] documents an operation without messages, and examples=[] turns off generated examples for that message. Tags take Tag objects rather than strings. To give an example headers, a name or a summary, pass a MessageExample; other values become the example's payload, and an explicit None payload is kept as null.

The same decorators describe Server-Sent Event routes, where only send operations are allowed. For contracts that have no route, see Contracts without a route.

Contracts without a route¶

A broker topic that the application consumes or publishes elsewhere has no route to decorate. Declare it with ChannelDefinition in AsyncAPIConfig.channels. This application answers price quote requests on a reply channel:

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

{
  "operations": {
    "handleQuoteRequest": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/quote-requests"
      },
      "messages": [
        {
          "$ref": "#/channels/quote-requests/messages/QuoteRequest"
        }
      ],
      "reply": {
        "channel": {
          "$ref": "#/channels/quotes"
        },
        "messages": [
          {
            "$ref": "#/channels/quotes/messages/Quote"
          }
        ]
      }
    },
    "sendQuote": {
      "action": "send",
      "channel": {
        "$ref": "#/channels/quotes"
      },
      "messages": [
        {
          "$ref": "#/channels/quotes/messages/Quote"
        }
      ]
    }
  }
}

A channel's key names it in the document and its address is the broker destination. Operations and messages take the same OperationDefinition and MessageDefinition values as the decorators, and payload dataclasses become schemas under components.schemas.

ReplyDefinition("quotes", messages=["Quote"]) names the reply channel by key and its messages by name; the document gets the matching references. Each reply message must belong to the reply channel, and an unknown channel or message name raises an error that lists the valid names. The reply channel declares its own send operation, so clients can tell which direction the reply travels. Pass address=ReplyAddress(...) when the reply goes to an address carried in the request.

Recap¶

Decorators go above the route decorator and always name an action.
asyncapi_operation declares an operation with named message choices; asyncapi_message adds one message.
ChannelDefinition declares channels without a route, and ReplyDefinition names a reply channel and its messages.
Next, Schema generation explains how payload types become schemas.