FastAPI×WebSocket入門ガイド：リアルタイム通信・チャット・ダッシュボードをつくる実装パターン

要約（最初に全体の流れをつかむ）

• WebSocket は、サーバーとクライアントが双方向・リアルタイムに通信できるプロトコルで、チャットや通知、ダッシュボード更新などに向いています。
• FastAPI は Starlette の機能を引き継いでいて、@app.websocket() デコレータと WebSocket クラスを使うことで、シンプルなWebSocketサーバーを実装できます。
• この記事では、単一クライアント用の最小WebSocket、複数クライアントへのブロードキャスト、クラスで管理するコネクションマネージャーといったパターンをステップごとに紹介します。
• セキュリティ（Origin・トークン認証）や、Testing WebSockets（TestClient でのテスト方法）、簡単なフロントエンド（JavaScript）との連携サンプルも含めて、実務で使いやすい形を目指します。
• 最後に、チャット・ダッシュボード・通知などユースケース別の設計ポイントと、「どこから導入するか」のロードマップをまとめます。

誰が読んで得をするか（具体的な読者像）

個人開発・学習者さん

• 「チャット風の画面」や「リアルタイムカウンター」などを、自作サービスに少し追加してみたい方。
• REST APIは作れるけれど、「WebSocketってどう生かせばいいの？」がぼんやりしている方。
• JavaScriptとFastAPIを一緒に触りながら、双方向通信の感覚を掴んでみたい方。

この方には、最小のWebSocketエンドポイントと、素朴なHTML＋JavaScriptを組み合わせたチャット風サンプルを通じて、「動くもの」を作るところまで一緒に進めていきます。

小規模チームのバックエンドエンジニアさん

• 管理画面やダッシュボードに「即時反映」したい情報が増えてきて、ポーリングでは厳しくなってきた方。
• 既存のFastAPIアプリに、通知・チャット・進捗表示などのリアルタイム要素を後付けしたい方。
• WebSocket接続の管理（切断検知・ブロードキャスト）や、テスト戦略を整理しておきたいと考えている方。

この方には、コネクションマネージャークラス、ルーム分け、テストの仕方など、現実寄りのパターンが役に立つはずです。

SaaS開発チーム・スタートアップの皆さま

• システム監視・在庫のリアルタイム反映・通知ハブなど、「即時性」がビジネス価値になる領域を手がけている方。
• 将来的なスケール（複数インスタンス・外部ブローカー連携）も視野に入れつつ、「今から大きく外さない」構成を考えたい方。
• WebSocketの認証・メッセージ形式・ルーム設計を、後々の拡張を見据えて決めておきたいと考えている方。

この方には、「1インスタンスでの基本形」をベースに、将来的にRedis Pub/Subやメッセージブローカーと組み合わせることを見越した設計のヒントをお届けします。

アクセシビリティ評価（読みやすさ・理解しやすさの観点）

• 情報設計
  • WebSocketの概念 → FastAPIでの基本構文 → シンプルな例 → コネクションマネージャー → セキュリティ → テスト → ユースケース、という段階構成にしています。
• 用語
  • 「接続」「メッセージ」「ブロードキャスト」など、できるだけ日常語に近い表現を用い、英単語は必要最小限＋初出で補足しています。
• コード
  • 1ブロックが長くなりすぎないよう、小さめの単位で分割しました。コメントも控えめにして、視線が迷わないようにしています。
• 読者想定
  • PythonとFastAPIの基礎チュートリアルを終えた方を想定しつつ、「この章だけ読んでもなんとなく雰囲気が分かる」ように、章ごとに完結した説明を心がけています。

全体として、多様な読者が読み進めやすいことを目指し、WCAG AA相当を意識したテキスト構造にしています。

1. WebSocketとは？HTTPとの違いをサクッと整理

まずは、「そもそもWebSocketって何者？」というところを軽く整理しておきますね。

1.1 WebSocketの特徴

