Grok Voice Think Fast 2.0 API-Tutorial: Baue einen Echtzeit-Voice-Agenten in Python

SpaceXAI's Grok Voice Think Fast 2.0 ist ein Speech-to-Speech-Modell. Du sendest Audio über einen WebSocket und bekommst Audio zurück. Dazwischen kann das Modell bereits weiterreden und parallel eine Funktion ausführen, für die es sich entschieden hat. Kein separates Speech-to-Text, kein separates Text-to-Speech.

SpaceXAI kündigte Think Fast 2.0 am 29. Juli 2026 an: schnelleres erstes Audio, stabileres Vollduplex-Verhalten (zuhören, während es spricht, statt strikter Wechsel) und Toolaufrufe, die früh in der Runde starten. Benchmark-Talk halte ich kurz – wichtig fürs Tutorial ist, was sich in deinem Code ändert.

Wir bauen einen Voice-Agenten für den Kundensupport eines Onlineshops. Anrufer können nach einer Bestellung fragen, eine Lieferanweisung ändern, stornieren, den Agenten mitten im Satz unterbrechen und nach einer unterbrochenen Verbindung wieder einsteigen. Das ist der API-Weg, nicht der No-Code-Builder aus unserem Grok Voice Agent Builder Tutorial. Für die Konsolenvariante fang dort an.

Was ist Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 ist SpaceXAI's neuestes Modell für die Speech to Speech API – der Produktname hinter dem, was die meisten einfach Grok Voice nennen. Falls du das Unternehmen noch als xAI kennst: Gleiche Truppe – am 6. Juli 2026 in SpaceX integriert und in SpaceXAI umbenannt. Die API hat das Rebranding nicht mitvollzogen, daher steht unten überall noch xai – von der XAI_API_KEY-Variable bis zum api.x.ai-Host.

Ein klassischer Voice-Stack kettet drei Services: Speech-to-Text, ein Sprachmodell, dann Text-to-Speech – jeder Hop erhöht die Latenz und birgt Kontextverlust. Think Fast 2.0 reduziert das auf ein Modell, das Audio oder Text entgegennimmt und über dieselbe Verbindung Audio oder Text ausgibt.

Speech-to-Speech-WebSocket versus Drei-Services-Architektur. Bild: Autor.

Für einen Agenten, der handelt statt nur zu reden, zählt, dass Reasoning und Sprache parallel laufen. SpaceXAI sagt, Toolaufrufe starten „meistens“ bevor der Agent seinen ersten Satz beendet – und dieses Wort trägt Gewicht.

Auf den von SpaceXAI zitierten Benchmarks von Artificial Analysis erreicht Think Fast 2.0 82,9 % im Speech-to-Speech-Index gegenüber 75,7 % für 1.0 und senkt die Time-to-First-Audio von 1,25 s auf 0,70 s. Anbieterzahlen auf einem allgemeinen Benchmark sind eine Hypothese über deinen Callflow, kein Testplan.

Es gibt drei Modell-Strings, die du sehen wirst: grok-voice-latest, grok-voice-think-fast-2.0 und grok-voice-think-fast-1.0. Der Alias ist beim Prototyping bequem und sonst nicht stabil genug.

Als ich am 4. August 2026 getestet habe, löste grok-voice-latest noch in grok-voice-think-fast-1.0 auf. Laut SpaceXAI's Release Notes sollte der Wechsel auf Think Fast 2.0 am nächsten Tag erfolgen. Der Wechsel ist so sehr eine Preis- wie eine Modelländerung: $0,08 pro Audiominute gegenüber $0,05 für 1.0. Ein unfixierter Alias wird teurer, ohne dass du eine Zeile Code änderst. Pinne die versionierte Zeichenkette in allem, was du ausrollst.

Was wir bauen

Der Agent deckt typische Supportfragen ab: Bestellung nachschlagen, über E-Mail finden, wenn keine Nummer da ist, Lieferanweisung ändern, stornieren, Ticket eröffnen oder prüfen und an eine Person übergeben. Unterbrechungen und eine getrennte Verbindung kommen unterwegs vor.

Es sind ein paar kleine Dateien statt eines Skripts, da jedes Teil eine andere Aufgabe hat und du sie separat testen willst. So ist das Layout:

• config.py lädt den API-Schlüssel und hält Modell-String, Samplerate und Endpunkt-URLs

