Grok Voice Agent Builder : guide pratique en Python

xAI a lancé Voice Agent Builder, une console pour créer des agents vocaux. Vous décrivez le parcours d’appel, joignez des documents et des outils, et choisissez une voix.

Quand j’évalue une console d’agent vocal, je m’intéresse moins à la note de lancement qu’aux parties à raccorder au code : configuration de la session WebSocket, circulation de l’audio, emplacement des appels d’outils, coût de l’appel et façon dont un autre service peut déclencher le workflow.

Le code ci-dessous reconstruit ce flux directement via l’API Voice Agent. Nous allons créer un assistant de prise de rendez-vous en clinique qui vérifie les disponibilités, répond à l’oral, suit les coûts, gère les échecs d’outils et expose un endpoint FastAPI.

Qu’est-ce que Grok Voice Agent Builder ?

Voice Agent Builder est la console d’xAI pour créer et déployer des agents vocaux sur Grok Voice. Le lancement en bêta a eu lieu le 1er juillet 2026. Au lieu d’assembler des services séparés de reconnaissance vocale, LLM et synthèse vocale, il propose une chaîne unifiée autour d’un modèle vocal.

La console inclut la téléphonie, la recherche documentaire, des outils et connecteurs, des garde-fous, des serveurs MCP distants, ainsi que des journaux d’appels avec enregistrements, transcriptions et traces.

La facturation audio se fait à la minute. La console étant encore en bêta, nous utilisons directement l’API.

Comment fonctionne l’API Grok Voice Agent sous le Builder

Sous la console se trouve l’API Voice Agent, une API WebSocket temps réel qui expose le même runtime que le Builder.

Le Builder repose sur l’API Voice. Image de l’auteur.

Le modèle utilisé ici est grok-voice-think-fast-1.0. L’alias grok-voice-latest pointe vers le modèle le plus récent. Je l’utilise ici, mais pour une application en production je figerais la version. xAI annonce un score de 67,3 % pour ce modèle sur le tableau de bord τ-voice Bench ; je le prends comme un indicateur, pas comme une garantie.

Note de compatibilité : l’API est compatible avec l’API Realtime d’OpenAI. Si votre code cible déjà l’endpoint realtime d’OpenAI, vous changez surtout l’URL de base et la clé.

Aperçu du projet : ce que nous allons construire

L’assistant de clinique prend une entrée vocale, répond avec une voix synthétisée, pose des questions de clarification, vérifie la disponibilité avant de proposer un créneau et transfère à un humain si nécessaire. L’exemple central utilise un seul outil ; la démo Streamlit ajoute les actions de réservation, transfert et fin d’appel.

Le tutoriel se découpe en quatre fichiers, chacun dédié à une tâche :

• voice_client.py : client WebSocket, utilitaires audio et suivi des coûts

• tools.py contient check_availability, plus des outils de démo supplémentaires pour Streamlit

• assistant.py : prompt système, configuration de session et workflow

• app.py sert l’ensemble via FastAPI

Ces quatre fichiers structurent l’article. Le dépôt inclut aussi app_streamlit.py pour la démo visuelle et run.py comme lanceur Windows ; nous y reviendrons une fois le flux de base opérationnel.

Prérequis

Avant d’exécuter le code, prévoyez Python 3.10 ou plus récent, un compte xAI, une clé API depuis console.x.ai, des crédits prépayés, et des bases sur les variables d’environnement, JSON et WebSockets.

Mise en place du projet

Créez un dossier et un environnement virtuel, puis installez les paquets :

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

Figez ces dépendances dans un requirements.txt pour reproduire l’environnement à l’identique.

Créez un fichier .env à côté des fichiers Python :

XAI_API_KEY=xai-your-key-here

Ajoutez .env à .gitignore. La clé API doit rester côté serveur.

Construire l’agent vocal

Commençons la construction.

Connexion à l’API Grok Voice Agent via WebSocket

Première étape : ouvrir la connexion. Passez le modèle en paramètre de requête et votre clé en jeton bearer lors du 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())

Avec une clé valide, le premier événement est session.created, confirmant que le socket est ouvert et prêt à être configuré.

L’événement de création de session confirme la connexion. Image de l’auteur.

Configuration de la session vocale

Un socket actif n’est pas encore un agent configuré. Vous le façonnez en envoyant un événement session.update avec un objet session.

Voix, format audio et instructions

Les trois réglages les plus courants sont la voix, le format audio et le prompt système. L’API temps réel propose cinq voix nommées, eve, ara, rex, sal et leo, plus d’éventuels clones personnalisés. Par défaut, l’audio est en audio/pcm à 24 000 Hz, avec des réglages séparés pour l’entrée et la sortie.

Voici la configuration de session utilisée par l’assistant, assemblée dans 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],
    }

Le champ instructions est le prompt système. Celui de la clinique reste concis : de longues réponses vocales s’écoutent mal.

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 ligne d’escalade évite que l’agent sorte de son rôle et donne des conseils médicaux. Les deux dernières lignes le gardent dans le périmètre et évitent les boucles quand l’appelant n’est pas clair. La config ajoute aussi la date du jour car, lors de tests réels, le modèle pouvait se tromper d’année pour des dates du type « 6 juillet ».

Ajuster la détection des tours

La détection de tour détermine quand l’agent considère que vous avez fini de parler. Définissez turn_detection.type sur server_vad pour que le serveur termine le tour au silence. Laissez-le à null et vous contrôlez les tours en validant le tampon audio, ce que j’utilise pour le flux basé sur fichier.

Le VAD serveur a trois réglages utiles : threshold définit le niveau minimal pour compter comme parole, silence_duration_ms la durée de pause qui termine le tour, et prefix_padding_ms conserve un léger préambule avant le début de la parole. Si votre agent coupe la parole, augmentez d’abord silence_duration_ms.

