FastAPI中POST请求向WebSocket传数据报错：连接已关闭问题排查

我开发了一个允许多个机器人通过**/ws/{bot_id}**端点连接的应用，将所有连接存储在websockets字典中。当有POST请求时，想要向对应的WebSocket连接传递数据，却报错：

RuntimeError: Unexpected ASGI message 'websocket.send', after sending 'websocket.close' or response already completed

服务端原代码 #

import asyncio
from fastapi import FastAPI, WebSocket

app = FastAPI()
websockets: dict[int, WebSocket] = {}


@app.websocket("/ws/{bot_id}")
async def websocket_endpoint(websocket: WebSocket, bot_id: int):
    websockets[bot_id] = websocket
    await websocket.accept()


@app.post("/bot/{bot_id}/start_lobby")
async def start_lobby(bot_id: int):
    websocket = websockets.get(bot_id)
    await websocket.send_text("START")


@app.post("/bot/{bot_id}/quit_lobby")
async def destroy_lobby(bot_id: int):
    websocket = websockets.get(bot_id)
    await websocket.send_text("DELETE")

机器人客户端原代码（标准输出始终为空） #

def websocket_connection():
    with connect("ws://localhost:8000/ws/1") as websocket:
        while True:
            message = websocket.recv()
            if message == "START":
                start_lobby()
            elif message == "DELETE":
                destroy_lobby()
            websocket.send("BOT 1: RECEIVED")
            print(f"Received: {message}")

问题原因 #

1. 服务端连接提前关闭：WebSocket端点函数websocket_endpoint在执行await websocket.accept()后直接返回，FastAPI会判定该连接已完成，自动关闭WebSocket连接。后续调用send_text时，连接已失效，触发报错。
2. 客户端未处理异常：服务端关闭连接后，客户端recv()会抛出异常，导致循环直接终止，因此看不到任何输出。

修复方案 #

服务端修复代码 #

import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()
websockets: dict[int, WebSocket] = {}


@app.websocket("/ws/{bot_id}")
async def websocket_endpoint(websocket: WebSocket, bot_id: int):
    await websocket.accept()
    websockets[bot_id] = websocket
    try:
        # 维持长连接：循环监听客户端消息（即使不需要处理，也需保持连接活跃）
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        # 连接断开时从字典移除无效连接
        del websockets[bot_id]


@app.post("/bot/{bot_id}/start_lobby")
async def start_lobby(bot_id: int):
    websocket = websockets.get(bot_id)
    if websocket:
        await websocket.send_text("START")


@app.post("/bot/{bot_id}/quit_lobby")
async def destroy_lobby(bot_id: int):
    websocket = websockets.get(bot_id)
    if websocket:
        await websocket.send_text("DELETE")

客户端修复代码 #

from websockets.sync.client import connect
import time

def websocket_connection():
    while True:
        try:
            with connect("ws://localhost:8000/ws/1") as websocket:
                while True:
                    message = websocket.recv()
                    if message == "START":
                        start_lobby()  # 保留原有业务逻辑
                        print("执行start_lobby")
                    elif message == "DELETE":
                        destroy_lobby()  # 保留原有业务逻辑
                        print("执行destroy_lobby")
                    websocket.send("BOT 1: RECEIVED")
                    print(f"Received: {message}")
        except Exception as e:
            print(f"连接断开，1秒后重试: {e}")
            time.sleep(1)

修复说明 #

• 服务端：通过while True循环监听客户端消息维持长连接，添加WebSocketDisconnect异常捕获，断开时自动清理字典中的无效连接；同时增加连接存在性判断，避免空指针错误。
• 客户端：外层增加循环实现自动重连，捕获连接异常避免程序崩溃，添加调试打印便于确认消息接收情况。

内容的提问来源于stack exchange，提问作者Andrew