Plugin Architecture For Your Code Using pyplugs in Python3

Intermediate

You build an analytics dashboard that shows live traffic numbers. Your users refresh the page every 30 seconds to see if anything changed. That is not a dashboard — that is polling with extra steps. Real-time means the server pushes updates the moment they happen, not when the client gets around to asking. WebSockets make this possible by keeping a persistent, bidirectional connection open between the browser and the server. One connection, two directions, no polling, no wasted requests.

FastAPI has first-class WebSocket support built in — no plugins, no third-party libraries required beyond what you already have installed. You define a WebSocket endpoint exactly like an HTTP route, accept the connection with await websocket.accept(), and start reading and sending messages. FastAPI handles the protocol upgrade from HTTP to WebSocket transparently. For broadcasting to multiple clients — the pattern behind every chat app and live feed — you manage a list of active connections and iterate over it. The pattern is a dozen lines of Python.

This article covers everything you need to build real-time features with FastAPI WebSockets: the basic connection pattern, sending and receiving text and JSON, broadcasting to multiple clients, managing disconnections cleanly, using query parameters for authentication, and a complete real-life chat application. By the end you will have a working multi-room chat server and the patterns to build any real-time feature on top of them.

FastAPI WebSockets: Quick Example

Here is the minimal working WebSocket endpoint in FastAPI. It accepts a connection, reads messages from the client in a loop, and echoes each one back. Run it with uvicorn quick_ws:app --reload and connect from a browser console or a WebSocket client like Hoppscotch.

# quick_ws.py
from fastapi import FastAPI, WebSocket

app = FastAPI()

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

Browser console test (open any tab on the same machine):

const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = e => console.log(e.data);
ws.send("hello");
// Output in console:
// Echo: hello
ws.send("fastapi websockets");
// Echo: fastapi websockets

The @app.websocket decorator declares a WebSocket route the same way @app.get declares an HTTP route. The await websocket.accept() call performs the WebSocket handshake — you must call this before reading or sending anything. The while True loop keeps the connection alive, reading one message at a time with receive_text() and responding with send_text(). When the client disconnects, receive_text() raises a WebSocketDisconnect exception which we will handle in the next sections.

What Are WebSockets and When Should You Use Them?

WebSockets are a communication protocol that upgrades an HTTP connection into a persistent, full-duplex channel. “Full-duplex” means both sides can send messages at any time, independently of each other — unlike HTTP where the client always speaks first and the server can only respond. The connection stays open until one side explicitly closes it or the network drops.

Think of HTTP like a walkie-talkie: one side talks, the other listens, then you swap. WebSockets are like a phone call: both sides can talk whenever they want, and the line stays open until someone hangs up. For anything that needs the server to push data without waiting for the client to ask, WebSockets are the right tool.

Use the comparison table below to decide which transport fits your use case:

Use Case	HTTP / Polling	Server-Sent Events	WebSockets
Chat application	Poor — high latency	Receive only	Best fit
Live dashboard (server pushes)	Acceptable with short poll	Good fit	Good fit
Collaborative editing	Poor	Receive only	Required
Real-time notifications	Wasteful	Good fit	Good fit
File upload / REST API	Best fit	Not applicable	Overkill
One-time data fetch	Best fit	Not applicable	Overkill

WebSockets add connection management complexity that plain HTTP does not have. Use them only when you genuinely need bidirectional or server-push communication. For simple server-push scenarios (like a progress bar or news feed), Server-Sent Events are simpler. For everything truly bidirectional — chat, collaborative tools, multiplayer games — WebSockets are the right choice.

Handling Disconnections Cleanly

When a client closes the browser tab, loses network access, or explicitly calls ws.close(), FastAPI raises a WebSocketDisconnect exception inside your endpoint. If you do not catch it, the exception propagates up and leaves server-side resources in an unknown state. Always wrap the message loop in a try/except block.

# ws_disconnect.py
from fastapi import FastAPI, WebSocket
from starlette.websockets import WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws/safe")
async def safe_websocket(websocket: WebSocket):
    await websocket.accept()
    client_id = id(websocket)
    print(f"Client {client_id} connected")
    try:
        while True:
            data = await websocket.receive_text()
            print(f"Received from {client_id}: {data}")
            await websocket.send_text(f"Got: {data}")
    except WebSocketDisconnect:
        print(f"Client {client_id} disconnected cleanly")
    except Exception as exc:
        print(f"Client {client_id} dropped with error: {exc}")

Server output when a client connects and then closes the tab:

Client 140234567890 connected
Received from 140234567890: hello
Client 140234567890 disconnected cleanly

