開発

FastAPIのWebSocket実装｜チャットの接続管理・切断処理とRedisで複数ワーカー対応

FastAPIのWebSocket実装｜チャットの接続管理・切断処理とRedisで複数ワーカー対応

FastAPIのWebSocketは、@app.websocketを1つ書けば動きます。手間がかかるのはその先で、接続の一覧をどこに持つか、切断をどこで拾うか、認証をいつ行うか、ワーカーを増やしたときに配信をどう届けるかの4点です。この記事では、FastAPI 0.162.5で小さなエコーサーバーを作り、ルーム付きのチャットへ広げ、Redis Pub/Subで複数ワーカーに対応させるまでをコードで進めます。nginxとUvicornの本番設定、TestClientでのテスト、自前で実装しない方がよい場面の判断も扱います。

まとめ：FastAPIのWebSocket実装で最初に決める接続管理と配信の経路

結論から書きます。公式ドキュメントのConnectionManagerは、接続をプロセスのメモリ上のリストに持つ設計です。公式自身の説明も「単一プロセスでしか動かない」という制約を明記したものです。Uvicornの--workersを2以上にした時点で、別のワーカーに繋がった利用者へメッセージが届かなくなります。本番を想定するなら、最初からRedis Pub/Subのような外部の中継を挟んだ構成で書きます。

公式が中継の手段として案内しているencode/broadcasterは、2025年8月19日にアーカイブされ、更新が止まっています。新規の案件では採用せず、redis-pyのredis.asyncioで購読タスクを自分で書く方が保守の見通しが立ちます。数十行で済む量です。

認証はaccept()の前に済ませ、失敗したらWebSocketExceptionで閉じます。Redis Pub/Subは切断中のメッセージを保存しないため、取りこぼしが許されない通知では、履歴をDBから取り直す経路を別に用意します。

FastAPIでWebSocketを動かす検証環境とエコーサーバーの最小コード

まずはクライアントから送った文字列をそのまま返すだけのサーバーを作り、接続から切断までの流れを確かめます。

Python 3.11でWebSocket用のFastAPI 0.162系を揃える手順

WebSocket用の部品はfastapi[standard]で揃えます。2026年11月13日時点のPyPIで、FastAPIの最新は0.162.5（2026年11月11日公開）です。FastAPI本体の動作要件はPython 3.10以上ですが、土台のStarlette 1.10.0と、Uvicornが使うwebsockets 17.4はどちらもPython 3.11以上を要求します。3.10で入れると古い版が選ばれるため、ここではPython 3.11以上に揃えます。FastAPIを初めて動かす場合はFastAPIの最短の起動手順とasync/defの使い分けを先に済ませてください。

# 仮想環境を作り、FastAPIとRedisクライアントを入れる
python3.11 -m venv .venv
source .venv/bin/activate        # Windowsは .venv\Scripts\activate
pip install "fastapi[standard]==0.162.5" "redis==8.1.0"
python -c "import fastapi, starlette, uvicorn, websockets; print(fastapi.__version__, starlette.__version__, uvicorn.__version__, websockets.__version__)"

[standard]を付けるとuvicorn[standard]が入り、その依存としてwebsocketsパッケージも入ります。[standard]を付けずにFastAPIと素のUvicornだけを入れると、UvicornはWebSocketを処理するライブラリを見つけられず、アップグレード要求を受け付けません。FastAPI公式のWebSocketsの章が冒頭でwebsocketsの追加を指示しているのはこのためです。

@app.websocketのエコーサーバーをPythonのクライアントから試す流れ

# main.py（FastAPI 0.162.5）
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def echo(websocket: WebSocket):
    await websocket.accept()          # ハンドシェイクを完了させる
    try:
        while True:
            text = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {text}")
    except WebSocketDisconnect as e:
        # クライアントが閉じると受信側で例外になる
        print("disconnected", e.code)

fastapi dev main.pyで起動し、別の端末から次のスクリプトで接続します。websocketsは13.0で新しいasyncio実装が加わり、17系のクライアントはwebsockets.asyncio.clientのconnectを使います（websocketsのasyncioクライアントのリファレンス）。

# client.py（websockets 17.4）
import asyncio
from websockets.asyncio.client import connect

async def main():
    async with connect("ws://127.0.0.1:8000/ws") as ws:
        await ws.send("こんにちは")
        print(await ws.recv())        # Message text was: こんにちは

asyncio.run(main())

