Grok Voice Think Fast 2.0 API Tutorial: Build a Real-Time Voice Agent in Python

SpaceXAI's Grok Voice Think Fast 2.0 is a speech-to-speech model. You send it audio over a WebSocket and it sends audio back, and in between it can reason and keep talking while a function call it decided to make is already running. No separate speech-to-text step, no separate text-to-speech step.

SpaceXAI announced Think Fast 2.0 on July 29, 2026: faster first audio, steadier full-duplex behavior (listening while it talks instead of taking strict turns), and tool calls that fire early in the turn. I will keep the benchmark talk brief, since what matters for a tutorial is what changes in your code.

We are going to build a customer-support voice agent for an online store. A caller can ask about an order, change a delivery instruction, cancel, interrupt the agent mid-sentence, and pick the conversation back up after a dropped connection. This is the API path, not the no-code Voice Agent Builder that our Grok Voice Agent Builder tutorial walks through. Start there for the console-first version.

What Is Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 is SpaceXAI's newest model for the Speech to Speech API, the product name behind what most people just call Grok Voice. If you still think of the company as xAI, same outfit: it was folded into SpaceX and rebranded as SpaceXAI on July 6, 2026. The API did not follow the rebrand, so every identifier below still says xai, from the XAI_API_KEY variable to the api.x.ai host.

A traditional voice stack chains three services: speech-to-text, a language model, then text-to-speech, and every hop adds latency and a place for context to get lost. Think Fast 2.0 collapses that into one model that takes audio or text in and produces audio or text out over the same connection.

Speech-to-speech WebSocket versus three-service pipeline architecture. Image by Author.

For an agent that acts rather than one that only talks, what matters is that reasoning and speech run in parallel. SpaceXAI says tool calls "usually" start executing before the agent finishes its first sentence, and that word is doing real work.

On the benchmarks SpaceXAI cites from Artificial Analysis, Think Fast 2.0 scores 82.9% on the Speech to Speech Index against 75.7% for 1.0, and cuts time to first audio from 1.25 seconds to 0.70 seconds. Vendor numbers on a general benchmark are a hypothesis about your caller flow, not a test plan.

There are three model strings you will see: grok-voice-latest, grok-voice-think-fast-2.0, and grok-voice-think-fast-1.0. The alias is convenient while prototyping and not stable enough for anything else.

When I tested on August 4, 2026, grok-voice-latest still resolved to grok-voice-think-fast-1.0, with SpaceXAI's release notes scheduling the move to Think Fast 2.0 for the next day. That switch is a price change as much as a model change, $0.08 per minute of audio against $0.05 for 1.0, so an unpinned alias gets more expensive without a line of your code changing. Pin the versioned string in anything you deploy.

What We'll Build

The agent covers what a support line gets asked: look up an order, find one from an email when the caller has no number, change a delivery instruction, cancel, open or check a ticket, and hand the call to a person. Interruptions and a dropped connection show up along the way.

It is a handful of small files rather than one script, since each piece has a different job and you will want to test them separately. Here is the layout:

• config.py loads the API key and holds the model string, sample rate, and endpoint URLs

• voice_client.py wraps the WebSocket, tracks billing, and exposes send/receive helpers

• tools.py defines the order functions and a small in-memory order store standing in for a real database

• assistant.py holds the system prompt, session configuration, and the event loop that ties everything together

• token_server.py is a small FastAPI endpoint that mints ephemeral tokens

• app_streamlit.py puts the same client behind a live browser call, which I will come back to after the testing section

The teaching path runs from a terminal. The demo adds the microphone.

Prerequisites

You need a SpaceXAI account with an API key, a funded billing arrangement (there is no permanent free tier, and new-account promotional credits will not carry you), and enough comfort with asyncio and WebSockets to follow along without a line-by-line explanation of await.

SpaceXAI's quick-start examples use the raw websockets package rather than a dedicated SDK, and so do we. The docs never state a required Python version. I tested on 3.11.

Keep the API key on the server. If a browser or mobile app talks to the Voice API directly, it gets an ephemeral token instead of your real key, covered in the security section below.

Setting up the project

Every file below lives in the project repo, so you can clone it instead of copying snippets:

