fastapi/fastapi
WebSockets
WebSocket support reuses most of the HTTP request machinery: the same dependency resolver, the same exception model, the same exit-stack scheme.
Public API
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"echo: {data}")
except WebSocketDisconnect:
passfastapi/websockets.py is a one-line shim:
from starlette.websockets import (
WebSocket as WebSocket,
WebSocketDisconnect as WebSocketDisconnect,
WebSocketState as WebSocketState,
) # noqafastapi.WebSocket and fastapi.WebSocketDisconnect are re-exported from fastapi/__init__.py for convenience.
Decorators
| Decorator | Source |
|---|---|
@app.websocket(path) / @router.websocket(path) |
APIRouter.websocket / FastAPI.websocket (fastapi/routing.py, fastapi/applications.py) |
app.add_api_websocket_route(path, endpoint, ...) |
Equivalent imperative form |
The decorator builds an APIWebSocketRoute (fastapi/routing.py). Like APIRoute, it owns a Dependant built from the endpoint signature, plus a websocket_session wrapper that opens the per-request AsyncExitStacks and runs the endpoint.
Dependency injection over WebSockets
The same dependency system applies. A WebSocket endpoint can declare:
websocket: WebSocket— the connection itself (special parameter;Dependant.websocket_param_name).Annotated[T, Cookie(...)],Annotated[T, Header(...)],Annotated[T, Query(...)]— read from the WebSocket handshake.Annotated[T, Depends(...)]— same resolver, same caching, same support foryield.- Security schemes — the OAuth2 / API-key classes work identically.
What you cannot use over WebSockets:
Pathparameters that are part of the URL still work.Body,Form,Filedo not — there is no HTTP body to parse.BackgroundTasksdoes not inject (noResponse).
Validation errors close the WebSocket with WS_1008_POLICY_VIOLATION and the JSON-encoded errors as the close reason — see websocket_request_validation_exception_handler in fastapi/exception_handlers.py.
Errors
fastapi/exceptions.py:WebSocketException(code, reason) mirrors HTTPException for WebSockets. Raise it from inside an endpoint or dependency; the route closes the socket with the supplied close code.
Lifespan and cleanup
websocket_session() (fastapi/routing.py) opens scope["fastapi_inner_astack"] and scope["fastapi_function_astack"] exactly as request_response() does for HTTP requests. Depends(...) callables that yield are cleaned up the same way after the WebSocket closes.
Tests
tests/test_ws_router.py— router and decorator behaviour.tests/test_ws_dependencies.py— dependency injection over WebSockets.tests/test_dependency_yield_scope_websockets.py— yield-scope behaviour.tests/test_dependency_after_yield_websockets.py— cleanup ordering.
Integration points
- Down to Starlette —
WebSocket,WebSocketDisconnect,WebSocketState, the underlying ASGI plumbing. - Down to dependency resolver — the same
solve_dependenciespipeline. - Down to exception handlers —
websocket_request_validation_exception_handlertranslates validation failures into close codes.
Entry points for modification
- Adding new "special" parameters: edit
analyze_paraminfastapi/dependencies/utils.pyand add a corresponding field onDependant(fastapi/dependencies/models.py). - Customising the validation-error close code: replace the handler via
app.add_exception_handler(WebSocketRequestValidationError, my_handler). - Adding hooks for connection accept/disconnect: not supported in framework — wrap the endpoint or use a dependency with
yield.
Built by Factory AutoWiki from public repository content. It is a generated preview for codebase exploration, not source-maintained documentation.
Previous
Background tasks
Next
API