Getting Started with Litestar WebSockets
Litestar is a new Python ASGI framework focused on performance and developer experience. Unlike other frameworks, it offers various WebSocket implementation patterns for different use cases.
Litestar's WebSocket offers a type-safe approach and automatic data serialization. It supports simple bidirectional communication or complex multi-client broadcasting with inter-process support, scaling from prototype to production without architectural changes.
This practical guide explores Litestar's WebSocket capabilities through hands-on examples.
Prerequisites
To use Litestar WebSockets effectively, ensure you have Python 3.8 or newer installed. Familiarity with Python's async/await syntax and type hints will help you take full advantage of Litestar's type-safe features. You don't need to know ASGI in detail, but understanding the basics of HTTP request handling can be helpful.
Setting up your Litestar WebSocket project
In this section, you will establish a clean project structure that illustrates Litestar's approach to WebSocket development.
To begin, run the following commands to set up your development environment with proper isolation, preventing dependency conflicts:
mkdir litestar_websockets && cd litestar_websockets
python3 -m venv venv
source venv/bin/activate
Install Litestar and a production-ready ASGI server:
pip install "litestar[standard]" uvicorn
Create your first Litestar WebSocket application:
from litestar import Litestar
from litestar.handlers.websocket_handlers import websocket_listener
@websocket_listener("/")
async def websocket_handler(data: str) -> str:
return f"Echo: {data}"
app = Litestar([websocket_handler])
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
Litestar's websocket_listener decorator transforms ordinary functions into WebSocket handlers with automatic data parsing and serialization. The type annotations (data: str -> str) tell Litestar how to handle incoming and outgoing data, eliminating manual JSON parsing and validation.
Create a WebSocket client interface to test your application:
mkdir templates
Create templates/index.html:
<!DOCTYPE html>
<html>
<head>
<title>Litestar WebSocket Demo</title>
<style>
body {
font-family: system-ui, -apple-system, sans-serif;
max-width: 800px;
margin: 0 auto;
padding: 20px;
background-color: #f8fafc;
}
.chat-container {
background: white;
border-radius: 12px;
box-shadow: 0 4px 20px rgba(0,0,0,0.08);
overflow: hidden;
}
.chat-header {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 20px 24px;
font-weight: 600;
}
#messages {
height: 400px;
overflow-y: auto;
padding: 24px;
border-bottom: 1px solid #e2e8f0;
}
.message {
margin-bottom: 12px;
padding: 12px 16px;
background: #f1f5f9;
border-radius: 8px;
max-width: 75%;
border-left: 4px solid #cbd5e1;
}
.message.own {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
margin-left: auto;
text-align: right;
border-left: 4px solid rgba(255,255,255,0.3);
}
.message.system {
background: #fef3c7;
color: #92400e;
font-style: italic;
text-align: center;
max-width: 100%;
border-left: 4px solid #f59e0b;
}
.input-container {
padding: 24px;
display: flex;
gap: 12px;
}
#messageInput {
flex: 1;
padding: 12px 16px;
border: 2px solid #e2e8f0;
border-radius: 8px;
font-size: 16px;
transition: border-color 0.2s;
}
#messageInput:focus {
outline: none;
border-color: #667eea;
}
#sendButton {
padding: 12px 24px;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
border: none;
border-radius: 8px;
cursor: pointer;
font-weight: 500;
transition: transform 0.1s;
}
#sendButton:hover {
transform: translateY(-1px);
}
#sendButton:disabled {
background: #9ca3af;
cursor: not-allowed;
transform: none;
}
.status {
padding: 12px 24px;
background: #fee2e2;
color: #991b1b;
font-size: 14px;
}
.status.connected {
background: #d1fae5;
color: #065f46;
}
</style>
</head>
<body>
<div class="chat-container">
<div class="chat-header">
<h1 style="margin: 0;">Litestar WebSocket Chat</h1>
</div>
<div class="status" id="status">Connecting...</div>
<div id="messages"></div>
<div class="input-container">
<input type="text" id="messageInput" placeholder="Type your message..." maxlength="500">
<button id="sendButton" disabled>Send</button>
</div>
</div>
<script>
let ws = null;
const messages = document.getElementById('messages');
const messageInput = document.getElementById('messageInput');
const sendButton = document.getElementById('sendButton');
const status = document.getElementById('status');
function connect() {
ws = new WebSocket('ws://localhost:8000/');
ws.onopen = function() {
status.textContent = 'Connected to Litestar server';
status.classList.add('connected');
sendButton.disabled = false;
messageInput.focus();
};
ws.onmessage = function(event) {
addMessage(event.data, 'system');
};
ws.onclose = function() {
status.textContent = 'Disconnected from server';
status.classList.remove('connected');
sendButton.disabled = true;
// Attempt reconnection after 3 seconds
setTimeout(connect, 3000);
};
ws.onerror = function() {
status.textContent = 'Connection error occurred';
status.classList.remove('connected');
};
}
function addMessage(message, sender) {
const messageElement = document.createElement('div');
messageElement.classList.add('message');
messageElement.textContent = message;
if (sender === 'self') {
messageElement.classList.add('own');
} else if (sender === 'system') {
messageElement.classList.add('system');
}
messages.appendChild(messageElement);
messages.scrollTop = messages.scrollHeight;
}
function sendMessage() {
const message = messageInput.value.trim();
if (message && ws && ws.readyState === WebSocket.OPEN) {
ws.send(message);
addMessage(message, 'self');
messageInput.value = '';
}
}
sendButton.addEventListener('click', sendMessage);
messageInput.addEventListener('keypress', function(e) {
if (e.key === 'Enter') {
sendMessage();
}
});
// Start connection
connect();
</script>
</body>
</html>
This interface features Litestar's signature gradient design and includes automatic reconnection logic. The WebSocket client sends raw text data and displays responses with visual distinction between sent and received messages.
Add a route to serve the HTML interface by updating your app.py:
from pathlib import Path
from litestar import Litestar, get
from litestar.response import Template
from litestar.contrib.jinja import JinjaTemplateEngine
from litestar.template.config import TemplateConfig
from litestar.handlers.websocket_handlers import websocket_listener
@get("/")
async def index() -> Template:
return Template("index.html")
@websocket_listener("/")
async def websocket_handler(data: str) -> str:
return f"Echo: {data}"
app = Litestar(
route_handlers=[index, websocket_handler],
template_config=TemplateConfig(
directory=Path("templates"),
engine=JinjaTemplateEngine,
),
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
The highlighted code include the template engine imports, a new HTTP route to serve the HTML interface, and the template configuration that tells Litestar where to find your HTML files and which template engine to use.
Install the required template dependency:
pip install jinja2
Start your Litestar development server:
python app.py
Navigate to http://localhost:8000 to see your WebSocket interface. The status should show "Connected to Litestar server" once the WebSocket establishes successfully.
Working with WebSocket listeners
Litestar's websocket_listener approach treats WebSocket communication like regular function calls—you receive typed data, process it, and return typed responses. This abstraction eliminates the complexity of manual connection management while providing full control when needed.
Let's enhance your listener with proper data validation and multiple data types:
from pathlib import Path
from typing import Dict, Any
from litestar import Litestar, get
from litestar.response import Template
from litestar.contrib.jinja import JinjaTemplateEngine
from litestar.template.config import TemplateConfig
from litestar.handlers.websocket_handlers import websocket_listener
@get("/")
async def index() -> Template:
return Template("index.html")
@websocket_listener("/echo")
async def echo_handler(data: str) -> str:
"""Simple text echo handler"""
if not data.strip():
return "Error: Empty message not allowed"
if len(data) > 500:
return "Error: Message too long (max 500 characters)"
return f"Echo: {data}"
@websocket_listener("/json")
async def json_handler(data: Dict[str, Any]) -> Dict[str, Any]:
"""JSON data handler with automatic serialization"""
message_type = data.get("type", "unknown")
content = data.get("content", "")
return {
"type": "response",
"original_type": message_type,
"processed_content": content.upper(),
"length": len(str(content)),
"timestamp": "2025-01-01T00:00:00Z"
}
@websocket_listener("/chat")
async def chat_handler(data: str) -> str:
"""Enhanced chat handler with commands"""
if data.startswith("/"):
# Handle commands
command = data[1:].lower()
if command == "help":
return "Available commands: /help, /time, /echo <message>"
elif command == "time":
from datetime import datetime
return f"Current time: {datetime.now().strftime('%H:%M:%S')}"
elif command.startswith("echo "):
return f"Command echo: {command[5:]}"
else:
return f"Unknown command: /{command}"
# Regular message handling
return f"You said: {data}"
app = Litestar(
route_handlers=[index, echo_handler, json_handler, chat_handler],
template_config=TemplateConfig(
directory=Path("templates"),
engine=JinjaTemplateEngine,
),
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
This demonstrates Litestar's type-safe WebSocket handling:
- String handlers work with raw text data and return processed strings
- JSON handlers automatically parse incoming JSON and serialize return dictionaries
- Command processing shows how to build interactive WebSocket applications
Create an enhanced client interface to test different endpoints:
<!DOCTYPE html>
<html>
<head>
<title>Litestar WebSocket Demo</title>
<style>
/* Keep existing styles and add */
...
</style>
</head>
<body>
<div class="chat-container">
<div class="chat-header">
<h1 style="margin: 0;">Litestar WebSocket Demo</h1>
</div>
<div class="endpoint-selector">
<div class="endpoint-buttons">
<button class="endpoint-btn active" data-endpoint="/echo">Echo</button>
<button class="endpoint-btn" data-endpoint="/json">JSON</button>
<button class="endpoint-btn" data-endpoint="/chat">Chat</button>
</div>
</div>
<div class="status" id="status">Connecting...</div>
<div id="messages"></div>
<div class="input-container">
<input type="text" id="messageInput" placeholder="Type your message..." maxlength="500">
<button id="sendButton" disabled>Send</button>
</div>
</div>
<script>
let ws = null;
let currentEndpoint = '/echo';
const messages = document.getElementById('messages');
const messageInput = document.getElementById('messageInput');
const sendButton = document.getElementById('sendButton');
const status = document.getElementById('status');
function connect(endpoint) {
if (ws) ws.close();
ws = new WebSocket(`ws://localhost:8000${endpoint}`);
ws.onopen = function() {
status.textContent = `Connected to ${endpoint}`;
status.classList.add('connected');
sendButton.disabled = false;
messageInput.focus();
};
ws.onmessage = function(event) {
let data = event.data;
try {
// Try to parse as JSON for better display
const parsed = JSON.parse(data);
data = JSON.stringify(parsed, null, 2);
} catch {
// Use as-is if not JSON
}
addMessage(data, 'system');
};
ws.onclose = function() {
status.textContent = 'Disconnected from server';
status.classList.remove('connected');
sendButton.disabled = true;
};
}
function addMessage(message, sender) {
const messageElement = document.createElement('div');
messageElement.classList.add('message');
if (sender === 'self') {
messageElement.classList.add('own');
} else if (sender === 'system') {
messageElement.classList.add('system');
}
// Preserve formatting for JSON
messageElement.style.whiteSpace = 'pre-wrap';
messageElement.textContent = message;
messages.appendChild(messageElement);
messages.scrollTop = messages.scrollHeight;
}
function sendMessage() {
const message = messageInput.value.trim();
if (!message || !ws || ws.readyState !== WebSocket.OPEN) return;
let dataToSend = message;
// Special handling for JSON endpoint
if (currentEndpoint === '/json') {
try {
// Try to parse as JSON first
JSON.parse(message);
dataToSend = message;
} catch {
// Wrap plain text in JSON structure
dataToSend = JSON.stringify({
type: "message",
content: message
});
}
}
ws.send(dataToSend);
addMessage(message, 'self');
messageInput.value = '';
}
// Endpoint switching
document.querySelectorAll('.endpoint-btn').forEach(btn => {
btn.addEventListener('click', function() {
document.querySelectorAll('.endpoint-btn').forEach(b => b.classList.remove('active'));
this.classList.add('active');
currentEndpoint = this.dataset.endpoint;
messages.innerHTML = '';
connect(currentEndpoint);
// Update placeholder based on endpoint
if (currentEndpoint === '/json') {
messageInput.placeholder = 'Type JSON or plain text...';
} else if (currentEndpoint === '/chat') {
messageInput.placeholder = 'Type message or /help for commands...';
} else {
messageInput.placeholder = 'Type your message...';
}
});
});
sendButton.addEventListener('click', sendMessage);
messageInput.addEventListener('keypress', function(e) {
if (e.key === 'Enter') sendMessage();
});
// Start with echo endpoint
connect(currentEndpoint);
</script>
</body>
</html>
The enhanced interface lets you test different WebSocket endpoints with appropriate input handling. The JSON endpoint automatically wraps plain text in JSON structure, while the chat endpoint shows command functionality.
Restart your server and test the different endpoints:
python app.py
Visit http://127.0.0.1:8000/:
The type annotations guide Litestar's automatic data handling, eliminating manual parsing while providing robust error handling and validation.
Broadcasting with WebSocket streams
Most real-time applications need to push data proactively to clients without waiting for incoming messages. Litestar's WebSocket streams use async generators to continuously send data, perfect for live dashboards, notifications, or real-time monitoring systems.
Create streaming handlers that demonstrate continuous data broadcasting:
from pathlib import Path
from typing import Dict, Any, AsyncGenerator
import asyncio
import time
import random
from datetime import datetime
from litestar import Litestar, get
from litestar.response import Template
from litestar.contrib.jinja import JinjaTemplateEngine
from litestar.template.config import TemplateConfig
from litestar.handlers.websocket_handlers import websocket_listener, websocket_stream
@get("/")
async def index() -> Template:
return Template("index.html")
@websocket_listener("/echo")
async def echo_handler(data: str) -> str:
if not data.strip():
return "Error: Empty message not allowed"
return f"Echo: {data}"
@websocket_stream("/stream")
async def time_stream() -> AsyncGenerator[Dict[str, Any], None]:
"""Stream current time and random metrics every 2 seconds"""
counter = 0
while True:
counter += 1
yield {
"counter": counter,
"time": datetime.now().strftime("%H:%M:%S"),
"cpu_usage": round(random.uniform(10, 90), 1),
"memory_usage": round(random.uniform(30, 80), 1),
"timestamp": datetime.now().isoformat()
}
await asyncio.sleep(2)
@websocket_stream("/news")
async def news_stream() -> AsyncGenerator[str, None]:
"""Stream text-based updates"""
news_items = [
"System performance improved",
"New feature deployed",
"Security update applied",
"Database optimization completed",
"Maintenance window scheduled"
]
while True:
item = random.choice(news_items)
timestamp = datetime.now().strftime("%H:%M:%S")
yield f"[{timestamp}] {item}"
await asyncio.sleep(random.uniform(3, 7))
app = Litestar(
route_handlers=[index, echo_handler, time_stream, news_stream],
template_config=TemplateConfig(
directory=Path("templates"),
engine=JinjaTemplateEngine,
),
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
WebSocket streams use AsyncGenerator type hints to define continuous data flow. Litestar automatically serializes yielded values and manages the connection lifecycle.
Create a client interface to display streaming data:
<!DOCTYPE html>
<html>
<head>
<title>Litestar WebSocket Streams</title>
<style>
body {
font-family: system-ui, -apple-system, sans-serif;
margin: 0;
padding: 20px;
background-color: #f8fafc;
}
.container {
max-width: 1000px;
margin: 0 auto;
display: grid;
grid-template-columns: 1fr 1fr;
gap: 20px;
}
.stream-panel {
background: white;
border-radius: 12px;
box-shadow: 0 4px 20px rgba(0,0,0,0.08);
overflow: hidden;
}
.panel-header {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 16px 20px;
font-weight: 600;
display: flex;
justify-content: space-between;
align-items: center;
}
.status-dot {
width: 8px;
height: 8px;
border-radius: 50%;
background: #ef4444;
}
.status-dot.connected {
background: #10b981;
}
.panel-content {
padding: 20px;
height: 300px;
overflow-y: auto;
}
.metric-item {
display: flex;
justify-content: space-between;
padding: 8px 0;
border-bottom: 1px solid #e2e8f0;
}
.metric-value {
font-weight: 600;
color: #667eea;
}
.news-item {
margin-bottom: 12px;
padding: 12px;
background: #f8fafc;
border-radius: 6px;
border-left: 4px solid #667eea;
}
.echo-section {
grid-column: 1 / -1;
background: white;
border-radius: 12px;
box-shadow: 0 4px 20px rgba(0,0,0,0.08);
padding: 20px;
}
.echo-input {
display: flex;
gap: 12px;
margin-top: 16px;
}
.echo-input input {
flex: 1;
padding: 12px;
border: 2px solid #e2e8f0;
border-radius: 8px;
font-size: 16px;
}
.echo-input button {
padding: 12px 24px;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
border: none;
border-radius: 8px;
cursor: pointer;
font-weight: 500;
}
#echoMessages {
height: 150px;
overflow-y: auto;
border: 1px solid #e2e8f0;
border-radius: 8px;
padding: 16px;
margin-bottom: 16px;
}
</style>
</head>
<body>
<div class="container">
<div class="stream-panel">
<div class="panel-header">
<span>Live Metrics</span>
<div class="status-dot" id="metricsStatus"></div>
</div>
<div class="panel-content" id="metricsPanel">
<div id="metricsDisplay">Waiting for data...</div>
</div>
</div>
<div class="stream-panel">
<div class="panel-header">
<span>News Stream</span>
<div class="status-dot" id="newsStatus"></div>
</div>
<div class="panel-content" id="newsPanel">
<div id="newsItems">Connecting to news feed...</div>
</div>
</div>
<div class="echo-section">
<h3>Echo Test</h3>
<div id="echoMessages">Echo responses will appear here...</div>
<div class="echo-input">
<input type="text" id="echoInput" placeholder="Type message for echo test...">
<button id="echoSend">Send</button>
</div>
</div>
</div>
<script>
const streams = {
metrics: { ws: null, endpoint: '/stream' },
news: { ws: null, endpoint: '/news' },
echo: { ws: null, endpoint: '/echo' }
};
function updateStatus(streamName, connected) {
const statusEl = document.getElementById(streamName + 'Status');
if (statusEl) {
statusEl.classList.toggle('connected', connected);
}
}
function connectStream(streamName) {
const stream = streams[streamName];
if (stream.ws) stream.ws.close();
stream.ws = new WebSocket(`ws://localhost:8000${stream.endpoint}`);
stream.ws.onopen = () => updateStatus(streamName, true);
stream.ws.onclose = () => updateStatus(streamName, false);
stream.ws.onmessage = (event) => {
handleStreamMessage(streamName, event.data);
};
}
function handleStreamMessage(streamName, data) {
switch (streamName) {
case 'metrics':
try {
const metrics = JSON.parse(data);
document.getElementById('metricsDisplay').innerHTML = `
<div class="metric-item">
<span>Update:</span>
<span class="metric-value">#${metrics.counter}</span>
</div>
<div class="metric-item">
<span>Time:</span>
<span class="metric-value">${metrics.time}</span>
</div>
<div class="metric-item">
<span>CPU Usage:</span>
<span class="metric-value">${metrics.cpu_usage}%</span>
</div>
<div class="metric-item">
<span>Memory Usage:</span>
<span class="metric-value">${metrics.memory_usage}%</span>
</div>
`;
} catch (e) {
console.error('Error parsing metrics:', e);
}
break;
case 'news':
const newsContainer = document.getElementById('newsItems');
const newsItem = document.createElement('div');
newsItem.className = 'news-item';
newsItem.textContent = data;
newsContainer.insertBefore(newsItem, newsContainer.firstChild);
while (newsContainer.children.length > 8) {
newsContainer.removeChild(newsContainer.lastChild);
}
break;
case 'echo':
const echoContainer = document.getElementById('echoMessages');
const echoItem = document.createElement('div');
echoItem.style.cssText = 'margin-bottom: 8px; padding: 8px; background: #f1f5f9; border-radius: 4px;';
echoItem.textContent = data;
echoContainer.appendChild(echoItem);
echoContainer.scrollTop = echoContainer.scrollHeight;
break;
}
}
// Echo input handling
document.getElementById('echoSend').addEventListener('click', function() {
const input = document.getElementById('echoInput');
const message = input.value.trim();
if (message && streams.echo.ws && streams.echo.ws.readyState === WebSocket.OPEN) {
streams.echo.ws.send(message);
input.value = '';
}
});
document.getElementById('echoInput').addEventListener('keypress', function(e) {
if (e.key === 'Enter') {
document.getElementById('echoSend').click();
}
});
// Auto-connect all streams
Object.keys(streams).forEach(connectStream);
</script>
</body>
</html>
This client interface creates a dashboard with three distinct sections: live metrics display, news feed, and interactive echo testing. The streams object manages multiple concurrent WebSocket connections, each handling different data types and update patterns.
The handleStreamMessage() function routes incoming data based on stream type, parsing JSON for metrics, displaying raw text for news, and formatting echo responses. Status indicators show connection health for each stream, while the interface automatically connects to all endpoints when the page loads.
Restart your server and open the streaming dashboard:
python app.py
Navigate to http://127.0.0.1:8000 to see live data streaming from multiple endpoints:
Type a message in the echo test section and press Enter to see the bidirectional communication:
The interface demonstrates concurrent WebSocket streams with different data types and update frequencies. Each stream operates independently while maintaining efficient resource usage, and the echo functionality shows how listeners and streams can coexist in the same application.
You've successfully implemented real-time data streaming with Litestar WebSockets!
Final thoughts
You've just built a complete real-time application with Litestar WebSockets! From simple echo handlers to live streaming dashboards, you've seen how Litestar's type-safe approach makes WebSocket development both powerful and straightforward.
To learn more, check out the Litestar WebSocket docs for advanced features, and don't forget to explore the dependency injection system
This work is licensed under a Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License.