| name | litestar-realtime |
| description | Auto-activate for @websocket, websocket_listener, websocket_stream, WebSocket, ServerSentEvent, ChannelsPlugin, AsyncPgChannelsBackend, RedisChannelsPubSubBackend, SQLSpecChannelsBackend, or fan-out. Not for polling APIs. |
Litestar Realtime
Use this skill for WebSockets, SSE, ChannelsPlugin backends, realtime event contracts, and fan-out from workers or services.
Code Style Rules
- Use plain WebSocket handlers for one-off streams.
- Use ChannelsPlugin when dynamic topics, history, or cross-process fan-out matter.
- Choose the backend that matches the existing stack.
- Treat WebSocket auth separately from HTTP header auth constraints.
Quick Reference
Workflow
- Choose WebSocket, SSE, or Channels based on delivery needs.
- Define event contracts before wiring transport.
- Select a backend that fits the project stack.
- Test connection, auth, fan-out, disconnect, and error paths.
Guardrails
- Do not force Redis into a PostgreSQL-only stack just for Channels.
- Do not assume browsers can set arbitrary WebSocket headers.
- Do not publish untyped event dicts across service boundaries.
- Do not use request-scoped resources after the socket lifecycle ends.
Validation Checkpoint
Example
from litestar import websocket
@websocket("/ws")
async def stream(socket: WebSocket) -> None:
await socket.accept()
await socket.send_json({"type": "ready"})
References Index
Official References
Shared Styleguide Baseline