Envoyer de l’audio à l’agent

Nous envoyons maintenant la voix de l’appelant. L’audio doit respecter le format de session : mono PCM 16 bits à 24 000 Hz, encodé en base64 et transmis par morceaux.

Le client diffuse le fichier par tranches, puis valide le tampon pour marquer la fin du tour :

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)

Si votre fréquence d’échantillonnage ou l’encodage ne correspondent pas à session.update, vous risquez d’obtenir des grésillements ou du silence plutôt qu’une erreur claire. L’audio passe par input_audio_buffer.append, donc la facturation se fait à la durée et non au message.

Recevoir les réponses vocales

Après avoir demandé une réponse, l’audio arrive via response.output_audio.delta, la transcription via response.output_audio_transcript.delta, et response.done clôt le tour.

Le client agrège tout cela dans une boucle async unique :

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

Décodez les deltas audio, concaténez-les dans l’ordre et écrivez le résultat dans un fichier response.wav. Pour capturer les paroles de l’appelant, activez audio.input.transcription et lisez conversation.item.input_audio_transcription.completed.

Construction du workflow de prise de rendez-vous

Nous transformons désormais ces briques en conversation : demande de réservation, question de clarification, vérification de disponibilité, proposition de créneaux, confirmation. Pour garder le contexte entre les tours, chaque nouveau tour se reconnecte avec l’ID de conversation et opte pour la reprise de session.

Ajouter l’appel d’outils à l’agent vocal

Dans la clinique, l’agent doit vérifier la disponibilité avant d’annoncer un horaire. Les outils personnalisés permettent au modèle d’atteindre votre code : il émet une requête, votre application exécute la fonction, puis vous renvoyez le résultat.

L’outil est une simple fonction accompagnée d’un schéma JSON inséré dans la configuration de session. Voici le schéma depuis 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"],
    },
}

La boucle suit une forme fixe. Quand le modèle veut l’outil, il envoie response.function_call_arguments.done avec les arguments. Vous exécutez la fonction, retournez un function_call_output, puis envoyez response.create pour que l’agent poursuive. Si vous oubliez ce dernier response.create, l’agent se tait.

L’aller-retour d’un appel d’outil, expliqué. Image de l’auteur.

Les fonctions personnalisées s’exécutent dans votre code. La démo Streamlit en enregistre trois autres depuis le même fichier : book_appointment, transfer_to_human et end_call. Les outils intégrés, comme la recherche web, la recherche sur X, la recherche dans les collections et les outils MCP distants, s’exécutent sur les serveurs d’xAI.

Gérer les échecs d’outils

Les outils échouent, et un agent vocal qui suppose la réussite peut promettre un créneau inexistant. Mon ToolRegistry.execute ne lève jamais d’exception : une recherche en échec renvoie un dictionnaire {"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)}

Un état d’erreur explicite évite à l’agent de traiter un appel d’outil échoué comme une réussite.

Ajouter le suivi des coûts

Avant toute mise à disposition, sachez combien coûte un appel. L’audio est facturé 0,05 $ par minute, en comptant ce que vous envoyez et ce que vous recevez. Les événements d’entrée texte sont facturés 0,004 $ pièce. Les résultats function_call_output et les événements response.create ne sont pas facturés.

Le client suit tout en temps réel, donc le coût est disponible à tout moment :

@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 numéro téléphonique fourni par xAI ajoute une surtaxe de 0,01 $ par minute, appliquée par l’utilitaire quand vous définissez telephony=True. Les outils hébergés chez xAI sont facturés séparément : la recherche web et la recherche X valent environ 5 $ pour mille appels, et la recherche de fichiers environ 2,50 $.

Gérer les erreurs et cas limites

La plupart des échecs se rangent dans une courte liste :

• Clé API manquante ou invalide : 401 au handshake ; vérifiez la clé en premier

• Équipe bloquée : 403 ; limitation de débit : 429, à réessayer avec backoff

• Configuration de session mal formée : 400, souvent une faute dans un nom de champ

• Format audio non pris en charge : grésillements au lieu d’une erreur ; alignez la fréquence de session

• Oubli de response.create après un résultat d’outil : l’agent reste bloqué

• Tentative de double réservation : ne relancez pas aveuglément

Relancer une lecture comme check_availability est sans risque, mais relancer une écriture (une vraie réservation) peut créer un doublon. Toute action qui modifie des données doit être protégée par un contrôle d’idempotence.

Utiliser des jetons éphémères pour les apps clientes

Jusqu’ici, on suppose que le code tourne sur votre serveur, là où la clé API doit rester. Si un navigateur ou une app mobile se connecte directement, utilisez des jetons éphémères.

Votre serveur appelle POST https://api.x.ai/v1/realtime/client_secrets avec votre clé, récupère un jeton, puis transmet la valeur au client. Dans mon test, la réponse contenait value et 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()

Les navigateurs ne peuvent pas définir d’en-têtes WebSocket personnalisés ; le jeton transite donc dans l’en-tête sec-websocket-protocol avec le préfixe xai-client-secret..

Transformer le workflow en endpoint FastAPI

Un endpoint permet à un frontend ou un autre service d’appeler le workflow. La route valide le corps de requête avec un modèle Pydantic, accepte un message texte typé ou un chemin audio, et renvoie la transcription, l’audio de réponse, le journal des outils, la latence et le coût estimé.

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

Lancez avec uvicorn app:app --reload et ouvrez http://localhost:8000/docs. Lisez XAI_API_KEY depuis l’environnement serveur et ne l’acceptez jamais dans le corps d’une requête.