WebSockets provide a persistent, bidirectional connection between client and server — ideal for chat, live dashboards, collaborative editing and notifications. FastAPI (via Starlette) supports them natively, and they only work on ASGI servers like Uvicorn.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws/echo")
async def websocket_echo(websocket: WebSocket):
await websocket.accept()
try:
while True:
message = await websocket.receive_text()
await websocket.send_text(f"Echo: {message}")
except WebSocketDisconnect:
print("client disconnected")
@app.websocket(path) declares a WS route (no HTTP method).await websocket.accept() completes the handshake; afterwards you can send_text / receive_text (also send_json, send_bytes).WebSocketDisconnect.from fastapi import FastAPI, WebSocket
class ConnectionManager:
def __init__(self):
self.active: list[WebSocket] = []
async def connect(self, ws: WebSocket):
await ws.accept()
self.active.append(ws)
def disconnect(self, ws: WebSocket):
self.active.remove(ws)
async def broadcast(self, message: str):
for ws in list(self.active):
try:
await ws.send_text(message)
except Exception:
self.disconnect(ws)
manager = ConnectionManager()
@app.websocket("/ws/chat/{room}")
async def chat(websocket: WebSocket, room: str):
await manager.connect(websocket)
await manager.broadcast(f"someone joined {room}")
try:
while True:
msg = await websocket.receive_text()
await manager.broadcast(f"[{room}] {msg}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("someone left")
const ws = new WebSocket("ws://localhost:8000/ws/chat/general");
ws.onopen = () => ws.send("hello!");
ws.onmessage = (event) => console.log("received:", event.data);
ws.onclose = () => console.log("closed");
from fastapi import Query, WebSocket, WebSocketDisconnect, status
@app.websocket("/ws/secure")
async def secure_ws(websocket: WebSocket, token: str = Query()):
user = validate_jwt_or_none(token)
if user is None:
await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
return
await websocket.accept()
... # normal loop
accept() only after validating — or accept and immediately close(1008) with the policy-violation code.Depends works too (a dependency may raise WebSocketException).A: The ConnectionManager is in-process memory. With --workers 4 (or multiple replicas behind a load balancer), clients connect to different processes and each manager only knows its own sockets. Scale with Redis pub/sub (or a message broker) so every process relays broadcasts to its local connections — that is the advanced-level pattern.
A: Keep the manager (or a registry keyed by user id) as a module-level singleton and call await manager.send_to(user_id, msg) from the HTTP route. Both HTTP and WS handlers run on the same event loop per worker process.
A: WebSockets are bidirectional and support binary — chat, games, collaborative apps. SSE (a StreamingResponse of text/event-stream) is unidirectional server→client, auto-reconnects for free over plain HTTP — simpler and often enough for notifications and live feeds. If the client only listens, prefer SSE.
WebSocket endpoints follow a simple lifecycle: accept, loop on receive, handle WebSocketDisconnect. A ConnectionManager gives you rooms and broadcasting; authentication goes through query params or cookies; and horizontal scaling needs a shared broker — keep that constraint in mind from day one.