Skip to content

WebSockets

Every source works on @app.websocket routes, the same way as on HTTP routes:

import os

from fastapi import FastAPI, Security, WebSocket

from fastapi_multiauth import APIKeyCookieAuth

SECRET_KEY = os.environ["SECRET_KEY"]


async def validate_session(username: str) -> str:
    return username


session = APIKeyCookieAuth(
    name="__Host-session",
    validator=validate_session,
    secret_key=SECRET_KEY,
    csrf=True,
    trusted_origins=["https://app.example.com"],
)

app = FastAPI()


@app.websocket("/ws")
async def ws(websocket: WebSocket, username: str = Security(session)):
    await websocket.accept()
    await websocket.send_json({"hello": username})
    await websocket.close()

A missing or invalid credential refuses the connection before accept().

Three differences from HTTP:

  • The credential is checked once, when the connection opens. A long-lived connection outlives a revoked session unless your route closes it.
  • Browsers cannot set headers on a WebSocket. Use a cookie, or a short-lived ticket in the query string (see the WebSocket tickets recipe).
  • CORS does not apply. Any page can open the socket, and the browser attaches the cookie. With a cookie, set csrf=True as above: the handshake must come from your origin or one in trusted_origins.