スクリプトを抜けるとクライアント側が接続を閉じ、サーバーのターミナルに出る表示はdisconnected 1000です。1000は正常終了を表すクローズコードで、RFC 6455の7.4.1節で定義されています。ハンドシェイクやフレームの仕組みそのものはハンドシェイクから運用判断まで扱ったWebSocketの解説にまとめています。

チャットを作るConnectionManagerの接続管理と切断処理の書き方

公式のConnectionManagerは全員に配る1部屋だけの例です。業務のチャットや通知では部屋（案件ごと、問い合わせごと）の単位が要るため、ルーム別に接続を持つ形に書き換えます。

ルーム単位で接続を持つConnectionManagerと送信失敗時の後始末

# rooms.py
from fastapi import WebSocket

class RoomManager:
    def __init__(self) -> None:
        # ルーム名 → そのルームに居る接続の集合
        self.rooms: dict[str, set[WebSocket]] = {}

    async def connect(self, room: str, ws: WebSocket) -> None:
        await ws.accept()
        self.rooms.setdefault(room, set()).add(ws)

    def disconnect(self, room: str, ws: WebSocket) -> None:
        members = self.rooms.get(room)
        if members is None:
            return
        members.discard(ws)               # 二重に呼ばれても例外にしない
        if not members:
            del self.rooms[room]          # 空のルームを残さない

    async def broadcast(self, room: str, text: str) -> None:
        dead = []
        for ws in list(self.rooms.get(room, ())):
            try:
                await ws.send_text(text)
            except Exception:
                # 切断を検知する前に送ると失敗するので、後でまとめて外す
                dead.append(ws)
        for ws in dead:
            self.disconnect(room, ws)

公式例との違いは3つです。リストを集合に変え、removeではなくdiscardで外すことで、受信側と送信側の両方から外そうとしても例外が出ません。空になったルームは辞書から消します。送信の失敗を拾って外すのは、相手が回線ごと落ちた場合、受信側でWebSocketDisconnectが出るより先に送信が失敗することがあるためです。

WebSocketDisconnectを受けたときの除去と退室通知を漏らさない構成

# chat_local.py（単一プロセス版）
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from rooms import RoomManager

app = FastAPI()
manager = RoomManager()

@app.websocket("/ws/{room}")
async def chat(websocket: WebSocket, room: str):
    await manager.connect(room, websocket)
    try:
        async for text in websocket.iter_text():
            await manager.broadcast(room, text)
    finally:
        # 正常な切断でも、例外で抜けた場合でも必ず外す
        manager.disconnect(room, websocket)
        await manager.broadcast(room, "1人が退室しました")

受信ループはiter_text()で書けます。Starlette公式のWebSocketsの章は、WebSocketDisconnectが送出されるとイテレータが終了すると説明しており、exceptを書かなくてもループを抜けます。後始末をfinallyに置くのは、メッセージ処理の途中で別の例外が起きた場合にも接続を一覧から外すためです。except WebSocketDisconnectだけに後始末を書くと、それ以外の例外で抜けた接続が一覧に残り、送信のたびに失敗する幽霊になります。

ブロードキャストは1人ずつawaitで送るため、回線の遅い相手がいると後ろの人が待たされます。参加者が数十人を超えるルームでは、接続ごとにasyncio.Queueと送信タスクを持たせ、ブロードキャストはキューに積むだけにする構成へ変えます。

accept前の認証とクローズコードでWebSocket接続を拒否する実装

WebSocketの認証は、HTTPのAPIと同じ感覚で書くと穴が開きます。ブラウザから送れる情報が限られるためです。

DependsとQueryでチケットを検証しWS_1008で閉じる依存関数の書き方

# auth.py
from typing import Annotated
from fastapi import Depends, Query, WebSocket, WebSocketException, status

# 実運用ではRedisなどに有効期限付きで置く。ここでは説明用の辞書
TICKETS: dict[str, str] = {"t-123": "user-1"}

async def ws_user(ticket: Annotated[str | None, Query()] = None) -> str:
    user = TICKETS.pop(ticket, None) if ticket else None   # 1回使ったら無効にする
    if user is None:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    return user

# エンドポイント側
# @app.websocket("/ws/{room}")
# async def chat(websocket: WebSocket, room: str, user: Annotated[str, Depends(ws_user)]):
#     await manager.connect(room, websocket)   # 認証が通った後でaccept

