WebSockets for Real-Time Communication
Build real-time features in FastAPI with WebSockets: accept connections, exchange messages, broadcast to many clients, and handle disconnects cleanly using async patterns.
WebSockets for Real-Time Communication is a free FastAPI Backend Development Bootcamp lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the FastAPI Backend Development Bootcamp learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
From Request/Response to Real-Time
Regular HTTP is request/response: the client asks, the server answers, the connection closes. WebSockets keep a single connection open for full-duplex, real-time messaging. Perfect for chat, live dashboards, and notifications.
Why Async Fits WebSockets
A WebSocket connection lives a long time and is mostly idle waiting for messages. FastAPI's async model lets one worker handle thousands of concurrent connections without blocking.
A Minimal WebSocket Endpoint
Declare a @app.websocket route. Accept the connection, then loop receiving and sending messages.
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket('/ws')
async def ws(websocket: WebSocket):
await websocket.accept()
while True:
msg = await websocket.receive_text()
await websocket.send_text(f'echo: {msg}')Sending JSON
You can exchange structured data with receive_json and send_json instead of raw text.
data = await websocket.receive_json()
await websocket.send_json({'received': data, 'status': 'ok'})Handling Disconnects
Clients drop off. Catch WebSocketDisconnect to clean up resources instead of crashing the handler.
from fastapi import WebSocketDisconnect
try:
while True:
await websocket.receive_text()
except WebSocketDisconnect:
print('client left')A Connection Manager
To broadcast, track active connections in a manager class. Add on connect, remove on disconnect.
class Manager:
def __init__(self):
self.active = []
async def connect(self, ws):
await ws.accept()
self.active.append(ws)
def disconnect(self, ws):
self.active.remove(ws)Broadcasting to Everyone
Loop the active connections and send each one the message. This is the core of a chat room.
async def broadcast(self, message):
for connection in self.active:
await connection.send_text(message)Broadcast Logic in Plain Python
The manager is just list bookkeeping. Here is the connect/disconnect/broadcast flow simulated synchronously.
active = []
def connect(c): active.append(c)
def disconnect(c): active.remove(c)
def broadcast(msg): return [f'{c}<-{msg}' for c in active]
connect('alice'); connect('bob')
print(broadcast('hi'))
disconnect('alice')
print(broadcast('bye'))A Chat Room Endpoint
Combine the pieces: connect, broadcast each received message, and disconnect on drop.
manager = Manager()
@app.websocket('/chat')
async def chat(ws: WebSocket):
await manager.connect(ws)
try:
while True:
text = await ws.receive_text()
await manager.broadcast(text)
except WebSocketDisconnect:
manager.disconnect(ws)Scaling Beyond One Process
An in-memory manager only knows connections on its worker. To broadcast across multiple workers or servers, use a pub/sub backend like Redis to fan out messages.
Security and Limits
Protect WebSocket endpoints:
- Authenticate during the handshake (e.g. token in query or header).
- Validate and size-limit incoming messages.
- Heartbeat/ping to detect dead connections.
Quick Check
Why does an in-memory connection manager fail to broadcast correctly when you run several Uvicorn workers?
Recap
You added real-time communication:
- Accepted WebSocket connections and exchanged text/JSON.
- Handled
WebSocketDisconnectcleanly. - Built a connection manager to broadcast to many clients.
- Learned to scale with Redis pub/sub and to secure connections.
Frequently asked questions
Is the “WebSockets for Real-Time Communication” lesson free?
Yes — the full text of “WebSockets for Real-Time Communication” is free to read here on the web, and the FastAPI Backend Development Bootcamp course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the FastAPI Backend Development Bootcamp course, upgrade to CoddyKit PRO.
What will I learn in “WebSockets for Real-Time Communication”?
Build real-time features in FastAPI with WebSockets: accept connections, exchange messages, broadcast to many clients, and handle disconnects cleanly using async patterns. You practise FastAPI Backend Development Bootcamp with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start FastAPI Backend Development Bootcamp?
No prior experience is required. FastAPI Backend Development Bootcamp on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “WebSockets for Real-Time Communication” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this FastAPI Backend Development Bootcamp lesson?
Yes. Every FastAPI Backend Development Bootcamp lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Async/Await in Python Refresher
- FastAPI and Async Operations
- Executing Background Tasks
- WebSockets for Real-Time Communication