Grok Voice Agent Builder: en praktisk guide i Python

xAI släppte Voice Agent Builder, en konsol för att skapa röstagenter. Du beskriver samtalsflödet, bifogar dokument och verktyg och väljer en röst.

När jag testar en röstagent-konsol bryr jag mig mindre om lanseringsnotisen och mer om delarna jag måste koppla in i kod: hur WebSocket-sessionen konfigureras, hur ljud rör sig, var verktygsanrop sker, vad samtalet kostar och hur en annan app skulle anropa arbetsflödet.

Koden nedan återskapar det flödet direkt mot Voice Agent API. Vi använder specifikt en klinikbokningsassistent som kontrollerar tillgänglighet, svarar med röst, spårar kostnad, hanterar verktygsfel och exponerar en FastAPI-endpoint.

Vad är Grok Voice Agent Builder?

Voice Agent Builder är xAI:s konsol för att skapa och driftsätta röstagenter på Grok Voice. Den lanserades i beta den 1 juli 2026. I stället för att använda separata tjänster för tal-till-text, språkmodell och text-till-tal, använder den en enda röstmodellkedja.

Konsolen inkluderar telefoni, dokumenthämtning, verktyg och anslutningar, skyddsräcken, fjärrstyrda MCP-servrar samt samtalsloggar med inspelningar, transkript och spårningar.

Ljud debiteras per minut. Konsolen är fortfarande beta, så vi använder API:et direkt.

Hur Grok Voice Agent API fungerar under Builder

Under konsolen finns Voice Agent API, ett WebSocket-API i realtid som exponerar samma runtime som Builder använder.

Builder ligger ovanpå Voice API. Bild av författaren.

Modellen som används här är grok-voice-think-fast-1.0. Aliaset grok-voice-latest pekar på den nyaste modellen. Jag använder det här, men för en driftsatt app skulle jag låsa till det versionssatta namnet. xAI rapporterar 67,3 % i poäng för den här modellen på τ-voice Bench-topplistan; jag ser det som en datapunkt, inte en garanti.

Kompatibilitetsnotis: API:et är kompatibelt med OpenAI Realtime API. Om du har kod som pratar med OpenAIs realtids-endpoint, ändrar du mestadels bas-URL:en och nyckeln.

Projektöversikt: Vad vi ska bygga

Klinikassistenten tar emot talat input, svarar med en genererad röst, ställer följdfrågor, kontrollerar tillgänglighet innan den erbjuder en tid och kopplar till en människa vid behov. Grundexemplet använder ett verktyg; Streamlit-demot lägger till åtgärder för bokning, överlämning och att avsluta samtal.

Grundtutorialen delas upp i fyra filer, var och en med en uppgift:

• voice_client.py innehåller WebSocket-klienten, ljudhjälpare och kostnadsspårning

• tools.py innehåller check_availability, plus extra demoverktyg som används av Streamlit

• assistant.py innehåller systemprompten, sessionskonfigurationen och arbetsflödet

• app.py serverar allt via FastAPI

Dessa fyra filer är stigen genom artikeln. Repon innehåller också app_streamlit.py för det visuella demot och run.py som en Windows-startare, men vi återkommer till dem när grundflödet fungerar.

Förkunskaper

Innan koden körs behöver du Python 3.10 eller senare, ett xAI-konto, en API-nyckel från console.x.ai, förbetalda krediter och grundläggande vana vid miljövariabler, JSON och WebSockets.

Ställa in projektet

Skapa en mapp och en virtuell miljö och installera sedan paketen:

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

Lås dessa paket i en requirements.txt så att en ny utcheckning använder samma uppsättning.

Skapa en .env-fil bredvid Python-filerna:

XAI_API_KEY=xai-your-key-here

Lägg till .env i .gitignore. API-nyckeln ska stanna på servern.

Bygga röstagenten

Låt oss börja bygga.

Ansluta till Grok Voice Agent API via WebSocket

Första steget är att öppna anslutningen. Skicka modellen som en query-parameter och din nyckel som en bearer-token vid handskakningen:

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

Mot en giltig nyckel är den första händelsen du ser session.created, vilket betyder att socketen är öppen och redo att konfigureras.

Händelsen session.created bekräftar anslutningen. Bild av författaren.

Konfigurera röstsessionen

En levande socket är inte en konfigurerad agent. Du formar den genom att skicka en session.update -händelse med ett session-objekt.

Röst, ljudformat och instruktioner

De tre inställningar du oftast rör är rösten, ljudformatet och systemprompten. Realtids-API:et exponerar fem namngivna röster, eve, ara, rex, sal, och leo, plus eventuella egna kloner. Ljud standardiseras till audio/pcm vid 24000 Hz, där in- och utdata konfigureras separat.

Här är sessionskonfigurationen som assistenten använder, sammansatt i 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],
    }

Fältet instructions är systemprompten. Den här klinikprompten hålls kort eftersom långa röstssvar är svåra att följa:

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.

Raden om eskalering håller klinikagenten borta från medicinsk rådgivning. De två sista raderna håller den inom ämnet och stoppar loopar när uppringaren är oklar. Konfigurationen lägger också till dagens datum eftersom modellen i mina live-tester kunde gissa fel år för datum som ”6 juli”.

Justera turdetektering

Turdetektering är hur agenten avgör att du har slutat prata. Sätt turn_detection.type till server_vad så avslutar servern turen vid tystnad. Lämna den som null och du styr turer genom att committa ljudbufferten, vilket är vad jag använder för filflödet.

