Grok Voice Agent Builder: een praktische gids in Python

xAI heeft Voice Agent Builder uitgebracht, een console om spraakagenten te maken. Je beschrijft de bel-flow, koppelt documenten en tools, en kiest een stem.

Als ik een spraakagent-console test, let ik minder op de launch-notes en meer op de onderdelen die ik in code moet aansluiten: hoe de WebSocket-sessie is geconfigureerd, hoe audio beweegt, waar tool-calls plaatsvinden, wat een gesprek kost en hoe een andere app de workflow zou aanroepen.

De code hieronder bouwt die flow direct na tegen de Voice Agent API. We gebruiken specifiek een assistent voor kliniekafspraken die beschikbaarheid controleert, met stem antwoordt, kosten bijhoudt, toolfouten afhandelt en een FastAPI-endpoint aanbiedt.

Wat is Grok Voice Agent Builder?

Voice Agent Builder is xAI's console voor het maken en uitrollen van spraakagenten op Grok Voice. Het lanceerde in beta op 1 juli 2026. In plaats van aparte speech-to-text-, taalmodel- en text-to-speech-services te gebruiken, volgt het één voice-modelpad.

De console bevat telefonie, documentopvraging, tools en connectors, guardrails, externe MCP-servers en belfLogs met opnames, transcripties en traces.

Audio wordt per minuut gefactureerd. De console is nog beta, dus we gebruiken direct de API.

Hoe de Grok Voice Agent API onder de Builder werkt

Onder de console zit de Voice Agent API, een realtime WebSocket-API die dezelfde runtime blootlegt als de Builder.

De Builder draait boven op de Voice API. Afbeelding door auteur.

Het model dat hier wordt gebruikt is grok-voice-think-fast-1.0. Het grok-voice-latest-alias wijst naar het nieuwste model. Ik gebruik dat hier, maar voor een uitgerolde app zou ik de versienaam vastzetten. xAI rapporteert een score van 67,3% voor dit model op het τ-voice Bench-leaderbord; ik zie dat als één datapunt, geen garantie.

Compatibiliteitsnotitie: de API is compatibel met de OpenAI Realtime API. Als je code al met OpenAI's realtime-endpoint praat, verander je meestal alleen de basis-URL en de sleutel.

Projectoverzicht: wat we gaan bouwen

De kliniekassistent neemt gesproken input aan, antwoordt in een gegenereerde stem, stelt vervolgvragen, controleert beschikbaarheid vóórdat er een tijdslot wordt aangeboden en draagt over aan een mens wanneer nodig. Het kernvoorbeeld gebruikt één tool; de Streamlit-demo voegt acties toe voor boeken, doorverbinden en gesprek beëindigen.

De kern-tutorial splitst in vier bestanden, elk met één taak:

• voice_client.py bevat de WebSocket-client, audiohelpers en kostentracking

• tools.py bevat check_availability, plus extra demotools gebruikt door Streamlit

• assistant.py bevat de system prompt, sessieconfig en de workflow

• app.py serveert het geheel via FastAPI

Die vier bestanden vormen het pad door het artikel. De repo bevat ook app_streamlit.py voor de visuele demo en run.py als Windows-starter, maar daar komen we op terug zodra de kernflow werkt.

Vereisten

Voordat de code draait, heb je Python 3.10 of nieuwer nodig, een xAI-account, een API-sleutel van console.x.ai, prepaid tegoed, en basiscomfort met omgevingsvariabelen, JSON en WebSockets.

Het project opzetten

Maak een map en een virtual environment, en installeer dan de pakketten:

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

Pin deze pakketten in een requirements.txt zodat een verse checkout dezelfde setup gebruikt.

Maak een .env-bestand naast de Python-bestanden:

XAI_API_KEY=xai-your-key-here

Voeg .env toe aan .gitignore. De API-sleutel hoort op de server te blijven.

De Voice Agent bouwen

Laten we beginnen met bouwen.

