Drive Responses API over WebSocket

Open wss://api.x.ai/v1/responses, send response.create for every turn, and chain with previous_response_id so follow-ups ship only new input — the open socket keeps prior state in memory.

First turn

import json, os
from websocket import create_connection

ws = create_connection(
    "wss://api.x.ai/v1/responses",
    header=[f"Authorization: Bearer {os.environ['XAI_API_KEY']}"],
)

ws.send(json.dumps({
    "type": "response.create",
    "model": "grok-4.6",
    "store": False,
    "input": [{
        "type": "message",
        "role": "user",
        "content": [{"type": "input_text", "text": "Find fizz_buzz()"}],
    }],
    "tools": [],
}))

Body shape matches Responses create minus transport-only fields (stream, background). Events stream on the socket in the same order as HTTP Responses streaming.

Warmup with generate: false

Prime tools, instructions, or system messages without running the model:

ws.send(json.dumps({
    "type": "response.create",
    "model": "grok-4.6",
    "store": False,
    "generate": False,
    "input": [],
    "tools": [],  # or your tool list
}))

The warmup still emits a response ID. Chain the real turn from that ID via previous_response_id.

Continue the chain

Send only new items — tool outputs plus the next user message. Do not resend history.

ws.send(json.dumps({
    "type": "response.create",
    "model": "grok-4.6",
    "store": False,
    "previous_response_id": "resp_123",
    "input": [
        {
            "type": "function_call_output",
            "call_id": "call_123",
            "output": "tool result",
        },
        {
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": "Now optimize it."}],
        },
    ],
    "tools": [],
}))

The connection caches the latest response in memory, so store=false and Zero Data Retention still chain while the socket stays up. Older IDs outside that cache: store=true may rehydrate from storage; store=false / ZDR fails with previous_response_not_found. A failed turn (4xx/5xx) evicts its ID from the cache.

Limits and reconnect

• One connection runs turns serially; a second response.create queues. Parallel turns need multiple sockets.
• Max open time: 25 minutes, then the server closes.
• After drop / 25-min cap: with store=true and a valid ID, continue on a new socket with previous_response_id. With store=false or previous_response_not_found, drop the ID and send full context for a fresh chain.

Errors to handle

previous_response_not_found — ID missing from connection cache and storage (ZDR, store=false, or eviction after failure):

{
  "type": "error",
  "status": 400,
  "error": {
    "code": "previous_response_not_found",
    "message": "Previous response with id 'resp_abc' not found.",
    "param": "previous_response_id"
  }
}

websocket_connection_limit_reached — fired right before the 25-minute close; open a new WebSocket and reconnect:

{
  "type": "error",
  "status": 400,
  "error": {
    "type": "invalid_request_error",
    "code": "websocket_connection_limit_reached",
    "message": "Responses websocket connection limit reached (25 minutes). Create a new websocket connection to continue."
  }
}

Pitfalls

• Keep API keys on the backend; do not open this socket from a browser with a long-lived key.
• Do not multiplex turns on one connection — open another for parallel work.
• On ZDR / store=false, treat socket loss as a full restart unless you can resend the full input context.
• Console API credits are separate from SuperGrok's weekly pool on grok.com.