首页
/ FastAPI WebSockets 实战指南：双向通信端点、依赖注入与多客户端广播

FastAPI WebSockets 实战指南：双向通信端点、依赖注入与多客户端广播

2026-09-06 15:21:50作者：滑思眉Philip
fastapi
FastAPI framework, high performance, easy to learn, fast to code, ready for production
项目地址：https://gitcode.com/GitHub_Trending/fa/fastapi

本文基于 FastAPI 官方文档 WebSockets 章节 整理并深入展开，带你掌握在 FastAPI 中构建 WebSocket 端点的全部核心技能：使用 @app.websocket() 创建端点、await 接收与发送消息、在 WebSocket 端点中复用 Depends/Cookie/Query 等依赖注入机制、处理 WebSocketDisconnect 断连事件，以及如何用 ConnectionManager 模式实现多客户端广播，并溯源到 FastAPI 源码中 WebSocket 路由的真实调用链。

安装 websockets 库

FastAPI 本身不直接实现 WebSocket 协议，而是依赖 Starlette（FastAPI 的底层 ASGI 框架）。要在项目中启用 WebSocket 支持，需要额外安装 Python 的 websockets 库——它是对 "WebSocket" 协议的底层实现，Uvicorn 等服务依赖它完成握手与帧解析：

$ uv add websockets

这是使用 FastAPI WebSockets 功能的前提；缺少该库时，Uvicorn 将无法升级 HTTP 连接为 WebSocket 连接。

客户端：生产环境与示例页面

生产环境中的客户端

在生产系统中，前端通常由 React、Vue.js 或 Angular 等现代框架构建，与后端的 WebSocket 通信一般直接使用前端框架自带的 WebSocket 工具；也可能是原生移动应用以原生代码直接连接 WebSocket 后端；当然也可能是任何其他方式与 WebSocket 端点通信。

聚焦服务端的简化示例

为了聚焦 WebSocket 的服务端逻辑并得到一个可运行的示例，官方教程使用了一段内嵌在长字符串中的极简 HTML + JavaScript 页面。当然，这并非最优方案，生产环境应使用上述方式，但它是最简单、最能突出服务端要点的做法。

完整示例（保存为 main.py，源自 docs_src/websockets_/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}")

客户端要点说明：

  • new WebSocket("ws://localhost:8000/ws") 发起握手，地址使用 ws:// 协议（若走 HTTPS 页面则需 wss://）；
  • ws.onmessage 处理服务端推送的每条消息，追加到 <ul id="messages"> 列表；
  • sendMessage() 在表单提交时调用 ws.send() 发送文本，并清空输入框、阻止页面刷新。

创建 websocket 路由

在 FastAPI 应用中，使用 @app.websocket() 装饰器创建一个 WebSocket 路由：

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

技术细节：WebSocket 来自 Starlette

你同样可以直接 from starlette.websockets import WebSocket。FastAPI 只是直接转发了 Starlette 的同一个 WebSocket 类，纯粹是出于对开发者的便利。这一点在源码中可以得到印证——fastapi/websockets.py 的全部内容就是对 Starlette 三个符号的再导出：

from starlette.websockets import WebSocket as WebSocket  # noqa
from starlette.websockets import WebSocketDisconnect as WebSocketDisconnect  # noqa
from starlette.websockets import WebSocketState as WebSocketState  # noqa

因此 WebSocket、WebSocketDisconnect、WebSocketState 都可直接从 fastapi 顶层导入使用。

路由装饰器的源码实现

从源码结构看，@app.websocket() 装饰器定义在 fastapi/routing.py 中：它调用 add_api_websocket_route()（fastapi/routing.py），后者为路由加上 Router 前缀与依赖，然后构造 APIWebSocketRoute 实例（fastapi/routing.py）。APIWebSocketRoute 在初始化时会：

  1. 用 compile_path() 把路径编译为正则、路径模板与参数转换器（支持 {client_id} 这类路径参数）；
  2. 用 _build_dependant_with_parameterless_dependencies() 分析端点签名，生成 Dependant 依赖树；
  3. 将端点包装为 websocket_session(get_websocket_app(...)) 形式的 ASGI 应用。

websocket_session()（fastapi/routing.py）是 Starlette 同名函数的修改版，额外引入两个 AsyncExitStack（分别存放在 scope 的 fastapi_inner_astack 与 fastapi_function_astack 键中）——这正是 WebSocket 端点支持带 yield 的依赖（依赖清理逻辑）的底层机制。

