Basic WebSockets in FastAPI

Capsule overview

HTTP works like a letter: you send a request, you get a response, and the conversation is over. If you want to know whether something changed, you have to send another letter. WebSocket works like a phone call: once you're connected, both sides can talk at any time without hanging up and dialing again.

FastAPI supports WebSockets natively with the @app.websocket() decorator. In this capsule you'll learn the full lifecycle of a WebSocket connection: the initial handshake, sending and receiving messages, and closing the connection cleanly. You'll build an echo server and a basic chat server.

By the end you'll know how to create WebSocket endpoints, handle the accept → send/receive → close cycle, send and receive both text and JSON, and handle disconnection errors.

The WebSocket protocol

How the connection is established

WebSocket starts out as a normal HTTP request with a special header:

GET /ws HTTP/1.1
Host: localhost:8000
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13

The server responds with 101 Switching Protocols:

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

From that moment on, the HTTP connection turns into a persistent WebSocket connection. There are no more requests and responses — there are messages flowing in both directions.

The lifecycle

1. Client requests an upgrade ──► Server accepts
2. Connection open (bidirectional)
3. Message exchange (text or binary)
4. Either side closes the connection

Your first WebSocket endpoint

Echo server: the "Hello World" of WebSockets

from fastapi import FastAPI, WebSocket

app = FastAPI(title="WebSocket Demo")


@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 Exception:
        pass

Let's break down each piece:

@app.websocket("/ws") — Defines a WebSocket route (not GET, POST, etc.)

ws: WebSocket — FastAPI injects the WebSocket object automatically

await ws.accept() — Completes the handshake. Without it, the connection is rejected

await ws.receive_text() — Waits for a message from the client (blocks until it arrives)

await ws.send_text() — Sends a message to the client

while True — Keeps the connection open for multiple messages

Testing it with Python

Create a file test_ws.py:

import asyncio
import websockets


async def test_echo():
    uri = "ws://localhost:8000/ws"
    async with websockets.connect(uri) as ws:
        await ws.send("Hello FastAPI")
        response = await ws.recv()
        print(f"Response: {response}")

        await ws.send("How are you?")
        response = await ws.recv()
        print(f"Response: {response}")


asyncio.run(test_echo())

# Terminal 1: the server
uvicorn app.main:app --reload

# Terminal 2: the client
pip install websockets
python test_ws.py

Output:

Response: Echo: Hello FastAPI
Response: Echo: How are you?

Testing it with JavaScript (the browser)

Open the browser console at http://localhost:8000/docs:

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

ws.onopen = () => {
    console.log("Connected!");
    ws.send("Hello from the browser");
};

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

ws.onclose = () => {
    console.log("Disconnected");
};

Sending and receiving JSON

In practice, you rarely send plain text. Messages are usually JSON:

import json
from fastapi import FastAPI, WebSocket

app = FastAPI()


@app.websocket("/ws/json")
async def websocket_json(ws: WebSocket):
    await ws.accept()

    try:
        while True:
            data = await ws.receive_json()

            response = {
                "type": "echo",
                "received": data,
                "message": f"I received a message of type: {data.get('type', 'unknown')}"
            }
            await ws.send_json(response)
    except Exception:
        pass

receive_json() — Parses the message as JSON automatically

send_json() — Serializes a dict to JSON and sends it

A client for JSON

import asyncio
import websockets
import json


async def test_json():
    async with websockets.connect("ws://localhost:8000/ws/json") as ws:
        message = {"type": "greeting", "content": "Hello", "user": "Mike"}
        await ws.send(json.dumps(message))

        response = json.loads(await ws.recv())
        print(f"Response: {response}")


asyncio.run(test_json())

Handling disconnections

The most important (and most ignored) part of WebSockets is handling disconnections correctly. A client can disconnect at any moment: they close the browser, lose their connection, or simply leave.

The problem

Without disconnect handling, your server crashes:

@app.websocket("/ws/bad")
async def websocket_bad(ws: WebSocket):
    await ws.accept()
    while True:
        data = await ws.receive_text()  # RuntimeError if the client disconnects
        await ws.send_text(f"Echo: {data}")

The solution: WebSocketDisconnect

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    print(f"Client connected: {ws.client}")

    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print(f"Client disconnected: {ws.client}")

WebSocketDisconnect — The exception FastAPI raises when the client disconnects. Catching it keeps the server from crashing.

ws.client — A (host, port) tuple that identifies the client. Useful for logging.

