FastAPI WebSockets 参考：从 WebSocket 对象到连接状态管理的完整接口指南

FastAPI 的 WebSocket 支持建立在 Starlette 之上，通过 fastapi/websockets.py 将 WebSocket、WebSocketDisconnect 与 WebSocketState 等核心类型原样再导出，让开发者只需一条 from fastapi import WebSocket 即可开始编写双向实时通信端点。本文以 FastAPI 官方参考文档为骨架，逐项梳理 WebSocket 对象上可用的属性、方法与连接生命周期控制手段，并结合本仓库源码说明 @app.websocket 装饰器、依赖注入与异常处理在底层是如何工作的。读完本文，你将掌握 WebSocket 端点的声明方式、收发文本/字节/JSON 数据的全部接口、断线检测与状态判定方法，以及同时兼容 HTTP 与 WebSocket 的依赖设计技巧。

FastAPI 中的 WebSocket 从哪来

WebSocket 并非 FastAPI 重复实现的协议封装，而是由 ASGI 生态中的 Starlette 提供。在 FastAPI 源码中，fastapi/websockets.py 的全部内容只有三行再导出语句：

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

随后 fastapi/init.py 又将 WebSocket 与 WebSocketDisconnect 提升到包顶层，因此你可以直接用最简洁的导入方式：

from fastapi import WebSocket
from fastapi import WebSocketDisconnect

WebSocketState 则不进入 fastapi 包顶层命名空间，需要从子模块导入：

from fastapi.websockets import WebSocketState

这种"顶层可用、子模块可查"的导出设计，保证了日常编写端点时导入最简，同时也为需要精确引用的场景保留了完整路径。

声明 WebSocket 端点与连接对象

编写 WebSocket 端点时，FastAPI 遵循与普通 HTTP 路径操作完全一致的装饰器风格。应用实例 FastAPI 在 applications.py 中暴露了 websocket 方法，其签名支持 path 与可选的 name、dependencies 参数。一个最典型的回声端点（该示例同样出现在 websocket 方法的 docstring 中）如下：

from fastapi import FastAPI, WebSocket

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 类型，FastAPI 在路由层识别该注解后，会为你构造一个绑定到当前连接的 WebSocket 对象，通过它你可以读取客户端发来的数据，也可以向客户端写回数据。仓库中的真实用例可参考 docs_src/websockets_ 目录下的教程代码，以及 tests/test_ws_dependencies.py 等测试文件。

底层：路由如何装配 WebSocket 端点

从源码结构看，WebSocket 端点的装配链路比 HTTP 路由更细：

1. FastAPI.websocket(...) 装饰器最终调用 self.add_api_websocket_route(...)；
2. 路由对象为 APIWebSocketRoute，它继承 Starlette 的 WebSocketRoute，在初始化时通过 _build_dependant_with_parameterless_dependencies 构建依赖树，并把最终处理器包装为 websocket_session(...)；
3. websocket_session 内部创建 session = WebSocket(scope, receive=receive, send=send)，同时注入两层 AsyncExitStack（分别存入 scope["fastapi_inner_astack"] 与 scope["fastapi_function_astack"]），用于支撑依赖在 yield 之后的清理逻辑；
4. get_websocket_app 调用依赖求解器 solve_dependencies(request=websocket, ...)，将解析出的参数解包后传给用户端点函数：await dependant.call(**solved_result.values)。

也就是说，你拿到的 WebSocket 参数本身就是 Starlette 的完整 WebSocket 会话对象，FastAPI 只在其外层叠加了依赖注入与校验能力。这也解释了为何接口参考文档会把连接元数据、原始收发能力与应用层扩展一并列在该对象名下。

WebSocket 对象的成员地图

参考文档为 WebSocket 类列出了完整的公开成员，它们大致可以分成四类，理解分组有助于按需取用。

连接元数据与属性

这些属性描述"我是谁、从哪来"：

成员	作用
scope	当前 ASGI 作用域字典，承载原始连接信息（类型、路径、查询串、ASGI 版本等），是 WebSocket 的"根"数据源
app	指向当前处理连接的 ASGI 应用对象
url	本次连接请求的完整 URL
base_url	不带路径的基准地址（scheme + host），供构建绝对链接使用
headers	只读的请求头映射（Headers 类型）
query_params	只读的查询参数映射（QueryParams 类型）
path_params	由路由路径模板解析出的路径参数
cookies	请求携带的 Cookie 映射
client	客户端地址信息，形如 (host, port)
state	单连接级状态字典，可在同一连接的请求生命周期内共享应用数据
client_state / application_state	与 state 对应的更细粒度状态访问入口
url_for(...)	按路由 name 反向生成 URL

