How to Create Chat Using FastAPI WebSocket

To create a chat using FastAPI and WebSocket, define a WebSocket endpoint with @app.websocket and manage connected clients to broadcast messages. Use await websocket.receive_text() to get messages and await websocket.send_text() to send messages to clients.

Syntax

Use @app.websocket decorator to create a WebSocket route. Inside the function, accept the connection with await websocket.accept(). Use await websocket.receive_text() to receive messages and await websocket.send_text() to send messages back. Manage multiple clients by storing their WebSocket connections.

python

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}")

Example

This example shows a simple chat server where multiple clients connect via WebSocket. Messages sent by one client are broadcast to all connected clients.

python

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/chat")
async def chat_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("A client left the chat")

Output

Server runs and accepts WebSocket connections at /ws/chat; messages from one client appear to all connected clients.

Common Pitfalls

• Not calling await websocket.accept() before receiving or sending messages causes errors.
• Forgetting to handle WebSocketDisconnect leads to stale connections and crashes.
• Not managing multiple connections properly means messages won't broadcast to all clients.
• Blocking code inside the WebSocket handler can freeze the server; always use async calls.

python

from fastapi import FastAPI, WebSocket

app = FastAPI()

# Wrong: Missing accept call
@app.websocket("/ws/wrong")
async def wrong_endpoint(websocket: WebSocket):
    data = await websocket.receive_text()  # This will raise an error
    await websocket.send_text(f"Echo: {data}")

# Right: Accept connection first
@app.websocket("/ws/right")
async def right_endpoint(websocket: WebSocket):
    await websocket.accept()
    data = await websocket.receive_text()
    await websocket.send_text(f"Echo: {data}")

Quick Reference

Remember these key points when creating a chat with FastAPI WebSocket:

• Use @app.websocket to define WebSocket routes.
• Always call await websocket.accept() before communication.
• Handle WebSocketDisconnect to clean up connections.
• Store active connections to broadcast messages.
• Use async functions to avoid blocking the server.

Key Takeaways

Use @app.websocket and await websocket.accept() to start WebSocket communication.

Manage connected clients to broadcast chat messages to all users.

Handle WebSocketDisconnect exceptions to remove disconnected clients cleanly.

Avoid blocking code inside WebSocket handlers by using async/await.

Test your chat with multiple clients to ensure messages broadcast correctly.