FastAPI公式は、WebSocketではHTTPExceptionではなくWebSocketExceptionを送出するよう書いています。1008は「ポリシー違反」を表すクローズコードです。依存関数はエンドポイントの本体より先に実行されるため、accept()の前に拒否できる仕組みです。Dependsで認証用の依存を組み立てる方法はFastAPIのDependsの使い方｜依存性注入・yield・テストの差し替えをコードで解説で説明しています。Starlette公式によれば、accept()の前に閉じた場合、サーバーはクライアントへHTTP 403を返します。ブラウザ側ではクローズコード1008ではなく、接続の失敗として見える点を前提にフロントエンドのエラー表示を作ります。

ブラウザのWebSocketがヘッダーを送れない制約とチケット方式の理由

MDNのWebSocket()コンストラクタの解説のとおり、ブラウザ標準のWebSocketが受け取る引数はURLとサブプロトコルの2つだけです。AuthorizationヘッダーにBearerトークンを載せる、という書き方はできません。選択肢はCookie、URLのクエリ、サブプロトコル欄への埋め込みに絞られます。

クエリに長期のアクセストークンをそのまま載せる方式は採りません。URLはnginxなどのアクセスログに残るためです。上のコードのように、通常のHTTPのAPIで認証済みの利用者に短命で1回限りのチケットを発行し、WebSocketの接続時にそれを渡します。ログに残っても再利用できない値にしておく設計です。Cookieで認証する場合は、別サイトからの接続を防ぐためOriginヘッダーを許可リストと照合します。

複数ワーカーで配信が届かない問題をRedis Pub/Subで解く構成

ここが本番化で最も詰まりやすい箇所です。開発中は動いていたチャットが、ワーカーを増やした途端に「届く人と届かない人がいる」状態になります。

Uvicornの–workersを増やすとメモリ上の接続一覧が分断される理由

Uvicornの--workers 4は、アプリを4つの独立したプロセスで起動します。RoomManagerの辞書もプロセスごとに別物です。利用者Aがワーカー1に、利用者Bがワーカー3に繋がると、Aの発言はワーカー1の辞書にしか配られず、Bには届きません。FastAPI公式も、メモリ上のリストで扱う例は「プロセスが動いている間だけ、単一のプロセスでしか動かない」と書いています。ワーカー数とプロセス構成の決め方はGunicornとの違いと本番設定を扱ったUvicornの解説を参照してください。

解き方は、発言を自分の辞書に直接配らず、いったんRedisへPUBLISHすることです。全ワーカーが同じチャネルを購読し、受け取ったメッセージを自分の辞書にいる接続へ配ります。

redis.asyncioのPub/Subでワーカー間に配る購読タスクの実装

# chat_redis.py（FastAPI 0.162.5 ／ redis-py 8.1.0）
import asyncio, json
from contextlib import asynccontextmanager
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import redis.asyncio as redis
from rooms import RoomManager

REDIS_URL = "redis://localhost:6379/0"
PREFIX = "prod:chat:"                  # 環境名を接頭辞にしてチャネルを分ける
manager = RoomManager()

async def relay(r: redis.Redis) -> None:
    # 全ルームを1本の購読でまとめて受け、自プロセスの接続へ配る
    async with r.pubsub() as pubsub:
        await pubsub.psubscribe(PREFIX + "*")
        while True:
            msg = await pubsub.get_message(ignore_subscribe_messages=True, timeout=None)
            if msg is None:
                continue
            room = msg["channel"].decode().removeprefix(PREFIX)
            await manager.broadcast(room, msg["data"].decode())

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = redis.from_url(REDIS_URL)
    task = asyncio.create_task(relay(app.state.redis))
    yield
    task.cancel()                          # 停止時に購読を止める
    await app.state.redis.aclose()         # asyncioでは明示的に閉じる

app = FastAPI(lifespan=lifespan)

@app.websocket("/ws/{room}")
async def chat(websocket: WebSocket, room: str):
    await manager.connect(room, websocket)
    r = websocket.app.state.redis
    try:
        async for text in websocket.iter_text():
            payload = json.dumps({"room": room, "text": text}, ensure_ascii=False)
            await r.publish(PREFIX + room, payload)   # 自分の辞書には直接配らない
    finally:
        manager.disconnect(room, websocket)

購読はredis-py公式のAsyncio Examplesと同じくget_message(ignore_subscribe_messages=True, timeout=None)で待ちます。同じページは、asyncioには自動で後始末をする仕組みが無いためaclose()で明示的に閉じるよう書いています。購読タスクをlifespanで起動・停止するのは、ワーカーごとに1本だけ購読を持たせるためです。接続ごとに購読を作ると、Redisへの接続数が利用者数に比例して増えます。

Redis公式は、Pub/SubがDB番号と無関係で、DB 10への発行がDB 1の購読者にも届くと説明しています。/0で分けたつもりでも、同じRedisを共有する検証環境と本番環境のメッセージは混ざります。上のコードでチャネルにprod:を付けたのはこのためです。

