Python WebSockets: Building Real-Time Applications With FastAPI

Most web applications still rely on the request-response cycle: the client asks, the server answers, and the connection closes. That model works well for fetching a page or submitting a form, but it breaks down the moment an application needs to push data to the client the instant something changes. Python WebSockets solve this problem by keeping a single connection open so the server and client can exchange messages in both directions without the client having to ask first. Paired with FastAPI's native async support, WebSockets give Python developers a practical way to build chat systems, live dashboards, and collaborative tools without reaching for a separate real-time stack.

This article walks through what WebSockets are, why FastAPI is a strong fit for building them in Python, and how to move from a basic echo endpoint to a connection manager that can support authentication, broadcasting, and horizontal scaling.

What are WebSockets?

The WebSocket protocol (RFC 6455) defines a persistent, full-duplex connection between a client and a server. Unlike HTTP, where each request opens a new connection and closes it once the response is sent, a WebSocket connection stays open for as long as both sides need it. Either party can send a message at any time, without waiting for a request.

A WebSocket connection starts as an HTTP request. The client sends an Upgrade: websocket header, and if the server accepts, the connection switches protocols during what's called the WebSocket handshake. From that point on, both the client and server can write and read frames over the same TCP connection until either side closes it.

This has a few practical consequences for the connection lifecycle:

• Open: The handshake completes and the connection is ready for messages.
• Message exchange: Either side can send text or binary frames at any point.
• Close: Either the client or server can close the connection, or it can drop due to a network issue, and the other side needs to detect and handle that.

Compare that to a typical HTTP request-response cycle, where the server has no way to reach the client unless the client polls or reconnects. WebSockets remove that constraint, which is why they show up so often in real-time communication scenarios.

Why use Python for WebSocket applications?

Python's asyncio library gives developers a mature foundation for handling many concurrent connections without spawning a thread per client. Async programming in Python lets a single process hold thousands of open WebSocket connections, each one waiting on I/O rather than blocking a worker.

Beyond the async runtime, Python's ecosystem matters here too. Backend teams building real-time features are usually already working with Python for APIs, data processing, or background jobs, so adding WebSocket endpoints to an existing Python service is often more practical than introducing a separate Node.js or Go process just for real-time messaging. Libraries for authentication, database access, and message queues integrate directly into the same codebase, which keeps the architecture simpler to reason about and maintain.

Why FastAPI for WebSocket development?

FastAPI is built on Starlette and runs on ASGI (Asynchronous Server Gateway Interface), which was designed from the start to support both HTTP and WebSocket connections in the same application. That matters because it means a single FastAPI app can expose REST endpoints for standard CRUD operations and WebSocket endpoints for real-time features, sharing the same dependency injection system, request context, and application state.

A few specific reasons FastAPI fits well here:

• Native WebSocket support through Starlette, with no extra library required for basic functionality.
• Async/await as a first-class pattern, matching how WebSocket connections need to be handled.
• Pydantic models for validating the structure of incoming and outgoing messages, which helps catch malformed payloads before they reach business logic.
• Automatic API documentation for the HTTP side of the application, even though WebSocket routes themselves aren't included in the OpenAPI schema.
• Straightforward integration with the same authentication and backend services used elsewhere in the application, since WebSocket routes are just another type of route in the same app.

Setting up a FastAPI WebSocket application

Getting a basic WebSocket endpoint running takes only a few steps.

1. Install FastAPI and an ASGI server

pip install fastapi uvicorn

1. Create the FastAPI application and a WebSocket endpoint

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"Message received: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

A few things are worth calling out in this small example:

• websocket.accept() completes the handshake. Nothing can be sent or received before this call.
• receive_text() and send_text() are async calls, and the endpoint awaits them inside a loop so the connection stays open for multiple messages rather than closing after one exchange.
• WebSocketDisconnect is raised when the client closes the connection or drops off the network, and catching it is what keeps an unhandled disconnection from crashing the coroutine.

1. Run the application

uvicorn main:app --reload

