FastAPI WebSocket Setup: Minimal Example
FastAPI simplifies WebSocket server development by wrapping Starlette's native WebSocket support in an ASGI-compatible async framework, allowing you to define socket endpoints as easily as HTTP routes. With just 25 lines of Python and the fastapi, websockets, and uvicorn packages, you can accept connections, receive and send messages, and handle disconnections. This article builds the minimal foundation on which all real-time apps rest.
Installation and Project Setup
Create a new Python 3.9+ project and install dependencies:
pip install fastapi uvicorn[standard] websockets
fastapi provides the framework; uvicorn[standard] is the ASGI server; websockets provides the protocol implementation. These are the only packages needed for a basic server.
Create a file main.py:
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
# Serve a simple HTML client
@app.get("/")
async def get():
return HTMLResponse("""
<html>
<body>
<h1>WebSocket Chat</h1>
<input type="text" id="msg" />
<button onclick="sendMsg()">Send</button>
<ul id="messages"></ul>
<script>
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = (event) => {
const li = document.createElement("li");
li.innerText = event.data;
document.getElementById("messages").appendChild(li);
};
function sendMsg() {
const msg = document.getElementById("msg").value;
ws.send(msg);
document.getElementById("msg").value = "";
}
</script>
</body>
</html>
""")
# WebSocket endpoint
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"You said: {data}")
except Exception:
await websocket.close()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Run it:
python main.py
Open http://localhost:8000 in your browser, type a message, and watch it echo back instantly. The server accepted your connection, received text, and sent a response—all in real time, over a single persistent connection.
Understanding the Endpoint Anatomy
The @app.websocket("/ws") decorator marks a route as WebSocket-capable. The decorated function receives a WebSocket object (from Starlette) with these key methods:
await websocket.accept(): Completes the HTTP upgrade handshake and transitions to OPEN state.await websocket.receive_text(): Blocks until a text message arrives; raisesWebSocketDisconnectif the client closes.await websocket.send_text(data): Sends a text frame to the client.await websocket.send_bytes(data): Sends binary data.await websocket.close(code=1000): Initiates a graceful close.
The try/except pattern above is critical: websocket.receive_text() raises WebSocketDisconnect when the client disconnects or the connection breaks, allowing you to clean up resources (close files, remove from a user list, etc.). Without catching it, the exception propagates and logs as an error; catching it lets you handle it gracefully.
The ConnectionManager Pattern
For a multi-client server (chat, broadcast), managing active connections becomes complex fast. The ConnectionManager pattern, standard in FastAPI docs, centralizes connection state:
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
from typing import List
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
try:
await connection.send_text(message)
except Exception:
# Connection failed; removed on next disconnect
pass
app = FastAPI()
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Client {client_id}: {data}")
except Exception:
manager.disconnect(websocket)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Now, any message from one client is broadcast to all connected clients. The manager.broadcast() method wraps each send in a try/except because a connection might fail mid-broadcast; rather than failing the whole operation, we log it and move on, removing the broken connection on its next disconnect attempt.
Async/Await and Event Loop
FastAPI routes are async functions, and await websocket.receive_text() is a non-blocking call. While one client's message is being processed, the event loop can handle messages from other clients, making high-concurrency real-time systems feasible. A single Python process can manage thousands of WebSocket clients without threads, because every I/O operation (socket read, database query, HTTP call) yields control back to the event loop via await.
If your application logic is CPU-bound (e.g., expensive calculation), use asyncio.to_thread() or concurrent.futures.ThreadPoolExecutor to avoid blocking the loop:
import asyncio
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
# Run slow function in a thread pool
result = await asyncio.to_thread(slow_computation, data)
await manager.broadcast(result)
except Exception:
manager.disconnect(websocket)
Testing with Multiple Clients
To test your server with multiple concurrent connections, open the HTML page in several browser tabs. Each tab will establish its own WebSocket connection. Send a message from one tab and confirm it appears in all tabs instantly—this confirms your broadcast is working.
For automated testing, use websockets library:
import asyncio
import websockets
async def test_broadcast():
async with websockets.connect("ws://localhost:8000/ws/client1") as ws1:
async with websockets.connect("ws://localhost:8000/ws/client2") as ws2:
await ws1.send("Hello")
msg = await ws2.recv()
print(msg) # "Client client1: Hello"
asyncio.run(test_broadcast())
Key Takeaways
- FastAPI's
@app.websocket()decorator accepts aWebSocketobject with methods to accept, receive, send, and close. - The
ConnectionManagerpattern stores active connections and broadcasts to all, enabling multi-client messaging. - Async/await allows thousands of concurrent clients on a single event loop without threads.
- Always wrap
receive_text()in try/except to handleWebSocketDisconnectgracefully. - Use
asyncio.to_thread()for CPU-bound operations to avoid blocking the loop.
Frequently Asked Questions
Can I use FastAPI's dependency injection with WebSocket endpoints?
Yes. WebSocket routes support path parameters and query parameters: async def websocket_endpoint(websocket: WebSocket, client_id: str, token: str = Query(...)). You can also create a dependency that validates the token before the route runs, enforcing authentication early.
What happens if I don't call websocket.accept()?
The client's connection attempt times out, and the client receives a 403 Forbidden response. The endpoint function can reject a connection based on authentication or other logic by omitting the accept call.
How do I broadcast to everyone except the sender?
Store the sender's websocket reference, then in broadcast(), skip it: if connection != sender: await connection.send_text(message). This is common in chat apps where the sender already displays the message locally.
Is there a maximum number of connections per FastAPI server?
Practically, yes, limited by OS file descriptors and available RAM. Each connection consumes ~30–50 KB; a 2 GB server might handle 40,000–60,000 concurrent connections. For higher scale, use load balancing and Redis pub/sub (covered in later articles).