WebSocketは、ブラウザとサーバー間で「長くつながりっぱなし」の双方向通信を行うためのプロトコルです。

• 一度接続を確立すると、クライアント→サーバー、サーバー→クライアントの双方から自由なタイミングでメッセージを送れます。
• HTTPのように「リクエスト→レスポンスで1往復しておしまい」ではなく、1本の接続を何度も使い回すイメージです。
• チャット、ゲーム、通知、ダッシュボードなど、リアルタイム性が求められる場面でよく使われます。

HTTPリクエストでこれを実現しようとすると「定期的なポーリング」（数秒おきにAPIを叩く）が必要で、無駄な通信が増えたり、リアルタイム性に限界があったりします。

1.2 FastAPIとWebSocketの関係

FastAPIは、ASGIフレームワーク Starlette の上に構築されていて、WebSocketもStarletteの機能として組み込まれています。

• HTTPエンドポイント：@app.get() / @app.post() など
• WebSocketエンドポイント：@app.websocket()

実はそんなに特別なものではなく、「HTTPの代わりにWebSocketオブジェクトを受け取る関数」を1つ定義してあげるだけ、という感じで始められます。

2. FastAPIでのWebSocketエンドポイントの基本

ここからは、FastAPIでの具体的な書き方を見ていきますね。

2.1 最小のWebSocketサーバー

まずは、単一クライアントとだけやりとりする、いちばん小さな例です。

# app/main.py
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws/echo")
async def websocket_echo(websocket: WebSocket):
    # クライアントからの接続を受け入れる
    await websocket.accept()

    while True:
        # メッセージを1件受信（テキスト）
        data = await websocket.receive_text()
        # そのまま送り返す
        await websocket.send_text(f"Echo: {data}")

ポイントは3つです。

1. @app.websocket("/ws/echo") でWebSocket用のルートを定義する。
2. 関数の引数に WebSocket 型を受け取る。
3. await websocket.accept() で接続を受け入れてから、receive_* / send_* を使う。

receive_text() / send_text() のほかに、バイナリ用の receive_bytes() / send_bytes() や、JSON用の receive_json() / send_json() も使えます。

2.2 簡単なクライアント（ブラウザ側）の例

ローカルで uvicorn app.main:app --reload を立ち上げたあと、
ブラウザのコンソール（開発者ツール）で次のように試せます。

// ブラウザコンソールで
const ws = new WebSocket("ws://localhost:8000/ws/echo");

ws.onopen = () => {
  console.log("connected");
  ws.send("hello");
};

ws.onmessage = (event) => {
  console.log("received:", event.data);
};

これで、コンソールに

connected
received: Echo: hello

のようなログが出れば、最小のWebSocketの世界に一歩踏み込めました。

3. 複数クライアントに対応する：コネクションマネージャーの基本

リアルタイムなUIを作りたい場面では、だいたい「複数クライアント」が同じ情報を共有することになります。
ここでは、「接続中のクライアント一覧を管理するコネクションマネージャー」という考え方を導入します。

3.1 コネクションマネージャークラス

# app/websocket_manager.py
from typing import List
from fastapi import 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 send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

• active_connections に、接続されている WebSocket を保持します。
• broadcast() で、接続中の全クライアントにメッセージを送れます。

3.2 コネクションマネージャーを使ったチャット風WebSocket

# app/main.py（抜粋）
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from app.websocket_manager import ConnectionManager

app = FastAPI()
manager = ConnectionManager()

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("a client disconnected")

WebSocketDisconnect が発生したときに disconnect() を呼び出すことで、
リストから削除し、残りのクライアントに「誰かが抜けたこと」を通知しています。

3.3 超シンプルなHTMLクライアント

最小のチャットUIをHTMLだけで作ってみます。

