FastAPI × WebSocket Beginner’s Guide: Implementation Patterns for Real-Time Communication, Chat, and Dashboards

---

Summary (Get the Big Picture First)

• WebSocket is a protocol that enables bi-directional, real-time communication between the server and the client, making it a great fit for chat, notifications, and live dashboard updates.
• FastAPI inherits capabilities from Starlette. By using the @app.websocket() decorator and the WebSocket class, you can implement a simple WebSocket server.
• This article introduces step-by-step patterns such as a minimal WebSocket for a single client, broadcasting to multiple clients, and a class-based connection manager.
• It also aims to be practical for real work by including security topics (Origin checks, token authentication), testing WebSockets (how to test with TestClient), and a small front-end sample (JavaScript).
• Finally, it summarizes design points by use case (chat, dashboards, notifications) and provides a roadmap for “where to start.”

---

Who Benefits from Reading This (Concrete Reader Personas)

Indie Developers / Learners

• You want to add something like a “chat-like UI” or a “real-time counter” to your personal service.
• You can build REST APIs, but you’re still fuzzy on “how WebSockets are useful.”
• You want to learn the feel of bi-directional communication while touching both JavaScript and FastAPI.

For these readers, we’ll build up to a working “moving thing” by combining a minimal WebSocket endpoint with a simple HTML + JavaScript chat-like sample.

Backend Engineers in Small Teams

• Your admin screens or dashboards need “instant updates,” and polling is starting to hurt.
• You want to retrofit real-time elements—notifications, chat, progress indicators—into an existing FastAPI app.
• You want to organize connection management (disconnect detection, broadcasting) and testing strategy.

For these readers, patterns like a connection manager class, room separation, and how to test will be especially useful.

SaaS Teams / Startups

• You work in areas where immediacy becomes business value: system monitoring, real-time inventory reflection, notification hubs, etc.
• You want a design that doesn’t paint you into a corner, keeping future scaling (multiple instances, external broker integration) in mind.
• You want to decide authentication, message formats, and room design in a way that stays extensible.

For these readers, we’ll start from a solid “single-instance baseline” and share design hints that anticipate pairing with Redis Pub/Sub or a message broker later.

---

Accessibility Evaluation (Readability and Comprehension)

• Information architecture
  • Concept of WebSocket → FastAPI basics → simple example → connection manager → security → testing → use cases, in staged steps.
• Terminology
  • Uses everyday wording where possible (connection, message, broadcast), with English terms kept minimal and explained at first mention.
• Code
  • Split into smaller blocks so no single snippet gets too long. Comments are kept light to avoid visual noise.
• Reader assumptions
  • Assumes you’ve finished a basic Python + FastAPI tutorial, while also trying to make each chapter understandable on its own.

Overall, the structure is written for a wide range of readers and aims for a WCAG AA-like clarity in text organization.

---

1. What Is WebSocket? A Quick Contrast with HTTP

Let’s briefly clarify “what WebSocket actually is.”

1.1 WebSocket Characteristics

WebSocket is a protocol for long-lived, two-way communication between the browser and the server.

• Once the connection is established, both client → server and server → client can send messages whenever they want.
• Unlike HTTP (“request → response and done”), WebSocket is like reusing a single connection repeatedly.
• It’s commonly used for chat, games, notifications, dashboards—anything requiring real-time updates.

If you try to do this with plain HTTP, you usually need “polling” (calling an API every few seconds), which increases unnecessary traffic and has real limits on real-time responsiveness.

1.2 FastAPI and WebSockets

FastAPI is built on the ASGI framework Starlette, and WebSockets come from Starlette’s functionality.

• HTTP endpoints: @app.get() / @app.post() etc.
• WebSocket endpoints: @app.websocket()

It’s not as special as it sounds—basically, you define one function that receives a WebSocket object “instead of HTTP.”

---

2. WebSocket Endpoints in FastAPI: The Basics

Now let’s look at actual code.

2.1 The Smallest WebSocket Server

First, the smallest example: talking to only one client.

# app/main.py
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws/echo")
async def websocket_echo(websocket: WebSocket):
    # Accept the connection from the client
    await websocket.accept()

    while True:
        # Receive one message (text)
        data = await websocket.receive_text()
        # Send it back as-is
        await websocket.send_text(f"Echo: {data}")

Three key points:

1. Define a WebSocket route with @app.websocket("/ws/echo").
2. Receive a WebSocket object as the function parameter.
3. Accept the connection with await websocket.accept(), then use receive_* / send_*.

Besides receive_text() / send_text(), you can also use binary methods like receive_bytes() / send_bytes(), or JSON helpers like receive_json() / send_json().

2.2 A Simple Client Example (Browser Side)

After starting the server with uvicorn app.main:app --reload, you can try this in the browser console (DevTools):

// In the browser console
const ws = new WebSocket("ws://localhost:8000/ws/echo");

ws.onopen = () => {
  console.log("connected");
  ws.send("hello");
};

ws.onmessage = (event) => {
  console.log("received:", event.data);
};

If you see logs like:

connected
received: Echo: hello

…then you’ve taken your first step into the WebSocket world.

---

3. Supporting Multiple Clients: The Basics of a Connection Manager

In real-time UIs, you almost always have “multiple clients” sharing the same information.
Here we introduce the idea of a “connection manager” that keeps track of active clients.

3.1 Connection Manager Class

# app/websocket_manager.py
from typing import List
from fastapi import WebSocket, WebSocketDisconnect

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_message(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)

• active_connections stores connected WebSocket objects.
• broadcast() sends a message to all active clients.

3.2 A Chat-Like WebSocket Using the Connection Manager

# app/main.py (excerpt)
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from app.websocket_manager import ConnectionManager

app = FastAPI()
manager = ConnectionManager()

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("a client disconnected")

When WebSocketDisconnect occurs, we call disconnect() to remove it from the list and notify remaining clients that someone left.

3.3 A Super Simple HTML Client

Let’s build the smallest chat UI using only HTML.

<!-- Example: static/chat.html -->
<!DOCTYPE html>
<html lang="ja">
  <head>
    <meta charset="UTF-8" />
    <title>FastAPI WebSocket Chat</title>
    <style>
      body { font-family: sans-serif; }
      #messages { border: 1px solid #ccc; height: 200px; overflow-y: scroll; padding: 4px; }
      #input-area { margin-top: 8px; }
    </style>
  </head>
  <body>
    <h1>Chat</h1>
    <div id="messages"></div>
    <div id="input-area">
      <input id="msg" type="text" placeholder="Message" />
      <button id="send">Send</button>
    </div>

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

      const ws = new WebSocket("ws://localhost:8000/ws/chat");

      ws.onmessage = (event) => {
        const div = document.createElement("div");
        div.textContent = event.data;
        messages.appendChild(div);
        messages.scrollTop = messages.scrollHeight;
      };

      sendBtn.onclick = () => {
        if (input.value) {
          ws.send(input.value);
          input.value = "";
        }
      };
    </script>
  </body>
</html>

Even with this simple structure, opening two browser windows lets you experience “a real-time chat where messages instantly appear for each other.”

---

4. Common Practical Patterns: Rooms, Users, and Message Formats

Next, we’ll briefly整理 some patterns that often appear in real chat/dashboards.

4.1 Splitting by Room

Often you want rooms, not one open global chat.
In that case, you can manage connections with a structure like dict[str, list[WebSocket]].

# app/websocket_manager_rooms.py
from collections import defaultdict
from typing import Dict, List
from fastapi import WebSocket, WebSocketDisconnect

class RoomManager:
    def __init__(self):
        self.rooms: Dict[str, List[WebSocket]] = defaultdict(list)

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

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

    async def broadcast(self, room: str, message: str):
        for ws in self.rooms.get(room, []):
            await ws.send_text(message)

You can receive the room name via the WebSocket path or query parameters, then broadcast only to clients in that room.

4.2 Using JSON Messages

If you use plain strings, it becomes hard to handle growth as message types increase.
In practice, it’s common to use JSON with fields like type and payload.

import json

async def handle_message(data: str, websocket: WebSocket):
    msg = json.loads(data)
    msg_type = msg.get("type")
    payload = msg.get("payload", {})

    if msg_type == "chat":
        text = payload.get("text", "")
        # Send a chat message to everyone
    elif msg_type == "join":
        # Handle joining a room
    elif msg_type == "leave":
        # Handle leaving a room
    else:
        await websocket.send_text("unknown message type")

If the front end also sends/receives JSON, you can add features later without changing the message format too much.

---

5. Combining with Security and Authentication

WebSockets also come with security concerns. Here are the essentials.

5.1 Origin and a CORS-Like Mindset

Browser WebSockets, like HTTP, require you to think about “which origins are allowed.”

• Restrict Origin via Nginx or a reverse proxy
• Or reject by checking websocket.headers (e.g., Origin, Sec-WebSocket-Protocol) on the server side

A practical approach is to pass a token as a query parameter at connection time and authenticate on the server.

5.2 Combining with JWT

For example, pass a JWT issued via HTTP login in the WebSocket query string.

// Front-end idea
const token = "your-existing-JWT";
const ws = new WebSocket(`wss://example.com/ws/chat?token=${token}`);

On the server side, you can read query params from FastAPI’s WebSocket.

# app/main.py (excerpt)
from fastapi import WebSocket, WebSocketDisconnect
from app.core.jwt import decode_access_token  # Assume a JWT utility from a previous article

@app.websocket("/ws/secure-chat")
async def websocket_secure_chat(websocket: WebSocket):
    token = websocket.query_params.get("token")
    if not token:
        await websocket.close(code=1008)  # Policy Violation
        return

    try:
        payload = decode_access_token(token)
    except Exception:
        await websocket.close(code=1008)
        return

    username = payload.get("sub", "anonymous")

    await manager.connect(websocket)
    await manager.broadcast(f"{username} joined")

    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"{username}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"{username} left")

This example uses “valid token = OK” for simplicity, but in production you’ll also consider scopes/roles, connection limits, and more.

