要让 Grok API 实现实时响应与服务器主动推送，传统的 HTTP 短连接显然无法满足需求——必须借助 WebSocket 建立一条持久化的双向通信通道。直接调用 client.generate() 只能获取单次结果，既无法接收流式 token，也无法中途发送中断信号。本质上，这是一次从“一问一答”到“实时对话”的架构升级，显著降低延迟并提升交互体验。
那么，具体需要满足哪些条件、如何实现？下面逐步拆解。
确认 Grok 服务端是否开放 WebSocket 端点
第一步，查阅官方文档或联系 API 提供商，确认是否存在类似 wss://api.pythongrok.com/v3.5/ws 的专用地址。HTTP 端点（比如 https://api.pythongrok.com/v3.5/generate）默认不支持升级为 WebSocket，如果强行发送 Upgrade: websocket 握手请求，大概率会收到 400 错误。
如果文档中未明确说明，可以尝试用 curl 探测：
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，它们无法处理 WebSocket 协议）。
第一步：安装依赖 pip install websockets。
第二步：构造带 JWT 令牌的握手请求。Grok 要求所有 WebSocket 连接在打开后 5 秒内发送 {"type":"auth","token":"eyJhbGciOi..."}，超时将立即断开。这个时间窗口非常紧凑，代码中务必优先处理认证。
第三步：代码示例：
import asyncio
import websockets
async def connect_grok_ws():
uri = "wss://api.pythongrok.com/v3.5/ws"
async with websockets.connect(uri) as ws:
# 必须在 open 后立刻发送认证帧
await ws.send('{"type":"auth","token":"your_jwt_here"}')
response = await ws.recv()
if not response.startswith('{"status":"ok"}'):
raise ConnectionRefusedError("认证失败")
print("WebSocket认证成功")
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":"算法"} 这样的结构化数据——每个 token 作为独立消息发送，而非拼接成完整字符串后再下发。这样下游可以逐字渲染，用户体验更加流畅。
方法二：中断正在运行的推理任务
任何时刻，只需发送一个控制帧即可终止流式输出：
{"type":"cancel","request_id":"abc123"}
注意：此操作不可逆。服务端收到后立即释放 GPU 资源并关闭当前推理上下文，如需恢复必须重新发送请求。
方法三：心跳保活与异常恢复
Grok 服务端要求每 30 秒内至少有一次数据交互（空 ping 帧也可），否则主动断开连接。建议在独立协程中执行心跳：
async def keep_alive(ws):
while True:
try:
await ws.ping()
except websockets.exceptions.ConnectionClosed:
break
await asyncio.sleep(25)
每 25 秒发送一次 ping，预留 5 秒余量，足以应对网络抖动。当然，如果应用场景本身就有频繁的消息收发，心跳频率可适当降低甚至无需额外处理——关键是要保证两次发送间隔不超过 30 秒。
总结：使用 WebSocket 连接 Grok 的核心要点包括——确认端点存在、5 秒内完成 JWT 认证、按格式发送请求并处理流式 token、以及定期心跳保活。将这些环节串联起来，即可实现真正低延迟的实时交互体验。