FastAPI WebSocket: уведомления в браузер

Сделаем маленькое FastAPI-приложение:.

Ниже — разделы про что мы соберем, когда FastAPI WebSocket уместен и подготовка, чтобы быстро понять устройство материала, практические ограничения и типовые точки отказа.

Содержание

Что мы соберем
Когда FastAPI WebSocket уместен
Подготовка
Первый WebSocket endpoint
Что делает accept
Отправляем JSON
Несколько подключенных клиентов
Уведомление из HTTP route
Что важно для продакшена
Авторизация
Частые ошибки
WebSocketDisconnect
Браузер не подключается к ws://localhost:8000/ws
На HTTPS-домене WebSocket не работает
Несколько workers теряют сообщения
Swagger не показывает WebSocket как обычный endpoint
Ответы на эти вопросы могут быть для вас полезными
FastAPI поддерживает WebSocket из коробки?
Можно ли отправлять JSON через FastAPI WebSocket?
Почему manager из урока не подходит для большого продакшена?
Нужно ли ставить библиотеку websockets?
Можно ли подключить React/Vue к FastAPI WebSocket?
Что почитать дальше по WebSockets

Что мы соберем

Сделаем маленькое FastAPI-приложение:

HTTP-страница открывается в браузере;
браузер подключается к /ws;
сервер принимает сообщения;
сервер отправляет уведомления клиенту;
несколько клиентов могут получать одно событие.

Это хороший мост между "я понял WebSocket на Python" и "я хочу встроить real-time в API"

Когда FastAPI WebSocket уместен

FastAPI WebSocket хорошо подходит, если:

backend уже написан на FastAPI;
нужны уведомления в админку;
нужно показать статус долгой задачи;
хочется отправлять события из Python-кода в браузер;
рядом уже есть REST API и авторизация;
не хочется поднимать отдельный Node.js real-time сервис.

Если вам нужен только один echo-сервер без FastAPI, можно начать с библиотеки websockets. Если у вас приложение на FastAPI, логичнее использовать встроенную поддержку WebSocket

Подготовка

Создайте проект:

mkdir fastapi-websocket-demo
cd fastapi-websocket-demo
python -m venv .venv

Активируйте окружение

macOS/Linux:

source .venv/bin/activate

.venv\Scripts\Activate.ps1

Установите зависимости:

pip install fastapi "uvicorn[standard]"

Первый WebSocket endpoint

Создайте main.py:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!doctype html>
<html lang="ru">
  <head>
    <meta charset="utf-8">
    <title>FastAPI WebSocket</title>
  </head>
  <body>
    <h1>FastAPI WebSocket</h1>
    <div id="status">Подключаемся...</div>
    <ul id="messages"></ul>
    <form id="form">
      <input id="input" autocomplete="off" placeholder="Сообщение">
      <button>Отправить</button>
    </form>

    <script>
      const statusEl = document.querySelector('#status');
      const messagesEl = document.querySelector('#messages');
      const form = document.querySelector('#form');
      const input = document.querySelector('#input');

      const socket = new WebSocket('ws://localhost:8000/ws');

      socket.addEventListener('open', () => {
        statusEl.textContent = 'Соединение открыто';
      });

      socket.addEventListener('message', (event) => {
        const li = document.createElement('li');
        li.textContent = event.data;
        messagesEl.append(li);
      });

      socket.addEventListener('close', () => {
        statusEl.textContent = 'Соединение закрыто';
      });

      form.addEventListener('submit', (event) => {
        event.preventDefault();
        const text = input.value.trim();

        if (text && socket.readyState === WebSocket.OPEN) {
          socket.send(text);
          input.value = '';
        }
      });
    </script>
  </body>
