WebSockets: chat, video signaling, JSON, text, and bytes
Mount a ready-made WebSocket service or handle low-level ASGI WebSocket connections directly.
Work through each step in order. The examples use the generated LenAPI project structure and are designed to be adapted to your app.
Configure trusted browser origins
WebSockets keep a two-way connection open for chat, live updates, notifications, and WebRTC signaling. Set WEBSOCKET_ALLOWED_ORIGINS in settings.py to the exact browser origins that may open a socket. The generated default is an empty list, so the handshake is rejected until you add an origin.
# settings.py
WEBSOCKET_ALLOWED_ORIGINS = [
"http://127.0.0.1:8009",
"https://app.example.com",
]An origin includes scheme, hostname, and optional port. Origin checks are not user authentication. Add an authenticator to private sockets, and always check whether the authenticated user may join the requested room.
Create and mount an authenticated chat service
WebSocketService configures the URL, accepted message shape, optional authentication callback, origin allowlist, payload-size limit, heartbeat, and connection limits. If allowed_origins is omitted, LenAPI reads WEBSOCKET_ALLOWED_ORIGINS from settings.py. An authenticator is a sync or async callable receiving a Request; return an identity when allowed, or None/False to reject.
from lenapi import LENAPI, WebSocketService
app = LENAPI()
def authenticate(request):
# Replace with real session/token validation.
session_id = request.cookies.get("session_id")
return {"user_id": "123"} if session_id == "valid-session" else None
chat = WebSocketService(
path="/ws/chat/{room_id}",
type="chat",
auth=authenticate,
max_message_size=64_000,
heartbeat_seconds=30,
max_connections=6,
max_total_connections=100,
)
app.include_router(chat.router())Browser WebSocket APIs cannot set arbitrary Authorization headers. Browser applications typically authenticate with a secure HttpOnly cookie or a custom authenticator that validates a short-lived credential. Do not treat a room ID as an access-control check.
Create a room and invite participants
Generate the room ID with an ordinary HTTP endpoint, then pass the returned WebSocket path to participants. The service adds the room to its in-memory map when the first socket connects. A random ID can make guessing harder, but it is not authorization and is not stored as a durable room record.
from uuid import uuid4
from pydantic import BaseModel
class RoomInvite(BaseModel):
room_id: str
websocket_path: str
@app.post("/rooms", response_model=RoomInvite)
def create_room():
room_id = uuid4().hex
return {
"room_id": room_id,
"websocket_path": f"/ws/chat/{room_id}",
}Pick a message type
chat relays JSON objects for room conversations. video accepts WebRTC signaling objects named offer, answer, and candidate. json relays any JSON value on a fixed path. message relays plain text. byte relays binary frames. Chat/video use a room ID; JSON, message, and byte can use fixed paths without one.
chat = WebSocketService(path="/ws/chat/{room_id}", type="chat")
video = WebSocketService(path="/ws/video/{room_id}", type="video")
json_feed = WebSocketService(path="/ws/json", type="json")
text_feed = WebSocketService(path="/ws/message", type="message")
bytes_feed = WebSocketService(path="/ws/byte", type="byte")
# Chat: {"type": "chat.message", "text": "Hi"}
# Video signaling: {"type": "offer", "sdp": "..."}
# JSON accepts any JSON value, including object, array, string, number, boolean, or null.Use the room relay from an HTTP route
A normal HTTP handler can publish to currently connected clients or disconnect everyone in a room. send() returns the number of clients that received the event. It does not queue messages for clients that connect later.
@app.post("/rooms/{room_id}/announce")
async def announce(room_id: str, data: dict):
delivered = await chat.send(room_id, data)
return {"room_id": room_id, "delivered": delivered}
@app.delete("/rooms/{room_id}/connections")
async def close_room(room_id: str):
disconnected = await chat.disconnect(room_id)
return {"room_id": room_id, "disconnected": disconnected}Protect these HTTP routes with authentication and authorization too. WebSocket room operations are separate from ordinary HTTP route security.
Set payload, heartbeat, and connection limits
max_message_size is measured in bytes and applies to incoming and service.send payloads. heartbeat_seconds controls an idle ping/pong check; set it to None to disable the application heartbeat. max_connections limits a room, while max_total_connections limits all rooms managed by this process.
chat = WebSocketService(
path="/ws/chat/{room_id}",
type="chat",
max_message_size=64_000,
max_connections=6,
max_total_connections=100,
heartbeat_seconds=30,
)The built-in video type caps rooms at two participants by default. Explicitly set max_connections to change that limit. Limits and heartbeat state are per worker because room membership is in local process memory.
Connect a browser chat client
Use ws:// on an HTTP page and wss:// on an HTTPS page. The service relays each message to every client currently connected to the same room, including the sender. Render incoming user content as text instead of inserting it as HTML.
const roomId = new URLSearchParams(location.search).get("room");
const scheme = location.protocol === "https:" ? "wss:" : "ws:";
const socket = new WebSocket(scheme + "//" + location.host + "/ws/chat/" + roomId);
socket.addEventListener("open", () => console.log("connected"));
socket.addEventListener("message", (event) => {
const message = JSON.parse(event.data);
if (message.type === "lenapi.ping") {
socket.send(JSON.stringify({ type: "lenapi.pong" }));
return;
}
const item = document.createElement("li");
item.textContent = event.data;
document.querySelector("#messages").append(item);
});
function sendChat(text) {
if (socket.readyState === WebSocket.OPEN) {
socket.send(JSON.stringify({ type: "chat.message", text }));
}
}
// Call sendChat(input.value) from your submit button handler.The browser WebSocket API cannot set arbitrary Authorization headers. For authenticated browser connections, use a secure cookie or a custom authenticator with a short-lived credential.
Send JSON, text, and binary frames
The service's type controls which WebSocket frame shapes are accepted. JSON clients send JSON.stringify(value), message clients send strings, and byte clients send ArrayBuffer or Blob values.
// type="json"
socket.send(JSON.stringify({ event: "updated", count: 3 }));
// type="message"
socket.send("plain text");
// type="byte"
socket.send(new Uint8Array([1, 2, 3]).buffer);
socket.binaryType = "arraybuffer";
socket.addEventListener("message", (event) => {
if (event.data instanceof ArrayBuffer) {
console.log(new Uint8Array(event.data));
}
});Understand deployment boundaries
Room membership and broadcasts live in process memory. Multiple workers or hosts cannot see each other's clients without shared pub/sub. Video mode relays only WebRTC offer/answer/candidate signaling; it does not carry camera or microphone media. Real calls need browser RTCPeerConnection code and ICE configuration; TURN is commonly needed across restrictive networks, and multi-party calls generally need an SFU.
Video rooms allow two participants by default. Set max_connections if you need a different room cap. The in-memory room manager and connection counters are per process, not shared across workers.
Use the low-level connection API when needed
Use a low-level app.websocket handler when you need custom authorization, validation, room behavior, or message routing. Accept the handshake before sending data, then receive frames in a loop and catch WebSocketDisconnect for cleanup.
from lenapi.service import WebSocket, WebSocketDisconnect
@app.websocket("/ws/{room_id}")
async def socket(websocket: WebSocket, room_id: str):
await websocket.accept()
try:
while True:
message = await websocket.receive_json()
await websocket.send_json({"room_id": room_id, "echo": message})
except WebSocketDisconnect:
passThe helper also has receive_text(), receive_bytes(), send_text(), send_bytes(), and close(code=...). WebSocket routes are not ordinary HTTP endpoints and are not shown as REST operations in Swagger.