How to Manage WebSocket Exceptions

To manage stateful connection errors in WebSockets, you can manually raise a WebSocketException to terminate a connection with a specific status code or override the default behavior for WebSocketRequestValidationError when client parameters fail validation.

Manually Closing Connections with WebSocketException

You can raise WebSocketException anywhere in your WebSocket endpoint or its dependencies to close the connection. This is typically used for business logic failures, such as missing authentication credentials or invalid session state.

from typing import Annotated
from fastapi import (
    Cookie,
    FastAPI,
    WebSocket,
    WebSocketException,
    status,
)

app = FastAPI()

@app.websocket("/items/{item_id}/ws")
async def websocket_endpoint(
    *,
    websocket: WebSocket,
    session: Annotated[str | None, Cookie()] = None,
    item_id: str,
):
    # Check for a session cookie before accepting the connection
    if session is None:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION, reason="Session cookie missing")
    
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")

Key Parameters

• code: An integer closing code defined in RFC 6455. Common codes include status.WS_1008_POLICY_VIOLATION and status.WS_1011_INTERNAL_ERROR.
• reason: An optional UTF-8 encoded string explaining why the connection was closed.

Customizing WebSocket Validation Errors

When a WebSocket endpoint requires parameters (like headers, query params, or cookies) that fail validation, FastAPI automatically raises a WebSocketRequestValidationError. By default, this closes the connection with code 1008 and the validation errors as the reason.

You can override this globally using an exception handler:

from fastapi import FastAPI, WebSocket, status
from fastapi.exceptions import WebSocketRequestValidationError
from fastapi.encoders import jsonable_encoder

app = FastAPI()

@app.exception_handler(WebSocketRequestValidationError)
async def websocket_validation_exception_handler(
    websocket: WebSocket, exc: WebSocketRequestValidationError
):
    # Custom logic: log the error or use a different close code
    await websocket.close(
        code=status.WS_1008_POLICY_VIOLATION, 
        reason="Invalid connection parameters"
    )

Accessing Error Context

The WebSocketRequestValidationError contains an endpoint_ctx attribute which provides metadata about the endpoint where the validation failed. This is useful for logging or debugging.

@app.exception_handler(WebSocketRequestValidationError)
async def websocket_validation_handler(
    websocket: WebSocket, exc: WebSocketRequestValidationError
):
    # exc.errors() contains the Pydantic validation errors
    # exc.endpoint_ctx contains info like the function name and path
    print(f"Validation failed for endpoint: {exc.endpoint_ctx}")
    await websocket.close(code=status.WS_1008_POLICY_VIOLATION)

Troubleshooting

Timing of Exceptions

If you raise a WebSocketException before calling await websocket.accept(), FastAPI handles the closure automatically. If you raise it after the connection is accepted, the connection will be closed immediately with the provided code and reason.

Reason String Limits

The WebSocket protocol typically limits the "reason" string to 123 bytes. If your WebSocketRequestValidationError contains many detailed Pydantic errors, the default handler (which uses jsonable_encoder(exc.errors())) might produce a string that exceeds this limit, leading to truncated messages or protocol errors in some clients. In such cases, it is recommended to use a custom exception handler to provide a shorter, summarized reason.

Custom Exception Handlers for Business Logic

You can also define and handle your own custom exceptions for WebSockets using the same pattern as WebSocketRequestValidationError:

class CustomBusinessError(Exception):
    pass

@app.exception_handler(CustomBusinessError)
async def custom_handler(websocket: WebSocket, exc: CustomBusinessError):
    await websocket.close(code=status.WS_1003_UNSUPPORTED_DATA, reason="Business logic failure")

@app.websocket("/custom")
async def custom_websocket(websocket: WebSocket):
    await websocket.accept()
    raise CustomBusinessError()