<!-- static/chat.html のイメージ -->
<!DOCTYPE html>
<html lang="ja">
  <head>
    <meta charset="UTF-8" />
    <title>FastAPI WebSocket Chat</title>
    <style>
      body { font-family: sans-serif; }
      #messages { border: 1px solid #ccc; height: 200px; overflow-y: scroll; padding: 4px; }
      #input-area { margin-top: 8px; }
    </style>
  </head>
  <body>
    <h1>Chat</h1>
    <div id="messages"></div>
    <div id="input-area">
      <input id="msg" type="text" placeholder="メッセージ" />
      <button id="send">送信</button>
    </div>

    <script>
      const messages = document.getElementById("messages");
      const input = document.getElementById("msg");
      const sendBtn = document.getElementById("send");

      const ws = new WebSocket("ws://localhost:8000/ws/chat");

      ws.onmessage = (event) => {
        const div = document.createElement("div");
        div.textContent = event.data;
        messages.appendChild(div);
        messages.scrollTop = messages.scrollHeight;
      };

      sendBtn.onclick = () => {
        if (input.value) {
          ws.send(input.value);
          input.value = "";
        }
      };
    </script>
  </body>
</html>

このくらいの構成でも、ブラウザを2つ開いて試せば「お互いのメッセージがリアルタイムに流れるチャット」が体験できます。

4. 実務でよくあるパターン：ルーム・ユーザー・メッセージ形式

もう少し実務寄りの話として、チャットやダッシュボードでよく出てくるパターンを軽く整理しておきますね。

4.1 ルーム（部屋）ごとに分ける

オープンなチャットだけでなく、部屋ごとにクライアントを分けたいことも多いです。
その場合、dict[str, list[WebSocket]] のような構造でコネクションを管理します。

# app/websocket_manager_rooms.py
from collections import defaultdict
from typing import Dict, List
from fastapi import WebSocket, WebSocketDisconnect

class RoomManager:
    def __init__(self):
        self.rooms: Dict[str, List[WebSocket]] = defaultdict(list)

    async def connect(self, room: str, websocket: WebSocket):
        await websocket.accept()
        self.rooms[room].append(websocket)

    def disconnect(self, room: str, websocket: WebSocket):
        self.rooms[room].remove(websocket)
        if not self.rooms[room]:
            del self.rooms[room]

    async def broadcast(self, room: str, message: str):
        for ws in self.rooms.get(room, []):
            await ws.send_text(message)

WebSocketのパスやクエリパラメータでルーム名を受け取り、
そのルームに属するクライアントだけに broadcast する形にします。

4.2 メッセージをJSONにする

メッセージを素の文字列にしていると、種類が増えたときに扱いづらくなります。
実務では、JSONで「type」「payload」などを分けておくのが定番です。

import json

async def handle_message(data: str, websocket: WebSocket):
    msg = json.loads(data)
    msg_type = msg.get("type")
    payload = msg.get("payload", {})

    if msg_type == "chat":
        text = payload.get("text", "")
        # みんなにチャットメッセージを送る
    elif msg_type == "join":
        # ルーム参加の処理
    elif msg_type == "leave":
        # ルーム退出の処理
    else:
        await websocket.send_text("unknown message type")

フロントエンド側も JSON で送受信することで、
将来的に機能が増えてもメッセージ形式をあまり変えずに済むようになります。

5. セキュリティ・認証との組み合わせ

WebSocketにも、当然ながらセキュリティの話がついてきます。
ここでは「最初に押さえておきたいポイント」だけを絞ってお話ししますね。

5.1 Origin・CORS的な考え方

ブラウザのWebSocketは、HTTPと同様に「どのオリジンから接続を許可するか」を意識する必要があります。

• Nginxやリバースプロキシの設定でOriginを制限する
• あるいはサーバー側で websocket.headers（Origin や Sec-WebSocket-Protocol）をチェックして弾く

といった方法があります。

簡易的には、WebSocket接続時にクエリパラメータで「トークン」を渡し、サーバー側で認証するのが扱いやすいです。