This is enough for local testing and for understanding the basic message flow, but it's not yet structured for production use. It has no way to track multiple clients, no reconnection handling, and no authentication. The rest of this article builds toward that.

Building a real-time application with FastAPI WebSockets

A chat application is a useful example because it touches most of the problems a production WebSocket service needs to solve: tracking multiple connections, broadcasting to a group, and handling clients that disconnect mid-session.

The basic message flow looks like this: a client connects and is added to a room, every message it sends gets broadcast to the other clients in that room, and when it disconnects it needs to be removed from tracking so the server doesn't try to write to a dead connection.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

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

The client_id in the route path is a simple way to identify who sent a message, though in a production system that identity would normally come from an authenticated session rather than a value the client supplies directly. This same pattern, a connection manager tracking active sockets and a broadcast method pushing messages out, extends to live notifications, collaborative editing, and dashboard updates. What changes between those use cases is mostly what triggers a broadcast and what the message payload contains.

Managing multiple WebSocket connections

A single global list, as in the example above, works for a small demo but has real limitations once an application has more than one room, needs to target specific users, or has to clean up stale connections that never sent a proper close frame.

A more realistic connection manager tracks connections by identity and supports targeted delivery:

from fastapi import WebSocket

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, WebSocket] = {}

    async def connect(self, client_id: str, websocket: WebSocket):
        await websocket.accept()
        self.active_connections[client_id] = websocket

    def disconnect(self, client_id: str):
        self.active_connections.pop(client_id, None)

    async def send_to_client(self, client_id: str, message: str):
        websocket = self.active_connections.get(client_id)
        if websocket:
            await websocket.send_text(message)

    async def broadcast(self, message: str, exclude: str | None = None):
        for client_id, connection in list(self.active_connections.items()):
            if client_id != exclude:
                await connection.send_text(message)

A few points that matter in practice:

• Track connections by an identifier, not just in a list, so messages can be routed to a specific user instead of only broadcast to everyone.
• Remove connections on disconnect, including in finally blocks, so a connection that raises an unexpected exception doesn't stay in the dictionary indefinitely.
• Handle send failures, since send_text can raise if the underlying socket is already closed. Wrapping broadcast sends in a try/except and disconnecting on failure prevents one dead connection from blocking delivery to the rest.
• Iterate over a copy of the connections when broadcasting, since disconnecting a client while iterating over the live dictionary can raise a runtime error.

WebSocket authentication and security

WebSocket connections need the same authentication and authorization scrutiny as any other endpoint, but the mechanics differ slightly because the handshake happens before any application-level messages are exchanged.

A common approach is to pass a token as a query parameter or header during the initial connection request, then validate it before calling accept():

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

app = FastAPI()

def verify_token(token: str) -> str | None:
    # Replace with real JWT validation logic
    if token == "valid-token":
        return "user_123"
    return None

@app.websocket("/ws/secure")
async def secure_websocket(websocket: WebSocket, token: str = Query(...)):
    user_id = verify_token(token)
    if user_id is None:
        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"{user_id}: {data}")
    except WebSocketDisconnect:
        pass

Points worth taking seriously here:

• Validate before accepting. Closing the connection with a policy violation code before accept() avoids doing any application work for an unauthenticated client.
• Use short-lived tokens and check for expiry during the connection, plus periodically if the connection stays open for a long time, since a JWT that was valid at connect time can expire hours into a long-running session.
• Validate the Origin header on the server side to reduce the risk of cross-site WebSocket hijacking, particularly for browser-based clients.
• Validate every incoming message, not just the initial token, since a connection being authenticated doesn't mean every message on it is well-formed or authorized for the action it's requesting.
• Rate limit per connection and per user to prevent a single client from flooding the server with messages.
• Use WSS in production. WebSocket traffic over plain WS is unencrypted, the same way HTTP is unencrypted relative to HTTPS, and WSS should be the default outside local development.

Passing tokens in the query string is common, but it does mean the token can end up in server logs or browser history, so some teams prefer sending the token as the first message after connecting, before treating the connection as authenticated. Either approach works; the important part is that no application logic runs before the token is checked.