WebSocket with path parameters

Just like HTTP endpoints, WebSocket endpoints support path parameters:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws/{room_name}")
async def websocket_room(ws: WebSocket, room_name: str):
    await ws.accept()
    await ws.send_json({
        "type": "system",
        "message": f"Connected to room: {room_name}"
    })

    try:
        while True:
            data = await ws.receive_json()
            response = {
                "type": "message",
                "room": room_name,
                "content": data.get("content", ""),
                "from": data.get("username", "anon")
            }
            await ws.send_json(response)
    except WebSocketDisconnect:
        print(f"A client left {room_name}")

This lets you create separate "rooms" or communication channels.

WebSocket with query parameters

You can also take query parameters for authentication or configuration:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query

app = FastAPI()


@app.websocket("/ws/chat")
async def websocket_chat(
    ws: WebSocket,
    username: str = Query(default="anon"),
    token: str = Query(default=None)
):
    if token != "secret123":
        await ws.close(code=4001, reason="Invalid token")
        return

    await ws.accept()
    await ws.send_json({
        "type": "welcome",
        "message": f"Welcome, {username}!"
    })

    try:
        while True:
            data = await ws.receive_text()
            await ws.send_json({
                "type": "message",
                "from": username,
                "content": data
            })
    except WebSocketDisconnect:
        print(f"{username} disconnected")

The client connects with: ws://localhost:8000/ws/chat?username=Mike&token=secret123

await ws.close(code, reason) — Closes the connection with a custom code. Codes 4000-4999 are reserved for application use.

Pattern: an interactive server with commands

A useful pattern is a WebSocket that responds to different message types:

import json
from datetime import datetime
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

tasks_db = [
    {"id": 1, "title": "Learn WebSockets", "status": "in_progress"},
    {"id": 2, "title": "Build an API", "status": "pending"},
]


@app.websocket("/ws/tasks")
async def websocket_tasks(ws: WebSocket):
    await ws.accept()

    try:
        while True:
            message = await ws.receive_json()
            action = message.get("action")

            if action == "list":
                await ws.send_json({
                    "type": "task_list",
                    "tasks": tasks_db,
                    "count": len(tasks_db)
                })
            elif action == "get":
                task_id = message.get("id")
                task = next((t for t in tasks_db if t["id"] == task_id), None)
                if task:
                    await ws.send_json({"type": "task_detail", "task": task})
                else:
                    await ws.send_json({"type": "error", "message": f"Task {task_id} not found"})
            elif action == "create":
                new_task = {
                    "id": max(t["id"] for t in tasks_db) + 1 if tasks_db else 1,
                    "title": message.get("title", "Untitled"),
                    "status": "pending"
                }
                tasks_db.append(new_task)
                await ws.send_json({"type": "task_created", "task": new_task})
            else:
                await ws.send_json({
                    "type": "error",
                    "message": f"Unknown action: {action}",
                    "valid_actions": ["list", "get", "create"]
                })
    except WebSocketDisconnect:
        pass

Testing the interactive server

import asyncio
import websockets
import json


async def test_interactive():
    async with websockets.connect("ws://localhost:8000/ws/tasks") as ws:
        await ws.send(json.dumps({"action": "list"}))
        print("List:", json.loads(await ws.recv()))

        await ws.send(json.dumps({"action": "create", "title": "New task"}))
        print("Created:", json.loads(await ws.recv()))

        await ws.send(json.dumps({"action": "get", "id": 3}))
        print("Get:", json.loads(await ws.recv()))

        await ws.send(json.dumps({"action": "invalid"}))
        print("Error:", json.loads(await ws.recv()))


asyncio.run(test_interactive())

Key differences: HTTP vs WebSocket in FastAPI

WebSocket does NOT show up in /docs. FastAPI doesn't generate automatic documentation for WebSocket endpoints. You'll have to document them manually or with external tools.

Exercises

Exercise 1: A WebSocket calculator

Create a WebSocket endpoint /ws/calc that takes math operations as JSON and returns the result:

// Input
{"operation": "add", "a": 5, "b": 3}
// Output
{"result": 8, "operation": "5 + 3 = 8"}

