如何通过python调用fastapi中的websocket

要在 FastAPI 中使用 WebSocket，需要使用 FastAPI 的 WebSocketEndpoint 类，该类是 WebSocket 协议的一个实现。以下是一个使用 FastAPI WebSocket 的简单示例：

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"You said: {data}")

在上面的代码中，定义了一个 WebSocket 端点 `/ws`，并使用 `async def` 定义了一个异步函数 `websocket_endpoint`。该函数的形参是一个 WebSocket，它将在连接建立时自动创建。

使用 `await websocket.accept()` 方法来接受 WebSocket 的连接请求，并在之后的循环中等待从客户端接收到的数据。使用 `await websocket.receive_text()` 方法接收数据，并使用 `await websocket.send_text()` 方法向客户端发送响应。

可以使用 Python 的 WebSocket 客户端来测试这个 WebSocket 端点，比如使用 `websockets` 库：

import asyncio
import websockets

async def send_message():
    async with websockets.connect('ws://localhost:8000/ws') as websocket:
        await websocket.send('Hello')
        response = await websocket.recv()
        print(response)

asyncio.get_event_loop().run_until_complete(send_message())

在上面的代码中，使用 `websockets.connect()` 方法连接到 WebSocket 端点，发送一个消息，并等待回复。最后，使用 `print()` 方法打印回复。