The WebSocketDisconnect exception comes from Starlette (which FastAPI is built on) and carries a code attribute with the WebSocket close code and a reason string. Normal browser tab closes produce code 1001 (“going away”). Explicit ws.close() calls typically use code 1000 (“normal closure”). Logging the code is useful for diagnosing unexpected disconnections in production.

Sending and Receiving JSON Messages

Text messages work fine for simple cases, but real applications exchange structured data. FastAPI’s WebSocket object has receive_json() and send_json() methods that handle serialization automatically. They use json.loads() and json.dumps() internally, so you work with Python dicts directly.

# ws_json.py
from fastapi import FastAPI, WebSocket
from starlette.websockets import WebSocketDisconnect
import datetime

app = FastAPI()

@app.websocket("/ws/json")
async def json_websocket(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            payload = await websocket.receive_json()
            # Validate that the expected fields are present
            action = payload.get("action", "unknown")
            value = payload.get("value", "")

            response = {
                "action": action,
                "echo": value,
                "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
                "status": "ok",
            }
            await websocket.send_json(response)
    except WebSocketDisconnect:
        pass

Browser console test:

const ws = new WebSocket("ws://localhost:8000/ws/json");
ws.onmessage = e => console.log(JSON.parse(e.data));

ws.send(JSON.stringify({ action: "ping", value: "hello" }));
// Output:
// { action: "ping", echo: "hello", timestamp: "2026-08-07T06:00:00.000Z", status: "ok" }

ws.send(JSON.stringify({ action: "update", value: "42" }));
// { action: "update", echo: "42", timestamp: "2026-08-07T06:00:00.123Z", status: "ok" }

Use .get("key", default) when reading from incoming JSON payloads. Clients can send malformed or incomplete messages, and a missing key without a default raises a KeyError that disconnects them unexpectedly. Defensive reads keep the connection stable even when client code has bugs. If the payload is so malformed that you cannot proceed, send an error response back via send_json() and keep the connection open — let the client decide whether to close it.

Broadcasting to Multiple Clients

Sending a message to one connected client is straightforward. Sending it to all connected clients — the core pattern behind every chat app, live feed, and collaborative tool — requires managing a list of active connections. The connection manager pattern below is the standard approach: a class holds a set of active WebSocket objects, and provides connect, disconnect, and broadcast methods.

# ws_broadcast.py
from fastapi import FastAPI, WebSocket
from starlette.websockets import WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active: List[WebSocket] = []

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

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

    async def broadcast(self, message: str):
        for ws in list(self.active):  # copy to avoid mutation during iteration
            try:
                await ws.send_text(message)
            except Exception:
                pass  # disconnected mid-broadcast; will be removed on next receive

manager = ConnectionManager()

@app.websocket("/ws/broadcast/{client_id}")
async def broadcast_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"[{client_id}]: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"[{client_id}] left the room")

Expected behavior with two browser tabs both connected to /ws/broadcast/alice and /ws/broadcast/bob:

-- Tab 1 (alice) sends "hello" --
Both tabs receive: [alice]: hello

-- Tab 2 (bob) sends "hey there" --
Both tabs receive: [bob]: hey there

-- Tab 1 closes --
Tab 2 receives: [alice] left the room

The list(self.active) copy inside broadcast is important. If a client disconnects mid-broadcast, iterating over the original list while modifying it raises a RuntimeError. The copy lets the loop finish safely. The silent except inside the broadcast loop eats the send failure — the client will be properly removed on their own disconnect path. This is intentional: a failed send to one client should never prevent the message from reaching the others.

Using Query Parameters for Authentication

WebSocket connections cannot carry custom HTTP headers the way REST API calls can — the browser’s WebSocket API only exposes the URL. The standard workaround is to pass a token as a query parameter. FastAPI’s query parameter injection works the same way in WebSocket endpoints as in HTTP routes.

# ws_auth.py
from fastapi import FastAPI, WebSocket, Query, HTTPException
from starlette.websockets import WebSocketDisconnect

app = FastAPI()

# Simulated token store -- in production, validate against your auth system
VALID_TOKENS = {"secret-token-alice": "alice", "secret-token-bob": "bob"}

async def get_user_from_token(token: str) -> str:
    """Return the username for a valid token, or raise an error."""
    user = VALID_TOKENS.get(token)
    if not user:
        raise ValueError(f"Invalid token: {token}")
    return user

@app.websocket("/ws/secure")
async def secure_websocket(
    websocket: WebSocket,
    token: str = Query(..., description="Auth token"),
):
    try:
        user = await get_user_from_token(token)
    except ValueError:
        # Reject the connection before accepting it
        await websocket.close(code=4001, reason="Unauthorized")
        return

    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"[{user}] {data}")
    except WebSocketDisconnect:
        pass

