WebSocket Routing

When building real-time applications in FastAPI, you can organize your WebSocket endpoints using the APIRouter class. This allows you to group related WebSocket logic into separate files and apply shared dependencies or path prefixes across multiple connections.

Basic WebSocket Routing

To define a WebSocket endpoint on a router, use the @router.websocket() decorator. This decorator registers an APIWebSocketRoute which, unlike standard Starlette routes, supports FastAPI's dependency injection system.

from fastapi import APIRouter, WebSocket

router = APIRouter()

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

To use this router in your main application, include it using app.include_router():

from fastapi import FastAPI

app = FastAPI()
app.include_router(router)

Injecting Dependencies

FastAPI's APIWebSocketRoute allows you to use Depends() in your WebSocket endpoints. Dependencies are resolved before the endpoint function is executed. If a dependency fails or raises an exception, the connection is handled according to the error type.

Parameter-level Dependencies

You can inject dependencies directly into the endpoint function parameters:

from fastapi import APIRouter, WebSocket, Depends

router = APIRouter()

async def get_cookie_or_token(websocket: WebSocket, session: str | None = None):
    if session is None:
        await websocket.close(code=1008)
    return session

@router.websocket("/items")
async def read_items(
    websocket: WebSocket, 
    session: str = Depends(get_cookie_or_token)
):
    await websocket.accept()
    await websocket.send_text(f"Session: {session}")
    await websocket.close()

Router-level Dependencies

You can also apply dependencies to every WebSocket route in a router by passing them to the APIRouter constructor:

async def verify_auth():
    # Shared logic for all routes in this router
    pass

router = APIRouter(dependencies=[Depends(verify_auth)])

Structuring Large Applications

For larger projects, you can use path prefixes to group WebSocket endpoints. When you include a router with a prefix, FastAPI automatically prepends that prefix to all WebSocket paths defined within that router.

# chat.py
chat_router = APIRouter(prefix="/chat")

@chat_router.websocket("/room/{room_id}")
async def chat_room(websocket: WebSocket, room_id: str):
    await websocket.accept()
    await websocket.send_text(f"Welcome to room {room_id}")
    # ... logic ...

In your main application file:

# main.py
from fastapi import FastAPI
from .chat import chat_router

app = FastAPI()
app.include_router(chat_router)
# The WebSocket is now available at ws://host:port/chat/room/{room_id}

Troubleshooting

Validation Errors

If a dependency uses Pydantic models or FastAPI's Query/Header parameters and the validation fails, FastAPI will automatically close the WebSocket connection with code 1008 (Policy Violation).

websocket() vs websocket_route()

APIRouter provides two methods for adding WebSocket routes:

1. .websocket(path): Recommended. Creates an APIWebSocketRoute which supports FastAPI features like dependency injection.
2. .websocket_route(path): A lower-level method inherited from Starlette. It creates a standard routing.WebSocketRoute which does not support FastAPI's Depends() or automatic parameter resolution.

Path Prefixes

When using APIRouter(prefix="/something") or app.include_router(router, prefix="/something"), ensure the prefix starts with / and does not end with /. FastAPI's APIRouter.__init__ and include_router methods include assertions to enforce this formatting:

# This will raise an AssertionError
router = APIRouter(prefix="/api/") 

# This is correct
router = APIRouter(prefix="/api")