而 get_websocket_app()（fastapi/routing.py）在每次连接时调用 solve_dependencies(request=websocket, ...) 求解全部依赖；若存在校验错误，则抛出 WebSocketRequestValidationError，而不是返回 HTTP 422 响应。

接收与发送消息：await 消息循环

WebSocket 路由的核心是一个消息循环——await 接收消息，再发送消息：

while True:
    data = await websocket.receive_text()
    await websocket.send_text(f"Message text was: {data}")
  • 必须先 await websocket.accept() 完成握手，连接才真正建立；
  • receive_text() 挂起等待下一条文本消息，客户端断开时抛出 WebSocketDisconnect；
  • 你可以接收和发送 二进制、文本和 JSON 数据，对应 receive_bytes()/send_bytes()、receive_text()/send_text()、receive_json()/send_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，你会看到一个简单的聊天页面。在输入框中键入消息并发送，FastAPI 的 WebSocket 端点会回显 Message text was: ...；连续发送多条消息时，它们全部复用同一条 WebSocket 连接，而非每次新建连接——这正是 WebSocket 与 HTTP 短连接的本质区别。

仓库中为这个示例配备了自动化测试 tests/test_tutorial/test_websockets/test_tutorial001.py，它使用 TestClient.websocket_connect("/ws") 模拟一次真实连接：

def test_websocket():
    with pytest.raises(WebSocketDisconnect):
        with client.websocket_connect("/ws") as websocket:
            message = "Message one"
            websocket.send_text(message)
            data = websocket.receive_text()
            assert data == f"Message text was: {message}"
            # ... 继续第二轮收发断言

测试还验证了退出连接上下文后会触发 WebSocketDisconnect，与浏览器中“关闭标签页”的行为一致。

在 WebSocket 端点中使用 Depends 等依赖注入

在 WebSocket 端点中，你可以从 fastapi 导入并使用与 HTTP 端点完全相同的参数机制：

  • Depends
  • Security
  • Cookie
  • Header
  • Path
  • Query

它们的工作方式与其他 FastAPI 端点（path operations）完全一致。以下示例（源自 docs_src/websockets_/tutorial002_an_py310.py）展示了一个需要 Cookie 或 Query 提供凭证的受保护端点：

from typing import Annotated

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

app = FastAPI()

# ...（前面省略 HTML 页面，与 tutorial001 类似，另含 Item ID 与 Token 输入框）


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}")

示例拆解：

  • 路径参数：/items/{item_id}/ws 中的 item_id 自动注入端点；
  • 查询参数：q: int | None = None 声明为可选整数，带类型校验；
  • 依赖函数：get_cookie_or_token 同时接收 websocket: WebSocket 以及 Cookie()/Query() 注解参数，优先取 session cookie，其次取 query 中的 token；
  • WebSocketException 与关闭码：由于这是 WebSocket 连接，抛出 HTTPException 没有意义，应改用 WebSocketException 并携带 RFC 6455 第 7.4.1 节定义的关闭码，例如 status.WS_1008_POLICY_VIOLATION（策略违规），连接会带着该码被服务端关闭。

运行带依赖的 WebSocket

运行 uv run fastapi dev 后，在 http://127.0.0.1:8000 页面可以设置：

  • Item ID（用于路径）；
  • Token（作为查询参数）。

注意：查询参数 token 是由依赖 get_cookie_or_token 来处理的，而不是端点签名本身——这正是 Depends 组合凭证校验逻辑的价值：连接建立前先完成鉴权，校验失败则直接以 WebSocket 关闭码断开，而不会返回 HTTP 错误。

对应的测试位于 tests/test_tutorial/test_websockets/test_tutorial002.py，覆盖了带凭证连接与参数校验失败的分支。

处理断连与多客户端广播

当某条 WebSocket 连接关闭时，await websocket.receive_text() 会抛出 WebSocketDisconnect 异常，捕获它即可执行清理逻辑。下面的完整示例（源自 docs_src/websockets_/tutorial003_py310.py）用一个 ConnectionManager 维护所有活跃连接，实现“私聊回复 + 全员广播”：

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

app = FastAPI()

# ...（HTML 页面省略，JS 中用 Date.now() 生成 client_id 并连接
#   ws://localhost:8000/ws/${client_id}）


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.get("/")
async def get():
    return HTMLResponse(html)


@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")

