Right, so you’ve graduated from the humble HTTP request-response cycle. Good for you. It’s a fine model, but it’s a bit like passing notes in class—you have to initiate every single conversation. Sometimes, you need a proper back-and-forth, a continuous stream of chatter between the client and server. That’s where WebSockets come in, and FastAPI, true to form, makes implementing them almost stupidly simple.

Let’s be clear: a WebSocket is a persistent, bidirectional communication channel over a single TCP connection. Once established, both you (the client) and I (the server) can send messages to each other at any time, without the overhead of HTTP headers for every single ping-pong. It’s the foundation for real-time stuff: chat apps, live notifications, collaborative editors, and, of course, incredibly frustrating multiplayer games.

The Absolute Basics: Your First Echo Server

FastAPI handles WebSockets in a way that will feel very familiar if you’ve used its HTTP endpoint machinery. You create a route and instead of a get or post decorator, you use websocket. Here’s the canonical “echo” example—the “Hello, World” of this domain.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

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

Let’s autopsy this. First, we await websocket.accept(). This is the server formally agreeing to the WebSocket handshake. You must call this; forgetting it is a classic rookie mistake that leaves the client hanging.

Then, we enter a loop. Inside, we await websocket.receive_text(). This call patiently waits, not blocking other requests thanks to async, until a message arrives from the client. When it does, we just send it right back with websocket.send_text(). The loop keeps this going forever.

The WebSocketDisconnect exception is how we know the client has gracefully (or ungracefully) closed the connection. Catching it lets us clean up resources. This is crucial. Without this try/except, a client disconnecting would crash your worker with a traceback. Not a good look.

Handling More Than Just Text

The world isn’t made of text. Sometimes you need to send JSON, or maybe even binary data like an image fragment. The WebSocket object has methods for these too.

@app.websocket("/ws/data")
async def websocket_data_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            # Wait for either text or bytes
            message = await websocket.receive()
            if message["type"] == "websocket.receive.text":
                json_data = json.loads(message["text"])
                # Process your JSON...
                await websocket.send_json({"status": "received"})
            elif message["type"] == "websocket.receive.bytes":
                binary_data = message["bytes"]
                # Process your bytes...
                await websocket.send_bytes(binary_data)
    except WebSocketDisconnect:
        pass

Here, we use the generic receive() method which returns a dict describing the event. This is more verbose but gives you maximum flexibility. For most cases, the convenience methods like receive_text() and receive_json() are what you’ll use. They handle the parsing for you.

# Much simpler for JSON
data = await websocket.receive_json()
await websocket.send_json({"response": "Got it!"})

The Real-World Complication: Managing Multiple Connections

An echo server is cute, but useless. The real power—and complexity—comes from managing multiple connected clients. Imagine a chat room: when one client sends a message, you need to broadcast it to all other connected clients. This means you need to keep track of who’s connected.

This is where people often reach for a list to store connections. Don’t. A naive list will cause you pain when you try to remove disconnected clients. Use a proper data structure. Here’s a robust pattern using a set and a manager class.

from fastapi import WebSocket
from typing import Set

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

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

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

    async def broadcast(self, message: str):
        # Disconnections can happen mid-broadcast, so we need to be safe.
        for connection in self.active_connections.copy():
            try:
                await connection.send_text(message)
            except RuntimeError:  # Catches issues like disconnected clients
                self.disconnect(connection)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            message = await websocket.receive_text()
            await manager.broadcast(f"Client says: {message}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("A client has left the chat.")

Notice the active_connections.copy() in the broadcast loop? That’s a critical best practice. You’re iterating over a set that other tasks might be modifying (by removing connections). Iterating over a copy prevents a RuntimeError as the set size changes during iteration.

The Rough Edges and Pitfalls

FastAPI’s WebSocket implementation is brilliant but has its quirks. For one, the automatic dependency injection you know and love for HTTP routes? It doesn’t work for WebSocket endpoints. You can’t just declare a Depends() in the function signature. This is the biggest gotcha. If you need a database session, you have to create it manually inside the endpoint.

Also, remember this is all async. Your receive and send calls must be awaited. If you block the event loop with a CPU-intensive task inside your WebSocket handler, you’re going to have a bad time.

Finally, scaling out is a whole other beast. The ConnectionManager above is in-memory. The second you run more than one server process, you have a problem: a user connected to Process A won’t receive broadcasts from a user connected to Process B. For that, you need a proper distributed pub/sub system like Redis. But that, my friend, is a topic for another section. For now, master the single-server case. It’s where 80% of the battles are fought.