git clone https://github.com/KhalidAbdelaty/grok-voice-think-fast-2.0.git
cd grok-voice-think-fast-2.0
pip install -r requirements.txt

websockets carries the realtime connection and python-dotenv reads your key. The rest covers the token endpoint and the browser demo. Put your key in .env:

XAI_API_KEY=xai-your-key-here

That is most of the setup. The connection is the interesting part.

Understanding the Grok Voice Realtime API

Grok Voice is the product name. The thing you actually write code against is a WebSocket endpoint at wss://api.x.ai/v1/realtime, and the whole conversation happens as a stream of JSON events over that one socket.

The event lifecycle

A connection follows a fixed shape: the server sends session.created and conversation.created as soon as you connect, you send session.update to configure voice and tools, the server confirms with session.updated, and from there you create conversation items and request responses. I ran this against a live key and the order matched the docs exactly.

• session.update (client) configures voice, instructions, tools, and audio format

• conversation.item.create (client) adds a user message, an assistant message, or a tool result

• response.create (client) asks the model to speak; server VAD sends this for you automatically

• response.output_audio.delta and response.output_audio_transcript.delta (server) stream the reply as it generates

• response.done (server) closes out the turn

Two things trip people up. The Speech to Speech docs page I linked above mentions a conversation.item.created event during session resumption, but the canonical event reference only lists conversation.item.added, and that is what arrived in every test I ran, so code against it. You will also see an undocumented ping event a few seconds into most connections, mentioned only so you do not read it as an error.

Audio formats and transport

Codec and transport are separate choices. The codec, set under audio.input.format and audio.output.format, is audio/pcm (Linear16, default 24000 Hz), audio/pcmu or audio/pcma (G.711 at 8 kHz, for telephony), or audio/opus (24 kHz). Transport is how those bytes travel on the wire:

• json (the default) sends audio as base64 text inside input_audio_buffer.append and response.output_audio.delta, easy to log and debug

• binary sends raw codec bytes as WebSocket binary frames, skipping the base64 overhead at the cost of a receive loop that has to branch on message type

Start with JSON. Every example in the docs uses it, it is trivial to inspect, and base64 overhead is not what bottlenecks a support-agent build. Move to binary if you measure a reason to.

Compatibility with the OpenAI Realtime API

Skip ahead if you have never touched OpenAI's Realtime API. For everyone else, the Speech to Speech API tracks the OpenAI Realtime API closely enough that most client code ports over by changing the base URL and the key, but it is not a perfect drop-in.

Transcripts arrive as conversation.item.input_audio_transcription.updated here instead of OpenAI's delta, a few OpenAI events go unsupported, and SpaceXAI adds its own extensions: force_message for a scripted disclosure line, resumption for reconnects, and replace for fixing mispronounced brand names before text-to-speech.

Building the Real-Time Voice Agent

Enough protocol. Here is the client that talks to it.

Connecting and configuring the session

The connection opens with a bearer token and a model query parameter, and the first message you send configures everything about how the agent behaves:

import asyncio
import json
import os
import websockets

MODEL = "grok-voice-think-fast-2.0"  # pin the version, not grok-voice-latest

async def connect():
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}"
    ws = await websockets.connect(
        url, additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"}
    )
    await ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "voice": "eve",
            "instructions": SYSTEM_PROMPT,
            "turn_detection": {"type": "server_vad"},
            "tools": ORDER_TOOLS,
            "resumption": {"enabled": True},
        }
    }))

return ws

instructions is the system prompt, and this model wants short ones. SpaceXAI's migration notes say to simplify prompts written for older GPT-era voice models rather than port them verbatim. Mine tells the agent to keep replies short, ask one question at a time, and read any write back before acting. A spoken confirmation is a UX nicety, not a security control. Your application still enforces authorization on the write itself.

One thing that surprised me: an unrecognized model string does not raise an error at connect time, it silently falls back to grok-voice-think-fast-1.0. Downgrading a paid request because of a typo, without saying so, is a strange default. Log session.created's session.model field once at startup and check you got what you asked for.

Terminal output showing session.created after connecting. Image by Author.

Streaming user audio

With turn_detection.type set to server_vad, you only need to keep appending audio. The server decides when the caller has stopped talking and triggers the response for you. Set it to null instead and you own that decision yourself, committing the buffer explicitly when you think the turn is over.

