FastAPI WebSockets 实战：从建立连接、依赖注入到多客户端广播

2026-09-06 12:52:38作者：钟日瑜

本篇基于 FastAPI 官方文档《WebSockets》整理并扩展，讲解如何在 FastAPI 应用中实现 WebSocket 实时通信：安装与配置、创建 websocket 端点、收发消息、在 WebSocket 中使用 Depends/Cookie/Query/Path 等参数依赖、以及处理连接断开与多客户端广播。读完本文，你可以从零搭建一个可运行的 WebSocket 聊天服务，理解 WebSocketDisconnect 异常处理机制，并能通过源码（fastapi/websockets.py、fastapi/routing.py）印证 FastAPI 对 Starlette WebSocket 能力的封装方式。

在项目中安装 `websockets` 库

WebSocket 协议需要一个 Python 实现库来支撑底层通信。FastAPI 要求你的项目中加入 websockets 库（它是对 WebSocket 协议使用的封装）：

$ uv add websockets

---> 100%

安装完成后，FastAPI（经由 Starlette/Uvicorn）即可处理 WebSocket 升级握手与帧传输。

WebSocket 客户端：生产环境与示例策略

在生产系统中，你通常会有一个用 React、Vue.js 或 Angular 等现代框架构建的 Frontend，通过前端自带的 WebSocket 能力与后端通信；也可能是原生移动端应用直接以原生代码与 WebSocket 后端交互；或者其他任意能与 WebSocket 端点通信的客户端。

为了便于演示，官方文档采用了一个极简方案：把一段 HTML 加 JavaScript 代码写成一个长字符串，由 FastAPI 直接返回。这显然不是生产做法（生产环境应使用上述任一前端方案），但它是聚焦服务端 WebSocket 逻辑、快速得到可运行示例的最简方式。

完整示例代码（完整源码见 tutorial001_py310.py）：

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

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""


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


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

客户端核心是浏览器原生的 WebSocket API：

new WebSocket("ws://localhost:8000/ws")：向 ws://localhost:8000/ws 发起连接；
ws.onmessage：收到消息时把内容追加到 <ul id="messages"> 列表；
ws.send(input.value)：提交表单时把输入内容经同一条 WebSocket 连接发出，并阻止表单的默认 HTTP 提交行为。

创建 `websocket` 端点

在你的 FastAPI 应用中用 @app.websocket(path) 装饰器声明一个 WebSocket 路由：

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

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

几个技术细节，可以从源码印证：

WebSocket 直接来自 Starlette。查看 fastapi/websockets.py 可以看到，WebSocket、WebSocketDisconnect、WebSocketState 全部是从 starlette.websockets 的原样再导出。也就是说，FastAPI 只是把 Starlette 的 WebSocket 类直接暴露出来方便开发者导入，你写成 from starlette.websockets import WebSocket 效果相同。
路由注册实现：装饰器背后创建的是 APIWebSocketRoute（见 fastapi/routing.py#L801），它继承自 Starlette 的 routing.WebSocketRoute。@app.websocket(...) 装饰器定义在 fastapi/applications.py#L1376，APIRouter.websocket(...) 定义在 fastapi/routing.py#L3057。这意味着 WebSocket 路由同样可以在 APIRouter 上声明后再 include_router 挂载，装饰器还接受 dependencies 参数为整个路由预置 Depends()。
await websocket.accept() 必不可少：它完成 WebSocket 握手确认，之后才能进入消息收发循环。

接收消息与发送消息

在 WebSocket 路由中，你可以 await 接收消息，也可以主动发送消息：

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

await websocket.receive_text() 会挂起协程直到客户端发来一条文本消息；
await websocket.send_text(...) 把处理结果回传给该客户端；
除了文本，WebSocket 还提供 receive_bytes()/send_bytes() 收发二进制数据，以及 receive_json()/send_json() 收发 JSON 数据，你可以根据业务场景自由选择数据类型。

运行并验证

把代码保存到 main.py，然后运行：

$ uv run fastapi dev

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

在浏览器打开 http://127.0.0.1:8000 ，你会看到一个简单的页面（一个聊天输入框 + 消息列表）：

在输入框输入消息并发送，服务端会把内容回显：

你可以连续发送多条消息（并同时接收多条消息），所有消息都复用同一条 WebSocket 长连接，这正是 WebSocket 与逐次 HTTP 请求的本质区别。

以上示例的自动化测试位于 test_tutorial001.py，使用 TestClient 的 WebSocket 支持（client.websocket_connect）验证消息往返行为。

在 WebSocket 端点中使用 `Depends` 和其他参数依赖

与普通 HTTP 端点不同，WebSocket 路由也支持从 fastapi 导入并使用以下声明式参数与依赖工具：

