WebSocket连接意外断开排查：Quart客户端+FastAPI服务端

为什么WebSocket会在run_interaction执行后断开？ #

1. 服务端侧：FastAPI的WebSocket端点处理函数执行完毕后，框架会自动关闭WebSocket连接。你的服务端代码在发送完JSON响应后，websocket_new_task函数就执行结束了，连接随之关闭，这是框架的默认行为。
2. 客户端侧：你使用async with websockets.connect(...)上下文管理器，当代码退出with块时（也就是return response之后），客户端会主动关闭连接，这也是上下文管理器的预期行为。

如果你的需求是保持连接以处理多次请求，或者你误以为当前的断开是“意外断开”，可以参考以下解决方案：

解决方案1：保持连接处理多轮请求 #

修改服务端代码，加入循环以持续接收客户端的消息，实现长连接：

@app.websocket("/new_task")
async def websocket_new_task(websocket: WebSocket):
    await websocket.accept()
    try:
        # 循环接收客户端消息，维持长连接
        while True:
            data = await websocket.receive_json()
            print("Incoming Data", data)
            context = data.get('context', '')
            org_id = data.get('org_id', '')
            url = data.get('url', '')  
            task = data.get("task", '')
            token = data.get('token', '')

            input = InteractionInput(
                url=url,
                user_input=task,
                token=token,
                base_prompt="" 
            )

            result = await run_interaction(input, "generalist_endpoint", context=context, task_description=task)
            print(f"Completed result: {result}")

            task_summary = f"Task created for org {org_id} with context: {context}. Result: {result.get('text_output', 'No output')}"
            await websocket.send_json({"response": 200, "task_summary": task_summary})
    except WebSocketDisconnect:
        print("WebSocket connection closed by client.")
    except Exception as e:
        print(f"An error occurred: {str(e)}")
        await websocket.send_json({"response": 500, "error": "Internal server error"})

解决方案2：避免因心跳超时导致的意外断开 #

如果run_interaction执行时间较长，可能触发客户端的ping超时（websockets库默认ping超时为20秒），导致客户端主动断开连接。可以在客户端连接时关闭ping超时：

async def perform_action(self, call_sid, context):
    uri = "wss://localhost:5515/new_task" 
    task = """request
    """

    # 关闭ping超时，避免长时间执行任务时触发断开
    async with websockets.connect(uri, ping_timeout=None) as websocket:
        task_details = {
            "task": task,
            "context": context,
            "org_id": "orgid-placeholder", 
            "url": "url-placeholder", 
            "token": "t-placeholder",  
        }

        await websocket.send(json.dumps(task_details))
        print("Task details sent, waiting for response...")

        response = await websocket.recv()
        print("Response received from the server:")
        print(response)

        if call_sid:
            self.track.create_or_update_track(call_sid, on_hold=False)

        return response

解决方案3：确保run_interaction不阻塞事件循环 #

如果run_interaction是同步阻塞函数，会卡住FastAPI的事件循环，导致服务端无法处理WebSocket的心跳包，客户端会误以为连接断开。这种情况下，需要用asyncio.to_thread将同步代码包装为异步执行：

import asyncio

# 替换原有的result赋值行
result = await asyncio.to_thread(
    run_interaction, 
    input, 
    "generalist_endpoint", 
    context=context, 
    task_description=task
)

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