5.2 JWTトークンとの組み合わせ

たとえば、HTTPログインで発行したJWTを、WebSocket接続時にクエリ文字列で渡すパターンです。

// フロントエンド側イメージ
const token = "取得済みJWT";
const ws = new WebSocket(`wss://example.com/ws/chat?token=${token}`);

サーバー側では、FastAPIのWebSocketオブジェクトからクエリを読めます。

# app/main.py（抜粋）
from fastapi import WebSocket, WebSocketDisconnect
from app.core.jwt import decode_access_token  # 以前の記事で登場したJWTユーティリティを想定

@app.websocket("/ws/secure-chat")
async def websocket_secure_chat(websocket: WebSocket):
    token = websocket.query_params.get("token")
    if not token:
        await websocket.close(code=1008)  # Policy Violation
        return

    try:
        payload = decode_access_token(token)
    except Exception:
        await websocket.close(code=1008)
        return

    username = payload.get("sub", "anonymous")

    await manager.connect(websocket)
    await manager.broadcast(f"{username} joined")

    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"{username}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"{username} left")

ここでは簡易的に「トークンが有効ならOK」としていますが、
実務ではスコープ・ロール・接続数制限などもあわせて検討します。

6. WebSocketのテスト：TestClientでの検証方法

FastAPIの公式ドキュメントには、WebSocketをテストするためのガイドが用意されています。

6.1 テストの基本構文

HTTPと同じ TestClient を使い、client.websocket_connect() を利用します。

# tests/test_websocket_echo.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_websocket_echo():
    with client.websocket_connect("/ws/echo") as websocket:
        websocket.send_text("hello")
        data = websocket.receive_text()
        assert data == "Echo: hello"

ポイントは2つです。

• with client.websocket_connect("/ws/echo") as websocket: で接続を開き、
  ブロックを抜けるときに自動的に閉じてくれる。
• send_text() / receive_text() など、サーバ側と対になるAPIが提供されている。

6.2 ブロードキャストのテスト

複数クライアントを同時に扱うテストも書けます。

def test_broadcast():
    with client.websocket_connect("/ws/chat") as ws1:
        with client.websocket_connect("/ws/chat") as ws2:
            ws1.send_text("hello")

            msg1 = ws1.receive_text()
            msg2 = ws2.receive_text()

            assert "hello" in msg1
            assert "hello" in msg2

テストがある程度揃っていると、

• コネクションマネージャーのロジックを変えたとき
• 認証周りを変更したとき

にも、「既存の挙動を壊していないか」がすぐ分かるようになります。

7. 簡単なユースケース別の設計メモ

最後に、WebSocketを使った典型的なユースケースをいくつか挙げて、
それぞれ「どんな感じで設計するとよいか」のヒントを簡単にまとめておきますね。

7.1 チャット（1対1・グループ）

• ルームID（チャットルーム）ごとにコネクションを管理する。
• メッセージはJSON形式で、「type=chat」「payload={text, sender_id, room_id}」のように分ける。
• 認証済みユーザーのみ参加させる場合は、JWTでユーザーIDを特定し、ルームへの参加権限をチェックする。

7.2 リアルタイムダッシュボード

• バックエンドで定期的にデータを集計し、変更があったときだけWebSocketで通知する。
• 「全ユーザー共通」の情報であれば、1つのルーム（dashboard）に接続させ、全員にブロードキャストする。
• 個別のユーザー専用のダッシュボード（例：自分のタスク状況）なら、「ユーザーIDごとのルーム」を作るか、メッセージ内でユーザーIDをチェックしてフィルタリングする。

7.3 通知・イベントストリーム

• 新規コメント・ステータス変更・システムアラートなどを、イベントとしてWebSocketで流す。
• WebSocketは「リアルタイムな通知チャンネル」、REST APIは「履歴の取得」に役割を分ける。
• 既読管理・再接続時の過去イベント同期などが必要であれば、WebSocketは「きっかけ」として使いつつ、実データはRESTやDBから取り直す設計にしておくとシンプルになります。