---

6. Testing WebSockets: How to Verify with TestClient

FastAPI’s official docs include a guide for testing WebSockets.

6.1 Basic Test Syntax

Use the same TestClient as HTTP and call client.websocket_connect().

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

client = TestClient(app)

def test_websocket_echo():
    with client.websocket_connect("/ws/echo") as websocket:
        websocket.send_text("hello")
        data = websocket.receive_text()
        assert data == "Echo: hello"

Two key points:

• with client.websocket_connect("/ws/echo") as websocket: opens a connection and closes it automatically when the block ends.
• The client API provides send_text() / receive_text() methods that mirror the server side.

6.2 Testing Broadcast

You can also write tests that involve multiple clients.

def test_broadcast():
    with client.websocket_connect("/ws/chat") as ws1:
        with client.websocket_connect("/ws/chat") as ws2:
            ws1.send_text("hello")

            msg1 = ws1.receive_text()
            msg2 = ws2.receive_text()

            assert "hello" in msg1
            assert "hello" in msg2

Having tests in place helps you quickly detect regressions when:

• You change connection manager logic
• You modify authentication behavior

---

7. Use-Case Design Notes (Quick Tips)

Finally, here are some typical WebSocket use cases and design hints.

7.1 Chat (1-to-1 / Group)

• Manage connections by room ID (chat room).
• Use JSON messages such as type=chat, payload={text, sender_id, room_id}.
• If only authenticated users can join, identify the user via JWT and check permission to join the room.

7.2 Real-Time Dashboards

• Aggregate data periodically on the backend and push updates only when changes occur.
• If the data is shared by all users, connect everyone to a single room (e.g., dashboard) and broadcast.
• If each user has a personalized dashboard, either create “room per user ID” or filter messages by user ID in the payload.

7.3 Notifications / Event Streams

• Stream events like new comments, status changes, and system alerts over WebSocket.
• Split responsibilities: WebSocket as the “real-time channel,” REST API as “history retrieval.”
• If you need read/unread tracking or replay after reconnect, keep WebSocket as the “trigger,” and re-fetch the actual data via REST/DB for simplicity.

---

8. Roadmap by Reader Type: Where to Start

Indie Developers / Learners

1. Implement the minimal @app.websocket("/ws/echo") and connect from the browser console.
2. Add a connection manager and build a broadcast-based simple chat.
3. Create a small chat UI with HTML + JavaScript and try chatting across two browser windows.
4. Once comfortable, switch to JSON messages and stream events beyond chat.

Backend Engineers in Small Teams

1. Add a single WebSocket endpoint to an existing FastAPI app for “notifications” or a “dashboard.”
2. Introduce a connection manager + room management so you can segment connections by user/project.
3. Integrate with your existing auth (JWT, etc.) to support “authenticated-only WebSockets.”
4. Add WebSocket tests with TestClient so auth and broadcast behavior are protected.

SaaS Teams / Startups

1. Build a PoC: a single-instance WebSocket server (chat/dashboard) inside FastAPI.
2. Design with abstraction so you can later extract broadcasting to an external broker (Redis Pub/Sub, message queues).
3. Define WebSocket security policy (auth, rate limiting, Origin restrictions) and align it with infra (LB/reverse proxy).
4. Add monitoring (connections, disconnects, error rates) and logging (event volume) to grow it into an operable real-time system.

---

9. References (Official Docs, Japanese Articles, etc.)

Note: The contents may change over time. Check each site for the latest information.

• Official documentation (FastAPI / Starlette)

  • WebSockets – FastAPI (English)
  • WebSockets – FastAPI Reference (English)
  • WebSockets – FastAPI Japanese Docs (GitHub)
  • WebSockets – Starlette
  • Testing WebSockets – FastAPI

• Japanese articles / tutorials

  • Tried Getting Started with WebSocket in FastAPI (Zenn)
  • WebSockets in FastAPI (Qiita)
  • FastAPI WebSocket Introduction (Qiita)
  • I Used the Wrong Decorator for FastAPI WebSocket (Blog)
  • Latest Guide: Using WebSocket with FastAPI (APIDog Japanese)

• English articles / advanced examples

  • Building a WebSocket Server with FastAPI (Medium)
  • Getting Started with FastAPI WebSockets (Better Stack)
  • Developing a Real-time Dashboard with FastAPI and WebSockets (TestDriven.io)

---

Conclusion

• WebSocket is a protocol for real-time, bi-directional communication—and FastAPI supports it quite simply.
• With @app.websocket() and the WebSocket class, you can start from a minimal echo server, then add a connection manager and room management to build multi-client chat or dashboards.
• By thinking about security (authentication, Origin restrictions) and adding tests (WebSocket tests via TestClient), you can keep expanding features safely.
• You don’t need to aim for a huge real-time system from day one. Start small—chat or notifications—and gradually get comfortable with the WebSocket ecosystem.

If this becomes a trigger for adding a bit of real-time “fun” to your FastAPI app, I’ll be very happy.
Try one step at a time, at a pace that feels comfortable.

---