• voice_client.py kapselt den WebSocket, trackt die Abrechnung und stellt Sende-/Empfangs-Helper bereit

• tools.py definiert die Bestellfunktionen und einen kleinen In-Memory-Order-Store als Platzhalter für eine echte Datenbank

• assistant.py enthält den System-Prompt, die Sitzungskonfiguration und die Ereignisschleife, die alles verbindet

• token_server.py ist ein kleiner FastAPI-Endpunkt, der kurzlebige Tokens ausstellt

• app_streamlit.py stellt denselben Client für Live-Anrufe im Browser bereit – darauf komme ich nach dem Testteil zurück

Die Lernstrecke läuft über das Terminal. Die Demo bringt das Mikrofon dazu.

Voraussetzungen

Du brauchst ein SpaceXAI-Konto mit API-Schlüssel, eine hinterlegte Zahlungsmethode (es gibt keinen permanenten Free-Tarif, und Promo-Guthaben für neue Accounts tragen dich nicht) und genug Routine mit asyncio und WebSockets, um ohne Zeile-für-Zeile-Erklärung von await mitzukommen.

SpaceXAI's Quickstart-Beispiele nutzen das rohe websockets-Paket statt eines dedizierten SDK – wir auch. Die Doku nennt keine erforderliche Python-Version. Ich habe mit 3.11 getestet.

Behalte den API-Schlüssel auf dem Server. Wenn ein Browser oder eine Mobile-App direkt mit der Voice-API spricht, erhält sie stattdessen ein ephemeres Token – unten im Security-Abschnitt erklärt.

Projekt einrichten

Alle Dateien unten liegen im Projekt-Repo. Du kannst es klonen statt Snippets zu kopieren:

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 trägt die Realtime-Verbindung und python-dotenv liest deinen Schlüssel. Der Rest deckt den Token-Endpunkt und die Browser-Demo ab. Pack deinen Schlüssel in .env:

XAI_API_KEY=xai-your-key-here

Das war fast alles fürs Setup. Die Verbindung ist der spannende Teil.

Die Grok Voice Realtime API verstehen

Grok Voice ist der Produktname. Der tatsächliche Code zielt auf einen WebSocket-Endpunkt unter wss://api.x.ai/v1/realtime, und das gesamte Gespräch läuft als Strom aus JSON-Events über genau diesen Socket.

Der Event-Lifecycle

Eine Verbindung folgt einem festen Muster: Der Server sendet session.created und conversation.created direkt nach dem Connect. Du sendest session.update zur Konfiguration von Stimme und Tools, der Server bestätigt mit session.updated, und danach erzeugst du Conversation-Items und forderst Antworten an. Ich habe das gegen einen Live-Key gefahren – die Reihenfolge entsprach exakt der Doku.

• session.update (Client) konfiguriert Voice, Instructions, Tools und Audioformat

• conversation.item.create (Client) fügt eine User-Message, eine Assistant-Message oder ein Tool-Result hinzu

• response.create (Client) fordert das Modell zum Sprechen auf; Server-VAD sendet das automatisch für dich

• response.output_audio.delta und response.output_audio_transcript.delta (Server) streamen die Antwort, während sie generiert wird

• response.done (Server) schließt die Runde ab

Zwei Stolpersteine: Die verlinkte Speech-to-Speech-Seite erwähnt ein conversation.item.created-Event bei Sitzungsfortsetzung, aber die kanonische Event-Referenz listet nur conversation.item.added – und genau das kam in allen meinen Tests an. Code also dagegen. Du wirst außerdem ein undokumentiertes ping -Event ein paar Sekunden nach Verbindungsstart sehen – nur erwähnt, damit du es nicht für einen Fehler hältst.

Audioformate und Transport

Codec und Transport sind getrennte Entscheidungen. Den Codec setzt du unter audio.input.format und audio.output.format: audio/pcm (Linear16, Standard 24000 Hz), audio/pcmu oder audio/pcma (G.711 bei 8 kHz, für Telefonie) oder audio/opus (24 kHz). Transport bestimmt, wie die Bytes übers Kabel gehen:

• json (Standard) sendet Audio als Base64-Text in input_audio_buffer.append und response.output_audio.delta, leicht zu loggen und zu debuggen

• binary sendet rohe Codec-Bytes als WebSocket-Binary-Frames, spart Base64-Overhead, erfordert aber einen Receive-Loop, der nach Nachrichtentyp verzweigt