Verbinden met de Grok Voice Agent API via WebSocket

De eerste stap is de verbinding openen. Geef het model door als queryparameter en je sleutel als bearer token bij de 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())

Met een live sleutel is het eerste event dat je ziet session.created, wat betekent dat de socket open is en klaar om te configureren.

Het event "Session created" bevestigt de verbinding. Afbeelding door auteur.

De spraaksessie configureren

Een live socket is nog geen geconfigureerde agent. Je vormt die door een session.update -event te sturen met een session-object.

Stem, audioformaat en instructies

De drie instellingen die je het vaakst aanraakt zijn de stem, het audioformaat en de system prompt. De realtime-API biedt vijf benoemde stemmen, eve, ara, rex, sal, en leo, plus elke eigen kloon. Audio is standaard audio/pcm op 24000 Hz, met input en output apart geconfigureerd.

Hier is de sessieconfig die de assistent gebruikt, samengesteld 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],
    }

Het veld instructions is de system prompt. Deze kliniekprompt blijft kort omdat lange stemantwoorden lastig te volgen zijn:

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.

De escalatieregel houdt de kliniekagent weg van medisch advies. De laatste twee regels houden hem bij de scope en voorkomen lussen wanneer de beller onduidelijk is. De config voegt ook de datum van vandaag toe omdat het model in mijn live tests het verkeerde jaar kon gokken bij datums als "6 juli".

Turn-detectie afstellen

Turn-detectie bepaalt hoe de agent beslist dat jij bent uitgepraat. Zet turn_detection.type op server_vad en de server beëindigt de beurt bij stilte. Laat het op null en je stuurt beurten door de audiobuffer te committen; dat is wat ik voor de bestandsflow gebruik.

Server-VAD heeft drie nuttige instellingen: threshold bepaalt hoe hard audio moet zijn om als spraak te tellen, silence_duration_ms bepaalt hoe lang een pauze de beurt beëindigt, en prefix_padding_ms houdt een beetje audio vóór spraak vast. Als je agent mensen onderbreekt, verhoog dan eerst silence_duration_ms.

Audio naar de agent sturen

Nu sturen we de stem van de beller. De audio moet overeenkomen met het sessieformaat: mono 16-bit PCM op 24000 Hz, gecodeerd als base64 en in chunks verzonden.

De client streamt het bestand in stukken en commit daarna de buffer om het einde van de beurt te markeren:

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)

Als je sample rate of codering niet overeenkomt met session.update, krijg je mogelijk ruis of stilte in plaats van een duidelijke fout. Audio gaat via input_audio_buffer.append, dus het wordt op duur gefactureerd in plaats van per bericht.

Stemantwoorden ontvangen

Nadat je een antwoord aanvraagt, komt audio binnen als response.output_audio.delta, de transcriptie als response.output_audio_transcript.delta, en response.done sluit de beurt af.

De client verzamelt dat alles in één async-lus:

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

Decodeer de audiodelta's, voeg ze in volgorde samen en schrijf het resultaat naar een response.wav-bestand. Om de woorden van de beller zelf vast te leggen, stel je audio.input.transcription in en lees je conversation.item.input_audio_transcription.completed.

De workflow voor de afsprakenassistent bouwen

Nu worden de onderdelen een gesprek: boekingsverzoek, verduidelijkende vraag, beschikbaarheidscheck, aangeboden slots, bevestiging. Om context over beurten mee te nemen, maakt elke nieuwe beurt opnieuw verbinding met het conversatie-ID en kiest voor sessiehervatting.

Tool-calling toevoegen aan de spraakagent

Voor de kliniek moet de agent beschikbaarheid controleren vóórdat er een tijd wordt beloofd. Aangepaste tools zijn hoe het model jouw code aanroept: het zendt een verzoek uit, jouw applicatie voert de functie uit en jij stuurt het resultaat terug.

