FastAPIでWebSocketとSSEを実装する方法：接続管理、再接続、Redis、監視

WebSocketとSSE（Server-Sent Events）は、画面を再読み込みせずにサーバーの更新を届けるための仕組みです。
FastAPIでリアルタイム通知を作るときは、ブラウザからの送信が必要か、切断中の通知を復元するか、何台のアプリケーションで配信するかを先に決めます。

双方向通信にはWebSocket、サーバーからブラウザへの片方向配信にはSSEが基本です。
ただし、本番運用では接続できることだけでは足りません。
認証、再接続、取りこぼし、遅いクライアント、プロキシのタイムアウト、監視まで一つの設計として扱う必要があります。

FastAPIそのものの特徴やAPI設計から確認したい場合は、FastAPIの基本と設計ポイントも参照してください。

WebSocketとSSEの選び方

方式は流行や実装量ではなく、通信の向きと再開要件で選びます。
通知バッジや処理進捗のようにサーバーから送るだけならSSEが扱いやすく、チャットや共同編集のようにクライアントから頻繁に送るならWebSocketが適しています。

SSEとJSON Lines、バイナリ配信の違いは、FastAPIのストリーミングAPI設計で詳しく整理しています。

実装前に分けて考える四つの責務

リアルタイム機能は、通信経路だけを実装すると障害時の挙動が曖昧になります。
少なくとも、次の四つを別の責務として決めます。

• 接続管理：どのユーザーがどのルームに接続しているかを、各プロセス内で管理します。
• 配信：一つのプロセス内、またはRedisなどを介した複数プロセス間でイベントを届けます。
• 再開：イベントIDと履歴を使い、切断中に発生した更新を必要に応じて再送します。
• 運用：認証、入力制限、バックプレッシャ、タイムアウト、ログ、メトリクスを扱います。

ここでいうルームは、同じイベントを受け取る接続のグループです。
プロジェクト、チャットチャンネル、ユーザーIDなどをルームIDに対応させますが、ルーム名を知っているだけで参加できる設計にはしません。

単一プロセスでWebSocket接続を管理する

次の接続マネージャーは、ルームごとにWebSocketを保持する最小例です。
FastAPI公式ドキュメントと同様に、接続を受け入れてから送受信し、切断時には管理対象から除外します。

from collections import defaultdict
from fastapi import WebSocket


class ConnectionManager:
    def __init__(self) -> None:
        self.rooms: dict[str, set[WebSocket]] = defaultdict(set)

    async def connect(self, websocket: WebSocket, room_id: str) -> None:
        await websocket.accept()
        self.rooms[room_id].add(websocket)

    def disconnect(self, websocket: WebSocket, room_id: str) -> None:
        members = self.rooms.get(room_id)
        if members is None:
            return
        members.discard(websocket)
        if not members:
            self.rooms.pop(room_id, None)

    async def broadcast_room(self, room_id: str, event: dict) -> None:
        disconnected: list[WebSocket] = []
        for websocket in tuple(self.rooms.get(room_id, ())):
            try:
                await websocket.send_json(event)
            except Exception:
                disconnected.append(websocket)

        for websocket in disconnected:
            self.disconnect(websocket, room_id)

この例の集合は一つのPythonプロセスにしか存在しません。
ワーカーやコンテナを増やすと、別プロセスの接続には届かないため、後述するプロセス間配信が必要です。

このままでは送信が直列なので、遅い接続が同じルームの配信を待たせる可能性もあります。
本番では接続ごとに上限付き送信キューを置き、キューが埋まった場合の扱いを明文化します。

認証、認可、入力検証を接続処理に入れる

ブラウザ標準のWebSocketコンストラクターが受け取るのはURLとサブプロトコルであり、任意のAuthorizationヘッダーを設定する一般的な引数はありません。
同一サイトのアプリケーションでは、安全な属性を付けたセッションCookieを使い、Originも検査する構成が候補になります。
クエリ文字列へトークンを入れる場合は、URLがログや履歴に残る可能性を考慮し、短時間だけ有効な値とログの伏字処理が必要です。

次の例では、セッション検証とルーム参加判定をプロジェクト固有の関数へ分離しています。
接続前に認証と認可を終え、受信JSONはPydanticモデルで検証します。

from typing import Literal
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from fastapi import WebSocketException, status
from pydantic import BaseModel, Field, ValidationError

router = APIRouter()
manager = ConnectionManager()
ALLOWED_ORIGINS = {'https://app.example.com'}


class ClientMessage(BaseModel):
    type: Literal['message']
    text: str = Field(min_length=1, max_length=2000)


@router.websocket('/ws/{room_id}')
async def websocket_endpoint(websocket: WebSocket, room_id: str) -> None:
    origin = websocket.headers.get('origin')
    if origin not in ALLOWED_ORIGINS:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)

    session = authenticate_session(websocket.cookies.get('session'))
    if session is None or not can_join_room(session.user_id, room_id):
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)

    await manager.connect(websocket, room_id)
    try:
        while True:
            try:
                message = ClientMessage.model_validate(
                    await websocket.receive_json()
                )
            except ValidationError:
                await websocket.send_json({
                    'type': 'error',
                    'code': 'invalid_message',
                })
                continue

            await manager.broadcast_room(room_id, {
                'type': 'message',
                'sender_id': session.user_id,
                'text': message.text,
            })
    except WebSocketDisconnect:
        pass
    finally:
        manager.disconnect(websocket, room_id)

