Module 5: WebSockets and File Uploads

ConnectionManager and Broadcasting

Next →

Capsule overview

In the previous capsule you built WebSocket endpoints that talk to a single client at a time. But the real power of WebSockets is one-to-many communication: when someone creates a task, every connected user should see it appear instantly. That's broadcasting.

The problem: FastAPI doesn't have a built-in system for managing multiple WebSocket connections. You have to build it yourself. The standard pattern is the ConnectionManager — a class that keeps a list of active connections and offers methods for broadcasting, targeted sending, and handling disconnections.

By the end of this capsule you'll know how to build a complete ConnectionManager, broadcast to every client, send messages to specific clients, handle rooms/channels, and manage disconnections without leaving zombie connections behind.

The problem: multiple clients

Without a ConnectionManager

Imagine 3 users are connected to your API:

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        pass

Each connection lives inside its own instance of the function. There's no way for one connection to say anything to the others. If User A creates a task, Users B and C don't find out until they poll over HTTP.

The solution: shared state

You need an object that lives outside the individual functions and that every connection shares:

ConnectionManager (global)
├── User A's connection
├── User B's connection
└── User C's connection

When A sends a message → the Manager sends it to B and C too

ConnectionManager: a basic implementation

from fastapi import WebSocket


class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

Let's break it down:

active_connections — A list that holds every active WebSocket connection

connect() — Accepts the connection and adds it to the list

disconnect() — Removes the connection from the list (it doesn't close it — it's already closed)

send_personal() — Sends to one specific client

broadcast() — Sends to every connected client

Using it in FastAPI

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()
manager = ConnectionManager()


