In FastAPI you declare a socket route with @app.websocket("/ws") and an async function that takes a WebSocket. Call await websocket.accept(), then loop over receive_text() and send_text(). Catch WebSocketDisconnect to clean up when a client leaves. Install the websockets package first, and test with TestClient.websocket_connect().
| Call | What it does |
|---|---|
| await websocket.accept() | Completes the handshake; nothing can be sent before it |
| await websocket.receive_text() | Waits for the next text message |
| await websocket.receive_json() | Waits for a message and parses it as JSON |
| await websocket.send_text(s) | Sends a text message |
| await websocket.send_json(obj) | Serialises an object and sends it |
| await websocket.close(code=1000) | Ends the connection with a close code |
Updated
The endpoint builder below writes a working main.py from a few choices: the route path, whether messages are plain text or JSON, whether every client hears every message, and whether a token must arrive in the query string. A matching pytest file comes with it. Every option maps to a pattern from the FastAPI documentation, rewritten here as a single runnable file.
Choose how the endpoint should behave and copy a complete main.py plus a pytest file that checks it.
Accept, then echo each message until the client disconnects.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws")
async def echo(websocket: WebSocket):
await websocket.accept()
try:
while True:
text = await websocket.receive_text()
await websocket.send_text(f"echo: {text}")
except WebSocketDisconnect:
print("client left")Works for a single process. Swap the list for a Redis-backed relay before you scale out.
class ConnectionManager:
def __init__(self) -> None:
self.active: list[WebSocket] = []
async def connect(self, websocket: WebSocket) -> None:
await websocket.accept()
self.active.append(websocket)
def disconnect(self, websocket: WebSocket) -> None:
if websocket in self.active:
self.active.remove(websocket)
async def broadcast(self, message: str) -> None:
for connection in list(self.active):
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/chat")
async def chat(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
await manager.broadcast(await websocket.receive_text())
except WebSocketDisconnect:
manager.disconnect(websocket)No server needed: the client talks to the app in process. Run it with pytest.
from fastapi.testclient import TestClient
from main import app
def test_echo():
client = TestClient(app)
with client.websocket_connect("/ws") as ws:
ws.send_text("hi")
assert ws.receive_text() == "echo: hi"A socket handler is a long-running coroutine. It accepts once, then sits on await receive_text() until the next message, so a single client costs almost nothing while idle. When the client goes away, the pending receive raises WebSocketDisconnect, and that is your moment to tidy up.
Because the handler is async, other clients keep being served while one is waiting. Avoid blocking calls such as time.sleep or synchronous database drivers inside the loop; they freeze every connection in the worker.
Read the token from the query string as a normal parameter. You then have two ways to refuse it. Closing before accept() turns into an HTTP 403 on the handshake, which a browser only reports as a failed connection. Accepting and then closing with code 1008 lets the client read the reason and show a sign-in prompt instead of retrying forever.
Start with the install: the FastAPI WebSocket docs ask you to add the websockets package alongside FastAPI. Then check the path, because a route declared with @app.websocket only answers socket upgrades, never plain HTTP GETs. Closing the socket before accept() also shows up as a refused handshake (HTTP 403).
Keep accepted sockets in a list inside a small manager class, add on connect, remove on disconnect, and loop over the list calling send_text. The builder above generates exactly this when you tick the broadcast option.
Not with an in-memory list. Each worker is a separate process with its own list. The FastAPI docs point to encode/broadcaster, which relays messages through Redis or PostgreSQL so every process hears every message.
Yes. WebSocket endpoints accept Depends, Query, Cookie, Header, Path and Security just like HTTP routes, so an authentication dependency can be shared between both.
Create a TestClient for the app and open the socket in a with block using client.websocket_connect("/ws"). Inside it, send_text and receive_text drive the conversation, and leaving the block closes the connection.
Related: WebSocket tester, WebSocket in Node.js, CORS explained, nginx load balancing, Webhook vs API.
Start for free or see all free tools. We.Inc is free to start, with no card needed. AI building is on paid plans from $20 a month.
We.Inc is an AI-powered website builder you can resell under your own brand. Launch a branded client dashboard, bill on Stripe Connect, and deliver AI-generated websites in minutes. White-label plans start at $99 a month for 25 client sites, with a 7-day free trial and no per-site fees.