WebSocket with FastAPI — Complete Guide

In this tutorial, you will learn about WebSocket with FastAPI. We cover key concepts, practical examples, and best practices to help you master this topic.

FastAPI supports WebSocket natively through Starlette's WebSocket class. You define WebSocket endpoints alongside regular HTTP routes, using the same Dependency Injection, authentication, and validation system.

What You'll Learn

• WebSocket endpoint definition with FastAPI
• WebSocket dependency injection
• Broadcasting to connected clients
• Authentication via dependencies
• Connection lifecycle management

Why It Matters

FastAPI's WebSocket support is lightweight and integrated with its async ecosystem. You can share dependencies between HTTP and WebSocket endpoints, use Pydantic models, and leverage async database libraries.

Real-World Use

FastAPI is used for real-time ML model inference with WebSocket streaming. The FastAPI documentation has official WebSocket chat examples. Many async Python services use FastAPI for combined HTTP+WebSocket APIs.

flowchart LR
    Client[WebSocket Client] --> FastAPI[FastAPI App]
    FastAPI --> WebSocket[WebSocket Endpoint]
    WebSocket --> Deps[Dependency Injection]
    WebSocket --> Receive[receive_text / receive_bytes]
    WebSocket --> Send[send_text / send_bytes]
    WebSocket --> Broadcast[Broadcast Manager]
    FastAPI --> HTTP[HTTP Routes]

Teacher Mindset

FastAPI WebSocket endpoints are async functions decorated with @app.websocket. Use WebSocket class methods (receive_text, send_json, etc.) for communication. Wrap broadcasting logic in a connection manager.

Code Examples

# Example 1: Basic WebSocket endpoint
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 disconnected")

# Example 2: Connection manager for broadcasting
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

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 broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

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

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client {client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client {client_id} disconnected")

# Example 3: Authenticated WebSocket with dependency
from fastapi import FastAPI, WebSocket, Depends, Query
from fastapi.security import HTTPBearer

app = FastAPI()

async def get_token(
    websocket: WebSocket,
    token: str = Query(...)
):
    if token != "secret-token":
        await websocket.close(code=1008)
        return None
    return token

@app.websocket("/ws")
async def authenticated_ws(
    websocket: WebSocket,
    token: str = Depends(get_token)
):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_json()
            await websocket.send_json({"echo": data})
    except WebSocketDisconnect:
        print("Authenticated client disconnected")

Common Mistakes

• Forgetting to call await websocket.accept() before receiving/sending
• Not catching WebSocketDisconnect exceptions in the receive loop
• Using synchronous blocking code inside async WebSocket handlers
• Not handling JSON Serialization manually when using send_json
• Running out of memory by storing all WebSocket objects without cleanup

Practice

1. Create a FastAPI WebSocket echo endpoint.
2. Implement a connection manager with broadcast support.
3. Add query parameter authentication to a WebSocket endpoint.
4. Use send_json and receive_json for structured messages.
5. Challenge: Build a FastAPI app with HTTP endpoints for message history and WebSocket for real-time updates.

FAQ

Mini Project

Build a FastAPI WebSocket chat application with: authentication via query token, ConnectionManager for broadcasting, HTTP endpoint for message history (stored in memory), JSON messages with username and text fields, and broadcast on connect/disconnect.

What's Next

Next, you will learn about WebSocket with Spring Boot and SockJS.