encode/broadcasterの2025年8月アーカイブと採用判断

FastAPI公式のWebSocketsの章は、より堅牢な構成としてencode/broadcasterを紹介しています。しかしGitHubのencode/broadcasterのリポジトリは2025年8月19日に所有者がアーカイブし、読み取り専用になりました。READMEはもともとAlpha版と位置づけ、broadcaster==0.3.0への厳密な固定を求めています。PyPIの最新は0.3.1（2024年8月1日公開）で、それ以降の版はありません。

判断は明確で、新規の案件には入れません。Redis、PostgreSQL、Kafkaを切り替えられる抽象化は魅力ですが、FastAPIとStarletteは0系・1系で更新が続いており、止まった部品は依存の更新で壊れたときに誰も直しません。必要なのがRedis Pub/Subだけなら、上のrelay関数の30行ほどで足ります。既に使っている案件は、版を固定したまま、次の改修でredis.asyncioへの置き換えを計画に入れます。

Pub/Subのat-most-once配信で切断中のメッセージが失われる前提

Redis公式のPub/Subの解説は、配信の性質をat-most-once（多くても1回）と明記しています。購読側がエラーや回線の切断でメッセージを処理できなければ、そのメッセージは永久に失われます。利用者側から見ると、スマートフォンが地下鉄で数十秒圏外になった間の発言は、再接続しても届きません。

実務ではまずこれだけ押さえれば足ります。発言はRedisへ流す前にDBへ保存し、クライアントは再接続の直後に「最後に受け取ったID以降」をHTTPのAPIで取り直す。WebSocketは「新着がある」ことを速く知らせる経路と割り切る設計です。サーバー側でも取りこぼしを許さない処理が要るなら、Redis公式が案内するとおりStreamsへ切り替え、at-least-onceの配信と再送を組みます。

本番でWebSocket接続を保つnginx・ping・メッセージ上限の設定値

ローカルで動いたWebSocketが、本番のリバースプロキシ越しで数十秒ごとに切れる。よくある相談の1つです。原因はたいていプロキシとサーバーのタイムアウトの組み合わせにあります。

nginxのUpgrade転送とproxy_read_timeoutの既定60秒

# nginx.conf（http ブロック内）
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

server {
    location /ws/ {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;             # 1.29.7より前のnginxでは必須
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        proxy_read_timeout 120s;            # 既定は60秒
    }
}

nginx公式のWebSocket proxyingの解説によれば、UpgradeとConnectionはホップごとのヘッダーで、明示的に転送しないと後ろのUvicornへ届きません。同じページは、後ろのサーバーが60秒間何も送らなければ既定で接続を切ると書き、proxy_read_timeoutを延ばすか、サーバーからpingフレームを定期的に送る方法を示しています。

Uvicornのws-ping-interval・ws-max-sizeの設定判断

Uvicorn公式のSettingsでは、--ws-ping-intervalと--ws-ping-timeoutの既定がどちらも20秒です。20秒ごとにpingが出るため、nginxの60秒タイムアウトには通常かかりません。切れる場合は、間にもう1段あるロードバランサーのアイドルタイムアウトが20秒より短くないかを先に確かめます。

# 4ワーカー、ping 20秒、1メッセージ64KBまで
uvicorn chat_redis:app --host 127.0.0.1 --port 8000 --workers 4 \
  --ws-ping-interval 20 --ws-ping-timeout 20 --ws-max-size 65536

見直すべきは--ws-max-sizeです。既定の16777216バイト（16MB）は、チャットの1発言に許す容量としては大きすぎるため、そのままにはしません。悪意のあるクライアントが16MBの発言を連続で送ると、各ワーカーにはその分のメモリ負担が生じるという問題です。テキストのチャットなら64KB程度に下げ、ファイルの受け渡しはWebSocketに載せず、HTTPのアップロードと署名付きURLに分けます。--workersを使う複数ワーカーの構成は、前章のRedisの中継を入れてからにします。

TestClientでのテストとFastAPIのWebSocketを採用しない場面の判断

最後に、壊れていないことを確かめる方法と、そもそも自前で持つべきかの判断を置きます。

TestClientのwebsocket_connectで送受信と接続拒否を検証

# test_ws.py（pytest）
import pytest
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect
from main import app          # エコーと、ws_user で守った /ws/secure を持つアプリ

client = TestClient(app)

def test_echo():
    with client.websocket_connect("/ws") as ws:
        ws.send_text("hello")
        assert ws.receive_text() == "Message text was: hello"

