做 Web 開發到一個階段，你一定會遇到需要「即時通訊」的需求——聊天室、即時通知、股票報價、協作編輯。這些場景用傳統的 HTTP 請求-回應模式是搞不定的，你需要 WebSocket。

而在 Python 生態系裡，FastAPI 對 WebSocket 的支援可以說是最優雅的。它繼承了 FastAPI 一貫的風格：型別提示、自動文件、簡潔的語法。我自己從 Flask-SocketIO 轉到 FastAPI WebSocket 之後，程式碼的可讀性跟可維護性提升了不少。今天就來從頭建一個即時聊天室。

WebSocket 基礎觀念

在動手寫程式之前，先快速搞懂 WebSocket 跟一般 HTTP 的差別：

HTTP：Client 發請求 → Server 回應 → 連線結束。每次都要重新建立連線
WebSocket：Client 跟 Server 建立一次連線後，雙方可以隨時互傳訊息，連線持續存在

想像你在餐廳點餐：HTTP 就像你每次要加點都得重新排隊；WebSocket 就像你跟服務生之間有一條專線，隨時可以溝通。

WebSocket 的連線流程是：

Client 發送一個 HTTP Upgrade 請求
Server 同意升級，回傳 101 Switching Protocols
連線升級成 WebSocket，雙方開始全雙工通訊

專案建置

先把環境設定好。如果你還在用 pip，建議試試看Python uv 專案管理工具，速度快非常多：

# 用 uv 建立專案
uv init fastapi-chat
cd fastapi-chat
uv add fastapi uvicorn websockets

# 或用傳統 pip
pip install fastapi uvicorn websockets

建立基本的專案結構：

fastapi-chat/
├── main.py          # FastAPI 應用主程式
├── chat_manager.py  # WebSocket 連線管理
├── static/
│   └── index.html   # 前端聊天介面
└── models.py        # 資料模型

基本 WebSocket 端點

先來寫一個最簡單的 WebSocket echo server：

# main.py
from fastapi import FastAPI, WebSocket
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"你說了：{data}")
    except Exception:
        print("連線已關閉")

這段程式碼做的事情很簡單：接受 WebSocket 連線，然後進入一個無限迴圈，收到什麼就回什麼。注意 await websocket.accept() 這步不能忘，它完成的是 WebSocket 的握手程序。

啟動 server：

uvicorn main:app --reload

這邊用到了 Python 的 async/await 語法。如果你對非同步程式設計還不太熟，強烈建議先讀一下Python asyncio 非同步程式設計教學。

連線管理器：處理多人連線

真正的聊天室不會只有一個人。我們需要一個管理器來追蹤所有連線：

# chat_manager.py
from fastapi import WebSocket
from typing import Dict, List
import json
from datetime import datetime

class ChatManager:
    def __init__(self):
        # room_id -> list of (websocket, username)
        self.rooms: Dict[str, List[tuple[WebSocket, str]]] = {}

    async def connect(self, websocket: WebSocket, room_id: str, username: str):
        await websocket.accept()
        if room_id not in self.rooms:
            self.rooms[room_id] = []
        self.rooms[room_id].append((websocket, username))
        # 通知房間內其他人
        await self.broadcast(
            room_id,
            {"type": "system", "message": f"{username} 加入了聊天室"},
            exclude=websocket
        )

    async def disconnect(self, websocket: WebSocket, room_id: str, username: str):
        if room_id in self.rooms:
            self.rooms[room_id] = [
                (ws, name) for ws, name in self.rooms[room_id]
                if ws != websocket
            ]
            await self.broadcast(
                room_id,
                {"type": "system", "message": f"{username} 離開了聊天室"}
            )

    async def broadcast(self, room_id: str, message: dict, exclude: WebSocket = None):
        if room_id not in self.rooms:
            return
        message["timestamp"] = datetime.now().isoformat()
        for ws, _ in self.rooms[room_id]:
            if ws != exclude:
                try:
                    await ws.send_text(json.dumps(message, ensure_ascii=False))
                except Exception:
                    pass  # 連線可能已斷開

manager = ChatManager()

這個 ChatManager 用一個字典來管理不同聊天室的連線。每個房間是一個清單，裡面存著 WebSocket 物件跟使用者名稱。broadcast 方法負責把訊息廣播給房間裡的所有人（可以排除發送者）。

完整聊天端點

把管理器整合到 FastAPI 的路由中：

# main.py（完整版）
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
from fastapi.staticfiles import StaticFiles
from chat_manager import manager
import json

app = FastAPI(title="FastAPI 即時聊天室")
app.mount("/static", StaticFiles(directory="static"), name="static")

@app.websocket("/ws/{room_id}")
async def chat_endpoint(
    websocket: WebSocket,
    room_id: str,
    username: str = Query(...)
):
    await manager.connect(websocket, room_id, username)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            await manager.broadcast(
                room_id,
                {
                    "type": "message",
                    "username": username,
                    "message": message.get("message", ""),
                }
            )
    except WebSocketDisconnect:
        await manager.disconnect(websocket, room_id, username)
    except Exception as e:
        print(f"錯誤：{e}")
        await manager.disconnect(websocket, room_id, username)