async def send_audio_chunk(ws, pcm_bytes: bytes):
    await ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": base64.b64encode(pcm_bytes).decode(),
    }))

Server VAD has three knobs, and getting them wrong is the most common way I have seen a voice agent feel broken while nothing in the logs errored. None of them appear in session.updated's echo by default, so check them against the docs rather than assume.

• threshold (0.1 to 0.9, default 0.85), how loud audio has to be to count as speech; raise it in noisy rooms, lower it if quiet speakers get missed

• silence_duration_ms, how long the caller goes quiet before the server ends their turn; too short cuts people off mid-thought, too long feels sluggish

• prefix_padding_ms (default 333), a slice of audio kept from just before speech was detected, so the first syllable does not get clipped

Tune silence_duration_ms first if callers keep getting cut off while pausing to think. It is the one I reach for before touching the other two.

Receiving and playing the response

Audio arrives in small pieces as response.output_audio.delta, and the point of streaming is that you play each piece the moment it lands instead of waiting for response.done.

async def play_response(ws):
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "response.output_audio.delta":
            chunk = base64.b64decode(event["delta"])
            speaker.write(chunk)  # your playback call goes here
        elif event["type"] == "response.output_audio_transcript.delta":
            print(event["delta"], end="", flush=True)

Keep the transcript around even in production. It is the cheapest debugging tool you have for when a caller says the agent "said something weird."

Adding Tools to the Voice Agent

A voice agent that only talks is a chatbot with a microphone.

Creating the order tools

Each tool is a JSON schema plus a plain Python function on our side. The model never touches the database, it only ever sees what our function returns.

ORDER_TOOLS = [
    {
        "type": "function",
        "name": "check_order_status",
        "description": "Look up the status, ETA, and delivery instructions for an order.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_number": {"type": "string", "description": "e.g. ORD-1042"},
            },
            "required": ["order_number"],
        },
    },
    # find_orders, update_delivery_instructions, cancel_order,
    # create_support_ticket, check_ticket_status and transfer_to_human
    # all follow the same shape
]

Read operations like check_order_status are safe to retry if something times out. Write operations are not: retrying update_delivery_instructions after an ambiguous timeout can apply the same change twice. A confirmation line in the prompt does not stop that, so give writes an idempotency key or a duplicate check instead.

Put the refusals in the function too. cancel_order returns a reason and an alternative instead of cancelling a shipped order, because a prompt saying "never cancel shipped orders" is a suggestion and a function that refuses is not.

Handling the tool-calling loop

Four steps, and the order matters more than it looks like it should. The model sends response.function_call_arguments.done, your code runs the function, you send the result back as a function_call_output item, and only then do you ask the model to continue.

async def handle_tool_call(ws, event):
    args = json.loads(event["arguments"])
    result = execute(event["name"], args)  # never raises; errors come back as {"error": ...}
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {
            "type": "function_call_output",
            "call_id": event["call_id"],
            "output": json.dumps(result),
        },
    }))

If the model needs more than one tool for a request, it fires multiple function_call_arguments.done events before any audio plays. Resolve all of them and send every result before a single response.create. Send it too early and the model answers without the context from the calls still in flight.

There is a gotcha here that SpaceXAI documents and I still hit on the first pass: sending response.create the instant your tool result goes out can overlap with the introductory sentence the agent is still playing. In one run it opened with "I'll check the status of order ORD-1042 right away" and called the tool mid-sentence, so an immediate response would have talked over its own opener.

Wait for the current turn's audio to finish, and show a short "thinking" state in between.

Tool call flow before continuing response. Image by Author.

Managing Interruptions and Conversation State

Two separate problems here. A caller talks over the agent mid-response, and a WebSocket drops and needs picking back up.

Supporting natural interruptions

With server_vad on, barge-in is automatic server-side: the moment it detects the caller speaking again, it signals input_audio_buffer.speech_started and stops generating the old response. Your job is the client half of that handshake, clearing whatever audio is already queued so the agent goes quiet instead of finishing a sentence nobody asked to hear.

if event["type"] == "input_audio_buffer.speech_started":
    playback_queue.clear()

For manual, non-VAD sessions, response.cancel does the same job on request. There is also conversation.item.truncate for trimming an assistant item down to what was actually heard. The docs confirm it exists but not when to fire it during a live barge-in, so test the timing yourself.

