要让Grok API支持实时响应和服务器主动推送，必须使用WebSocket建立持久化双向通道；HTTP端点不支持升级，需确认官方是否提供wss://地址，认证需JWT且含ws_access作用域，发送generate请求可获流式token，cancel可中断任务，需每30秒ping保活。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
要让Grok API支持实时响应和服务器主动推送，必须绕过HTTP短连接限制，用WebSocket建立持久化双向通道；直接调用client.generate()只能获得单次响应，无法接收流式token或中断信号。
确认Grok服务端是否开放WebSocket端点
访问官方文档或联系API提供商，确认是否存在类似wss://api.pythongrok.com/v3.5/ws的WebSocket专用地址。HTTP端点（如https://api.pythongrok.com/v3.5/generate）默认不支持升级为WebSocket连接，强行发起Upgrade: websocket握手会返回400错误。
若文档未明确标注WebSocket支持，可尝试发送一次标准HTTP Upgrade请求探测：
curl -i -N -H "Connection: Upgrade" -H "Upgrade: websocket" -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" -H "Sec-WebSocket-Version: 13" https://api.pythongrok.com/v3.5/ws
收到HTTP/1.1 101 Switching Protocols且Header含Upgrade: websocket，才代表服务端真正启用WebSocket协议栈。
客户端建立WebSocket连接并完成身份认证
使用Python标准库websockets连接（不要用requests或httpx）：
第一步：安装依赖：pip install websockets
第二步：构造带JWT令牌的握手请求——Grok要求所有WebSocket连接在打开后5秒内发送{"type":"auth","token":"eyJhbGciOi..."}，超时未发则立即关闭连接。
第三步：连接代码示例：
import asyncio<br>import websockets<br><br>async def connect_grok_ws():<br> uri = "wss://api.pythongrok.com/v3.5/ws"<br> async with websockets.connect(uri) as ws:<br> # 必须在open后立刻发送认证帧<br> await ws.send(\'{"type":"auth","token":"your_jwt_here"}\')<br> response = await ws.recv()<br> if not response.startswith(\'{"status":"ok"}\'):<br> raise ConnectionRefusedError("认证失败")<br> print("WebSocket认证成功")<br><br>asyncio.run(connect_grok_ws())
【JWT令牌必须是短期有效且含ws_access作用域的凭证，普通API Key不可用于WebSocket认证】
发送推理请求并接收流式响应
方法一：发送标准推理指令
连接认证通过后，向WebSocket发送JSON对象：
{"type":"generate","prompt":"量子计算如何影响密码学","max_tokens":200,"stream":true}
服务端将逐帧返回{"type":"token","value":"Shor"}、{"type":"token","value":"算法"}等结构化数据，而非拼接后的完整字符串。
方法二：中断正在运行的推理任务
在任意时刻发送控制帧即可终止流式输出：
{"type":"cancel","request_id":"abc123"}
注意：该操作不可逆，服务端收到后立即释放GPU资源并关闭当前推理上下文。
方法三：心跳保活与异常恢复
Grok WebSocket服务端要求每30秒内至少有一次数据交互（可以是空ping帧），否则主动断连。建议在独立协程中执行：
async def keep_alive(ws):<br> while True:<br> try:<br> await ws.ping()<br> except websockets.exceptions.ConnectionClosed:<br> break<br> await asyncio.sleep(25)
