Grok Voice Agent Builder: Ein praxisnaher Guide in Python

xAI hat den Voice Agent Builder veröffentlicht – eine Konsole zum Erstellen von Sprachagenten. Du beschreibst den Call-Flow, hängst Dokumente und Tools an und wählst eine Stimme.

Wenn ich eine Voice-Agent-Konsole teste, interessieren mich weniger die Launch-Notizen und mehr die Teile, die ich ins Codegerüst einbinden muss: wie die WebSocket-Session konfiguriert wird, wie Audio fließt, wo Toolaufrufe passieren, was der Anruf kostet und wie ein anderes System den Workflow anstößt.

Der folgende Code bildet diesen Flow direkt gegen die Voice Agent API nach. Konkret bauen wir einen Assistenten für Kliniktermine, der Verfügbarkeiten prüft, per Sprache antwortet, Kosten trackt, Toolfehler abfängt und einen FastAPI-Endpoint bereitstellt.

Was ist der Grok Voice Agent Builder?

Der Voice Agent Builder ist xAIs Konsole zum Erstellen und Ausrollen von Sprachagenten auf Grok Voice. Der Beta-Start war am 1. Juli 2026. Statt getrennte Dienste für Speech-to-Text, Sprachmodell und Text-to-Speech zu nutzen, läuft alles über einen Voice-Model-Pfad.

Die Konsole umfasst Telefonie, Dokumentenabfrage, Tools und Connectoren, Guardrails, Remote-MCP-Server sowie Call-Logs mit Aufzeichnungen, Transkripten und Traces.

Audio wird pro Minute abgerechnet. Da die Konsole noch Beta ist, nutzen wir direkt die API.

So funktioniert die Grok Voice Agent API unter dem Builder

Unter der Konsole liegt die Voice Agent API, eine Realtime-WebSocket-API, die dasselbe Runtime-System wie der Builder bereitstellt.

Der Builder sitzt auf der Voice API. Bild: Autor.

Das hier genutzte Modell ist grok-voice-think-fast-1.0. Der grok-voice-latest Alias zeigt auf das neueste Modell. Ich nutze ihn hier, aber für eine produktive App würde ich die versionierte Bezeichnung pinnen. xAI meldet für dieses Modell 67,3% auf dem τ-voice Bench-Leaderboard; das ist für mich ein Datenpunkt, keine Garantie.

Kompatibilitätshinweis: Die API ist kompatibel mit der OpenAI Realtime API. Wenn dein Code bereits mit OpenAIs Realtime-Endpoint spricht, änderst du im Wesentlichen Basis-URL und Key.

Projektüberblick: Was wir bauen

Der Klinikassistent nimmt gesprochene Eingaben entgegen, antwortet mit generierter Stimme, stellt Rückfragen, prüft Verfügbarkeit, bevor er Slots anbietet, und übergibt bei Bedarf an Menschen. Das Kernbeispiel nutzt ein Tool; die Streamlit-Demo ergänzt Buchen, Weiterleiten und Gespräch beenden.

Das Tutorial teilt sich in vier Dateien, jede mit einer klaren Aufgabe:

• voice_client.py enthält den WebSocket-Client, Audio-Helfer und Kostentracking

• tools.py enthält check_availability plus zusätzliche Demo-Tools für Streamlit

• assistant.py enthält den Systemprompt, die Session-Konfiguration und den Workflow

• app.py stellt alles per FastAPI bereit

Diese vier Dateien führen durch den Artikel. Im Repo liegen außerdem app_streamlit.py für die visuelle Demo und run.py als Windows-Starter. Zu denen kommen wir zurück, sobald der Kernflow läuft.

Voraussetzungen

Bevor der Code läuft, brauchst du Python 3.10 oder neuer, ein xAI-Konto, einen API-Key aus console.x.ai, Prepaid-Guthaben und Grundkenntnisse in Umgebungsvariablen, JSON und WebSockets.

Projekt einrichten

Erstelle einen Ordner und eine virtuelle Umgebung und installiere dann die Pakete:

mkdir appointment-agent
cd appointment-agent
python -m venv .venv
.venv\Scripts\activate       # macOS/Linux: source .venv/bin/activate
pip install websockets python-dotenv fastapi uvicorn pydantic httpx numpy streamlit

Fixiere diese Pakete in einer requirements.txt, damit ein frischer Checkout dieselbe Umgebung nutzt.