</html>
"""

@app.get("/")
async def index():
    return HTMLResponse(html)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(f"Сервер получил: {message}")
    except WebSocketDisconnect:
        print("Клиент отключился")

Запуск:

uvicorn main:app --reload

Откройте:

http://localhost:8000

Введите сообщение. Сервер должен вернуть ответ

Что делает `accept`

В FastAPI WebSocket-соединение нужно принять:

await websocket.accept()

Без этого сервер не подтверждает handshake. Это похоже на явное "да, я принимаю WebSocket-соединение"

Отправляем JSON

Текстовые сообщения хороши для старта, но в приложении удобнее JSON

Изменим серверный цикл:

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    try:
        while True:
            data = await websocket.receive_json()

            await websocket.send_json({
                "type": "reply",
                "text": f"Получил событие {data.get('type')}",
                "payload": data,
            })
    except WebSocketDisconnect:
        print("Клиент отключился")

На клиенте:

socket.send(JSON.stringify({
  type: 'lead_created',
  name: text
}));

И при получении:

const data = JSON.parse(event.data);

Так легче расширять протокол: notification, task_finished, lead_created, typing, error

Несколько подключенных клиентов

Для уведомлений часто нужно отправить событие всем открытым вкладкам админки. Сделаем простой manager

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

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)

app = FastAPI()
manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)

    try:
        while True:
            message = await websocket.receive_text()
            await manager.broadcast(f"Новое сообщение: {message}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Откройте страницу в двух вкладках и отправьте сообщение из одной. Вторая тоже должна его увидеть

Уведомление из HTTP route

Теперь добавим HTTP endpoint, который рассылает событие:

@app.post("/notify")
async def notify():
    await manager.broadcast("Появилось новое уведомление")
    return {"ok": True}

Проверка:

curl -X POST http://localhost:8000/notify

Браузерные клиенты получат сообщение. Это уже очень похоже на реальную админку: REST-запрос создал событие, WebSocket доставил его открытым клиентам

Что важно для продакшена

Учебный manager хранит подключения в памяти одного процесса. Это нормально для урока, но не для масштабирования

Если процессов несколько, один процесс не знает о клиентах другого. Тогда нужны:

Redis pub/sub;
message broker;
отдельный real-time сервис;
sticky sessions, если архитектура это допускает;
продуманная схема каналов.

Для первого проекта не нужно начинать с распределенной архитектуры. Но знать ограничение важно

Авторизация

WebSocket тоже должен проверять пользователя. Возможные варианты:

cookie с сессией;
token в query string;
token в первом сообщении;
header, если клиент и библиотека это позволяют.

Нельзя оставлять /ws открытым, если через него идут приватные уведомления

Частые ошибки

`WebSocketDisconnect`

Это не всегда ошибка. Пользователь закрыл вкладку, интернет пропал, сервер перезапустился. Обрабатывайте отключение спокойно

Браузер не подключается к `ws://localhost:8000/ws`

Проверьте, что uvicorn запущен, путь /ws совпадает, порт 8000 верный

На HTTPS-домене WebSocket не работает

Используйте wss://, а не ws://. И проверьте proxy-настройки

Несколько workers теряют сообщения

In-memory список подключений живет внутри процесса. Для нескольких workers нужен общий слой доставки событий

Swagger не показывает WebSocket как обычный endpoint

Это нормально. WebSocket не является обычным REST route. Проверяйте его браузером, Postman или отдельным клиентом

Ответы на эти вопросы могут быть для вас полезными

FastAPI поддерживает WebSocket из коробки?

Да. В FastAPI есть WebSocket, а практические примеры есть в официальной документации

Можно ли отправлять JSON через FastAPI WebSocket?

Да. Есть методы receive_json и send_json. Для реальных событий это удобнее, чем просто строки

Почему manager из урока не подходит для большого продакшена?

Он хранит подключения в памяти одного процесса. Если процессов или серверов несколько, нужен общий механизм рассылки

Нужно ли ставить библиотеку `websockets`?

Для запуска через Uvicorn часто достаточно стандартных зависимостей FastAPI/Uvicorn, но в учебных и документационных примерах библиотека websockets может использоваться как клиентская или протокольная зависимость. Следуйте требованиям вашего окружения

Можно ли подключить React/Vue к FastAPI WebSocket?

Да. На frontend используется обычный браузерный WebSocket, а backend остается FastAPI

Что почитать дальше по WebSockets

Если вы собираете тему по шагам, рядом лучше открыть:

WebSocket на Python: быстрый пример клиента и сервера — понять базовый Python WebSocket без фреймворка.
WebSocket test online: как проверить соединение — проверить endpoint /ws до подключения интерфейса.
Nginx и WebSocket proxy: почему локально работает, а на сервере нет — подготовить FastAPI WebSocket к работе за Nginx.
Spring, Go и Django WebSocket: какой путь выбрать — сравнить FastAPI с Spring, Go и Django Channels.