Grok Voice Agent Builder: una guida pratica in Python

xAI ha rilasciato Voice Agent Builder, una console per creare agenti vocali. Descrivi il flusso di chiamata, alleghi documenti e strumenti, e scegli una voce.

Quando provo una console per agenti vocali, mi interessa meno l’annuncio di lancio e più le parti da collegare al codice: come si configura la sessione WebSocket, come si muove l’audio, dove avvengono le chiamate agli strumenti, quanto costa la chiamata e come un’altra app richiamerebbe il workflow.

Il codice qui sotto ricostruisce quel flusso direttamente contro la Voice Agent API. In particolare, useremo un assistente per appuntamenti in clinica che controlla la disponibilità, risponde a voce, traccia i costi, gestisce gli errori degli strumenti ed espone un endpoint FastAPI.

Che cos’è Grok Voice Agent Builder?

Voice Agent Builder è la console di xAI per creare e distribuire agenti vocali su Grok Voice. È stata lanciata in beta il 1° luglio 2026. Invece di usare servizi separati di speech to text, modello linguistico e text to speech, utilizza un unico percorso di modello vocale.

La console include telefonia, recupero documenti, strumenti e connettori, guardrail, server MCP remoti e log delle chiamate con registrazioni, trascrizioni e tracce.

L’audio è fatturato al minuto. La console è ancora in beta, quindi usiamo direttamente l’API.

Come funziona la Grok Voice Agent API sotto il Builder

Sotto la console c’è la Voice Agent API, una API realtime via WebSocket che espone lo stesso runtime usato dal Builder.

Il Builder è sopra la Voice API. Immagine dell’Autore.

Il modello usato qui è grok-voice-think-fast-1.0. L’alias grok-voice-latest punta al modello più recente. Lo uso qui, ma per un’app in produzione fisserei il nome versionato. xAI riporta un punteggio del 67,3% per questo modello nella classifica τ-voice Bench; lo considero un dato tra i tanti, non una garanzia.

Nota di compatibilità: l’API è compatibile con la OpenAI Realtime API. Se hai codice che parla con l’endpoint realtime di OpenAI, nella maggior parte dei casi cambi solo la base URL e la chiave.

Panoramica del progetto: cosa costruiremo

L’assistente di clinica riceve input parlati, risponde con una voce generata, fa domande di approfondimento, controlla la disponibilità prima di proporre una fascia oraria e passa a un umano quando serve. L’esempio principale usa uno strumento; la demo Streamlit aggiunge azioni di prenotazione, trasferimento e chiusura chiamata.

Il tutorial principale è suddiviso in quattro file, ognuno con un compito:

• voice_client.py contiene il client WebSocket, gli helper audio e il tracciamento dei costi

• tools.py contiene check_availability, più altri strumenti demo usati da Streamlit

• assistant.py contiene il prompt di sistema, la configurazione di sessione e il workflow

• app.py serve il tutto tramite FastAPI

Questi quattro file sono il filo conduttore dell’articolo. Il repository include anche app_streamlit.py per la demo visuale e run.py come launcher Windows, ma ci torneremo dopo che il flusso principale funziona.

Prerequisiti

Prima di eseguire il codice, ti servono Python 3.10 o successivo, un account xAI, una chiave API da console.x.ai, crediti prepagati e un po’ di dimestichezza con variabili d’ambiente, JSON e WebSocket.

Configurazione del progetto

Crea una cartella e un ambiente virtuale, quindi installa i pacchetti:

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

Fissa questi pacchetti in un requirements.txt così un nuovo checkout usa la stessa configurazione.

Crea un file .env accanto ai file Python:

XAI_API_KEY=xai-your-key-here

Aggiungi .env a .gitignore. La chiave API dovrebbe restare sul server.

Costruire il Voice Agent

Iniziamo a costruire.

Connessione alla Grok Voice Agent API via WebSocket

Il primo passo è aprire la connessione. Passa il modello come parametro di query e la tua chiave come bearer token nell’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())

Con una chiave attiva, il primo evento che vedi è session.created, che significa che il socket è aperto e pronto per essere configurato.

L’evento di creazione sessione conferma la connessione. Immagine dell’Autore.

Configurare la sessione vocale

Un socket attivo non è un agente configurato. Lo modelli inviando un evento session.update con un oggetto session.

Voce, formato audio e istruzioni

Le tre impostazioni che tocchi più spesso sono la voce, il formato audio e il prompt di sistema. L’API realtime espone cinque voci nominate, eve, ara, rex, sal, e leo, oltre a eventuali cloni personalizzati. L’audio è di default audio/pcm a 24000 Hz, con input e output configurati separatamente.

Ecco la configurazione di sessione usata dall’assistente, assemblata 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],
    }

Il campo instructions è il prompt di sistema. Questo prompt per la clinica resta breve perché le risposte vocali lunghe sono difficili da seguire:

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.

La riga sull’escalation evita che l’agente della clinica dia consigli medici. Le ultime due righe lo mantengono in ambito e impediscono loop quando il chiamante non è chiaro. La configurazione aggiunge anche la data di oggi perché, nei miei test live, il modello poteva indovinare l’anno sbagliato per date come "6 luglio".

Regolare la rilevazione dei turni

La rilevazione dei turni determina come l’agente decide che hai smesso di parlare. Imposta turn_detection.type su server_vad e il server termina il turno al silenzio. Lascialo null e controlli tu i turni impegnando il buffer audio, che è quello che uso per il flusso da file.

Il VAD del server ha tre impostazioni da conoscere: threshold imposta quanto deve essere forte l’audio per contare come parlato, silence_duration_ms imposta quanto deve durare una pausa per terminare il turno, e prefix_padding_ms mantiene un po’ di audio prima dell’inizio del parlato. Se il tuo agente interrompe le persone, alza prima silence_duration_ms.