Lege eine .env Datei neben den Python-Dateien an:

XAI_API_KEY=xai-your-key-here

Füge .env zu .gitignore hinzu. Der API-Key gehört auf den Server.

Den Voice-Agent bauen

Los geht's.

Verbindung zur Grok Voice Agent API per WebSocket

Der erste Schritt ist das Öffnen der Verbindung. Übergebe das Modell als Query-Parameter und deinen Key als Bearer-Token beim Handshake:

import asyncio
import json
import os
import websockets

async def voice_agent():
    url = "wss://api.x.ai/v1/realtime?model=grok-voice-latest"
    async with websockets.connect(
        url,
        additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
    ) as ws:
        async for message in ws:
            print(json.loads(message)["type"])

asyncio.run(voice_agent())

Mit einem gültigen Key ist das erste Event session.created – die Verbindung steht und kann konfiguriert werden.

Das Event „Session created“ bestätigt die Verbindung. Bild: Autor.

Die Voice-Session konfigurieren

Eine offene Socket-Verbindung ist noch kein konfigurierter Agent. Du formst ihn, indem du ein session.update Event mit einem session Objekt sendest.

Stimme, Audioformat und Anweisungen

Die drei häufigsten Einstellungen sind Stimme, Audioformat und Systemprompt. Die Realtime-API bietet fünf benannte Stimmen, eve, ara, rex, sal und leo, plus eigene Klone. Audio ist standardmäßig audio/pcm mit 24000 Hz, Ein- und Ausgabe separat konfiguriert.

Hier ist die Session-Konfiguration des Assistenten, zusammengebaut in assistant.py:

def build_session_config(voice="ara", instructions=SYSTEM_PROMPT, sample_rate=24000):
    # The model needs to know "today" or it guesses the year for a date like "July 6th".
    instructions = f"{instructions}\nToday's date is {date.today().isoformat()}."
    return {
        "voice": voice,
        "instructions": instructions,
        "turn_detection": None,  # manual turns for file-based input
        "audio": {
            "input": {"format": {"type": "audio/pcm", "rate": sample_rate}},
            "output": {"format": {"type": "audio/pcm", "rate": sample_rate}},
        },
        "tools": [CHECK_AVAILABILITY_TOOL],
    }

Das Feld instructions ist der Systemprompt. Dieser Klinik-Prompt bleibt kurz, weil lange Sprachantworten am Telefon schwer zu folgen sind:

You are a voice appointment assistant for a small clinic. Help callers book,
reschedule, cancel, or ask questions about appointments, services, and hours.
Answer whatever the caller asks that relates to the clinic. Keep responses short
and natural for a phone conversation. Ask one question at a time. Confirm
important details before taking action. Use the availability tool before offering
a time slot. Escalate to a human for medical, urgent, sensitive, or unclear
requests. If a caller asks about something unrelated to the clinic, say briefly
that it is outside what you can help with, then steer back to booking. If you
cannot make out what the caller said, ask them to repeat it instead of repeating
your last message.

Die Eskalationszeile hält den Klinikagenten aus medizinischer Beratung heraus. Die letzten beiden Zeilen sorgen für Fokus und verhindern Schleifen, wenn der Anrufer unklar ist. Außerdem hängt die Config das heutige Datum an, weil das Modell in meinen Tests bei Datumsangaben wie „6. Juli“ das Jahr falsch raten konnte.

Turn-Detection feinjustieren

Turn-Detection legt fest, wann der Agent erkennt, dass du fertig gesprochen hast. Setze turn_detection.type auf server_vad, dann beendet der Server die Runde bei Stille. Lässt du es auf null, steuerst du die Runden, indem du den Audio-Puffer commitest – so mache ich es im Datei-Flow.

Server-VAD hat drei sinnvolle Settings: threshold legt fest, wie laut Audio sein muss, um als Sprache zu zählen, silence_duration_ms wie lang eine Pause sein darf, bis die Runde endet, und prefix_padding_ms behält etwas Audio vor Sprachbeginn. Wenn dein Agent Leute ins Wort fällt, erhöhe zuerst silence_duration_ms.

Audio an den Agenten senden

Jetzt schicken wir die Stimme des Anrufers. Das Audio muss zur Session passen: mono, 16 Bit PCM, 24000 Hz, als Base64 kodiert und in Chunks gesendet.

Der Client streamt die Datei in Scheiben und commitet den Puffer, um das Rundenende zu markieren:

