WebSockets

WebSocket APIs

# Full-duplex persistent connection
from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/{client_id}")
async def ws_endpoint(ws: WebSocket, client_id: str):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text("Echo: " + data)
    except WebSocketDisconnect:
        pass

# Broadcast manager
class Manager:
    def __init__(self): self.connections = []
    async def connect(self, ws):
        await ws.accept(); self.connections.append(ws)
    async def broadcast(self, msg: str):
        for c in self.connections: await c.send_text(msg)

# JS client
const ws = new WebSocket("wss://api.example.com/ws/1");
ws.send(JSON.stringify({type:"msg",text:"Hi"}));