Valid connection URL:

ws://localhost:8000/ws/secure?token=secret-token-alice
# Server accepts, messages are prefixed with [alice]

ws://localhost:8000/ws/secure?token=wrong-token
# Server sends close frame with code 4001 before the connection opens

Closing with code=4001 (a custom application-level code in the 4000-4999 range) signals to the client that authentication failed, so it can display a meaningful error rather than a generic “connection closed” message. Note that you call await websocket.close() BEFORE await websocket.accept() to reject during the handshake. This is one of the few times in a WebSocket endpoint where you do not call accept() first.

WebSockets vs Regular HTTP Endpoints in FastAPI

FastAPI WebSocket endpoints look similar to HTTP endpoints but behave differently in ways that matter for architecture decisions. Understanding the differences prevents common mistakes like expecting dependency injection to work identically, or trying to return a value from a WebSocket route.

Feature	HTTP Route (@app.get, @app.post)	WebSocket Route (@app.websocket)
Connection lifecycle	One request, one response, connection closed	Persistent until explicitly closed
Return value	Return value becomes the response body	No return value — send via websocket.send_*()
Dependencies (Depends)	Fully supported	Supported, but without response object
Path parameters	Yes — @app.get("/{id}")	Yes — @app.websocket("/{id}")
Query parameters	Yes — injected automatically	Yes — same injection mechanism
Error responses	Raise HTTPException	Close with code — no HTTP status codes
Middleware	Applied	Applied during upgrade phase only
Background tasks	Yes via BackgroundTasks	Use asyncio.create_task() instead

The most important practical difference: HTTP exceptions do not work inside WebSocket connections. Once the connection is accepted, you cannot raise HTTPException and expect the client to see a 4xx status code — the protocol is already WebSocket, not HTTP. Close the connection with an appropriate code instead.

Real-Life Example: Multi-Room Chat Server

The example below builds a complete multi-room chat server. Clients connect to /ws/chat/{room_name} with a username query parameter. Messages broadcast to everyone in the same room. The server tracks which rooms are active and how many clients are in each one.

# chat_server.py
from fastapi import FastAPI, WebSocket, Query
from starlette.websockets import WebSocketDisconnect
from collections import defaultdict
from typing import Dict, List
import datetime

app = FastAPI()

class RoomManager:
    def __init__(self):
        # room_name -> list of (websocket, username) tuples
        self.rooms: Dict[str, List[tuple]] = defaultdict(list)

    async def join(self, room: str, websocket: WebSocket, username: str):
        await websocket.accept()
        self.rooms[room].append((websocket, username))
        await self._broadcast(room, f"*** {username} joined {room} ***", sender=None)

    async def leave(self, room: str, websocket: WebSocket, username: str):
        self.rooms[room] = [
            (ws, u) for ws, u in self.rooms[room] if ws is not websocket
        ]
        if not self.rooms[room]:
            del self.rooms[room]
        else:
            await self._broadcast(room, f"*** {username} left {room} ***", sender=None)

    async def send_message(self, room: str, sender: str, text: str):
        ts = datetime.datetime.utcnow().strftime("%H:%M:%S")
        message = f"[{ts}] {sender}: {text}"
        await self._broadcast(room, message, sender=sender)

    async def _broadcast(self, room: str, message: str, sender: str | None):
        dead = []
        for ws, username in list(self.rooms.get(room, [])):
            try:
                await ws.send_text(message)
            except Exception:
                dead.append((ws, username))
        # Clean up dead connections
        for item in dead:
            self.rooms[room] = [x for x in self.rooms[room] if x != item]

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

manager = RoomManager()

@app.websocket("/ws/chat/{room}")
async def chat_endpoint(
    websocket: WebSocket,
    room: str,
    username: str = Query(..., min_length=1, max_length=32),
):
    await manager.join(room, websocket, username)
    try:
        while True:
            text = await websocket.receive_text()
            if text.strip():
                await manager.send_message(room, username, text.strip())
    except WebSocketDisconnect:
        await manager.leave(room, websocket, username)

@app.get("/rooms")
async def list_rooms():
    return {
        "rooms": {
            room: manager.room_count(room)
            for room in manager.rooms
        }
    }

Example session — two clients in “python” room, one in “general”:

-- alice connects to /ws/chat/python?username=alice --
alice sees: *** alice joined python ***

-- bob connects to /ws/chat/python?username=bob --
alice sees: *** bob joined python ***
bob sees:   *** bob joined python ***

