Create a Real-Time Chat App with Python, WebSockets, and FastAPI

by Didin J. on Aug 22, 2025 Create a Real-Time Chat App with Python, WebSockets, and FastAPI

Build a real-time chat app with FastAPI and WebSockets. Includes frontend, broadcast manager, rooms, auth-lite, Redis scaling, and deployment tips.

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 and Upgrade/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:

Thanks!