ChannelsPlugin¶

This application lets ChannelsPlugin create subscriber routes for two topics, gives the news topic a payload type and adds a 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])

Start it and publish a message:

uv run litestar --app docs.examples.channels_plugin.app:app run
curl -X POST http://localhost:8000/publish/news -H 'Content-Type: application/json' -d '{"message":"hello"}'

The generated channels¶

{
  "channels": {
    "/alerts": {
      "address": "/alerts",
      "messages": {
        "alerts_send": {
          "payload": {},
          "contentType": "text/plain",
          "x-websocket-mode": "text"
        }
      },
      "bindings": {},
      "x-litestar-source": "channels"
    },
    "/news": {
      "address": "/news",
      "messages": {
        "news_send": {
          "payload": {
            "$ref": "#/components/schemas/PublishPayload"
          },
          "contentType": "application/json",
          "x-websocket-mode": "text"
        }
      },
      "bindings": {},
      "x-litestar-source": "channels"
    },
    "/ws/echo": {
      "address": "/ws/echo",
      "messages": {
        "echo_receive": {
          "payload": {
            "$ref": "#/components/schemas/ListenerhandlerEchoPayloadRequestBody"
          },
          "contentType": "application/json",
          "x-websocket-mode": "text"
        },
        "echo_send": {
          "payload": {
            "$ref": "#/components/schemas/ListenerhandlerEchoPayloadResponseBody"
          },
          "contentType": "application/json",
          "x-websocket-mode": "text"
        }
      },
      "bindings": {
        "ws": {}
      },
      "x-litestar-source": "websocket"
    }
  }
}

Generated subscriber routes¶

With create_ws_route_handlers=True, ChannelsPlugin adds a WebSocket route per topic. Each one becomes a channel keyed by its path with a single send operation: the application publishes to subscribers, and the route discards incoming frames. The echo listener at /ws/echo is a separate handler and is documented as one.

Topic payloads¶

Topic messages are untyped unless channel_payloads maps the topic name to a type. Here news refers to the generated PublishPayload schema with contentType: application/json, while alerts keeps an unconstrained payload. Generated routes keep their x-websocket-mode.

A key in channel_payloads must name a configured topic, or ImproperlyConfiguredException is raised, even when ChannelsPlugin discovery is turned off. When a ChannelsPlugin allows arbitrary channels, an unlisted key instead declares a new topic. The arbitrary-channel route, whose path ends in {channel_name}, stays untyped.

Topic-only channels¶

A topic without a generated route, which includes every topic under Litestar's default create_ws_route_handlers=False, becomes a channel keyed by the topic name. It has the description "ChannelsPlugin broker topic" and one send operation named {topic}_send. It describes a broker topic, not an address a client can connect to. The concepts example shows one. Turn these channels off with AsyncAPIConfig(include_channels_plugin_topics=False), or all ChannelsPlugin discovery with include_channels_plugin=False.

Topics come from the names passed to the ChannelsPlugin constructor, read once at startup, so runtime subscriptions never change the document. A plugin with arbitrary_channels_allowed=True and no configured names adds no topic channels. A topic named like another channel key, or configured on two ChannelsPlugin instances, raises ImproperlyConfiguredException. An AsyncAPIGenerator used without AsyncAPIPlugin documents no topics of a plugin that allows arbitrary channels, because only the plugin records the configured names at startup.

Markers¶

Every ChannelsPlugin channel carries x-litestar-source: channels and an empty bindings object. The empty bindings tell it apart from a native WebSocket handler, even when both live under the same path prefix.

Clients subscribe to the generated routes directly. For a broker that the application reaches without ChannelsPlugin, declare explicit channels; declaring them does not create a broker client.