I tested this with a mid-response delivery-instruction change: start the request, interrupt with a different address partway through the agent's confirmation. What matters is whether the agent applies the corrected instruction instead of quietly finishing the old one, not whether the audio stopped. Assert against the order record, not the silence. The browser demo at the end lets you hear this one.

Resuming a disconnected session

Session resumption is opt-in and it is not memory. Set resumption.enabled: true on session.update, grab the ID from the conversation.created event, and if the socket drops, reconnect with ?conversation_id=<id> in the URL and opt in again on the new connection.

async def reconnect(conversation_id):
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}&conversation_id={conversation_id}"
    ws = await websockets.connect(url, additional_headers=auth_header)
    await ws.send(json.dumps({"type": "session.update", "session": {"resumption": {"enabled": True}}}))
    return ws

The cached turns, transcripts, tool calls, and tool results replay before your next question, and the cache disappears after 30 minutes of inactivity. I tested it by asking about an order, dropping the connection, and reconnecting for a follow-up without repeating myself; the agent picked the ETA back up correctly.

One undocumented catch: the replay does not land instantly, so a question fired the moment the socket opens can beat it and come back with no memory of the earlier turn. Give it a second before you blame resumption.

Terminal log of a resumed session. Image by Author.

Do not use this in place of saving order state to your own database. If the cache expires or the caller rings back tomorrow, you start from zero context, and that is by design.

Securing and Monitoring the Agent

Never put a permanent API key in browser or mobile code. If a client connects directly instead of going through your server, mint a short-lived token:

from fastapi import FastAPI
import httpx, os

app = FastAPI()

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.x.ai/v1/realtime/client_secrets",
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()  # {"value": "xai-realtime-client-secret-...", "expires_at": ...}

A browser cannot set a custom Authorization header on a WebSocket handshake, so it passes the token through the sec-websocket-protocol header instead, prefixed with xai-client-secret..

Server mints token, browser joins call. Image by Author.

Billing runs on two meters. Audio, sent or received, runs at the $0.08 per minute I mentioned earlier, which is $4.80 an hour, and every conversation.item.create that is not audio and not a function_call_output costs a flat $0.004. response.create is not billed at all. Each response.done carries a usage object, which in my test reported output_audio_seconds alongside a separate billable_audio_seconds. Bill from those, not from estimates.

Documented limits on the Speech to Speech API are 10 concurrent sessions per team and a 120-minute session cap, both in us-east-1. Do not plan capacity from the Voice Agent API's numbers, which are different.

On privacy, be precise. SpaceXAI's security FAQ says API requests and responses are retained encrypted for 30 days for abuse monitoring and not used for training without permission, and that teams can turn on Zero Data Retention, though ZDR drops persisted voice-agent conversation history and therefore does not work with resumption.

If you are disclosing that a call is recorded or AI-handled, that is what the force_message extension I listed earlier does. The line plays exactly as written instead of whatever the model paraphrases it into.

Testing the Voice Agent

A 200 status on the WebSocket handshake tells you nothing about whether the agent did the right thing. Test the outcome, not just the connection.

• A clean order lookup, checking the spoken answer against the record, not just that a response arrived
• An interrupted response, confirming playback stops and the agent addresses the new request
• A delivery update that requires confirmation, checked against the order record
• A refusal, like cancelling a shipped order, where the agent has to explain the rule rather than apologise
• An unknown order number, making sure the agent says so instead of inventing a status
• A tool that returns an error, checking the agent speaks it instead of stalling
• Reconnection and resumption, including that replay window I hit earlier
• Noisy audio, fast speech, and a caller who spells out numbers and addresses

I ran most of these against a live key while writing this. The interesting failures were behavioral, not errors: the resumption timing above, and an out-of-range VAD threshold accepted instead of rejected, the kind of thing that ships silently broken if you only test the happy path. Add a multilingual test too, and see the FAQs for a wrinkle in how you name the language.

Two of those you cannot test by typing. app_streamlit.py is a Streamlit page that puts a live call in the browser: the microphone streams into the same WebSocket over WebRTC, the agent's voice streams back, and the socket stays open throughout.

streamlit run app_streamlit.py