注意这里要区分两组概念：headers、query_params、path_params、cookies 等属于"请求侧只读元数据"，主要用来做鉴权、参数读取或日志记录；而 state 一族则用于在端点内部和依赖之间传递运行期上下文。

底层 ASGI 收发原语

• receive()：从 ASGI 通道读取下一个事件（如 websocket.receive、websocket.disconnect）；
• send(message)：把 ASGI 消息（如 websocket.accept、websocket.send.text）写入通道。

这两个方法是所有高层方法的地基，日常开发很少直接使用，只有在需要自定义协议行为时才需要触碰。

面向业务的高层收发与连接管理方法

参考文档列出的高层方法形成了 WebSocket 编程的"主菜单"：

方法	行为
await accept(...)	接受连接握手，可传 subprotocol 等参数
await receive_text()	阻塞式接收下一条文本消息
await receive_bytes()	阻塞式接收下一条二进制消息
await receive_json(...)	阻塞式接收并解析下一条 JSON 消息
async for ... in iter_text()	以异步迭代方式逐条消费文本消息
async for ... in iter_bytes()	以异步迭代方式逐条消费二进制消息
async for ... in iter_json()	以异步迭代方式逐条消费 JSON 消息
await send_text(data)	发送一条文本消息
await send_bytes(data)	发送一条二进制消息
await send_json(data, ...)	序列化并发送 JSON 消息
await close(code=..., reason=...)	主动关闭连接，可指定 WebSocket 关闭码与原因字符串

在基于 while True 的接收循环中，最常见的组合是 await websocket.receive_text() 配合 await websocket.send_text(...)；而 iter_* 系列迭代器让"收到即处理、处理完再收"的流水线写法更自然，例如：

@app.websocket("/echo-json")
async def echo_json(websocket: WebSocket):
    await websocket.accept()
    async for data in websocket.iter_json():
        await websocket.send_json({"echo": data})

处理客户端断开：WebSocketDisconnect

真实场景中客户端可能随时离开，此时继续等待消息会抛异常。Starlette 约定：当对端断开时，receive_* 或 iter_* 内部会抛出 WebSocketDisconnect，其中带一个 code 属性，指示断开原因（例如 1000 表示正常关闭，1001 表示客户端离开）。因此稳健的端点通常把收发循环包进异常处理：

from fastapi import WebSocket, WebSocketDisconnect

@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"Message text was: {data}")
    except WebSocketDisconnect:
        print("client disconnected")

WebSocketDisconnect 同样可以从 fastapi 顶层导入（见 fastapi/init.py 的导出），其定义最终源自 starlette.websockets。多客户端聊天室等场景常以它为"退订时机"：在某客户端断开时把它从在线集合中移除，并广播离开消息。

需要补充的是，FastAPI 还提供了比裸 ASGI 断开更高一层的语义化异常 WebSocketException，其定义在 fastapi/exceptions.py，允许你在自己的代码中主动抛出并向客户端展示错误，例如代码 1008（策略违规）：

from fastapi import WebSocketException
from starlette.status import WS_1008_POLICY_VIOLATION

raise WebSocketException(code=WS_1008_POLICY_VIOLATION, reason="Not allowed")

WebSocketState：连接状态的枚举

WebSocketState 是描述连接可能处于哪些状态的枚举，同样源自 Starlette。典型取值包括 CONNECTING（握手尚未完成）、CONNECTED（已 accept，可正常收发）、DISCONNECTED（已断开）。WebSocket 对象上通常暴露 client_state 与 application_state 来查询对应维度的当前状态，例如在依赖清理或异常兜底时判断"连接是否仍然打开"，避免向已断开的对端写数据。

由于该枚举在参考文档中被单列为"附加类"，实现位置是 fastapi.websockets 子模块而非包顶层，导入时请使用完整路径：

from fastapi.websockets import WebSocketState

if websocket.client_state == WebSocketState.CONNECTED:
    await websocket.send_text("still alive")

同时兼容 HTTP 与 WebSocket 的依赖：HTTPConnection

参考文档给出的关键技巧是：如果你希望某段依赖既服务于普通 HTTP 请求、又服务于 WebSocket 连接，就不要把依赖参数声明为具体的 Request 或 WebSocket，而是声明为二者的共同基类 HTTPConnection。这样同一套鉴权、日志、上下文解析逻辑可以无差别地在两类端点中复用。

