WebSocket Usage

Create a file websocket.py:

from fastapi import Depends, FastAPI, HTTPException, Request, WebSocket
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from fastapi_paseto import AuthPASETO
from fastapi_paseto.exceptions import AuthPASETOException

app = FastAPI()


class User(BaseModel):
    username: str
    password: str


@AuthPASETO.load_config
def get_config():
    """Return the application auth configuration."""

    return {
        # Demo only. Generate a strong secret and load it from secure storage in
        # production. See the key-management documentation.
        "authpaseto_secret_key": "secret"
    }


@app.exception_handler(AuthPASETOException)
def authpaseto_exception_handler(request: Request, exc: AuthPASETOException):
    """Return auth exceptions as JSON for HTTP endpoints."""

    return JSONResponse(status_code=exc.status_code, content={"detail": exc.message})


@app.post("/login")
def login(user: User, Authorize: AuthPASETO = Depends()):
    """Issue an access token for the demo user."""

    if user.username != "test" or user.password != "test":
        raise HTTPException(status_code=401, detail="Bad username or password")

    access_token = Authorize.create_access_token(subject=user.username)
    return {"access_token": access_token}


@app.websocket("/ws/header")
async def websocket_header(
    websocket: WebSocket,
    Authorize: AuthPASETO = Depends(),
) -> None:
    """Authorize the websocket using the configured auth header."""

    Authorize.paseto_required()
    await websocket.accept()
    await websocket.send_json({"user": Authorize.get_subject()})
    await websocket.close()


@app.websocket("/ws/query")
async def websocket_query(
    websocket: WebSocket,
    Authorize: AuthPASETO = Depends(),
) -> None:
    """Authorize the websocket using a query parameter fallback."""

    Authorize.paseto_required(location="query")
    await websocket.accept()
    await websocket.send_json({"user": Authorize.get_subject()})
    await websocket.close()

The example uses a placeholder authpaseto_secret_key only to keep the snippet short. For production websocket deployments, retrieve the key from secure storage in load_config() the same way you would for HTTP endpoints. See Key Management.

Websocket endpoints use the same dependency injection pattern as HTTP endpoints:

@app.websocket("/ws/header")
async def websocket_header(
    websocket: WebSocket,
    Authorize: AuthPASETO = Depends(),
) -> None:
    Authorize.paseto_required()
    await websocket.accept()

The important rule is to call paseto_required() before accept(). If auth fails, the handshake is rejected with websocket close code 1008 and the auth error message is used as the close reason.

Header Authorization

By default, websocket authorization reads the token from the configured auth header, reusing authpaseto_header_name and authpaseto_header_type.

Authorization: Bearer <access_token>

Query Authorization

Browsers often cannot attach custom auth headers during the websocket handshake. For those clients, you can either enable websocket query transport globally:

@AuthPASETO.load_config
def get_config():
    return {
        "authpaseto_secret_key": "secret",  # Demo only.
        "authpaseto_websocket_token_location": ["query"],
    }

or opt into query transport for a single route:

Authorize.paseto_required(location="query")

The default query parameter is token, and you can customize it with authpaseto_websocket_query_key or per-route with token_key.

Parity and Limits

• optional=True, fresh=True, refresh_token=True, custom type, denylist checks, issuer validation, audience validation, base64 decoding, footers, implicit assertions, and token= overrides all work the same way for websocket handlers.
• JSON body token transport is not supported for websocket handshakes, because there is no request body available before the connection is accepted.