async def send_audio(self, pcm_bytes, chunk_ms=100, commit=True):
    bytes_per_chunk = int(self._sample_rate * 2 * chunk_ms / 1000)
    for start in range(0, len(pcm_bytes), bytes_per_chunk):
        chunk = pcm_bytes[start:start + bytes_per_chunk]
        await self._t.send({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode(),
        })
    if commit:
        await self._t.send({"type": "input_audio_buffer.commit"})
    self.cost.audio_seconds += pcm_seconds(pcm_bytes, self._sample_rate)

Wenn Sample-Rate oder Kodierung nicht zu session.update passen, kommt statt einer klaren Fehlermeldung Rauschen oder Stille. Audio geht über input_audio_buffer.append – dadurch wird nach Dauer und nicht pro Nachricht abgerechnet.

Sprachausgaben empfangen

Nachdem du eine Antwort angefordert hast, kommt Audio als response.output_audio.delta, das Transkript als response.output_audio_transcript.delta, und response.done schließt die Runde ab.

Der Client sammelt alles in einer asynchronen Schleife:

async def _collect_response(self):
    audio = bytearray()
    transcript, calls = [], []
    while True:
        event = await self._recv()
        etype = event["type"]
        if etype == "response.output_audio.delta":
            audio += base64.b64decode(event["delta"])
        elif etype == "response.output_audio_transcript.delta":
            transcript.append(event.get("delta", ""))
        elif etype == "response.function_call_arguments.done":
            calls.append(event)
        elif etype == "response.done":
            break
    return bytes(audio), "".join(transcript), calls

Dekodiere die Audio-Deltas, hänge sie in Reihenfolge an und schreibe das Ergebnis in eine response.wav Datei. Um die Worte des Anrufers zu erfassen, setze audio.input.transcription und lies conversation.item.input_audio_transcription.completed aus.

Den Workflow des Terminassistenten bauen

Jetzt wird aus den Teilen ein Gespräch: Buchungswunsch, Rückfrage, Verfügbarkeitsprüfung, angebotene Slots, Bestätigung. Um Kontext über Runden zu halten, verbindet sich jede neue Runde mit der Conversation-ID und nutzt Session-Resumption.

Toolaufrufe zum Voice-Agent hinzufügen

In der Klinik muss der Agent vor einer Zusage die Verfügbarkeit prüfen. Eigene Tools sind die Brücke in deinen Code: Das Modell sendet eine Anfrage, deine Anwendung führt die Funktion aus und du schickst das Ergebnis zurück.

Das Tool ist eine normale Funktion plus ein JSON-Schema, das in die Session-Konfiguration kommt. Hier das Schema aus tools.py:

CHECK_AVAILABILITY_TOOL = {
    "type": "function",
    "name": "check_availability",
    "description": "Look up open appointment slots for a service on a given date. "
                   "Always call this before offering the caller a time.",
    "parameters": {
        "type": "object",
        "properties": {
            "service": {"type": "string", "description": "Service requested."},
            "date": {"type": "string", "description": "Requested date as YYYY-MM-DD."},
        },
        "required": ["service", "date"],
    },
}

Die Schleife hat immer dieselbe Form. Wenn das Modell das Tool will, sendet es response.function_call_arguments.done mit den Argumenten. Du führst die Funktion aus, lieferst ein function_call_output zurück und sendest dann response.create damit der Agent fortfahren kann. Fehlt dieses finale response.create, wird der Agent stumm.

Der Tool-Call Roundtrip erklärt. Bild: Autor.

Solche Custom-Funktionen laufen in deinem Code. Die Streamlit-Demo registriert drei weitere aus derselben Datei: book_appointment, transfer_to_human und end_call. Eingebaute Tools wie Websuche, X-Suche, Collections-Suche und Remote-MCP-Tools laufen auf xAIs Servern.

Toolfehler behandeln

Tools fallen aus, und ein Sprachagent, der Erfolg voraussetzt, kann Slots versprechen, die es nicht gibt. Mein ToolRegistry.execute wirft nie Fehler: Ein fehlgeschlagener Lookup kommt als {"error": ...}-Dict zurück.

def execute(self, name, arguments):
    handler = self._handlers.get(name)
    if handler is None:
        return {"error": f"unknown tool: {name}"}
    try:
        return handler(**arguments)
    except ToolError as exc:
        return {"error": str(exc)}

