Build a production-ready real-time chat with FastAPI and WebSockets. You’ll implement a lightweight backend, a minimal HTML/JS client, message broadcasting, and optional upgrades like JWT auth and Redis Pub/Sub for horizontal scaling.
What You’ll Build
A multi-user chat where each browser tab connects over a WebSocket, sends messages, and receives real-time updates from everyone else. We’ll start with a single global room and then show how to extend it to multiple rooms.
Prerequisites
-
Python 3.11+
-
Basic FastAPI + HTTP concepts
-
Familiarity with WebSockets
-
Node.js is optional (we’ll serve a simple HTML/JS client from FastAPI)
1) Project Setup
Create a project folder
mkdir fastapi-realtime-chat
cd fastapi-realtime-chat
Create a virtual environment & activate
python3 -m venv .venv
# macOS/Linux
source .venv/bin/activate
# Windows (PowerShell)
# .venv\Scripts\Activate.ps1
Add dependencies
Create requirements.txt:
fastapi==0.115.0
uvicorn[standard]==0.30.6
python-dotenv==1.0.1
itsdangerous==2.2.0
Notes:
uvicorn[standard]
pulls in uvloop & httptools for speed.
itsdangerous
is used later for a tiny signed-token example (optional JWT-like).
Install:
pip install -r requirements.txt
Recommended structure
fastapi-realtime-chat/
├─ app/
│ ├─ __init__.py
│ ├─ main.py
│ ├─ manager.py
│ ├─ utils.py
│ └─ static/
│ ├─ index.html
│ └─ styles.css
├─ requirements.txt
└─ .env # optional for secrets
Create folders/files as shown. We’ll fill them next.
2) WebSocket Connection Manager
We’ll keep track of connected clients, broadcast messages, and handle join/leave events.
Create app/manager.py:
from typing import Set
from fastapi import WebSocket
class ConnectionManager:
"""Keeps a set of active WebSocket connections and broadcasts messages."""
def __init__(self) -> None:
self.active_connections: Set[WebSocket] = set()
async def connect(self, websocket: WebSocket) -> None:
await websocket.accept()
self.active_connections.add(websocket)
def disconnect(self, websocket: WebSocket) -> None:
self.active_connections.discard(websocket)
async def broadcast(self, message: str) -> None:
to_remove = []
for connection in list(self.active_connections):
try:
await connection.send_text(message)
except Exception:
# connection likely closed
to_remove.append(connection)
for conn in to_remove:
self.disconnect(conn)
3) FastAPI App with WebSocket Route
Create app/main.py:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from pathlib import Path
import json
from .manager import ConnectionManager
app = FastAPI(title="FastAPI Real-Time Chat")
manager = ConnectionManager()
# (Optional) CORS if you serve a separate frontend
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Serve static files (our simple chat UI)
static_dir = Path(__file__).parent / "static"
app.mount("/static", StaticFiles(directory=static_dir), name="static")
@app.get("/")
async def root():
# Return the HTML client
html_path = static_dir / "index.html"
return HTMLResponse(html_path.read_text(encoding="utf-8"))
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
# Expect a query parameter "username" for display
username = websocket.query_params.get("username", "Guest")
await manager.connect(websocket)
await manager.broadcast(json.dumps({
"type": "system",
"message": f"{username} joined the chat",
}))
try:
while True:
text = await websocket.receive_text()
await manager.broadcast(json.dumps({
"type": "chat",
"user": username,
"message": text,
}))
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(json.dumps({
"type": "system",
"message": f"{username} left the chat",
}))
4) Minimal Frontend (HTML + JS)
Create app/static/index.html:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>FastAPI WebSocket Chat</title>
<link rel="stylesheet" href="/static/styles.css" />
</head>
<body>
<div class="chat-container">
<header>
<h1>FastAPI WebSocket Chat</h1>
<div class="user">
<label>Username</label>
<input id="username" placeholder="Type a name…" />
<button id="connectBtn">Connect</button>
<button id="disconnectBtn" disabled>Disconnect</button>
</div>
</header>
<main id="messages" class="messages"></main>
<footer>
<input
id="input"
placeholder="Write a message and hit Enter…"
disabled
/>
<button id="sendBtn" disabled>Send</button>
</footer>
</div>
<script>
function connect() {
const username = usernameInput.value.trim() || "Guest";
if (ws) ws.close();
ws = new WebSocket(
`ws://${location.host}/ws?username=${encodeURIComponent(username)}`
);
ws.addEventListener("open", () => {
input.disabled = false;
sendBtn.disabled = false;
connectBtn.disabled = true;
disconnectBtn.disabled = false;
appendMessage("Connected ✔", "system");
});
ws.addEventListener("message", (event) => {
try {
const data = JSON.parse(event.data);
if (data.type === "system") {
appendMessage(data.message, "system");
} else if (data.type === "chat") {
appendMessage(`${data.user}: ${data.message}`);
} else {
appendMessage(event.data);
}
} catch (e) {
appendMessage(event.data);
}
});
ws.addEventListener("close", () => {
input.disabled = true;
sendBtn.disabled = true;
connectBtn.disabled = false;
disconnectBtn.disabled = true;
appendMessage("Disconnected ✖", "system");
});
ws.addEventListener("error", () => {
appendMessage("WebSocket error", "system");
});
}
function disconnect() {
if (ws) {
ws.close();
}
}
sendBtn.addEventListener("click", () => {
if (ws && input.value.trim()) {
ws.send(input.value.trim());
input.value = "";
}
});
input.addEventListener("keydown", (e) => {
if (e.key === "Enter") {
sendBtn.click();
}
});
connectBtn.addEventListener("click", connect);
disconnectBtn.addEventListener("click", disconnect);
</script>
</body>
</html>
Create app/static/styles.css (optional but nice):
:root {
font-family: system-ui, -apple-system, Segoe UI, Roboto, sans-serif;
}
body {
margin: 0;
background: #0f172a;
color: #e2e8f0;
}
.chat-container {
max-width: 800px;
margin: 0 auto;
min-height: 95vh;
display: grid;
grid-template-rows: auto 1fr auto;
gap: 12px;
padding: 16px;
}
header {
display: flex;
align-items: center;
justify-content: space-between;
gap: 12px;
}
header h1 {
margin: 0;
font-size: 1.25rem;
}
.user {
display: flex;
align-items: center;
gap: 8px;
}
.user input {
padding: 8px;
border-radius: 8px;
border: 1px solid #334155;
background: #0b1220;
color: #e2e8f0;
}
.user button {
padding: 8px 12px;
border-radius: 8px;
border: 0;
background: #22c55e;
color: #0b1220;
cursor: pointer;
font-weight: 600;
}
.user button[disabled] {
background: #475569;
color: #94a3b8;
cursor: not-allowed;
}
.messages {
background: #0b1220;
border: 1px solid #334155;
border-radius: 12px;
padding: 12px;
overflow-y: auto;
}
.message {
padding: 6px 8px;
margin: 6px 0;
background: #111827;
border-radius: 8px;
}
.message.system {
background: #1f2937;
color: #a3e635;
font-weight: 600;
}
footer {
display: flex;
gap: 8px;
}
footer input {
flex: 1;
padding: 10px 12px;
border-radius: 10px;
border: 1px solid #334155;
background: #0b1220;
color: #e2e8f0;
}
footer button {
padding: 10px 12px;
border-radius: 10px;
border: 0;
background: #22c55e;
color: #0b1220;
font-weight: 700;
cursor: pointer;
}
footer button[disabled] {
background: #475569;
color: #94a3b8;
cursor: not-allowed;
}
5) Run & Test
Start the server:
uvicorn app.main:app --reload --port 8000
Open http://localhost:8000 in two browser tabs, set usernames, click Connect, and start chatting.
If you deploy behind a reverse proxy on HTTPS, your WebSocket URL will be
wss://your-domain/ws
.
6) Optional: Multiple Rooms
Add a room
query param and store connections per-room.
Update app/manager.py to support rooms:
from typing import Dict, Set
from fastapi import WebSocket
class RoomedConnectionManager:
def __init__(self) -> None:
self.rooms: Dict[str, Set[WebSocket]] = {}
async def connect(self, room: str, websocket: WebSocket) -> None:
await websocket.accept()
self.rooms.setdefault(room, set()).add(websocket)
def disconnect(self, room: str, websocket: WebSocket) -> None:
if room in self.rooms:
self.rooms[room].discard(websocket)
if not self.rooms[room]:
self.rooms.pop(room, None)
async def broadcast(self, room: str, message: str) -> None:
for ws in list(self.rooms.get(room, [])):
try:
await ws.send_text(message)
except Exception:
self.disconnect(room, ws)
Then in app/main.py:
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
username = websocket.query_params.get("username", "Guest")
room = websocket.query_params.get("room", "lobby")
await manager.connect(room, websocket)
await manager.broadcast(room, json.dumps({"type": "system", "message": f"{username} joined {room}"}))
try:
while True:
text = await websocket.receive_text()
await manager.broadcast(room, json.dumps({"type": "chat", "user": username, "message": text}))
except WebSocketDisconnect:
manager.disconnect(room, websocket)
await manager.broadcast(room, json.dumps({"type": "system", "message": f"{username} left {room}"}))
On the client, connect with ws://host/ws?username=Ann&room=general
.
7) Optional: Add a Simple Signed Token (Auth Lite)
For a lightweight auth gate without a full JWT provider:
Create app/utils.py:
import os
from itsdangerous import URLSafeSerializer
SECRET = os.getenv("CHAT_SECRET", "change-me")
signer = URLSafeSerializer(SECRET, salt="chat")
def issue_token(username: str) -> str:
return signer.dumps({"u": username})
def verify_token(token: str) -> str | None:
try:
data = signer.loads(token)
return data.get("u")
except Exception:
return None
Protect the WebSocket endpoint in app/main.py:
from .utils import verify_token
@app.websocket("/ws-secure")
async def ws_secure(websocket: WebSocket):
token = websocket.query_params.get("token")
username = verify_token(token) if token else None
if not username:
await websocket.close(code=4401) # 4401 Unauthorized
return
await manager.connect(websocket)
await manager.broadcast(json.dumps({"type": "system", "message": f"{username} joined"}))
try:
while True:
text = await websocket.receive_text()
await manager.broadcast(json.dumps({"type": "chat", "user": username, "message": text}))
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(json.dumps({"type": "system", "message": f"{username} left"}))
You can expose a tiny endpoint to mint tokens (in a real app, issue tokens after login):
from fastapi import Depends
@app.get("/token")
async def token(username: str):
from .utils import issue_token
return {"token": issue_token(username)}
Client connects using: ws://host/ws-secure?token=<issued-token>
.
8) Optional: Scale with Redis Pub/Sub
When running multiple Uvicorn workers or replicas on Kubernetes, in-process broadcasting won’t reach clients connected to other pods. Use Redis Pub/Sub to fan-out messages across instances.
Install:
pip install redis==5.0.8
Example Redis-backed broadcaster (append in a new file app/redis_broadcast.py):
import asyncio
import json
import os
from typing import Set
from fastapi import WebSocket
import redis.asyncio as redis
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
class RedisBroadcaster:
def __init__(self) -> None:
self.local_clients: Set[WebSocket] = set()
self.r = redis.from_url(REDIS_URL, decode_responses=True)
self.channel = "chat:global"
self._sub_task: asyncio.Task | None = None
async def start(self) -> None:
async def reader():
pubsub = self.r.pubsub()
await pubsub.subscribe(self.channel)
async for msg in pubsub.listen():
if msg.get("type") == "message":
data = msg.get("data")
# fan-out to local websockets
dead = []
for ws in list(self.local_clients):
try:
await ws.send_text(data)
except Exception:
dead.append(ws)
for d in dead:
self.local_clients.discard(d)
self._sub_task = asyncio.create_task(reader())
async def stop(self) -> None:
if self._sub_task:
self._sub_task.cancel()
async def connect(self, ws: WebSocket):
await ws.accept()
self.local_clients.add(ws)
def disconnect(self, ws: WebSocket):
self.local_clients.discard(ws)
async def broadcast(self, message: str):
await self.r.publish(self.channel, message)
Wire it in app/main.py:
from .redis_broadcast import RedisBroadcaster
broadcaster = RedisBroadcaster()
@app.on_event("startup")
async def on_start():
await broadcaster.start()
@app.on_event("shutdown")
async def on_stop():
await broadcaster.stop()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
username = websocket.query_params.get("username", "Guest")
await broadcaster.connect(websocket)
await broadcaster.broadcast(json.dumps({"type": "system", "message": f"{username} joined"}))
try:
while True:
text = await websocket.receive_text()
await broadcaster.broadcast(json.dumps({"type": "chat", "user": username, "message": text}))
except WebSocketDisconnect:
broadcaster.disconnect(websocket)
await broadcaster.broadcast(json.dumps({"type": "system", "message": f"{username} left"}))
docker-compose.yml for local Redis:
version: "3.9"
services:
redis:
image: redis:7-alpine
ports:
- "6379:6379"
Run: docker compose up -d
then start FastAPI with REDIS_URL=redis://localhost:6379/0
.
9) Deployment Notes
-
Uvicorn workers (multi-core):
pip install gunicorn
gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 2 -b 0.0.0.0:8000
-
HTTPS & Proxy: Put Nginx/Caddy in front; enable
proxy_http_version 1.1
andUpgrade
/Connection
headers for WebSockets. -
Kubernetes: Use sticky sessions or Redis Pub/Sub.
-
Health checks: Expose a simple GET
/healthz
returning{status: ok}
.
Example Nginx snippet:
location /ws {
proxy_pass http://app_upstream;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_read_timeout 3600;
}
10) Troubleshooting
-
CORS errors: if serving frontend elsewhere, add its origin to
allow_origins
. -
Mixed content: use
wss://
when your site is on HTTPS. -
No messages on multi-replica: you likely need Redis Pub/Sub.
-
Port blocked: ensure 8000 is open locally or via your proxy.
11) Next Steps & Enhancements
-
User list presence (heartbeat pings + set of usernames)
-
Message persistence (SQLite/Postgres) + REST history endpoint
-
Typing indicators (broadcast
typing
events) -
File/image message uploads (multipart + object storage)
-
Full JWT auth with
pyjwt
or an IdP (Auth0, Cognito) -
Multiple rooms + private DMs
Final Thoughts
You now have a concise but scalable real-time chat built with FastAPI and WebSockets, ready to evolve into a full-featured messaging app. Start simple, then layer in auth, persistence, and horizontal scaling as your traffic grows.
You can get the full source code on our GitHub.
That's just the basics. If you need more deep learning about Python and the frameworks, you can take the following cheap course:
-
Edureka's Django course helps you gain expertise in Django REST framework, Django Models, Django AJAX, Django jQuery etc. You'll master Django web framework while working on real-time use cases and receive Django certification at the end of the course.
-
Unlock your coding potential with Python Certification Training. Avail Flat 25% OFF, coupon code: TECHIE25
-
Database Programming with Python
-
Python Programming: Build a Recommendation Engine in Django
-
Python Course:Learn Python By building Games in Python.
-
Learn API development with Fast API + MySQL in Python
-
Learn Flask, A web Development Framework of Python
Thanks!