Last modified: Dec 01, 2025 By Alexander Williams
FastAPI WebSockets Tutorial: Real Time Python
Modern apps need live data. Think chat apps or live dashboards. FastAPI makes this easy with WebSockets.
This guide teaches you real-time Python. You will build a live chat server. It is perfect for beginners.
What Are WebSockets?
Traditional HTTP is request-response. A client asks, a server answers. The connection then closes.
WebSockets are different. They create a persistent, two-way channel. Both client and server can talk anytime.
This is ideal for real-time features. Use it for notifications, live feeds, or multiplayer games.
Why Use FastAPI for WebSockets?
FastAPI is a modern Python framework. It is fast and easy to learn. It has built-in WebSocket support.
You get automatic documentation. You also get data validation. It uses standard Python type hints.
If you are new, start with a basic FastAPI REST API tutorial. Then add WebSockets.
Project Setup and Installation
First, ensure Python is installed. Then, install FastAPI and an ASGI server. We use Uvicorn here.
If you see an import error, check our guide on fixing the FastAPI module error.
# Install the required packages
pip install fastapi uvicorn websockets
Creating Your First WebSocket Endpoint
Create a file named main.py. Import FastAPI. Then, define a WebSocket route.
The @app.websocket decorator defines the path. The function handles the connection.
from fastapi import FastAPI, WebSocket
app = FastAPI()
# WebSocket endpoint
@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}")
This code does a few things. First, it accepts the connection with accept.
Then, it enters a loop. It waits for text data with receive_text. It echoes it back with send_text.
Running the Application
Start the server with Uvicorn. Use the command below. The app will reload on code changes.
uvicorn main:app --reload
Your server runs at http://localhost:8000. The WebSocket endpoint is at ws://localhost:8000/ws.
Testing the WebSocket Connection
You can test with a simple HTML page. Create an index.html file. Use JavaScript to connect.
<!DOCTYPE html>
<html>
<body>
<script>
const ws = new WebSocket('ws://localhost:8000/ws');
ws.onopen = () => {
console.log('Connected!');
ws.send('Hello Server');
};
ws.onmessage = (event) => {
console.log('Message from server:', event.data);
};
</script>
</body>
</html>
Open this file in a browser. Check the developer console. You should see the echoed message.
Handling Multiple Clients
A single echo server is limited. Real apps handle many users. You need to manage connections.
Store active connections in a list. Add clients when they connect. Remove them on disconnect.
from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect
app = FastAPI()
# Store active connections
active_connections = []
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
active_connections.append(websocket)
try:
while True:
data = await websocket.receive_text()
# Broadcast message to all clients
for connection in active_connections:
await connection.send_text(f"Client says: {data}")
except WebSocketDisconnect:
active_connections.remove(websocket)
This is a key concept. The list tracks all live WebSocket objects. The broadcast loop sends data to everyone.
The WebSocketDisconnect exception handles client leave. It cleans up the connection list.
Building a Simple Chat Room
Let's improve our example. We will build a basic chat room. Clients can send and receive messages.
We will use a more robust connection manager. This is a common pattern in FastAPI apps.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
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 send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@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"Message: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("A client left the chat.")
The ConnectionManager class organizes the logic. It handles connect, disconnect, and broadcast.
This makes the main endpoint cleaner. It also makes the code easier to test and extend.
Adding Data Validation with Pydantic
Real apps need structured data. You can send JSON over WebSockets. Validate it with Pydantic.
Learn more in our guide on FastAPI Pydantic validation.
from pydantic import BaseModel
class ChatMessage(BaseModel):
username: str
content: str
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_json()
message = ChatMessage(**data) # Validate
await manager.broadcast(f"{message.username}: {message.content}")
except WebSocketDisconnect:
manager.disconnect(websocket)
This ensures incoming data has a username and content. It makes your app more robust.
Next Steps and Deployment
You now have a working real-time app. You can add more features. Try user authentication.
For secure apps, implement JWT authentication in FastAPI.
When ready for production, learn to deploy FastAPI with Docker.
Conclusion
FastAPI WebSockets enable powerful real-time apps. They are easy to set up and use.
You learned to create endpoints, handle connections, and broadcast messages. You built a simple chat server.
Remember to manage connections properly. Always clean up on disconnect. Validate your data.
Now you can build live notifications, dashboards, or collaborative tools. The possibilities are endless.