Build a multi-room chat server where clients connect over WebSockets and receive messages instantly — no polling required.

What You’ll Build 

  WebSocket /ws/{room}     Join a room, send/receive messages
GET    /                  Simple HTML chat client
GET    /rooms             List active rooms and user counts

Features:

  • Multiple chat rooms
  • Broadcast messages to all clients in a room
  • Connection/disconnection notifications
  • In-memory message history (last 50 messages per room)

Prerequisites

Setup 

  mkdir chat-app && cd chat-app
python -m venv .venv && source .venv/bin/activate
pip install "fastapi[standard]" uvicorn

Project Structure 

  chat-app/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── manager.py
│   └── static/
│       └── index.html
└── tests/
    └── test_chat.py

Connection Manager

Central class to track active WebSocket connections per room:

  # app/manager.py
from fastapi import WebSocket
from collections import defaultdict
import json

class ConnectionManager:
    def __init__(self):
        self.active: dict[str, list[WebSocket]] = defaultdict(list)
        self.history: dict[str, list[dict]] = defaultdict(list)
        self.MAX_HISTORY = 50

    async def connect(self, room: str, websocket: WebSocket, username: str):
        await websocket.accept()
        self.active[room].append(websocket)
        await self.broadcast(room, {
            "type": "system",
            "message": f"{username} joined the room",
        }, exclude=websocket)
        # Send recent history to the new client
        for msg in self.history[room][-self.MAX_HISTORY:]:
            await websocket.send_json(msg)

    def disconnect(self, room: str, websocket: WebSocket):
        if websocket in self.active[room]:
            self.active[room].remove(websocket)

    async def broadcast(self, room: str, data: dict, exclude: WebSocket | None = None):
        self.history[room].append(data)
        if len(self.history[room]) > self.MAX_HISTORY:
            self.history[room] = self.history[room][-self.MAX_HISTORY:]

        dead = []
        for connection in self.active[room]:
            if connection is exclude:
                continue
            try:
                await connection.send_json(data)
            except Exception:
                dead.append(connection)
        for conn in dead:
            self.disconnect(room, conn)

    def room_stats(self) -> dict[str, int]:
        return {room: len(conns) for room, conns in self.active.items() if conns}

manager = ConnectionManager()

WebSocket Endpoint 

  # app/main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from pathlib import Path
from app.manager import manager

app = FastAPI(title="Chat App")
app.mount("/static", StaticFiles(directory=Path(__file__).parent / "static"), name="static")

@app.get("/")
async def home():
    html = (Path(__file__).parent / "static" / "index.html").read_text()
    return HTMLResponse(html)

@app.get("/rooms")
async def list_rooms():
    return manager.room_stats()

@app.websocket("/ws/{room}")
async def websocket_endpoint(
    websocket: WebSocket,
    room: str,
    username: str = Query(default="Anonymous"),
):
    await manager.connect(room, websocket, username)
    try:
        while True:
            text = await websocket.receive_text()
            await manager.broadcast(room, {
                "type": "message",
                "username": username,
                "message": text,
            })
    except WebSocketDisconnect:
        manager.disconnect(room, websocket)
        await manager.broadcast(room, {
            "type": "system",
            "message": f"{username} left the room",
        })

HTML Client 

  <!-- app/static/index.html -->
<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <title>Chat Room</title>
  <style>
    body { font-family: sans-serif; max-width: 600px; margin: 2rem auto; }
    #messages { border: 1px solid #ccc; height: 300px; overflow-y: auto; padding: 1rem; }
    .system { color: #888; font-style: italic; }
    input, button { margin-top: 0.5rem; padding: 0.5rem; }
  </style>
</head>
<body>
  <h1>Python Chat</h1>
  <label>Username: <input id="username" value="Alice"></label>
  <label>Room: <input id="room" value="general"></label>
  <button onclick="connect()">Connect</button>
  <div id="messages"></div>
  <input id="input" placeholder="Type a message..." disabled>
  <button id="send" onclick="sendMessage()" disabled>Send</button>

  <script>
    let ws;
    const messages = document.getElementById("messages");
    const input = document.getElementById("input");
    const sendBtn = document.getElementById("send");

    function connect() {
      const username = document.getElementById("username").value;
      const room = document.getElementById("room").value;
      ws = new WebSocket(`ws://${location.host}/ws/${room}?username=${encodeURIComponent(username)}`);

      ws.onmessage = (event) => {
        const data = JSON.parse(event.data);
        const div = document.createElement("div");
        if (data.type === "system") {
          div.className = "system";
          div.textContent = data.message;
        } else {
          div.textContent = `${data.username}: ${data.message}`;
        }
        messages.appendChild(div);
        messages.scrollTop = messages.scrollHeight;
      };

      ws.onopen = () => {
        input.disabled = false;
        sendBtn.disabled = false;
      };
    }

    function sendMessage() {
      if (ws && input.value.trim()) {
        ws.send(input.value);
        input.value = "";
      }
    }

    input.addEventListener("keydown", (e) => {
      if (e.key === "Enter") sendMessage();
    });
  </script>
</body>
</html>

Run the Server 

  uvicorn app.main:app --reload
# Open http://127.0.0.1:8000 in two browser tabs

Join the same room with different usernames and chat in real time.

Testing 

  # tests/test_chat.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_list_rooms_empty():
    response = client.get("/rooms")
    assert response.status_code == 200
    assert isinstance(response.json(), dict)

def test_websocket_chat():
    with client.websocket_connect("/ws/general?username=Alice") as ws1:
        with client.websocket_connect("/ws/general?username=Bob") as ws2:
            ws1.send_text("Hello Bob!")
            data = ws2.receive_json()
            assert data["type"] == "message"
            assert data["username"] == "Alice"
            assert data["message"] == "Hello Bob!"

Run: pytest tests/ -v

How It Works 

  Client A ──WebSocket──┐
                      ├── ConnectionManager ── broadcast ──► all clients in room
Client B ──WebSocket──┘
  1. Client opens WebSocket to /ws/{room}?username=...
  2. Server accepts connection and adds to room’s active list
  3. Incoming text is broadcast as JSON to every client in the room
  4. On disconnect, server removes client and notifies others

Concepts Applied

Bonus Challenges

  1. Persist messages — store history in Redis or PostgreSQL instead of memory
  2. Authentication — require JWT before WebSocket upgrade
  3. Private messages — direct messaging between two users
  4. Rate limiting — throttle messages per user (prevent spam)
  5. Typing indicators — broadcast {type: "typing"} events
  6. Deploy with Docker — see DevOps

Real-time features like chat, live dashboards, and notifications all build on this WebSocket pattern.