注意幾個重點：

路由用 /ws/{room_id}，支援多個聊天室
username 透過 Query 參數傳入
用 WebSocketDisconnect 例外來捕捉斷線事件

前端聊天介面

簡單做一個前端頁面（用原生 JavaScript，不依賴框架）：

<!-- static/index.html -->
<!DOCTYPE html>
<html lang="zh-TW">
<head>
  <meta charset="UTF-8">
  <title>FastAPI 聊天室</title>
  <style>
    #messages { height: 400px; overflow-y: auto; border: 1px solid #ccc; padding: 10px; }
    .system { color: #888; font-style: italic; }
    .message { margin: 5px 0; }
    .username { font-weight: bold; color: #2563eb; }
  </style>
</head>
<body>
  <h1>FastAPI 即時聊天室</h1>
  <div id="messages"></div>
  <input type="text" id="input" placeholder="輸入訊息..." />
  <button onclick="sendMessage()">送出</button>
  <script>
    const username = prompt('請輸入你的暱稱') || '匿名';
    const roomId = 'general';
    const ws = new WebSocket(
      `ws://localhost:8000/ws/${roomId}?username=${encodeURIComponent(username)}`
    );
    ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      const div = document.createElement('div');
      if (data.type === 'system') {
        div.className = 'system';
        div.textContent = data.message;
      } else {
        div.className = 'message';
        div.innerHTML = `<span class="username">${data.username}</span>: ${data.message}`;
      }
      document.getElementById('messages').appendChild(div);
    };
    function sendMessage() {
      const input = document.getElementById('input');
      if (input.value) {
        ws.send(JSON.stringify({ message: input.value }));
        input.value = '';
      }
    }
    document.getElementById('input').addEventListener('keypress', (e) => {
      if (e.key === 'Enter') sendMessage();
    });
  </script>
</body>
</html>

進階功能

心跳機制

WebSocket 連線可能因為各種原因無聲無息地斷掉。心跳機制可以偵測這種情況：

import asyncio

async def heartbeat(websocket: WebSocket, interval: int = 30):
    while True:
        try:
            await asyncio.sleep(interval)
            await websocket.send_text(json.dumps({"type": "ping"}))
        except Exception:
            break

在建立連線後啟動一個背景 task 定期發送 ping，如果發送失敗就知道連線已斷開。

訊息歷史

新加入的使用者看不到之前的訊息，體驗不好。可以在 ChatManager 裡加一個訊息緩衝：

from collections import deque

class ChatManager:
    def __init__(self, history_size: int = 50):
        self.rooms = {}
        self.history: Dict[str, deque] = {}
        self.history_size = history_size

    async def connect(self, websocket, room_id, username):
        await websocket.accept()
        # 發送歷史訊息
        if room_id in self.history:
            for msg in self.history[room_id]:
                await websocket.send_text(json.dumps(msg, ensure_ascii=False))
        # ... 其他邏輯

    async def broadcast(self, room_id, message, exclude=None):
        # 存入歷史
        if room_id not in self.history:
            self.history[room_id] = deque(maxlen=self.history_size)
        self.history[room_id].append(message)
        # ... 廣播邏輯

測試與除錯

WebSocket 的測試比一般 API 稍微麻煩一點。FastAPI 的 TestClient 支援 WebSocket 測試：

from fastapi.testclient import TestClient

def test_websocket_chat():
    client = TestClient(app)
    with client.websocket_connect("/ws/test-room?username=Alice") as ws:
        ws.send_text(json.dumps({"message": "Hello!"}))
        # 因為是廣播，自己也會收到
        data = ws.receive_text()
        msg = json.loads(data)
        assert msg["username"] == "Alice"
        assert msg["message"] == "Hello!"

除了自動化測試，開發時推薦用 Postman 或 websocat 這個命令列工具來手動測試。如果你想寫更完整的 CLI 工具，Python Click CLI 開發教學很值得一看。

部署注意事項

WebSocket 應用的部署跟一般 REST API 有些不同：

反向代理設定：Nginx 需要額外設定 proxy_set_header Upgrade 跟 proxy_set_header Connection "upgrade"
多 Worker 問題：如果你用 Gunicorn 跑多個 Worker，WebSocket 的連線只存在於一個 Worker 裡。要跨 Worker 廣播，需要用 Redis Pub/Sub
連線數限制：每個 WebSocket 連線都佔一個 file descriptor，注意調高系統的 ulimit
SSL/TLS：生產環境務必使用 wss://（WebSocket over TLS），跟 HTTPS 一樣重要

善用Python Type Hints 與 mypy 可以幫你在開發階段就抓出很多型別相關的 bug。