WebSocket vs REST API

Factor	WebSockets	REST API
Communication	Full-duplex, either side can send at any time	Half-duplex, client initiates every exchange
Connection	Persistent, stays open across many messages	New connection (or reused HTTP connection) per request
Real-time updates	Native, no polling required	Requires polling or a separate mechanism like SSE
Server push	Supported directly	Not supported; server can only respond to a request
Typical use cases	Chat, live dashboards, multiplayer features, collaborative editing	CRUD operations, resource-based APIs, most standard web traffic
Complexity	Higher: connection state, reconnection, and message ordering to manage	Lower: stateless requests are simpler to reason about and cache
Scalability considerations	Connections are stateful and tied to a specific server process	Requests are stateless and easy to distribute across servers

REST is the better default for most application traffic, since it's stateless, cacheable, and simpler to scale and debug. WebSockets are worth the added complexity when the application genuinely needs the server to push data without the client asking, such as a live price feed or a collaborative document. For updates that are frequent but one-directional, Server-Sent Events are often a lighter-weight alternative to a full WebSocket connection.

Scaling FastAPI WebSocket applications

A REST API can scale horizontally by putting a load balancer in front of several stateless instances, since any instance can handle any request. WebSocket connections complicate this because each connection is tied to a specific server process for its entire duration. If a client connects to instance A, instance B has no direct way to send that client a message.
This is where a pub/sub layer becomes useful. Redis Pub/Sub is a common choice: when a message needs to reach a client, the originating instance publishes it to a Redis channel, and every instance subscribed to that channel receives it and forwards the message to any locally connected clients that need it.

import redis.asyncio as redis
import json

class RedisPubSubManager:
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)
        self.pubsub = self.redis.pubsub()

    async def publish(self, channel: str, message: dict):
        await self.redis.publish(channel, json.dumps(message))

    async def subscribe(self, channel: str):
        await self.pubsub.subscribe(channel)
        async for message in self.pubsub.listen():
            if message["type"] == "message":
                yield json.loads(message["data"])

Each FastAPI instance still keeps its own local connection manager for the sockets it directly holds, but instead of broadcasting only to its own connections, it publishes to Redis, and every instance's subscriber picks the message up and delivers it locally. This is a common pattern for distributed WebSocket systems, and it applies to other message brokers as well, not just Redis.

A few other things to plan for when scaling:

• Sticky sessions, so a load balancer keeps a given client connected to the same server for the life of that connection, since a WebSocket can't be transparently handed off mid-connection the way an HTTP request can.
• Stateless application design where possible, keeping session and user data in Redis or a database rather than only in process memory, so a client can reconnect to any instance if its original one goes down.
• Monitoring concurrent connections per instance, since memory and file descriptor limits will eventually cap how many connections a single process can hold.
• Containerized deployment with autoscaling based on connection count rather than just CPU, since a WebSocket-heavy workload can hold many idle-but-open connections without much CPU load.

Teams building this kind of distributed architecture in Python AI often end up structuring the WebSocket layer as one of several Python microservices, separate from the services handling background processing or data storage, so each piece can scale independently.

Common Python WebSocket challenges

A few problems come up repeatedly in production WebSocket systems, and most of them have established solutions.

Unexpected disconnections and network instability. Mobile clients in particular lose connectivity often. Catch WebSocketDisconnect on the server, and on the client side implement a reconnection strategy with exponential backoff rather than retrying immediately in a tight loop.

Connection timeouts. Some proxies and load balancers close idle connections after a fixed period. Sending periodic ping/pong frames, or an application-level heartbeat message, keeps the connection active and gives the server a way to detect a client that's gone silent without a clean close.

Message ordering. WebSocket frames arrive in order over a single connection, but across a distributed system with multiple publishers, messages can still arrive out of the order they were generated. Including a sequence number or timestamp in the payload lets the client detect and handle out-of-order delivery where it matters.