De tool is een gewone functie plus een JSON-schema dat in de sessieconfig gaat. Hier is het schema uit 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"],
    },
}

De lus heeft een vaste vorm. Wanneer het model de tool wil, stuurt het response.function_call_arguments.done met de argumenten. Jij draait de functie, retourneert een function_call_output en stuurt daarna response.create zodat de agent verder kan. Mis je die laatste response.create, dan valt de agent stil.

De roundtrip van de tool-call uitgelegd. Afbeelding door auteur.

Aangepaste functies zoals deze draaien in jouw code. De Streamlit-demo registreert er nog drie uit hetzelfde bestand: book_appointment, transfer_to_human, en end_call. Ingebouwde tools, zoals websearch, X-zoekopdracht, collections-search en externe MCP-tools, draaien op de servers van xAI.

Tool-failures afhandelen

Tools falen, en een spraakagent die succes veronderstelt kan een slot beloven dat niet bestaat. Mijn ToolRegistry.execute gooit nooit een exceptie: een mislukte lookup komt terug als een {"error": ...}-dict.

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

Een expliciete foutstatus voorkomt dat de agent mislukte tool-calls als succes behandelt.

Kosten bijhouden

Voordat je dit aan iemand serveert, wil je weten wat een gesprek kost. Audio wordt gefactureerd tegen $0,05 per minuut, voor zowel wat je verstuurt als wat je ontvangt. Tekstinputevenementen kosten $0,004 per stuk. function_call_output-resultaten en response.create-events worden niet gefactureerd.

De client houdt het gaandeweg bij, dus kosten zijn een eigenschap die je op elk moment kunt uitlezen:

@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

Een door xAI verstrekt nummer voegt de telefonietoeslag van $0,01 per minuut toe, die de helper toepast wanneer je telephony=True zet. Tools die door xAI worden gehost, worden apart gefactureerd: websearch en X-search kosten ongeveer $5 per duizend calls, en bestandszoekopdracht circa $2,50.

Fouten en randgevallen afhandelen

De meeste problemen vallen in een korte lijst:

• Ontbrekende of ongeldige API-sleutel geeft 401 bij de handshake, dus controleer eerst de sleutel

• Een geblokkeerd team geeft 403, en een rate limit geeft 429; die herprobeer je met backoff

• Ongeldige sessieconfig geeft 400, meestal een typfout in een veldnaam

• Niet-ondersteund audioformaat geeft ruis, geen fout, dus match de sessierate

• Een ontbrekende response.create na een toolresultaat laat de agent hangen

• Een dubbele boekingspoging kan echte problemen veroorzaken, dus herprobeer niet blindelings

Een mislukte read zoals check_availability herproberen is veilig, maar een mislukte write zoals een echte boeking herhalen kan een dubbele reservering veroorzaken. Elke actie die data wijzigt heeft eerst een idempotency-check nodig.

Ephemeral tokens gebruiken voor client-apps

Tot nu toe gaan we ervan uit dat de code op jouw server draait, waar de API-sleutel hoort. Als een browser- of mobiele app direct verbindt, gebruik dan ephemeral tokens.

Je server roept POST https://api.x.ai/v1/realtime/client_secrets aan met je sleutel, krijgt een tokenresponse terug en geeft de tokenwaarde door aan de client. In mijn run bevatte de response value en 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()

Browsers kunnen geen aangepaste WebSocket-headers zetten, dus het token gaat mee in de sec-websocket-protocol -header met een xai-client-secret.-prefix.

De workflow omzetten in een FastAPI-endpoint

Een endpoint laat een frontend of andere service de workflow aanroepen. De route valideert de request body met een Pydantic-model, neemt een getypte boodschap of een audiopad aan en retourneert de transcriptie, antwoordaudio, toollog, latency en geschatte kosten.

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

Draai het met uvicorn app:app --reload en open http://localhost:8000/docs. Lees XAI_API_KEY uit de serveromgeving en accepteer die nooit via een request body.