@app.websocket("/ws/{client_id}")
async def websocket_endpoint(ws: WebSocket, client_id: str):
    await manager.connect(ws)
    await manager.broadcast(f"🟢 {client_id} connected. Total: {len(manager.active_connections)}")

    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast(f"{client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(ws)
        await manager.broadcast(f"🔴 {client_id} disconnected. Total: {len(manager.active_connections)}")

Now when one client sends a message, everyone receives it.

ConnectionManager with JSON

In production, messages are always JSON:

from fastapi import WebSocket
from datetime import datetime


class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        if websocket in self.active_connections:
            self.active_connections.remove(websocket)

    async def send_personal(self, data: dict, websocket: WebSocket):
        await websocket.send_json(data)

    async def broadcast(self, data: dict):
        disconnected = []
        for connection in self.active_connections:
            try:
                await connection.send_json(data)
            except Exception:
                disconnected.append(connection)

        for conn in disconnected:
            self.disconnect(conn)

    async def broadcast_event(self, event_type: str, payload: dict):
        message = {
            "type": event_type,
            "data": payload,
            "timestamp": datetime.now().isoformat()
        }
        await self.broadcast(message)

Important improvements:

if websocket in active_connections — A defensive check before removing

Try/except in broadcast — If a client disconnected without warning, it doesn't crash the whole broadcast

Cleaning up the disconnected — Connections that fail get removed automatically

broadcast_event() — A convenience method that adds a type and a timestamp

Using it with events

app = FastAPI()
manager = ConnectionManager()


@app.post("/tasks", status_code=201)
async def create_task(title: str):
    task = {"id": 1, "title": title, "status": "pending"}

    await manager.broadcast_event("task_created", task)

    return task


@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await manager.connect(ws)
    await manager.send_personal(
        {"type": "connected", "message": "Connected to the event stream"},
        ws
    )

    try:
        while True:
            await ws.receive_text()
    except WebSocketDisconnect:
        manager.disconnect(ws)

Now your HTTP endpoints (POST, PUT, DELETE) can notify every connected WebSocket client. This is the real pattern behind real-time notifications.

Pattern: rooms

Often you don't want a global broadcast. You want to send messages only to a group of users — for example, the members of a team or the people assigned to a project.

from fastapi import WebSocket
from datetime import datetime


class RoomManager:
    def __init__(self):
        self.rooms: dict[str, list[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room: str):
        await websocket.accept()
        if room not in self.rooms:
            self.rooms[room] = []
        self.rooms[room].append(websocket)

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

    async def broadcast_to_room(self, room: str, data: dict):
        if room not in self.rooms:
            return

        disconnected = []
        for connection in self.rooms[room]:
            try:
                await connection.send_json(data)
            except Exception:
                disconnected.append(connection)

        for conn in disconnected:
            self.disconnect(conn, room)

    def get_room_count(self, room: str) -> int:
        return len(self.rooms.get(room, []))

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

Using RoomManager

app = FastAPI()
room_manager = RoomManager()


@app.websocket("/ws/room/{room_name}")
async def websocket_room(ws: WebSocket, room_name: str):
    await room_manager.connect(ws, room_name)
    await room_manager.broadcast_to_room(room_name, {
        "type": "system",
        "message": f"New user in {room_name}",
        "users_in_room": room_manager.get_room_count(room_name)
    })

    try:
        while True:
            data = await ws.receive_json()
            data["room"] = room_name
            data["timestamp"] = datetime.now().isoformat()
            await room_manager.broadcast_to_room(room_name, data)
    except WebSocketDisconnect:
        room_manager.disconnect(ws, room_name)
        await room_manager.broadcast_to_room(room_name, {
            "type": "system",
            "message": f"A user left {room_name}",
            "users_in_room": room_manager.get_room_count(room_name)
        })


@app.get("/rooms")
def list_rooms():
    return room_manager.get_all_rooms()

Pattern: targeted sending to a specific user

Sometimes you need to send a message to one specific user, not to everyone. For that you need to map users to connections:

from fastapi import WebSocket
from datetime import datetime


class UserConnectionManager:
    def __init__(self):
        self.connections: dict[str, WebSocket] = {}

    async def connect(self, user_id: str, websocket: WebSocket):
        await websocket.accept()
        self.connections[user_id] = websocket

    def disconnect(self, user_id: str):
        self.connections.pop(user_id, None)

    async def send_to_user(self, user_id: str, data: dict):
        ws = self.connections.get(user_id)
        if ws:
            try:
                await ws.send_json(data)
                return True
            except Exception:
                self.disconnect(user_id)
        return False

    async def broadcast(self, data: dict, exclude: str | None = None):
        disconnected = []
        for user_id, ws in self.connections.items():
            if user_id == exclude:
                continue
            try:
                await ws.send_json(data)
            except Exception:
                disconnected.append(user_id)

        for uid in disconnected:
            self.disconnect(uid)

    def is_online(self, user_id: str) -> bool:
        return user_id in self.connections

    def online_users(self) -> list[str]:
        return list(self.connections.keys())

Using it: a notification when a task is assigned to you

app = FastAPI()
user_manager = UserConnectionManager()


@app.websocket("/ws/user/{user_id}")
async def websocket_user(ws: WebSocket, user_id: str):
    await user_manager.connect(user_id, ws)
    await user_manager.broadcast(
        {"type": "user_online", "user": user_id, "online": user_manager.online_users()},
        exclude=user_id
    )

    try:
        while True:
            await ws.receive_text()
    except WebSocketDisconnect:
        user_manager.disconnect(user_id)
        await user_manager.broadcast(
            {"type": "user_offline", "user": user_id}
        )


@app.post("/tasks/{task_id}/assign/{user_id}")
async def assign_task(task_id: int, user_id: str):
    sent = await user_manager.send_to_user(user_id, {
        "type": "task_assigned",
        "task_id": task_id,
        "message": f"Task {task_id} was assigned to you"
    })

    return {
        "assigned_to": user_id,
        "notification_sent": sent,
        "user_online": user_manager.is_online(user_id)
    }

Integrating with APIRouter

In a modular app, the WebSocket manager is global but the endpoints live in a router:

# app/websocket_manager.py
from fastapi import WebSocket
from datetime import datetime


class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        if websocket in self.active_connections:
            self.active_connections.remove(websocket)

    async def broadcast(self, data: dict):
        disconnected = []
        for conn in self.active_connections:
            try:
                await conn.send_json(data)
            except Exception:
                disconnected.append(conn)
        for conn in disconnected:
            self.disconnect(conn)


ws_manager = ConnectionManager()
# app/routers/websocket.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from app.websocket_manager import ws_manager

router = APIRouter(tags=["WebSocket"])


@router.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws_manager.connect(ws)
    try:
        while True:
            await ws.receive_text()
    except WebSocketDisconnect:
        ws_manager.disconnect(ws)
# app/routers/tasks.py
from fastapi import APIRouter
from app.websocket_manager import ws_manager

router = APIRouter(prefix="/tasks", tags=["Tasks"])


@router.post("/", status_code=201)
async def create_task(title: str):
    task = {"id": 1, "title": title}
    await ws_manager.broadcast({"type": "task_created", "task": task})
    return task

Exercises

Exercise 1: A multi-user chat

Build a ConnectionManager and a /ws/chat/{username} endpoint where:

  • On connect, everyone is notified that the user joined
  • Messages go out to everyone with the sender's name
  • On disconnect, everyone is notified that the user left
  • Add a GET /chat/stats endpoint that returns the number of connected users
See solution 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from datetime import datetime


class ChatManager:
    def __init__(self):
        self.connections: dict[str, WebSocket] = {}

    async def connect(self, username: str, ws: WebSocket):
        await ws.accept()
        self.connections[username] = ws

    def disconnect(self, username: str):
        self.connections.pop(username, None)

    async def broadcast(self, data: dict, exclude: str | None = None):
        for user, ws in list(self.connections.items()):
            if user == exclude:
                continue
            try:
                await ws.send_json(data)
            except Exception:
                self.connections.pop(user, None)

    @property
    def user_count(self) -> int:
        return len(self.connections)

    @property
    def users(self) -> list[str]:
        return list(self.connections.keys())


app = FastAPI()
chat = ChatManager()


@app.websocket("/ws/chat/{username}")
async def ws_chat(ws: WebSocket, username: str):
    await chat.connect(username, ws)
    await chat.broadcast(
        {"type": "join", "user": username, "online": chat.users},
        exclude=username
    )
    await ws.send_json({"type": "welcome", "online": chat.users})

    try:
        while True:
            text = await ws.receive_text()
            await chat.broadcast({
                "type": "message",
                "from": username,
                "content": text,
                "timestamp": datetime.now().isoformat()
            })
    except WebSocketDisconnect:
        chat.disconnect(username)
        await chat.broadcast({"type": "leave", "user": username, "online": chat.users})


@app.get("/chat/stats")
def chat_stats():
    return {"users_online": chat.user_count, "users": chat.users}

Exercise 2: Rooms with a limit

Create a RoomManager that limits each room to a maximum of 5 users. If the room is full, reject the connection with code 4002.

See solution 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

MAX_ROOM_SIZE = 5


class LimitedRoomManager:
    def __init__(self):
        self.rooms: dict[str, list[WebSocket]] = {}

    def room_size(self, room: str) -> int:
        return len(self.rooms.get(room, []))

    async def connect(self, ws: WebSocket, room: str) -> bool:
        if self.room_size(room) >= MAX_ROOM_SIZE:
            await ws.close(code=4002, reason=f"Room {room} is full ({MAX_ROOM_SIZE} max)")
            return False

        await ws.accept()
        if room not in self.rooms:
            self.rooms[room] = []
        self.rooms[room].append(ws)
        return True

    def disconnect(self, ws: WebSocket, room: str):
        if room in self.rooms and ws in self.rooms[room]:
            self.rooms[room].remove(ws)
            if not self.rooms[room]:
                del self.rooms[room]

    async def broadcast(self, room: str, data: dict):
        for conn in self.rooms.get(room, []):
            try:
                await conn.send_json(data)
            except Exception:
                pass


app = FastAPI()
manager = LimitedRoomManager()


@app.websocket("/ws/room/{room}")
async def ws_room(ws: WebSocket, room: str):
    joined = await manager.connect(ws, room)
    if not joined:
        return

    await manager.broadcast(room, {
        "type": "join",
        "users": manager.room_size(room),
        "max": MAX_ROOM_SIZE
    })

    try:
        while True:
            data = await ws.receive_json()
            await manager.broadcast(room, data)
    except WebSocketDisconnect:
        manager.disconnect(ws, room)

Exercise 3: Selective broadcasting

Modify the ConnectionManager to support tags per connection. Each client connects with one or more tags (for example, "frontend", "admin", "mobile"). Add a broadcast_to_tag(tag, data) method that sends only to the clients carrying that tag.

See solution 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
from typing import Optional


class TaggedManager:
    def __init__(self):
        self.connections: dict[WebSocket, set[str]] = {}

    async def connect(self, ws: WebSocket, tags: set[str]):
        await ws.accept()
        self.connections[ws] = tags

    def disconnect(self, ws: WebSocket):
        self.connections.pop(ws, None)

    async def broadcast_to_tag(self, tag: str, data: dict):
        for ws, tags in list(self.connections.items()):
            if tag in tags:
                try:
                    await ws.send_json(data)
                except Exception:
                    self.connections.pop(ws, None)

    async def broadcast_all(self, data: dict):
        for ws in list(self.connections.keys()):
            try:
                await ws.send_json(data)
            except Exception:
                self.connections.pop(ws, None)


app = FastAPI()
manager = TaggedManager()


@app.websocket("/ws/tagged")
async def ws_tagged(ws: WebSocket, tags: str = Query(default="general")):
    tag_set = set(tags.split(","))
    await manager.connect(ws, tag_set)

    try:
        while True:
            await ws.receive_text()
    except WebSocketDisconnect:
        manager.disconnect(ws)


@app.post("/notify/{tag}")
async def notify_tag(tag: str, message: str):
    await manager.broadcast_to_tag(tag, {"type": "notification", "tag": tag, "message": message})
    return {"sent_to_tag": tag}

Exercise 4: A manager with a heartbeat

Create a ConnectionManager that sends a "ping" to every connection every 30 seconds. If one fails, remove it. Use asyncio.create_task() for the heartbeat.

See solution 
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, WebSocket, WebSocketDisconnect


class HeartbeatManager:
    def __init__(self):
        self.connections: list[WebSocket] = []
        self._heartbeat_task: asyncio.Task | None = None

    async def start_heartbeat(self, interval: int = 30):
        self._heartbeat_task = asyncio.create_task(self._heartbeat_loop(interval))

    async def stop_heartbeat(self):
        if self._heartbeat_task:
            self._heartbeat_task.cancel()

    async def _heartbeat_loop(self, interval: int):
        while True:
            await asyncio.sleep(interval)
            dead = []
            for ws in self.connections:
                try:
                    await ws.send_json({"type": "ping"})
                except Exception:
                    dead.append(ws)
            for ws in dead:
                if ws in self.connections:
                    self.connections.remove(ws)

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.connections.append(ws)

    def disconnect(self, ws: WebSocket):
        if ws in self.connections:
            self.connections.remove(ws)


manager = HeartbeatManager()


@asynccontextmanager
async def lifespan(app):
    await manager.start_heartbeat(interval=30)
    yield
    await manager.stop_heartbeat()


app = FastAPI(lifespan=lifespan)


@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            await ws.receive_text()
    except WebSocketDisconnect:
        manager.disconnect(ws)

Troubleshooting

The broadcast doesn't reach every client

Check that every client is in active_connections. If a client disconnected without going through disconnect(), it's still in the list but fails silently. Add try/except inside the broadcast loop.

list.remove(x): x not in list

You're trying to remove a connection that was already removed. Use if websocket in self.active_connections before removing.

Messages arrive duplicated

You're probably creating multiple instances of the ConnectionManager. It should be a single global instance — a module-level singleton.

The manager loses connections when the server reloads

With --reload, uvicorn restarts the Python process. Every connection is lost. That's normal in development. In production, you don't use --reload.

Slow broadcast with many clients

broadcast() sends sequentially. For hundreds of clients, use asyncio.gather():

async def broadcast(self, data: dict):
    tasks = [conn.send_json(data) for conn in self.active_connections]
    results = await asyncio.gather(*tasks, return_exceptions=True)

Resources

  1. FastAPI WebSockets — The official chat example
  2. Starlette WebSocket — The base API
  3. Real-time apps with FastAPI — A full tutorial
  4. WebSocket scaling patterns — Architecture for production
  5. asyncio.gather — Concurrency for broadcasting

What's next?

In Capsule 04 you'll learn file uploads in FastAPI: how to receive files via UploadFile, validate type and size, save to disk, and combine files with JSON data in the same request.

← PreviousBasic WebSockets in FastAPINext →File Uploads and Validation