authenticate_sessionとcan_join_roomは説明用の境界であり、実際の認証基盤と権限モデルに接続します。
認証済みかどうかと、そのルームへ参加できるかどうかは別の判定です。

FastAPIでSSEを配信する

FastAPI 0.135.0以降では、fastapi.sseのEventSourceResponseとServerSentEventを利用できます。
イベント本文だけでなく、種類、ID、再接続までの待ち時間も明示できます。

from collections.abc import AsyncIterable
from typing import Annotated
from fastapi import APIRouter, Header
from fastapi.sse import EventSourceResponse, ServerSentEvent

router = APIRouter()


@router.get('/events', response_class=EventSourceResponse)
async def stream_events(
    last_event_id: Annotated[str | None, Header()] = None,
) -> AsyncIterable[ServerSentEvent]:
    async for event in event_store.follow(after=last_event_id):
        yield ServerSentEvent(
            data=event.payload,
            event=event.type,
            id=event.id,
            retry=3000,
        )

event_store.follow()は、保存済みイベントの読み出しと新着待機を担うアプリケーション固有のインターフェースです。
単純な通知だけならメモリ上のキューでも動きますが、切断中の更新を復元するなら、イベントIDで読み直せる保存先が必要です。

現在のEventSourceResponseは、無通信時のキープアライブ、Cache-Control: no-cache、Nginxのバッファリングを抑えるX-Accel-Buffering: noを扱います。
古いFastAPIを使う場合はStreamingResponseでSSE形式を組み立てられますが、改行、キープアライブ、切断、ヘッダーを実装側で確認します。

再接続とイベントの取りこぼしを分ける

再接続は通信を張り直す処理であり、再開は届かなかったイベントを復元する処理です。
再接続に成功しても、サーバーが履歴を保存していなければ切断中のイベントは戻りません。

1. サーバーは、順序を判断できるイベントIDを各イベントへ付けます。
2. クライアントは、正常に反映した最後のイベントIDを保持します。
3. 切断後は、指数バックオフと小さなランダム待機を組み合わせて再接続します。
4. SSEではブラウザがLast-Event-IDを送ります。WebSocketでは接続後のメッセージなど、アプリケーションの手順としてIDを送ります。
5. サーバーは保存先から未配信分を返し、その後に新着配信へ切り替えます。

同じイベントが再送される構成では、クライアント側もイベントIDを使って重複反映を避けます。
切断中の通知を捨ててもよい画面と、監査や決済のように失えない処理を同じ配信方式で扱わないことが大切です。

Redisで複数インスタンスへ配信する

複数のFastAPIインスタンスで接続を受ける場合、Redis Pub/Subをプロセス間の配信バスとして使えます。
各インスタンスは同じチャネルを購読し、受け取ったイベントを自分が保持するWebSocketへ配信します。

発行元でもRedisから受信してローカル配信する形に統一すると、発行時の直接配信とRedis経由の配信が重複しません。
チャネル名には本番、検証などの環境を識別できる接頭辞を含めます。

import json
from redis.asyncio import Redis

CHANNEL = 'production:realtime:v1'


async def publish_event(redis: Redis, room_id: str, payload: dict) -> None:
    envelope = {'room_id': room_id, 'payload': payload}
    await redis.publish(
        CHANNEL,
        json.dumps(envelope, ensure_ascii=False),
    )


async def receive_events(redis: Redis, manager: ConnectionManager) -> None:
    pubsub = redis.pubsub()
    await pubsub.subscribe(CHANNEL)
    try:
        async for message in pubsub.listen():
            if message.get('type') != 'message':
                continue
            envelope = json.loads(message['data'])
            await manager.broadcast_room(
                envelope['room_id'],
                envelope['payload'],
            )
    finally:
        await pubsub.aclose()

外部入力をeval()で復元してはいけません。
JSONとして読み取り、必要ならイベントの種類、ルームID、本文をスキーマ検証してから配信します。

Pub/Subで失われる通知を許容できるか

Redis Pub/Subの配信保証はat-most-onceです。
購読側が切断している間や処理に失敗したとき、イベントは再送されません。

Pub/Subは接続中のクライアントへ素早く通知する経路として使い、正しい現在値はデータベースやAPIから再取得できる設計にすると、通知と業務データの役割が混ざりません。

バックプレッシャを設計する

バックプレッシャは、受信側より送信側が速いときに、待ち行列が増え続ける問題です。
一つの遅いブラウザのためにメモリを使い切らないよう、接続ごとのキューには上限を設けます。

• 画面の現在値は、古い更新を捨てて最新値へまとめます。
• 高頻度のメトリクスは、一定間隔でまとめて配信します。
• 順番と欠落が許されないイベントは、破棄せず永続化し、再開手順で読み直します。
• 上限到達時に接続を閉じる場合は、クライアントが再同期できる終了理由と手順を用意します。