Server-VAD har tre inställningar värda att känna till: threshold anger hur högt ljud måste vara för att räknas som tal, silence_duration_ms anger hur lång paus som avslutar turen, och prefix_padding_ms behåller lite ljud före talstart. Om din agent avbryter folk, höj silence_duration_ms först.

Skicka ljud till agenten

Nu skickar vi uppringarens röst. Ljudet måste matcha sessionsformatet: mono 16-bitars PCM vid 24000 Hz, kodat som base64 och skickat i bitar.

Klienten strömmar filen i skivor och committar sedan bufferten för att markera slutet på turen:

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)

Om din samplingsfrekvens eller kodning inte matchar session.update kan du få brus eller tystnad i stället för ett tydligt fel. Ljud går via input_audio_buffer.append, så det debiteras per varaktighet i stället för per meddelande.

Ta emot röstsvaret

Efter att du begärt ett svar kommer ljud som response.output_audio.delta, transkriptet kommer som response.output_audio_transcript.delta, och response.done stänger turen.

Klienten samlar allt detta i en async-loop:

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

Avkoda ljuddelarna, lägg till dem i ordning och skriv resultatet till en response.wav-fil. För att fånga uppringarens egna ord sätter du audio.input.transcription och läser conversation.item.input_audio_transcription.completed.

Bygga arbetsflödet för bokningsassistenten

Nu blir delarna till en konversation: bokningsförfrågan, förtydligande fråga, tillgänglighetskontroll, erbjudna tider, bekräftelse. För att bära kontext över turer återansluter varje ny tur med konversations-ID:t och väljer sessionsåterupptagning.

Lägga till verktygsanrop i röstagenten

För kliniken måste agenten kontrollera tillgänglighet innan den lovar en tid. Anpassade verktyg är hur modellen når din kod: den skickar en begäran, din applikation kör funktionen och du skickar tillbaka resultatet.

Verktyget är en vanlig funktion plus ett JSON-schema som går in i sessionskonfigurationen. Här är schemat från 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"],
    },
}

Loopen har en fast form. När modellen vill använda verktyget skickar den response.function_call_arguments.done med argumenten. Du kör funktionen, returnerar ett function_call_output och skickar sedan response.create så att agenten kan fortsätta. Missar du det sista response.create blir agenten tyst.

Varvet för verktygsanrop förklarat. Bild av författaren.

Anpassade funktioner som denna körs i din kod. Streamlit-demot registrerar tre till från samma fil: book_appointment, transfer_to_human, och end_call. Inbyggda verktyg, såsom webbsökning, X-sökning, collections-sökning och fjärr-MCP-verktyg, körs på xAI:s servrar.

Hantering av verktygsfel

Verktyg misslyckas, och en röstagent som förutsätter framgång kan utlova en tid som inte finns. Min ToolRegistry.execute kastar aldrig undantag: en misslyckad sökning kommer tillbaka som en {"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)}

Ett explicit feltilstånd hindrar agenten från att behandla misslyckade verktygsanrop som lyckade.

Lägga till kostnadsspårning

Innan du serverar detta till någon, ta reda på vad ett samtal kostar. Ljud debiteras $0,05 per minut, både för det du skickar och det du tar emot. Textinmatningshändelser debiteras $0,004 styck. function_call_output-resultat och response.create-händelser debiteras inte.

Klienten spårar det löpande, så kostnaden är en egenskap du kan läsa när som helst:

@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

Ett xAI-provisionerat nummer lägger till ett telefonipåslag på $0,01 per minut, vilket hjälpfunktionen tillämpar när du sätter telephony=True. Verktyg som hostas av xAI debiteras separat: webbsökning och X-sökning ligger runt $5 per tusen anrop, och filsökning runt $2,50.

Hantering av fel och kantfall

De flesta fel faller i en kort lista:

• Saknad eller ogiltig API-nyckel returnerar 401 vid handskakningen, så kontrollera nyckeln först

• Ett blockerat team returnerar 403, och en frekvensbegränsning returnerar 429, vilket du försöker igen med backoff

• Felaktig sessionskonfiguration returnerar 400, oftast ett stavfel i ett fältnamn

• Ett icke-stött ljudformat ger brus, inte ett fel, så matcha sessionsfrekvensen

• Ett saknat response.create efter ett verktygsresultat lämnar agenten hängande

• Ett dubbelt bokningsförsök kan ställa till verkliga problem, så försök inte igen på måfå

Att försöka igen en misslyckad läsning som check_availability är säkert, men att försöka igen en misslyckad skrivning som en faktisk bokning kan dubbelboka en uppringare. Alla åtgärder som ändrar data behöver först en idempotenskontroll.

Använda kortlivade token för klientappar

Allt hittills förutsätter att koden körs på din server, där API-nyckeln hör hemma. Om en webbläsare eller mobilapp ansluter direkt, använd kortlivade token.

Din server anropar POST https://api.x.ai/v1/realtime/client_secrets med din nyckel, får tillbaka ett token-svar och skickar token-värdet till klienten. I min körning inkluderade svaret value och 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()

Webbläsare kan inte sätta egna WebSocket-rubriker, så token skickas i headern sec-websocket-protocol med prefixet xai-client-secret..

Göra om arbetsflödet till en FastAPI-endpoint

En endpoint låter ett frontend eller en annan tjänst anropa arbetsflödet. Routingen validerar begärans body med en Pydantic-modell, tar emot ett typat meddelande eller en ljudsökväg och returnerar transkript, svars-ljud, verktygslogg, latens och uppskattad kostnad.

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

Kör den med uvicorn app:app --reload och öppna http://localhost:8000/docs. Läs XAI_API_KEY från serverns miljö och acceptera den aldrig från en request-body.