1
HTTP（寄信模式）:
2
  你：写了封信"天气怎么样？"→ 贴邮票 → 邮局 → 服务器
3
  服务器：看信 → 回信"晴天"→ 邮局 → 你
4
  你：【通信结束，下次再问要重新写信贴邮票】
5

6
WebSocket（打电话模式）:
7
  你：拨号"喂？"                   ← 一次握手
8
  服务器："通了"                    ← 101 Switching Protocols
9
  你：    ─═══════ 线路保持 ════════─  服务器
10
  你："天气怎么样？"  →
11
                    ←  "晴天"
12
  你："那明天呢？"    →
13
                    ←  "下雨"
14
  你："挂了"         →
15
                    ←  "拜拜"
16
  【挂断，线路关闭】

1
客户端 → 服务器:
2
  GET /ws/chat HTTP/1.1
3
  Host: 127.0.0.1:8000
4
  Upgrade: websocket              ← "我要升级"
5
  Connection: Upgrade
6
  Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==  ← 随机密钥
7
  Sec-WebSocket-Version: 13
8

9
服务器 → 客户端:
10
  HTTP/1.1 101 Switching Protocols  ← "好，给你换卡"
11
  Upgrade: websocket
12
  Connection: Upgrade
13
  Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=  ← 用客户端Key+SHA1算出

1
@router.websocket("/chat")
2
async def chat(
3
    websocket: WebSocket,
4
    token: str = Query(...),       # 从 URL 参数取 Token
5
    db: Session = Depends(get_db), # 依赖注入照样能用
6
):
7
    await websocket.accept()
8
    ...

1
# 最小模板（从这抄）
2
@router.websocket("/{path}")
3
async def handler(ws: WebSocket):
4
    await ws.accept()
5
    try:
6
        while True:
7
            msg = await ws.receive_text()
8
            await ws.send_text(f"回: {msg}")
9
    except WebSocketDisconnect:
10
        pass
11

12
# 带身份验证
13
@router.websocket("/secure")
14
async def secure(ws: WebSocket, token: str = Query(...)):
15
    if token != "secret":
16
        await ws.close(code=4001, reason="认证失败")
17
        return
18
    await ws.accept()
19
    ...

1
class ConnectionManager:
2
    """广播室：管理所有在线连接"""
3
    def __init__(self):
4
        self.active_connections: list[WebSocket] = []
5

6
    async def connect(self, websocket: WebSocket):
7
        await websocket.accept()
8
        self.active_connections.append(websocket)
9

10
    def disconnect(self, websocket: WebSocket):
11
        self.active_connections.remove(websocket)
12

13
    async def broadcast(self, message: str):
14
        for connection in self.active_connections:
15
            await connection.send_text(message)
16

17
manager = ConnectionManager()
18

19
@router.websocket("/room/{room_id}")
20
async def chat_room(websocket: WebSocket, room_id: str):
21
    await manager.connect(websocket)
22
    try:
23
        while True:
24
            data = await websocket.receive_text()
25
            await manager.broadcast(f"[{room_id}] 有人说: {data}")
26
    except WebSocketDisconnect:
27
        manager.disconnect(websocket)
28
        await manager.broadcast(f"有人离开了 {room_id}")

1
# 连接管理器骨架
2
class ConnectionManager:
3
    def __init__(self):
4
        self.connections: list[WebSocket] = []
5

6
    async def connect(self, ws): await ws.accept(); self.connections.append(ws)
7
    def disconnect(self, ws): self.connections.remove(ws)
8
    async def broadcast(self, msg):
9
        for ws in self.connections:
10
            await ws.send_text(msg)

1
from fastapi.responses import EventSourceResponse, ServerSentEvent
2

3
# 你的项目已经可以升级成这种写法：
4
return EventSourceResponse(
5
    (ServerSentEvent(data=chunk) for chunk in generate()),
6
    media_type="text/event-stream"
7
)

1
// 浏览器控制台可直接测试
2
const ws = new WebSocket("ws://127.0.0.1:8000/ws/ai-chat");
3

4
ws.onopen = () => ws.send(JSON.stringify({type: "chat", content: "你好"}));
5

6
ws.onmessage = (e) => {
7
    const msg = JSON.parse(e.data);
8
    if (msg.type === "chunk") process.stdout.write(msg.data);  // 流式输出
9
    if (msg.type === "done") console.log("\n✅ 完成:", msg.data);
10
    if (msg.type === "stopped") console.log("\n⏹️ 已停止");
11
};
12

13
// 打断 AI
14
ws.send(JSON.stringify({type: "stop"}));

1
# 认证 + 拒绝连接
2
@router.websocket("/secure")
3
async def secure(ws: WebSocket, token: str = Query(...)):
4
    if token != VALID_TOKEN:
5
        await ws.close(code=4001)  # 在 accept 前 close = HTTP 403
6
        return
7
    await ws.accept()
8
    ...
9

10
# 心跳（简单版）
11
import asyncio
12

13
@router.websocket("/ping")
14
async def ping(ws: WebSocket):
15
    await ws.accept()
16
    try:
17
        while True:
18
            try:
19
                msg = await asyncio.wait_for(ws.receive_text(), timeout=30)
20
                await ws.send_text(f"回: {msg}")
21
            except asyncio.TimeoutError:
22
                await ws.send_text("ping")  # 30秒没收到消息就发 ping
23
    except WebSocketDisconnect:
24
        pass