WebSocket

WebSocket class for handling real-time bidirectional communication in FastAPI

The WebSocket class provides an interface for handling WebSocket connections in FastAPI. It allows bidirectional, real-time communication between the client and server. FastAPI uses Starlette’s WebSocket class.

Importing

from fastapi import WebSocket, WebSocketDisconnect

Basic Usage

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.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}")

Class Reference

WebSocket

The WebSocket class provides methods for accepting connections, sending and receiving data, and closing connections.

Methods

accept()

async accept(subprotocol=None, headers=None)

Accept the WebSocket connection.

subprotocol · str | None · default: "None"

WebSocket subprotocol to use.

headers · Iterable[tuple[bytes, bytes]] | None · default: "None"

Additional headers to send in the accept response.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    # Connection is now established

Note: You must call accept() before sending or receiving any messages.

receive_text()

async receive_text()

Receive a text message from the client.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    message = await websocket.receive_text()
    print(f"Received: {message}")

Returns: str - The received text message

Raises: WebSocketDisconnect - If the client disconnects

receive_bytes()

async receive_bytes()

Receive binary data from the client.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    data = await websocket.receive_bytes()
    print(f"Received {len(data)} bytes")

Returns: bytes - The received binary data

Raises: WebSocketDisconnect - If the client disconnects

receive_json()

async receive_json(mode='text')

Receive and parse JSON data from the client.

mode · str · default: "'text'"

Either “text” or “binary” to specify how to receive the data.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    data = await websocket.receive_json()
    print(f"Received JSON: {data}")

Returns: Any - The parsed JSON data

Raises: WebSocketDisconnect - If the client disconnects

send_text()

async send_text(data)

Send a text message to the client.

data · str · required

The text message to send.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Hello, client!")

send_bytes()

async send_bytes(data)

Send binary data to the client.

data · bytes · required

The binary data to send.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_bytes(b"Binary data")

send_json()

async send_json(data, mode='text')

Serialize and send JSON data to the client.

data · Any · required

The data to serialize and send as JSON.

mode · str · default: "'text'"

Either “text” or “binary” to specify how to send the data.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_json({"message": "Hello", "count": 42})

close()

async close(code=1000, reason=None)

Close the WebSocket connection.

code · int · default: "1000"

WebSocket close code (1000 = normal closure).

reason · str | None · default: "None"

Optional reason for closing the connection.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        await websocket.close(code=1000, reason="Client disconnected")

Common close codes:

• 1000 - Normal closure
• 1001 - Going away
• 1002 - Protocol error
• 1003 - Unsupported data
• 1011 - Internal server error

Attributes

client · Address | None

Client address information (host and port).

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    if websocket.client:
        print(f"Client: {websocket.client.host}:{websocket.client.port}")

url · URL

The WebSocket URL.

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    print(f"URL: {websocket.url}")

headers · Headers

Request headers from the WebSocket handshake.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    auth_header = websocket.headers.get("authorization")

query_params · QueryParams

Query parameters from the WebSocket URL.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    token = websocket.query_params.get("token")

path_params · dict[str, Any]

Path parameters from the WebSocket URL.

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await websocket.accept()
    # Or access via websocket.path_params["client_id"]

cookies · dict[str, str]

Cookies from the WebSocket handshake request.

state · State

State object for storing data during the WebSocket connection lifecycle.

client_state · WebSocketState

Current state of the WebSocket connection.

States:

• WebSocketState.CONNECTING - Connection is being established
• WebSocketState.CONNECTED - Connection is established
• WebSocketState.DISCONNECTED - Connection is closed

WebSocketDisconnect Exception

Raised when the client disconnects or the connection is lost.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

Attributes:

code · int

WebSocket close code.

reason · str | None

Reason for disconnection.

Common Patterns

Echo Server

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

Broadcasting to Multiple Clients

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client {client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client {client_id} left")

JSON Message Handling

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from pydantic import BaseModel

app = FastAPI()

class Message(BaseModel):
    type: str
    content: str

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            # Receive JSON
            data = await websocket.receive_json()
            message = Message(**data)
            
            # Process message
            response = {
                "type": "response",
                "content": f"Received: {message.content}"
            }
            
            # Send JSON
            await websocket.send_json(response)
    except WebSocketDisconnect:
        print("Client disconnected")

Authentication

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, status

app = FastAPI()

async def verify_token(token: str) -> bool:
    # Verify token logic
    return token == "valid_token"

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    # Get token from query params or headers
    token = websocket.query_params.get("token")
    
    if not token or not await verify_token(token):
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

Path and Query Parameters

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(
    websocket: WebSocket,
    room_id: str,
):
    await websocket.accept()
    
    # Access query parameters
    username = websocket.query_params.get("username", "Anonymous")
    
    await websocket.send_text(f"Welcome {username} to room {room_id}")
    
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"[{room_id}] {username}: {data}")
    except WebSocketDisconnect:
        print(f"{username} left room {room_id}")

Dependencies with WebSockets

from fastapi import FastAPI, WebSocket, Depends, HTTPException, status

app = FastAPI()

async def get_token(websocket: WebSocket) -> str:
    token = websocket.query_params.get("token")
    if not token:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
    return token

async def verify_token(token: str = Depends(get_token)) -> str:
    if token != "valid_token":
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
    return token

@app.websocket("/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    token: str = Depends(verify_token),
):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Authenticated echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

Client Example (JavaScript)

// Connect to WebSocket
const ws = new WebSocket("ws://localhost:8000/ws");

// Connection opened
ws.onopen = (event) => {
    console.log("Connected to WebSocket");
    ws.send("Hello Server!");
};

// Listen for messages
ws.onmessage = (event) => {
    console.log("Message from server:", event.data);
};

// Connection closed
ws.onclose = (event) => {
    console.log("Disconnected from WebSocket");
};

// Error handling
ws.onerror = (error) => {
    console.error("WebSocket error:", error);
};

// Send JSON
ws.send(JSON.stringify({ type: "message", content: "Hello" }));