キュー長、破棄数、送信待ち時間を記録しなければ、通知が遅れ始めても運用側が気付けません。

購読タスクをlifespanで管理する

Redisの購読タスクはアプリケーション起動時に開始し、終了時にキャンセルして接続を閉じます。
FastAPIではstartup処理へ置きっぱなしにするより、開始と後始末を同じlifespan関数にまとめると所有関係が明確です。

import asyncio
from contextlib import asynccontextmanager, suppress
from fastapi import FastAPI
from redis.asyncio import Redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    redis = Redis.from_url(
        'redis://redis:6379/0',
        decode_responses=True,
    )
    task = asyncio.create_task(receive_events(redis, manager))
    app.state.redis = redis
    try:
        yield
    finally:
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task
        await redis.aclose()


app = FastAPI(lifespan=lifespan)

共有リソースの初期化、終了処理、複数ワーカーでの扱いは、FastAPIのlifespanによる共有リソース管理でも解説しています。

Nginxとプロキシの設定を確認する

WebSocketのUpgradeとConnectionはhop-by-hopヘッダーなので、Nginxから上流へ明示的に渡します。
無通信時間がproxy_read_timeoutを超えると接続が閉じるため、タイムアウトとハートビートの間隔を対にして決めます。

map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

location /realtime/ws/ {
    proxy_pass http://fastapi_upstream;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection $connection_upgrade;
    proxy_read_timeout 75s;
}

location /realtime/events {
    proxy_pass http://fastapi_upstream;
    proxy_buffering off;
    proxy_cache off;
}

SSEでは、アプリケーションが正しくイベントを出していても、NginxやCDNが応答をまとめると画面への反映が遅れます。
レスポンスヘッダー、バッファリング、キャッシュ、アイドルタイムアウトを、実際に通るすべての中継で確認します。

セキュリティで確認する項目

• 暗号化：公開環境ではHTTPSとWSSを使います。
• 認証：接続時にセッションや短期トークンを検証し、失効後の扱いも決めます。
• 認可：ルーム参加、送信、購読の権限を別々に判定します。
• Origin検査：Cookieを使うWebSocketでは、許可したWebオリジンからの接続かを確認します。
• 入力制限：メッセージの型、文字数、全体サイズ、送信頻度を制限します。
• ログ保護：トークン、Cookie、本文、個人情報を無条件に記録しません。
• 障害分離：不正な一接続や一ルームが、全接続の送信処理を止めない構造にします。

監視とテストで運用可能性を確かめる

HTTPの成功率だけでは、長時間接続の劣化を把握できません。
接続、配信、再接続、キュー、プロセス間連携を別々に観測します。

テストでは正常な送受信だけでなく、クライアント切断、アプリケーション再起動、Redis切断、遅い受信、プロキシのタイムアウト、イベント再送も再現します。
入力モデルと依存関係を含むテスト設計は、FastAPIで壊れにくいAPIを作るための型定義とテストも参考になります。

リアルタイム更新を利用者へ伝える

通信がリアルタイムでも、画面の変化が利用者へ伝わらなければ機能として不十分です。
接続状態、最終更新時刻、再接続中、情報が古い状態を文字でも示し、色だけに依存させません。

処理完了や接続状態のような短い通知にはrole='status'とaria-live='polite'を利用できます。
チャットのように項目が順に増える領域にはrole='log'が候補です。
ただし、高頻度の値を一件ずつ読み上げると操作を妨げるため、意味のある単位へまとめ、更新の一時停止や頻度調整も用意します。

<p role='status' aria-live='polite' aria-atomic='true'>
  接続しました。最終更新は10時30分です。
</p>

ライブリージョンを付ければ自動的に適合を保証できるわけではありません。
キーボード操作、フォーカス、読み上げ順、更新頻度を実際の支援技術と操作条件で確認します。

導入の順序

1. 通知の向き、許容遅延、欠落の可否を決め、WebSocketかSSEを選びます。
2. 単一プロセスで認証、認可、入力検証、切断処理まで実装します。
3. 再接続と再開を分け、イベントIDと保存先の要否を決めます。
4. 複数インスタンスが必要なら、Redis Pub/Subなどでプロセス間配信を追加します。
5. 上限付きキュー、レート制限、プロキシ設定、監視を追加します。
6. 切断、再起動、遅い接続、負荷を含む試験を行い、運用手順を整えます。

リアルタイム機能の成否は、接続速度だけでは決まりません。
切れたときにどう戻るか、失ってよいイベントは何か、遅い接続をどう隔離するかを先に決めると、通信方式と保存方式を無理なく選べます。

公式資料

• FastAPI WebSockets
• FastAPI Server-Sent Events
• FastAPI Lifespan Events
• Starlette WebSocket
• Redis Pub/Sub
• MDN WebSocketコンストラクター
• MDN Server-sent events
• Nginx WebSocket proxying
• W3C WAIのstatusロール解説
• W3C WAIのlogロール解説