Starte mit JSON. Alle Beispiele in der Doku nutzen es, es ist trivial zu inspizieren, und Base64-Overhead ist nicht der Flaschenhals in einem Support-Agenten. Wechsle zu Binary, wenn du einen messbaren Grund hast.

Kompatibilität mit der OpenAI Realtime API

Überspring, wenn du nie mit OpenAI's Realtime API gearbeitet hast. Für alle anderen: Die Speech-to-Speech-API folgt der OpenAI Realtime API nah genug, dass die meisten Clients mit geändertem Base-URL und Key portieren – aber nicht 1:1.

Transkripte kommen hier als conversation.item.input_audio_transcription.updated statt OpenAI's delta. Einige OpenAI-Events fehlen, und SpaceXAI ergänzt eigene Erweiterungen: force_message für eine fest vorgegebene Offenlegung, resumption für Reconnects und replace zum Korrigieren falsch ausgesprochener Markennamen vor TTS.

Den Echtzeit-Voice-Agenten bauen

Genug Protokoll. Hier ist der Client, der damit spricht.

Verbinden und die Session konfigurieren

Die Verbindung öffnet mit Bearer-Token und einem Model-Query-Parameter. Die erste Nachricht, die du sendest, konfiguriert das gesamte Agentenverhalten:

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 ist der System-Prompt – und dieses Modell mag kurze. SpaceXAI's Migrationshinweise raten, Prompts für ältere GPT-Voice-Modelle zu vereinfachen statt 1:1 zu portieren. Meiner bittet um kurze Antworten, eine Frage nach der anderen, und eine Rücklese-Bestätigung vor Schreibaktionen. Eine gesprochene Bestätigung ist UX, kein Security-Control. Deine App erzwingt die Autorisierung beim Schreiben selbst.

Eine Sache hat mich überrascht: Ein nicht erkannter Modell-String wirft beim Connect keinen Fehler, sondern fällt still auf grok-voice-think-fast-1.0 zurück. Einen bezahlten Request wegen Tippfehler downzugraden, ohne es zu sagen, ist ein seltsamer Standard. Logge session.created's Feld session.model einmal beim Start und prüfe, dass du bekommen hast, was du wolltest.

Terminalausgabe mit session.created nach dem Connect. Bild: Autor.

User-Audio streamen

Mit turn_detection.type auf server_vad musst du nur weiter Audio anhängen. Der Server entscheidet, wann der Anrufer aufgehört hat zu sprechen, und triggert die Antwort. Setze es auf null und du triffst die Entscheidung selbst – du commitest den Buffer explizit, wenn du die Runde für beendet hältst.

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 hat drei Stellschrauben – falsch eingestellt fühlt sich der Agent kaputt an, obwohl keine Logs fehlschlagen. Sie erscheinen standardmäßig nicht im Echo von session.updated. Also mit der Doku abgleichen, nicht raten.

• threshold (0,1–0,9, Standard 0,85): wie laut Audio sein muss, um als Sprache zu zählen; in lauten Räumen erhöhen, bei überhörten Leisesprechern senken

• silence_duration_ms: wie lange der Anrufer still ist, bevor der Server seine Runde beendet; zu kurz schneidet Gedanken ab, zu lang wirkt träge

• prefix_padding_ms (Standard 333): ein Stück Audio kurz vor Spracherkennung, damit die erste Silbe nicht abgeschnitten wird

Stell zuerst silence_duration_ms ein, wenn Anrufer beim Nachdenken ständig abgeschnitten werden. Das ist mein erster Hebel vor den anderen beiden.

Antwort empfangen und abspielen

Audio kommt in kleinen Stücken als response.output_audio.delta. Der Sinn von Streaming ist, jedes Stück sofort abzuspielen statt auf response.done zu warten.

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)

Behalte das Transkript auch in Produktion. Es ist dein günstigstes Debugging-Tool, wenn jemand sagt, der Agent habe „etwas Komisches gesagt“.

Tools zum Voice-Agenten hinzufügen

Ein Voice-Agent, der nur redet, ist ein Chatbot mit Mikrofon.

Die Order-Tools erstellen

Jedes Tool ist ein JSON-Schema plus eine einfache Python-Funktion bei uns. Das Modell berührt die Datenbank nie, es sieht nur, was unsere Funktion zurückgibt.

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
]