Ein expliziter Fehlerzustand verhindert, dass der Agent fehlgeschlagene Toolaufrufe als Erfolg wertet.

Kosten tracking hinzufügen

Bevor du das jemandem anbietest, solltest du wissen, was ein Anruf kostet. Audio wird mit 0,05 $ pro Minute abgerechnet, sowohl für gesendetes als auch empfangenes Audio. Text-Input-Events kosten 0,004 $ pro Stück. function_call_output Ergebnisse und response.create-Events sind kostenfrei.

Der Client trackt die Kosten laufend, du kannst sie jederzeit auslesen:

@property
def audio_usd(self):
    rate = 0.05 + (0.01 if self.telephony else 0.0)
    return self.audio_seconds / 60 * rate

@property
def total_usd(self):
    return self.audio_usd + self.text_usd + self.tool_usd

Eine von xAI bereitgestellte Nummer fügt 0,01 $ pro Minute Telefonieaufschlag hinzu, den der Helfer anwendet, wenn du telephony=True setzt. Von xAI gehostete Tools werden separat abgerechnet: Websuche und X-Suche liegen etwa bei 5 $ pro 1000 Aufrufen, Dateisuche bei ca. 2,50 $.

Fehler und Edge-Cases behandeln

Die meisten Fehler fallen in eine kurze Liste:

• Fehlender oder ungültiger API-Key gibt 401 beim Handshake – also zuerst den Key prüfen

• Gesperrtes Team liefert 403, Rate-Limit 429 – mit Backoff neu versuchen

• Fehlerhafte Session-Config gibt 400, meist ein Tippfehler im Feldnamen

• Nicht unterstütztes Audioformat erzeugt Rauschen statt Fehler – also Session-Rate matchen

• Ein fehlendes response.create nach einem Tool-Resultat lässt den Agenten hängen

• Ein doppelter Buchungsversuch kann echte Probleme machen – also nicht blind retryen

Ein fehlgeschlagener Read wie check_availability ist unkritisch zu wiederholen, ein fehlgeschriebener Write wie eine echte Buchung kann hingegen doppelt buchen. Jede Aktion mit Seiteneffekt braucht vorher eine Idempotenzprüfung.

Ephemere Tokens für Client-Apps nutzen

Bisher nehmen wir an, dass der Code auf deinem Server läuft – dort gehört der API-Key hin. Wenn ein Browser oder eine Mobile-App direkt verbindet, nutze ephemere Tokens.

Dein Server ruft POST https://api.x.ai/v1/realtime/client_secrets mit deinem Key auf, erhält eine Token-Antwort und gibt den Token-Wert an den Client weiter. In meinem Lauf enthielt die Antwort value und expires_at:

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            CLIENT_SECRETS_URL,
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()

Browser können keine eigenen WebSocket-Header setzen, daher wird der Token im Header sec-websocket-protocol mit dem Präfix xai-client-secret. übergeben.

Den Workflow als FastAPI-Endpoint bereitstellen

Ein Endpoint erlaubt es einem Frontend oder Service, den Workflow aufzurufen. Die Route validiert den Request-Body mit einem Pydantic-Modell, nimmt eine getypte Nachricht oder einen Audiopfad entgegen und gibt Transkript, Antwortaudio, Tool-Log, Latenz und geschätzte Kosten zurück.

@app.post("/appointments/voice")
async def appointments_voice(body: VoiceRequest):
    fail = {"check_availability"} if body.simulate_tool_failure else None
    assistant = AppointmentAssistant(voice=body.voice, telephony=body.telephony, fail_tools=fail)
    if body.text:
        result = await assistant.run_live(text=body.text, conversation_id=body.conversation_id)
    else:
        pcm = load_wav_as_pcm(body.audio_path, 24000)
        result = await assistant.run_live(pcm, conversation_id=body.conversation_id)
    return {
        "transcript": result.transcript,
        "audio_wav_base64": base64.b64encode(encode_wav_bytes(result.audio, 24000)).decode(),
        "tool_calls": result.tool_calls,
        "latency_seconds": round(result.latency_s, 3),
        "estimated_cost_usd": round(result.cost.total_usd, 6),
        "audio_seconds": round(result.cost.audio_seconds, 2),
        "conversation_id": result.conversation_id,
    }

Starte mit uvicorn app:app --reload und öffne http://localhost:8000/docs. Lies XAI_API_KEY aus der Server-Umgebung und akzeptiere ihn niemals aus dem Request-Body.