FastAPI 之所以支持这种写法，是因为依赖求解器在 dependencies/utils.py 中会根据传入请求的实际类型进行匹配：当类型注解可兼容 HTTPConnection（Request 与 WebSocket 的共同父类）时即视为满足，且该文件同时导入了 WebSocket 与 Request 作为判型依据。而 WebSocket 端点本身在路由装配阶段也会走同一套 solve_dependencies 逻辑（见 routing.py），这保证了 HTTP 与 WebSocket 两条路径依赖语义的一致性。

from fastapi import FastAPI, WebSocket, Depends
from starlette.requests import HTTPConnection

app = FastAPI()

def extract_token(conn: HTTPConnection) -> str:
    # 对 Request 与 WebSocket 均适用
    return conn.headers.get("x-token", "")

@app.get("/items")
async def read_items(token: str = Depends(extract_token)):
    return {"token": token}

@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket, token: str = Depends(extract_token)):
    await websocket.accept()
    await websocket.send_text(f"token={token}")
    await websocket.close()

WebSocket 端点上的依赖注入

与 HTTP 路径操作一样，@app.websocket 装饰器接受 dependencies 参数，用于在端点执行前统一解析一批前置依赖。仓库中的 tests/test_ws_dependencies.py 展示了应用级、路由级、前缀级与端点级依赖如何叠加生效：

from fastapi import APIRouter, Depends, FastAPI, WebSocket
from fastapi.testclient import TestClient

app = FastAPI(dependencies=[create_dependency("app")])

@app.websocket("/", dependencies=[create_dependency("index")])
async def index(websocket: WebSocket, deps: DepList):
    await websocket.accept()
    await websocket.send_text(json.dumps(deps))
    await websocket.close()

该测试随后用 TestClient 发起 websocket_connect("/") 并断言收到的依赖顺序为 ["app", "router", "prefix_router", "index"]（按路由嵌套层级顺序收集），说明 FastAPI 的 WebSocket 依赖系统与 HTTP 完全同构。你可以用同样的方式在 WebSocket 端点上执行数据库连接注入、Token 校验、限流等横切逻辑。

校验失败时如何收尾：WebSocket 专用异常处理

在 HTTP 场景下，请求体校验失败会返回 422 JSON 响应；但对 WebSocket 而言没有"响应体"可言，因此 FastAPI 走的是另一条路径。当依赖求解或参数校验失败时，get_websocket_app 会抛出 WebSocketRequestValidationError，而 fastapi/exception_handlers.py 中的默认处理器会以 1008（策略违规）关闭码关闭连接，并把错误详情编码进 reason：

async def websocket_request_validation_exception_handler(
    websocket: WebSocket, exc: WebSocketRequestValidationError
) -> None:
    await websocket.close(
        code=WS_1008_POLICY_VIOLATION, reason=jsonable_encoder(exc.errors())
    )

从源码结构可以看出，这构成了 WebSocket 编程的一条完整原则：任何无法通过 HTTP 状态码表达的失败，在 WebSocket 世界里都必须翻译成"关闭码 + reason 字符串"。这与你使用 WebSocketDisconnect.code 判断对端为何离开，以及用 close(code, reason) 主动说明离开原因，是同一套语义体系的两个侧面。若要完全掌控行为，可以对 websocket_request_validation_exception_handler 注册自定义版本，替换默认的关闭码与提示。

小结

结合官方参考文档与本仓库源码可以看到，FastAPI 的 WebSocket 编程模型非常收敛：对外，你只需要 WebSocket 对象上的十余个属性和方法——accept 开启通道，receive_*/iter_* 读、send_* 写，close 收尾，辅以 WebSocketDisconnect 感知异常断开、WebSocketState 判断当前状态；对内，FastAPI 通过 APIWebSocketRoute 与 websocket_session 把 Starlette 的会话对象无缝接入了自己的依赖注入与参数校验体系，并借助 HTTPConnection 让同一份依赖逻辑在 HTTP 与 WebSocket 两类端点间自由复用。

参考文件索引

• 官方接口参考：本页即 docs/en/docs/reference/websockets.md
• 类型再导出： fastapi/websockets.py、fastapi/init.py
• 端点注册与底层装配：fastapi/applications.py、fastapi/routing.py
• WebSocket 依赖兼容判定：fastapi/dependencies/utils.py
• 异常与校验处理：fastapi/exceptions.py、fastapi/exception_handlers.py
• 可运行的示例与测试：docs_src/websockets_、tests/test_ws_dependencies.py