Lese-Operationen wie check_order_status sind bei Timeouts unkritisch zu wiederholen. Schreib-Operationen nicht: Ein erneutes update_delivery_instructions nach unklarem Timeout kann dieselbe Änderung doppelt anwenden. Eine Bestätigungszeile im Prompt verhindert das nicht – gib Writes daher einen Idempotenz-Key oder einen Duplikat-Check.

Pack die Ablehnungen auch in die Funktion. cancel_order gibt bei bereits versandten Bestellungen einen Grund und eine Alternative zurück, statt zu stornieren – denn ein Prompt „storniere nie versandte Bestellungen“ ist ein Vorschlag, eine Funktion, die ablehnt, ist es nicht.

Die Tool-Call-Schleife handhaben

Vier Schritte, und die Reihenfolge ist wichtiger, als sie wirkt. Das Modell sendet response.function_call_arguments.done, dein Code führt die Funktion aus, du sendest das Ergebnis als function_call_output -Item zurück und erst dann bittest du das Modell weiterzumachen.

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),
        },
    }))

Braucht das Modell für eine Anfrage mehrere Tools, feuert es mehrere function_call_arguments.done -Events, bevor Audio spielt. Löse sie alle auf und sende jedes Ergebnis, bevor du ein einziges response.create schickst. Zu früh gesendet, antwortet das Modell ohne Kontext der noch laufenden Calls.

Ein dokumentierter Stolperstein, den ich trotzdem zuerst traf: response.create genau in dem Moment zu schicken, in dem dein Tool-Result rausgeht, kann sich mit dem Einleitungssatz überlappen, den der Agent noch spielt. In einem Lauf begann er mit „Ich prüfe sofort den Status der Bestellung ORD-1042“ und rief das Tool mitten im Satz – eine sofortige Antwort hätte sich selbst überlagert.

Warte, bis das Audio der aktuellen Runde fertig ist, und zeige dazwischen einen kurzen „Denke nach“-Status.

Tool-Call-Flow vor der Fortsetzung der Antwort. Bild: Autor.

Unterbrechungen und Gesprächsstatus managen

Zwei getrennte Probleme: Ein Anrufer redet dem Agenten ins Wort. Und ein WebSocket fällt ab und muss wieder aufgenommen werden.

Natürliche Unterbrechungen unterstützen

Mit server_vad ist Barge-in serverseitig automatisch: Sobald es erkennt, dass der Anrufer wieder spricht, signalisiert es input_audio_buffer.speech_started und stoppt die alte Antwort. Deine Aufgabe ist der Client-Teil dieses Handshakes: vorhandenes, bereits gequeutes Audio leeren, damit der Agent verstummt statt einen Satz zu Ende zu sprechen, den niemand mehr hören will.

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

Für manuelle, nicht-VAD-Sessions erledigt response.cancel dasselbe auf Zuruf. Es gibt auch conversation.item.truncate zum Kürzen eines Assistant-Items auf den tatsächlich gehörten Teil. Die Doku bestätigt, dass es existiert, aber nicht, wann man es bei einem Live-Barge-in feuert – teste das Timing selbst.

Getestet habe ich das mit einer Lieferanweisung, die mitten in der Bestätigung geändert wird: Anfrage starten, während der Agent bestätigt, mit einer anderen Adresse unterbrechen. Wichtig ist, ob der Agent die korrigierte Anweisung anwendet statt leise die alte zu beenden – nicht nur, ob das Audio stoppte. Prüfe gegen den Order-Datensatz, nicht gegen die Stille. Die Browser-Demo am Ende lässt dich das hören.

Eine getrennte Session fortsetzen

Sitzungsfortsetzung ist opt-in – und keine Erinnerung. Setze resumption.enabled: true in session.update, greife die ID aus dem Event conversation.created ab und, wenn der Socket abbricht, verbinde dich neu mit ?conversation_id=<id> in der URL und opt-in erneut auf der neuen Verbindung.

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

Die gecachten Runden, Transkripte, Toolaufrufe und Toolergebnisse spielen vor deiner nächsten Frage wieder ein, und der Cache verschwindet nach 30 Minuten Inaktivität. Ich habe gefragt, Verbindung getrennt und für die Nachfrage ohne Wiederholung neu verbunden; der Agent hat die ETA korrekt wieder aufgegriffen.

