FastAPI makes building APIs in Python fast and simple. While it's known for REST APIs, it also supports WebSockets for real-time applications, such as chat and live updates.
FastAPI handles WebSocket connections with built-in tools for managing connections, broadcasting messages, and injecting dependencies—all using the same clean design it's known for.
In this guide, you’ll learn how to build WebSocket-based features from scratch. We'll cover basic setup, room-based messaging, and authentication. By the end, you'll be ready to build real-time features into any FastAPI app.
Prerequisites
Before getting started with WebSockets in FastAPI, ensure you have Python 3.8 or later and pip installed. This guide assumes you already have a basic understanding of FastAPI and a general knowledge of asynchronous programming in Python.
Setting up your FastAPI WebSocket project
To follow along smoothly, create a separate project just for working with WebSockets. Start by setting up a new folder and a virtual environment:
mkdir fastapi-websockets && cd fastapi-websockets
python3 -m venv venv
source venv/bin/activate
Install FastAPI along with the WebSocket dependencies and an ASGI server for running your application. You'll use Uvicorn, which provides excellent WebSocket support:
pip install fastapi "uvicorn[standard]" "fastapi[standard]" websockets
Create your main application file and add the basic FastAPI setup:
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
This initial setup establishes a FastAPI application that you'll extend with WebSocket functionality throughout this guide.
Next, create a simple HTML client to test your WebSocket connections. This will help you visualize the real-time communication as you build your features:
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>FastAPI WebSocket Test</title>
</head>
<body>
<h1>WebSocket Test Client</h1>
<div id="messages"></div>
<input type="text" id="messageText" placeholder="Type a message...">
<button onclick="sendMessage()">Send</button>
<script>
var ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages');
var message = document.createElement('div');
message.textContent = event.data;
messages.appendChild(message);
};
function sendMessage() {
var input = document.getElementById('messageText');
ws.send(input.value);
input.value = '';
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
Start your development server to verify everything works correctly:
fastapi dev main.py
Navigate to http://localhost:8000 in your browser to see the test client interface:
The WebSocket connection won't work yet since you haven't implemented the WebSocket endpoint.
Creating your first WebSocket endpoint
With your FastAPI app and HTML client in place, you need to define a WebSocket endpoint that handles live connections. This allows your browser client to send and receive real-time messages.
Add the following route to main.py:
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
# ... html template code from above ...
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"You said: {data}")
The @app.websocket("/ws") decorator creates a WebSocket route at /ws. When a client connects, websocket.accept() accepts the connection. The while True loop keeps listening for incoming messages, and for each message received, it sends a response back prefixed with "You said:".
Restart your FastAPI server with fastapi dev main.py and go back to http://localhost:8000. Type a message and click Send:
You should see your message echoed back in real time under the messages section:
You've just created your first working WebSocket with FastAPI. Next, you'll handle connection errors gracefully.
Understanding WebSocket connection lifecycle
WebSocket connections in FastAPI follow a predictable lifecycle that's important to understand for building reliable real-time applications. The connection begins with an HTTP upgrade request that FastAPI automatically handles when you call websocket.accept().
Your current endpoint works, but it doesn't handle connection errors gracefully. Let's improve it by adding proper error handling:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
app = FastAPI()
# ... html template code from above ...
@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"You said: {data}")
except WebSocketDisconnect:
print("Client disconnected")
except Exception as e:
print(f"Error: {e}")
The try-except block handles connection closures gracefully. WebSocket connections can terminate for various reasons, including network issues, client-side navigation, or an explicit disconnection.
The WebSocketDisconnect exception is raised when a client closes the connection normally.
Restart your server and test again at http://localhost:8000.
Now, when you close the browser tab or refresh the page, you'll see "Client disconnected" printed in your terminal instead of an unhandled error crashing your application.
This error handling is essential for production applications where clients frequently connect and disconnect. Next, you'll learn how to manage multiple clients connecting simultaneously.
Managing multiple WebSocket connections
Real-world applications often require handling multiple clients simultaneously. Currently, your WebSocket only communicates with one client at a time. To broadcast messages between multiple users, you need a connection manager.
Create a simple connection manager class that keeps track of all active connections:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections = []
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()
# ... html template code from above ...
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Someone said: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
print("Client disconnected")
The ConnectionManager class maintains a list of active WebSocket connections. When a client connects, it gets added to the list. When they disconnect, they're removed. The broadcast() method sends a message to all connected clients.
Now, restart your server and open multiple browser tabs to http://localhost:8000. Type a message in one tab and watch it appear in all the other tabs instantly:
You've just built a basic real-time chat system.
Final thoughts
You’ve built a real-time WebSocket app using FastAPI. You started with a simple echo server and expanded it into a multi-client chat with broadcasting.
The concepts you used, such as WebSocket endpoints, connection management, and message handling, are the foundation for real-time features like notifications and collaborative tools.
FastAPI provides built-in async support and a clean structure, making it easy to add real-time functionality. From here, you can add authentication, connect to a database, or scale with tools like Redis.
For more advanced features, visit the FastAPI WebSocket documentation.
This work is licensed under a Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License.
Make your mark
Join the writer's program
Are you a developer and love writing and sharing your knowledge with the world? Join our guest writing program and get paid for writing amazing technical guides. We'll get them to the right readers that will appreciate them.
Write for us
Build on top of Better Stack
Write a script, app or project on top of Better Stack and share it with the world. Make a public repository and share it with us at our email.
community@betterstack.comor submit a pull request and help us build better products for everyone.
See the full list of amazing projects on github