Depends
Security
Cookie
Header
Path
Query

它们的用法与其他 FastAPI 路径操作完全一致。完整示例（源码见 tutorial002_an_py310.py）的核心部分：

from typing import Annotated

from fastapi import (
    Cookie,
    Depends,
    FastAPI,
    Query,
    WebSocket,
    WebSocketException,
    status,
)
from fastapi.responses import HTMLResponse

app = FastAPI()

async def get_cookie_or_token(
    websocket: WebSocket,
    session: Annotated[str | None, Cookie()] = None,
    token: Annotated[str | None, Query()] = None,
):
    if session is None and token is None:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    return session or token


@app.websocket("/items/{item_id}/ws")
async def websocket_endpoint(
    *,
    websocket: WebSocket,
    item_id: str,
    q: int | None = None,
    cookie_or_token: Annotated[str, Depends(get_cookie_or_token)],
):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(
            f"Session cookie or query token value is: {cookie_or_token}"
        )
        if q is not None:
            await websocket.send_text(f"Query parameter q is: {q}")
        await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")

这段代码展示了三类依赖能力：

路径参数：item_id: str 从 /items/{item_id}/ws 中解析；
查询参数：q: int | None = None 直接声明在端点函数上，带类型校验且可选；
依赖注入：get_cookie_or_token 是一个可读取 WebSocket 对象本身的依赖，它从 Cookie（session）或 Query（token）中获取凭证，两者皆无时抛出异常。

注意：由于这是 WebSocket 而不是 HTTP 请求，抛出 HTTPException 没有意义，应当抛出 WebSocketException，并使用 WebSocket 规范（RFC 6455 第 7.4.1 节）中定义的关闭码（Closing Code），例如 status.WS_1008_POLICY_VIOLATION（策略违规）。WebSocketException 与 WebSocketDisconnect 一样，都来自 Starlette（见 fastapi/websockets.py）。

对应的示例页面在浏览器中会额外提供 “Item ID” 和 “Token” 两个输入框，供你设置路径中的 item ID 和作为查询参数的 token，之后点击 “Connect” 建立连接并收发消息：

其中 token 这个查询参数正是由依赖函数 get_cookie_or_token 处理的，这验证了查询参数可以被“依赖”声明式地解析。对应的自动化测试见 test_tutorial002.py。

处理连接断开与多个客户端

当一个 WebSocket 连接被关闭时，await websocket.receive_text() 会抛出 WebSocketDisconnect 异常。你可以捕获并处理它（源码见 tutorial003_py310.py）：

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

app = FastAPI()


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)


manager = ConnectionManager()


@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_personal_message(f"You wrote: {data}", websocket)
            await manager.broadcast(f"Client #{client_id} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

（示例还包含一个 @app.get("/") 路由，返回带 client_id 标识的 HTML 聊天页面，每个浏览器标签页会以 Date.now() 生成自己的 ID 并连接到 ws://localhost:8000/ws/{client_id}。）

ConnectionManager 的设计要点：

connect()：先 accept() 握手，再把连接追加进 active_connections 列表；
disconnect()：从列表移除已断开的连接；
send_personal_message()：只回发给当前客户端（私聊效果）；
broadcast()：遍历所有活跃连接逐一发送（群发效果）；
端点函数用 try/except WebSocketDisconnect 包住接收循环：某个客户端断开时，把它移出列表，并向所有其他客户端广播 Client #xxx left the chat。

验证方式：用多个浏览器标签页打开应用，在各标签页里发送消息，然后关掉其中一个标签页——这就会触发 WebSocketDisconnect 异常，其余客户端都会收到类似下面的消息：

Client #1596980209979 left the chat

对应的自动化测试见 test_tutorial003.py。

适用边界提醒：上述应用是刻意保持最小化的示例——所有状态都存放在单一进程内存中的一条列表里，因此只在该进程存活期间有效，且只适用于单进程部署（多 worker / 多实例时状态互不共享）。如果需要与 FastAPI 良好集成、但更健壮、由 Redis、PostgreSQL 等外部存储支撑的广播方案，可以调研 Starlette 生态的 encode/broadcaster 项目。

更多资料

FastAPI 的 WebSocket、WebSocketDisconnect、WebSocketState 均直接来自 Starlette，更完整的选项（如类 WebSocket 协议方法、基于类的 WebSocketEndpoint 写法等）请查阅 Starlette 官方文档。
三个示例的完整源码分别在 tutorial001_py310.py、tutorial002_an_py310.py、tutorial003_py310.py，可在 docs_src/websockets_/ 目录下直接查看与运行。
路由层实现细节可参考 fastapi/routing.py 中的 APIWebSocketRoute 与 websocket() 装饰器，以及 fastapi/applications.py 中 FastAPI.websocket() 的实现。