Ein undokumentierter Haken: Das Replay landet nicht instant. Eine Frage, die du im Moment des Socket-Öffnens sendest, kann es schlagen und ohne Erinnerung an die vorherige Runde zurückkommen. Gib ihm eine Sekunde, bevor du Resumption beschuldigst.

Terminal-Log einer fortgesetzten Session. Bild: Autor.

Nutze das nicht anstelle eigener Speicherung des Bestellstatus. Wenn der Cache abläuft oder der Anrufer morgen wiederklingelt, startest du ohne Kontext – so vorgesehen.

Agent absichern und überwachen

Lege niemals einen permanenten API-Schlüssel in Browser- oder Mobile-Code. Wenn ein Client direkt statt über deinen Server verbindet, stelle ein kurzlebiges Token aus:

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": ...}

Ein Browser kann beim WebSocket-Handshake keinen eigenen Authorization -Header setzen. Er übergibt das Token daher im sec-websocket-protocol-Header, mit dem Präfix xai-client-secret..

Server stellt Token aus, Browser tritt dem Call bei. Bild: Autor.

Abrechnung läuft über zwei Zähler. Audio – gesendet oder empfangen – kostet die erwähnten $0,08 pro Minute, also $4,80 pro Stunde. Und jedes conversation.item.create, das weder Audio noch ein function_call_output ist, kostet pauschal $0,004. response.create wird nicht berechnet. Jedes response.done trägt ein usage-Objekt, das in meinem Test output_audio_seconds und separat billable_audio_seconds meldete. Rechne mit diesen, nicht mit Schätzungen.

Dokumentierte Limits der Speech-to-Speech-API sind 10 gleichzeitige Sessions pro Team und 120 Minuten Session-Cap, beides in us-east-1. Plane Kapazitäten nicht nach den Zahlen der Voice Agent API – die sind anders.

Zur Privatsphäre: Sei präzise. SpaceXAI's Security-FAQ sagt, API-Requests und -Responses werden 30 Tage verschlüsselt zur Missbrauchsüberwachung aufbewahrt und ohne Zustimmung nicht zum Training genutzt. Teams können Zero Data Retention aktivieren – dadurch fällt jedoch die persistierte Voice-Agent-Verlaufsspeicherung weg und Resumption funktioniert nicht.

Wenn du offenlegst, dass ein Anruf aufgezeichnet wird oder KI-gesteuert ist, ist das genau der Job der Erweiterung force_message. Die Zeile wird exakt so abgespielt, statt vom Modell paraphrasiert.

Den Voice-Agenten testen

Ein 200-Status beim WebSocket-Handshake sagt nichts darüber, ob der Agent das Richtige getan hat. Teste das Ergebnis, nicht nur die Verbindung.

• Ein sauberer Bestelllookup – die gesprochene Antwort gegen den Datensatz prüfen, nicht nur, dass etwas kam
• Eine unterbrochene Antwort – Playback stoppt und der Agent geht auf die neue Anfrage ein
• Ein Lieferupdate mit Bestätigungspflicht – gegen den Order-Datensatz prüfen
• Eine Ablehnung – z. B. Stornierung einer versandten Bestellung – bei der der Agent die Regel erklärt statt sich zu entschuldigen
• Eine unbekannte Bestellnummer – der Agent muss das sagen statt einen Status zu erfinden
• Ein Tool, das einen Fehler zurückgibt – der Agent spricht ihn aus statt zu hängen
• Reconnect und Resumption – inkl. des Replay-Fensters von oben
• Laute Umgebung, schnelles Sprechen und ein Anrufer, der Zahlen/Adressen buchstabiert

Ich habe das meiste mit Live-Key während des Schreibens getestet. Die interessanten Fehler waren Verhaltensfehler, keine Errors: das Resumption-Timing oben und ein VAD-Threshold außerhalb des Bereichs, der akzeptiert statt abgelehnt wurde – genau die Sorte, die still kaputt ausliefert, wenn du nur den Happy Path testest. Ergänze einen mehrsprachigen Test und sieh ins FAQ für eine Besonderheit bei der Sprachnennung.

Zwei davon kannst du nicht tippend testen. app_streamlit.py ist eine Streamlit-Seite für den Live-Call im Browser: Das Mikrofon streamt über WebRTC in denselben WebSocket, die Agentenstimme streamt zurück und der Socket bleibt offen.

streamlit run app_streamlit.py