How to trigger message send of Fastapi websocket outside of Fastapi app

FastAPI

WebSocket

Messaging

Python

Backend Development

How to trigger message send of Fastapi websocket outside of Fastapi app

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design

Networking & APIs Python

Introduction

FastAPI WebSocket connections are managed inside the ASGI application context, so you cannot directly call websocket.send_text() from outside a route handler. To send messages from background tasks, external services, or other parts of your application, you need a shared connection manager that stores active WebSocket connections and exposes a method to broadcast or send targeted messages. Common patterns include an in-memory connection manager, Redis pub/sub for multi-process setups, or an asyncio queue bridge.

Basic WebSocket Setup in FastAPI

python

1from fastapi import FastAPI, WebSocket, WebSocketDisconnect
2
3app = FastAPI()
4
5@app.websocket("/ws")
6async def websocket_endpoint(websocket: WebSocket):
7    await websocket.accept()
8    try:
9        while True:
10            data = await websocket.receive_text()
11            await websocket.send_text(f"Echo: {data}")
12    except WebSocketDisconnect:
13        print("Client disconnected")

The websocket object only exists inside this handler. Once the handler returns, the connection is closed. To send messages from elsewhere, you need to store a reference to the connection.

Connection Manager Pattern

Create a class that tracks active connections and provides send/broadcast methods:

python

1from fastapi import FastAPI, WebSocket, WebSocketDisconnect
2from typing import Dict, List
3
4class ConnectionManager:
5    def __init__(self):
6        self.active_connections: Dict[str, WebSocket] = {}
7
8    async def connect(self, client_id: str, websocket: WebSocket):
9        await websocket.accept()
10        self.active_connections[client_id] = websocket
11
12    def disconnect(self, client_id: str):
13        self.active_connections.pop(client_id, None)
14
15    async def send_to(self, client_id: str, message: str):
16        websocket = self.active_connections.get(client_id)
17        if websocket:
18            await websocket.send_text(message)
19
20    async def broadcast(self, message: str):
21        for websocket in self.active_connections.values():
22            await websocket.send_text(message)
23
24manager = ConnectionManager()
25app = FastAPI()
26
27@app.websocket("/ws/{client_id}")
28async def websocket_endpoint(websocket: WebSocket, client_id: str):
29    await manager.connect(client_id, websocket)
30    try:
31        while True:
32            await websocket.receive_text()
33    except WebSocketDisconnect:
34        manager.disconnect(client_id)

Sending from an HTTP Endpoint

The simplest way to trigger a WebSocket message externally — call an HTTP endpoint that uses the connection manager:

python

1from fastapi import FastAPI
2from pydantic import BaseModel
3
4class Message(BaseModel):
5    client_id: str
6    text: str
7
8@app.post("/send")
9async def send_message(msg: Message):
10    await manager.send_to(msg.client_id, msg.text)
11    return {"status": "sent"}
12
13@app.post("/broadcast")
14async def broadcast_message(msg: dict):
15    await manager.broadcast(msg["text"])
16    return {"status": "broadcast sent"}

External services can trigger WebSocket messages by making HTTP POST requests to these endpoints.

Sending from a Background Task

Use FastAPI's BackgroundTasks or a standalone asyncio task:

python

1import asyncio
2from fastapi import BackgroundTasks
3
4async def monitor_prices():
5    """Background task that sends updates to connected clients."""
6    while True:
7        price = await fetch_current_price()
8        await manager.broadcast(f'{{"price": {price}}}')
9        await asyncio.sleep(5)
10
11@app.on_event("startup")
12async def startup():
13    asyncio.create_task(monitor_prices())

Since the background task runs in the same asyncio event loop as FastAPI, it can directly call manager.broadcast().

Using an Asyncio Queue Bridge

For decoupling message producers from the WebSocket handler:

python

1import asyncio
2
3message_queue: asyncio.Queue = asyncio.Queue()
4
5async def queue_consumer():
6    """Reads from queue and broadcasts to WebSocket clients."""
7    while True:
8        message = await message_queue.get()
9        await manager.broadcast(message)
10
11@app.on_event("startup")
12async def startup():
13    asyncio.create_task(queue_consumer())
14
15# Any code in the same process can put messages on the queue
16@app.post("/notify")
17async def notify(data: dict):
18    await message_queue.put(data["message"])
19    return {"status": "queued"}

Multi-Process Setup with Redis Pub/Sub

When running multiple FastAPI workers (e.g., with Gunicorn), an in-memory connection manager only knows about connections on its own worker. Use Redis pub/sub to broadcast across workers:

python

1import aioredis
2import asyncio
3import json
4
5redis = aioredis.from_url("redis://localhost:6379")
6
7async def redis_listener():
8    """Subscribe to Redis channel and forward to local WebSocket clients."""
9    pubsub = redis.pubsub()
10    await pubsub.subscribe("ws_broadcast")
11    async for message in pubsub.listen():
12        if message["type"] == "message":
13            data = message["data"].decode()
14            await manager.broadcast(data)
15
16@app.on_event("startup")
17async def startup():
18    asyncio.create_task(redis_listener())
19
20# Any process can publish to Redis
21async def send_from_anywhere(message: str):
22    await redis.publish("ws_broadcast", message)

This works from any process that can connect to Redis — background workers, Celery tasks, separate microservices, or management scripts.

Common Pitfalls

Calling websocket.send_text() from a different thread: WebSocket objects are not thread-safe. If you call send_text() from a thread (e.g., a synchronous background job), use asyncio.run_coroutine_threadsafe(coro, loop) to schedule the send on the event loop.
In-memory manager with multiple workers: Gunicorn with multiple workers runs separate Python processes, each with its own connection manager. Client A connected to worker 1 cannot receive messages sent through worker 2's manager. Use Redis pub/sub or a shared message broker.
Not handling WebSocketDisconnect: If a client disconnects and you try to send a message, send_text() raises an exception. Always handle WebSocketDisconnect and remove the connection from the manager.
Blocking the event loop: Calling synchronous (blocking) functions in the WebSocket handler or background task blocks the entire event loop, preventing all WebSocket connections from sending or receiving. Use await for I/O or run blocking code in asyncio.to_thread().
Memory leaks from stale connections: If disconnect() is not called when clients drop (network failure without clean close), the connection manager accumulates dead WebSocket references. Implement periodic health checks or heartbeat pings to detect and clean up stale connections.

Summary

Store active WebSocket connections in a ConnectionManager class accessible throughout the app
Send messages from HTTP endpoints by calling manager.send_to() or manager.broadcast()
Background tasks in the same event loop can call the manager directly
Use asyncio.Queue to decouple message producers from WebSocket handlers
For multi-worker deployments, use Redis pub/sub to broadcast across processes
Always handle client disconnection and clean up stale connections

System Design Fundamentals

Build a strong foundation in designing scalable, reliable distributed systems.

View the course

Track what you have practised

A free account saves your progress, solutions and study plan across every problem on Codemia.

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design

How to trigger message send of Fastapi websocket outside of Fastapi app

System Design practice on Codemia

Introduction

Basic WebSocket Setup in FastAPI

Connection Manager Pattern

Sending from an HTTP Endpoint

Sending from a Background Task

Using an Asyncio Queue Bridge

Multi-Process Setup with Redis Pub/Sub

Common Pitfalls

Summary

Related reading

System Design Fundamentals

Track what you have practised

System Design practice on Codemia