Support: add, subtract, multiply, divide (with division-by-zero handling).

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws/calc")
async def websocket_calc(ws: WebSocket):
    await ws.accept()

    operations = {
        "add": lambda a, b: a + b,
        "subtract": lambda a, b: a - b,
        "multiply": lambda a, b: a * b,
        "divide": lambda a, b: a / b if b != 0 else None,
    }

    symbols = {"add": "+", "subtract": "-", "multiply": "×", "divide": "÷"}

    try:
        while True:
            data = await ws.receive_json()
            op = data.get("operation")
            a = data.get("a", 0)
            b = data.get("b", 0)

            if op not in operations:
                await ws.send_json({
                    "error": f"Unknown operation: {op}",
                    "valid": list(operations.keys())
                })
                continue

            if op == "divide" and b == 0:
                await ws.send_json({"error": "Division by zero"})
                continue

            result = operations[op](a, b)
            symbol = symbols[op]
            await ws.send_json({
                "result": result,
                "operation": f"{a} {symbol} {b} = {result}"
            })
    except WebSocketDisconnect:
        pass

Exercise 2: Chat with timestamps

Create a WebSocket endpoint /ws/chat/{username} that:

• Sends a welcome message on connect
• Adds a timestamp to every message it receives
• Counts how many messages the user has sent

from datetime import datetime
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws/chat/{username}")
async def websocket_chat(ws: WebSocket, username: str):
    await ws.accept()
    message_count = 0

    await ws.send_json({
        "type": "system",
        "message": f"Welcome, {username}!",
        "timestamp": datetime.now().isoformat()
    })

    try:
        while True:
            text = await ws.receive_text()
            message_count += 1
            await ws.send_json({
                "type": "message",
                "from": username,
                "content": text,
                "message_number": message_count,
                "timestamp": datetime.now().isoformat()
            })
    except WebSocketDisconnect:
        print(f"{username} sent {message_count} messages before disconnecting")

Exercise 3: WebSocket with token validation

Create a WebSocket endpoint that validates a token query parameter. If the token is invalid, close the connection with code 4001. If it's valid, accept and work as an echo server.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query

app = FastAPI()

VALID_TOKENS = {"token_abc", "token_xyz", "admin_token"}


@app.websocket("/ws/secure")
async def websocket_secure(ws: WebSocket, token: str = Query()):
    if token not in VALID_TOKENS:
        await ws.close(code=4001, reason="Invalid or missing token")
        return

    await ws.accept()
    await ws.send_json({"type": "auth", "status": "ok", "message": "Authenticated"})

    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(f"[Authenticated] Echo: {data}")
    except WebSocketDisconnect:
        print(f"Token {token[:8]}... disconnected")

Exercise 4: A WebSocket that sends data periodically

Create a WebSocket endpoint /ws/clock that sends the current time to the client every 2 seconds, without the client having to ask for it.

import asyncio
from datetime import datetime
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws/clock")
async def websocket_clock(ws: WebSocket):
    await ws.accept()

    try:
        while True:
            now = datetime.now()
            await ws.send_json({
                "type": "tick",
                "time": now.strftime("%H:%M:%S"),
                "timestamp": now.isoformat()
            })
            await asyncio.sleep(2)
    except WebSocketDisconnect:
        pass
    except Exception:
        pass

Troubleshooting

"403 Forbidden" when connecting a WebSocket

Your CORS middleware is probably interfering. FastAPI's CORSMiddleware doesn't block WebSockets directly, but if you have a reverse proxy (nginx), you need to configure WebSocket passthrough.

"Connection refused" on ws://localhost:8000/ws

Check that the server is running (uvicorn app.main:app --reload). Check that the path is exact — there's no automatic trailing slash in WebSockets.

The server crashes when the client disconnects

You're not catching WebSocketDisconnect. Always use try/except:

try:
    while True:
        data = await ws.receive_text()
except WebSocketDisconnect:
    pass

receive_json() throws an error with a text message

receive_json() expects valid JSON. If the client sends plain text, use receive_text() and parse it manually with json.loads().

WebSocket doesn't show up in /docs

That's normal. FastAPI doesn't document WebSocket endpoints in Swagger/OpenAPI. Document them manually in your README or use a tool like AsyncAPI.

Summary

Resources

1. FastAPI WebSockets Tutorial — The official docs
2. Starlette WebSockets — The underlying API
3. MDN WebSocket API — The browser reference
4. RFC 6455 — The protocol specification
5. websockets library — For testing with Python
6. websocat — A CLI for WebSocket testing

What's next?

In Capsule 03 you'll learn to handle multiple simultaneous connections with the ConnectionManager pattern. That's what lets you broadcast — send a message to every connected client at the same time. It's the fundamental pattern for real-time notifications.