def test_reject_without_ticket():
    # accept前に閉じられると接続時点で例外になる
    with pytest.raises(WebSocketDisconnect) as e:
        with client.websocket_connect("/ws/secure"):
            pass
    assert e.value.code == 1008

FastAPI公式のTesting WebSocketsの章のとおり、HTTPのテストと同じTestClientでwebsocket_connectをwith文で使います。拒否されるべき接続が拒否されることのテストは、認証の依存関数を書き換えたときの事故を防ぐための検証です。Redisを挟んだ複数ワーカーの配信はTestClientの1プロセスでは再現できないため、docker composeでRedisとワーカー2つを立てた結合テストを別に持ちます。テストを置くディレクトリ構成はディレクトリ構成とテストを実測で解説したFastAPIのベストプラクティスを土台にできます。

FastAPIで自前実装する条件とSocket.IOやマネージドへ任せる条件

FastAPIのWebSocketで自前実装するのは、次の条件が揃う場合です。既にFastAPIでAPIを持っていて、認証やDBの処理を共有したい。同時接続が数千程度までで、サーバーとRedisを自分たちで運用できる。通知の取りこぼしを、DBからの取り直しで補える。社内の業務システムでの進捗通知や、問い合わせ窓口のチャットがこれに当たります。

見送る場面もはっきりしています。クライアントがサーバーから受け取るだけの通知なら、WebSocketは過剰で、SSEで足ります（SSEとWebSocketの違いと使い分け）。自動再接続やルーム機能をクライアント側のライブラリごと揃えたいなら切断対策とスケールまで扱ったSocket.IOの解説の方が近道です。サーバーの常駐を持ちたくない場合はAPI GatewayのWebSocket APIで接続管理を実装する手順のようなマネージドに寄せます。既存のAPIにリアルタイム通知を足す設計や、どの方式を選ぶかの判断から関わる場合は、API開発・システム連携の受託開発で相談を受けています。

よくある質問

FastAPIのWebSocketについて、実装の途中で出てくることの多い質問に、2026年11月時点の版で答えます。

FastAPIのWebSocketでJSONを送受信するにはどう書きますか？

JSONの受信に使うのはawait websocket.receive_json()で、送信に使うのはawait websocket.send_json(data)です。Starlette公式によれば、send_jsonは既定でテキストフレームとして送り、mode="binary"を付けるとバイナリフレームになります。受け取った辞書はPydanticのモデルでmodel_validateして形を確かめると、HTTPのAPIと同じ水準で検証できます。

WebSocketのエンドポイントはSwagger UIに表示されますか？

表示されません。FastAPIが自動生成するOpenAPIの仕様書は、HTTPのリクエストと応答を記述する形式で、@app.websocketのルートは含まれません。フロントエンドの担当者には、接続URL、クエリの項目、送受信するメッセージの形とクローズコードの意味を、別の文書で渡します。

クライアントの切断はどのタイミングで検知できますか？

受信待ちのreceive_text()などがWebSocketDisconnectを送出した時点です。クライアントが正しく閉じればすぐに分かりますが、回線ごと落ちた場合は閉じる合図が届きません。その場合はUvicornのpingに応答が無く、既定ではping送信から20秒のタイムアウトで切断として扱われます。

FastAPIのWebSocketは1プロセスで何接続まで扱えますか？

公式に決まった上限はありません。接続ごとにファイルディスクリプタとメモリを使うため、OSのulimit -n、メッセージの大きさ、ブロードキャストの頻度で決まります。案件ごとの上限は、負荷試験による実測で決める値です。少なくとも、async defの中で同期のDBドライバやrequestsを呼ぶと、その間は全接続の処理が止まる点は避けます。

Django ChannelsとFastAPIのWebSocketはどう違いますか？

Django ChannelsはChannel Layerというワーカー間の中継の仕組みを持ち、Redisを設定すればグループ配信がそのまま使えます。FastAPIは中継を持たず、この記事のようにRedisの購読を自分で書きます。Djangoで画面と管理機能を作っているならASGIとChannel Layerの構成を扱ったDjango Channelsの解説の方が素直な選択です。

関連記事

お気に入りに入れた記事の一覧

一創の公式 SNS

X では毎朝の新着記事と IT ニュースの速報を、YouTube・TikTok・Instagram・LinkedIn・Facebook では記事の解説動画を配信しています。

この記事は以下の記事からリンクされています

この記事に関連するサービス

資料請求

RELATED POSTS 関連記事

一覧へ戻る

目次