关键点：

  • ConnectionManager：以内存列表维护活跃连接；connect() 负责 accept() 并登记，disconnect() 移除，broadcast() 遍历所有连接逐条推送；
  • 断连检测：while True 循环中的 receive_text() 在连接断开时抛出 WebSocketDisconnect，except 分支中移除该连接并向其余在线客户端广播类似 Client #1596980209979 left the chat 的消息；
  • 手动验证：用多个浏览器标签页打开应用，从不同标签页发消息，再关闭其中一个标签页，其余标签页即可收到“某客户端离开了聊天室”的通知。

生产环境注意事项

上述应用是一个极简示例，用于演示如何处理消息并向多个 WebSocket 连接广播。但要记住：所有状态都保存在内存的单条列表中，它只在进程存活期间有效，且只适用于单进程部署。若需要易于集成到 FastAPI、又由 Redis、PostgreSQL 等后端支撑的更健壮广播方案，可以参考官方生态项目 encode/broadcaster。多进程/多节点部署时，连接状态必须外置到共享存储，否则不同工作进程无法互相感知连接。

进阶：了解 WebSocket 的更多能力

FastAPI 的 WebSocket 就是 Starlette 的 WebSocket，因此 Starlette 提供的所有 WebSocket 能力都可直接使用，包括：

  • WebSocket 类的完整 API（receive()/send() 底层方法、client_state/application_state 状态属性、close(code) 等）；
  • 基于类（class-based）的 WebSocket 端点处理方式（WebSocketEndpoint）。

建议直接查阅 Starlette 官方文档中的这两部分：WebSocket 类参考与“Class-based WebSocket handling”。

小结

  • 安装 websockets 库后，用 @app.websocket() 即可声明 WebSocket 路由，端点签名中声明 websocket: WebSocket 注入连接对象；
  • await websocket.accept() 握手后，用 while True + receive_text()/send_text() 构建收发循环，支持文本、二进制与 JSON 三种数据；
  • WebSocket 端点与 HTTP 端点共享同一套依赖注入体系（Depends、Security、Cookie、Header、Path、Query），鉴权失败时用 WebSocketException 携带 RFC 6455 关闭码断开，而非抛 HTTPException；
  • 多客户端场景用 ConnectionManager 集中管理连接，捕获 WebSocketDisconnect 完成清理与广播；生产级部署需将连接状态外置（Redis/PostgreSQL）以支持多进程。

相关源码与文档索引：

内容 路径
官方 WebSockets 文档 docs/en/docs/advanced/websockets.md
示例一：最简回显端点 docs_src/websockets_/tutorial001_py310.py
示例二：依赖注入与凭证校验 docs_src/websockets_/tutorial002_an_py310.py
示例三：断连处理与广播 docs_src/websockets_/tutorial003_py310.py
WebSocket 等符号再导出 fastapi/websockets.py
WebSocket 路由与依赖求解 fastapi/routing.py、fastapi/routing.py、fastapi/routing.py
配套自动化测试 tests/test_tutorial/test_websockets/test_tutorial001.py
fastapi
FastAPI framework, high performance, easy to learn, fast to code, ready for production
项目地址：https://gitcode.com/GitHub_Trending/fa/fastapi
登录后查看全文
热门项目推荐
相关项目推荐

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件，TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件，使 PyTorch 框架能够直接调用昇腾 NPU，为开发者提供昇腾 AI 处理器的超强算力。
Python
855
1.34 K
kernelkernel
openEuler内核是openEuler操作系统的核心，既是系统性能与稳定性的基石，也是连接处理器、设备与服务的桥梁。
C
528
589
leetcodeleetcode
🔥LeetCode solutions in any programming language | 多种编程语言实现 LeetCode、《剑指 Offer（第 2 版）》、《程序员面试金典（第 6 版）》题解
Markdown
77
23
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库，实现网络在NPU上加速计算。
C++
1.13 K
2.73 K
docsdocs
暂无描述
Markdown
895
5.79 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent，它能够将大语言模型的强大能力，通过你日常使用的各类通讯应用，直接延伸至你的指尖。
Python
3.56 K
1.01 K
cann-learning-hubcann-learning-hub
CANN 学习中心仓，支持在线互动运行、边学边练，提供教程、示例与优化方案，一站式助力昇腾开发者快速上手。
Jupyter Notebook
998
511
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库，实现网络在NPU上加速计算。
C++
908
1.83 K
openGauss-serveropenGauss-server
openGauss kernel ~ openGauss is an open source relational database management system
C++
213
313