Inviare audio all’agente

Ora inviamo la voce del chiamante. L’audio deve corrispondere al formato di sessione: mono PCM 16 bit a 24000 Hz, codificato in base64 e inviato a chunk.

Il client esegue lo streaming del file a fette, poi conferma il buffer per segnare la fine del turno:

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)

Se il tuo sample rate o la codifica non corrispondono a session.update, potresti ottenere fruscii o silenzio invece di un errore chiaro. L’audio passa tramite input_audio_buffer.append, quindi viene fatturato per durata e non per messaggio.

Ricevere le risposte vocali

Dopo che hai richiesto una risposta, l’audio arriva come response.output_audio.delta, la trascrizione arriva come response.output_audio_transcript.delta e response.done chiude il turno.

Il client raccoglie tutto in un unico loop async:

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

Decodifica i delta audio, concatenali in ordine e scrivi il risultato in un file response.wav. Per catturare le parole del chiamante, imposta audio.input.transcription e leggi conversation.item.input_audio_transcription.completed.

Costruire il workflow dell’assistente per gli appuntamenti

Ora i pezzi diventano una conversazione: richiesta di prenotazione, domanda di chiarimento, controllo disponibilità, slot proposti, conferma. Per mantenere il contesto tra i turni, ogni nuovo turno si ricollega con l’id della conversazione e opta per la ripresa di sessione.

Aggiungere le chiamate agli strumenti all’agente vocale

Per la clinica, l’agente deve controllare la disponibilità prima di promettere un orario. Gli strumenti personalizzati sono il modo in cui il modello raggiunge il tuo codice: emette una richiesta, la tua applicazione esegue la funzione e tu rimandi il risultato.

Lo strumento è una semplice funzione più uno schema JSON che va nella configurazione di sessione. Ecco lo schema da 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"],
    },
}

Il loop ha una forma fissa. Quando il modello vuole lo strumento, invia response.function_call_arguments.done con gli argomenti. Tu esegui la funzione, restituisci un function_call_output e poi invii response.create così l’agente può continuare. Se ti dimentichi l’ultimo response.create, l’agente resta in silenzio.

Il giro di chiamata allo strumento spiegato. Immagine dell’Autore.

Funzioni personalizzate come questa girano nel tuo codice. La demo Streamlit ne registra altre tre dallo stesso file: book_appointment, transfer_to_human, e end_call. Gli strumenti integrati, come la ricerca web, la ricerca su X, la ricerca nelle raccolte e gli strumenti MCP remoti, vengono eseguiti sui server di xAI.

Gestire i guasti degli strumenti

Gli strumenti falliscono, e un agente vocale che presume il successo può promettere uno slot che non esiste. Il mio ToolRegistry.execute non solleva mai eccezioni: una ricerca non riuscita torna come un dict {"error": ...}.

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

Uno stato di errore esplicito impedisce all’agente di trattare le chiamate a strumenti fallite come successi.

Aggiungere il tracciamento dei costi

Prima di servirlo a chiunque, sappi quanto costa una chiamata. L’audio è fatturato a $0,05 al minuto, contando sia ciò che invii sia ciò che ricevi. Gli eventi di input testuale sono fatturati a $0,004 ciascuno. I risultati di function_call_output e gli eventi response.create non sono fatturati.

Il client lo traccia in corso d’opera, quindi il costo è una proprietà che puoi leggere in qualsiasi momento:

@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

Un numero fornito da xAI aggiunge il sovrapprezzo di telefonia di $0,01 al minuto, che l’helper applica quando imposti telephony=True. Gli strumenti ospitati da xAI sono fatturati separatamente: la ricerca web e la ricerca su X costano circa $5 per mille chiamate, e la ricerca sui file circa $2,50.

Gestire errori e casi limite

La maggior parte dei problemi rientra in un breve elenco:

• Chiave API mancante o non valida restituisce 401 all’handshake, quindi verifica prima la chiave

• Un team bloccato restituisce 403, e un rate limit restituisce 429, che ritenti con backoff

• Configurazione di sessione malformata restituisce 400, di solito un refuso nel nome di un campo

• Formato audio non supportato produce fruscii, non un errore, quindi allinea il sample rate di sessione

• Un response.create mancante dopo un risultato dello strumento lascia l’agente in sospeso

• Un tentativo di doppia prenotazione può creare veri problemi, quindi non ritentare alla cieca

Ritentare una lettura fallita come check_availability è sicuro, ma ritentare una scrittura fallita come una prenotazione reale può creare doppie prenotazioni. Qualsiasi azione che modifica i dati necessita prima di un controllo di idempotenza.

Usare token effimeri per app client

Finora abbiamo assunto che il codice giri sul tuo server, dove appartiene la chiave API. Se un browser o un’app mobile si connette direttamente, usa token effimeri.

Il tuo server chiama POST https://api.x.ai/v1/realtime/client_secrets con la tua chiave, riceve una risposta con il token e passa il valore del token al client. Nella mia esecuzione, la risposta includeva value e 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()

I browser non possono impostare header WebSocket personalizzati, quindi il token viaggia nell’header sec-websocket-protocol con un prefisso xai-client-secret..

Trasformare il workflow in un endpoint FastAPI

Un endpoint permette a un frontend o a un altro servizio di richiamare il workflow. La route valida il body della richiesta con un modello Pydantic, accetta un messaggio tipizzato o un percorso audio e restituisce trascrizione, audio di risposta, log degli strumenti, latenza e costo stimato.

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

Eseguilo con uvicorn app:app --reload e apri http://localhost:8000/docs. Leggi XAI_API_KEY dall’ambiente del server e non accettarlo mai dal body di una richiesta.