8. 読者別ロードマップ：どこから導入するか

個人開発・学習者さん

1. 最小の @app.websocket("/ws/echo") を実装して、ブラウザコンソールから接続してみる。
2. コネクションマネージャーを導入し、ブロードキャスト式の簡易チャットを作る。
3. HTML＋JavaScriptで小さなチャットUIを作って、実際にブラウザ2枚で会話してみる。
4. 慣れてきたら、JSON形式のメッセージにして、チャット以外のイベントも流してみる。

小規模チームのバックエンドエンジニアさん

1. 既存のFastAPIアプリに、「通知専用」または「ダッシュボード専用」のWebSocketエンドポイントを1つ追加する。
2. コネクションマネージャー＋ルーム管理を導入し、ユーザー／プロジェクト単位で接続を分けられるようにする。
3. JWTなど既存の認証基盤と組み合わせて、「認証済みユーザーだけ接続できるWebSocket」を整備する。
4. TestClient を使ったWebSocketテストを追加し、認証やブロードキャストの挙動をテストで守る。

SaaS開発チーム・スタートアップの皆さま

1. PoCとして、1インスタンスで完結するWebSocketサーバー（チャット・ダッシュボード）をFastAPI内に実装する。
2. 将来的なスケールを見据えて、コネクションマネージャーから外部ブローカー（Redis Pub/Subやメッセージキュー）に切り出せるよう、抽象化を意識する。
3. 認証・レート制限・Origin制限などを含めたWebSocket用のセキュリティポリシーを整理し、インフラ（ロードバランサ・リバースプロキシ）と整合を取る。
4. モニタリング（接続数・切断数・エラー率）やログ（どのイベントがどのくらい流れているか）を整え、運用しやすいリアルタイムシステムに育てていく。

9. 参考リンク（公式ドキュメント・日本語記事など）

※内容は執筆時点のものです。最新情報は各サイトでご確認ください。

• 公式ドキュメント（FastAPI / Starlette）

  • WebSockets – FastAPI（英語）
  • WebSockets – FastAPI リファレンス（英語）
  • WebSockets – FastAPI 日本語ドキュメント（GitHub）
  • WebSockets – Starlette
  • Testing WebSockets – FastAPI

• 日本語記事・チュートリアル

  • FastAPIでWebSocketに入門してみた（Zenn）
  • FastAPIでWebSockets（Qiita）
  • FastAPIでWebSocket入門（Qiita）
  • FastAPIのWebSocketで利用するデコレータを間違えた（ブログ）
  • 最新ガイド：FastAPIでWebsocketを使用する（APIDog日本語）

• 英語記事・応用例

  • Building a WebSocket Server with FastAPI（Medium）
  • Getting Started with FastAPI WebSockets（Better Stack）
  • Developing a Real-time Dashboard with FastAPI and WebSockets（TestDriven.io）

まとめ

• WebSocketは、FastAPIでも比較的簡単に扱える、リアルタイム・双方向通信のためのプロトコルです。
• @app.websocket() と WebSocket クラスを使って、最小のエコーサーバーから始め、コネクションマネージャーやルーム管理を導入することで、複数クライアント向けのチャットやダッシュボードを実装できます。
• 認証やOrigin制限など、セキュリティ面も意識しつつ、テスト（TestClient でのWebSocketテスト）を用意しておくことで、安心して機能追加や仕様変更を続けられます。
• いきなり大規模なリアルタイムシステムを目指す必要はありません。まずは、小さなチャットや通知機能から始めて、少しずつWebSocketの世界に慣れていっていただければと思います。

あなたのFastAPIアプリに、ちょっとしたリアルタイムの「楽しさ」が加わるきっかけになれたら、とてもうれしいです。
どうぞ、無理のないペースで1つずつ試してみてくださいね。