-- alice sends "anyone using FastAPI?" --
alice sees: [06:01:23] alice: anyone using FastAPI?
bob sees:   [06:01:23] alice: anyone using FastAPI?

-- GET /rooms --
{ "rooms": { "python": 2 } }

-- alice disconnects --
bob sees: *** alice left python ***

The RoomManager class separates connection state management from the route handler. The route handler stays thin — it joins, loops on messages, and leaves on disconnect. All the broadcasting and cleanup logic lives in the manager. To extend this into a production-ready chat server, add a Redis pub/sub layer (so the manager works across multiple Uvicorn worker processes), persist message history to a database, and add rate limiting to the send_message path to prevent spam.

Frequently Asked Questions

How do I scale WebSockets across multiple workers?

FastAPI with Uvicorn runs in a single process by default. When you run multiple workers with --workers 4, each worker has its own ConnectionManager instance in memory — a client connected to worker 1 will not receive broadcasts sent by a client connected to worker 2. The standard solution is to use Redis pub/sub as a shared message bus: workers subscribe to a Redis channel and forward messages to their local connections. Libraries like broadcaster (pip install broadcaster[redis]) wrap this pattern cleanly and integrate with FastAPI.

How do I detect a dead WebSocket connection?

Networks drop silently. A client can disappear without sending a close frame, and the server will not notice until it tries to send something and gets an error. The standard fix is to implement a ping/pong heartbeat: send a WebSocket ping frame every 30 seconds with await websocket.send_text("__ping__") (or use Starlette’s lower-level websocket.send({"type": "websocket.ping"})) and remove the connection if you do not receive a response within a timeout window. Most browser WebSocket clients automatically respond to protocol-level ping frames, so the lower-level approach is more reliable than application-level pings.

Why can’t I send auth headers in a WebSocket connection?

The browser’s WebSocket API does not expose a headers argument — you can only set the URL and optionally a subprotocol. This is a deliberate design choice in the WebSocket specification. The practical workaround is to pass the token as a query parameter (?token=...) or as a cookie. If you use cookies, the browser sends them automatically with the WebSocket upgrade request, and you can read them from FastAPI’s Request object passed alongside the WebSocket parameter. For short-lived tokens, some applications do a two-step handshake: exchange credentials via HTTPS first, receive a one-time WebSocket token, then connect with that token.

Can I use FastAPI Depends() inside a WebSocket endpoint?

Yes, with one caveat: dependencies that rely on the HTTP response object (like those that set cookies or headers on the response) will not work, because WebSocket endpoints do not have an HTTP response after the handshake. Dependencies that only read data — database sessions, current user from a token, config values — work normally. Inject them the same way you would in an HTTP route: def secure_ws(websocket: WebSocket, user: User = Depends(get_current_user)). FastAPI resolves them before calling the endpoint function.

Is it safe to call send_text() from multiple coroutines simultaneously?

No. A WebSocket object is not concurrency-safe for writes — sending from multiple concurrent coroutines on the same connection can cause corrupted frames or errors. If you need to push messages from a background task and the main receive loop simultaneously, use an asyncio.Queue: the background task puts messages in the queue, and a separate coroutine drains the queue and calls send_text(). This serializes all writes through a single coroutine. Alternatively, use asyncio.Lock to guard the send call explicitly.

What WebSocket close codes should I use in my app?

Codes 1000-1015 are defined by the WebSocket specification and used by the protocol itself. Codes 4000-4999 are reserved for application use — use them to signal application-specific error conditions. Common conventions: 4000 for generic application error, 4001 for authentication failure, 4003 for authorization failure (authenticated but not allowed), 4008 for rate limit exceeded. Clients can inspect the close code in the onclose event (event.code) to display an appropriate message rather than a generic “disconnected” notice.

Conclusion

FastAPI’s WebSocket support gives you real-time bidirectional communication with the same clean, type-annotated API you use for HTTP routes. The core patterns — accept, receive, send, disconnect handling — are straightforward. Broadcasting to multiple clients requires a ConnectionManager class that maintains a list of active connections. Authentication passes through query parameters or cookies since WebSocket connections cannot carry custom headers. The multi-room chat example shows how all of these patterns combine into a working application.

To take the chat server further, add Redis pub/sub via the broadcaster library for multi-process broadcasting, add a heartbeat loop to detect dead connections, and add a message history endpoint so clients can replay recent messages when they reconnect. Each of those extensions plugs into the RoomManager class without touching the route handler.

For official documentation and deeper reference, see FastAPI WebSockets docs and MDN WebSocket API reference.

Related Articles

• How To Use FastAPI Background Tasks in Python
• How To Use Dependency Injection in FastAPI with Depends
• How To Build a REST API with FastAPI in Python