Duplicate messages. Retry logic on the client or reconnection handling can result in the same message being sent twice. An idempotency key on each message lets the receiving side deduplicate.

Connection leaks and memory consumption. A connection that's removed from tracking without actually being closed, or vice versa, will leak over time. Always close the WebSocket and remove it from the manager in the same code path, ideally in a finally block.

Authentication expiry mid-connection. For long-running connections, check token expiry periodically rather than only at connect time, and close the connection with an appropriate code if the token has expired.

Server restarts and deployments. Every open connection drops when a server process restarts. Clients need reconnection logic regardless of how well the server is built, and rolling deployments with connection draining reduce how many clients get dropped at once.

Python WebSockets best practices

• Use async/await consistently through the WebSocket handling code; a blocking call inside an async endpoint stalls every other connection sharing that event loop.
• Keep the connection manager as a single, well-tested component responsible for tracking, adding, and removing connections, rather than scattering that logic across endpoints.
• Wrap message handling in try/except blocks that specifically catch WebSocketDisconnect, plus a general exception handler that logs unexpected errors without crashing the connection loop.
• Authenticate before accepting the connection, and re-validate tokens for long-lived sessions.
• Validate every message against an expected schema, using Pydantic models, before acting on it.
• Log connection events, including connect, disconnect, and errors, with enough context (user ID, connection duration) to debug issues after the fact.
• Monitor active connection counts and message throughput so capacity issues show up before they cause outages.
• Rate limit messages per connection to prevent abuse and accidental flooding from a buggy client.
• Build reconnection logic into the client, not just error handling on the server.
• Implement graceful shutdown, closing WebSocket connections cleanly with an appropriate close code when the server is shutting down, rather than letting connections drop abruptly.
• Plan for horizontal scaling early, since retrofitting a pub/sub layer onto a WebSocket system that assumed a single process is more work than designing for it from the start.
• Test WebSocket endpoints using FastAPI's TestClient, which supports WebSocket connections for integration testing without needing a running server.

When should you use FastAPI WebSockets?

WebSockets are a good fit when the server needs to push data to the client without the client asking first, and when that needs to happen frequently or with low latency. Common examples include:

• Chat applications, where messages need to appear for other users instantly.
• Real-time notifications, such as alerts or status changes.
• Live dashboards showing metrics, logs, or monitoring data that updates continuously.
• Collaboration tools, like shared documents or whiteboards, where multiple users edit the same state.
• Multiplayer applications, where game state needs to sync across clients with minimal delay.
• Trading interfaces, where price or order book updates need to reach the client as they happen.
• Monitoring systems tracking infrastructure or application health in real time.

WebSockets can be unnecessary overhead in other cases. If updates are infrequent, a REST endpoint with client-side polling is often simpler to build and debug. If updates only flow from server to client and never the other way, server-sent events avoid the complexity of managing a full-duplex connection. And for anything that doesn't need real-time delivery at all, a standard REST API remains the simpler and more maintainable choice.

Conclusion

Python's async ecosystem, combined with FastAPI's native WebSocket support on ASGI, gives developers a practical path to building real-time features without stepping outside their existing Python stack. A working WebSocket endpoint takes only a few lines of code, but a production-ready one needs a proper connection manager, authentication before the handshake completes, validated messages, and a plan for scaling connections across multiple server instances, typically through Redis Pub/Sub or a similar broker.

The practical takeaway for developers evaluating this for their own project: reach for WebSockets when the application genuinely needs bidirectional, low-latency communication, and be honest about whether polling or server-sent events would solve the problem with less operational complexity. When WebSockets are the right tool, FastAPI provides a solid, async-native foundation for building them, and teams that need help scaling out a Python backend around this pattern, whether that's structuring the API integration services between WebSocket and REST layers or extending the system into distributed Python microservices, are increasingly turning to specialized Python development services and backend development services to get the architecture right the first time. For teams without deep in-house async Python experience, it's often more efficient to hire Python developers with production WebSocket experience or to hire dedicated developers for the broader backend build than to learn these scaling patterns under production pressure.