Grok Voice Agent Builder: guía práctica en Python

xAI ha lanzado Voice Agent Builder, una consola para crear agentes de voz. Describes el flujo de la llamada, adjuntas documentos y herramientas, y eliges una voz.

Al probar una consola de agentes de voz, me importa menos la nota de lanzamiento y más las piezas que tengo que conectar en código: cómo se configura la sesión WebSocket, cómo viaja el audio, dónde se hacen las llamadas a herramientas, cuánto cuesta la llamada y cómo otra app invocaría el flujo.

El código de abajo reconstruye ese flujo directamente contra la API de Voice Agent. En concreto, usaremos un asistente de citas de una clínica que comprueba disponibilidad, responde por voz, controla el coste, maneja fallos de herramientas y expone un endpoint en FastAPI.

¿Qué es Grok Voice Agent Builder?

Voice Agent Builder es la consola de xAI para crear y desplegar agentes de voz en Grok Voice. Se lanzó en beta el 1 de julio de 2026. En vez de usar servicios separados de reconocimiento de voz, modelo de lenguaje y síntesis, utiliza una única ruta de modelo de voz.

La consola incluye telefonía, recuperación de documentos, herramientas y conectores, guardarraíles, servidores MCP remotos y registros de llamadas con grabaciones, transcripciones y trazas.

El audio se factura por minuto. Como la consola sigue en beta, aquí usamos la API directamente.

Cómo funciona la API de Grok Voice Agent bajo el Builder

Bajo la consola está la Voice Agent API, una API WebSocket en tiempo real que expone el mismo runtime que usa el Builder.

El Builder se apoya en la Voice API. Imagen del autor.

El modelo usado aquí es grok-voice-think-fast-1.0. El alias grok-voice-latest apunta al modelo más reciente. Lo uso aquí, pero para una app en producción fijaría la versión concreta. xAI informa de una puntuación del 67,3% para este modelo en el ranking τ-voice Bench; lo tomo como un dato más, no como garantía.

Nota de compatibilidad: la API es compatible con la OpenAI Realtime API. Si ya tienes código que habla con el endpoint realtime de OpenAI, básicamente cambias la URL base y la clave.

Resumen del proyecto: qué vamos a construir

El asistente de la clínica recibe voz, responde con una voz generada, hace preguntas de seguimiento, comprueba disponibilidad antes de ofrecer una franja y deriva a una persona cuando hace falta. El ejemplo base usa una herramienta; la demo en Streamlit añade acciones de reserva, transferencia y fin de llamada.

El tutorial base se divide en cuatro archivos, cada uno con una función:

• voice_client.py contiene el cliente WebSocket, utilidades de audio y control de costes

• tools.py incluye check_availability y otras herramientas de demo usadas por Streamlit

• assistant.py define el prompt del sistema, la configuración de sesión y el workflow

• app.py sirve todo a través de FastAPI

Esos cuatro archivos marcan el camino del artículo. El repo también incluye app_streamlit.py para la demo visual y run.py como lanzador de Windows, pero volveremos a ellos cuando funcione el flujo principal.

Requisitos previos

Antes de ejecutar el código, necesitas Python 3.10 o superior, una cuenta de xAI, una clave de API desde console.x.ai, créditos prepago y soltura básica con variables de entorno, JSON y WebSockets.

Configuración del proyecto

Crea una carpeta y un entorno virtual, luego instala los paquetes:

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

Fija estas dependencias en un requirements.txt para que un clon nuevo use la misma configuración.

Crea un archivo .env junto a los archivos de Python:

XAI_API_KEY=xai-your-key-here

Añade .env a .gitignore. La clave de API debe quedarse en el servidor.

Creando el agente de voz

Vamos a ponernos manos a la obra.

Conexión a la API de Grok Voice Agent por WebSocket

El primer paso es abrir la conexión. Pasa el modelo como parámetro de consulta y tu clave como bearer token en el 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 clave válida, el primer evento que verás es session.created, que confirma que el socket está abierto y listo para configurar.

El evento de creación de sesión confirma la conexión. Imagen del autor.

Configuración de la sesión de voz

Un socket activo no es un agente configurado. Lo defines enviando un evento session.update con un objeto session.

Voz, formato de audio e instrucciones

Los tres ajustes que más tocarás son la voz, el formato de audio y el prompt del sistema. La API en tiempo real expone cinco voces con nombre, eve, ara, rex, sal, y leo, además de cualquier clon personalizado. Por defecto el audio es audio/pcm a 24000 Hz, con entrada y salida configuradas por separado.

Esta es la configuración de sesión que usa el asistente, montada en 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],
    }

El campo instructions es el prompt del sistema. Este prompt de la clínica se mantiene corto porque las respuestas largas por voz son difíciles de seguir:

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 línea de escalado evita que el agente de la clínica dé consejos médicos. Las dos últimas líneas lo mantienen dentro de alcance y evitan bucles cuando el interlocutor no se entiende. La configuración también añade la fecha de hoy porque, en pruebas reales, el modelo podía adivinar mal el año para fechas como "6 de julio".

Ajuste de la detección de turnos

La detección de turnos decide cuándo has dejado de hablar. Configura turn_detection.type como server_vad y el servidor cierra el turno al detectar silencio. Déjalo en null y controlarás los turnos confirmando el búfer de audio, que es lo que uso para el flujo con archivos.

El VAD del servidor tiene tres ajustes clave: threshold marca cuán alto debe sonar para contar como voz, silence_duration_ms define cuánto dura la pausa que cierra el turno y prefix_padding_ms conserva un poco de audio antes de que empiece la voz. Si tu agente interrumpe a la gente, sube primero silence_duration_ms.

Envío de audio al agente

Ahora enviamos la voz de quien llama. El audio debe coincidir con el formato de sesión: mono PCM de 16 bits a 24000 Hz, codificado en base64 y enviado por fragmentos.

El cliente transmite el archivo por trozos y luego confirma el búfer para marcar el fin 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)

Si tu frecuencia de muestreo o codificación no coincide con session.update, puedes obtener ruido o silencio en lugar de un error claro. El audio pasa por input_audio_buffer.append, así que se factura por duración y no por mensaje.

Recepción de respuestas por voz

Tras solicitar una respuesta, el audio llega como response.output_audio.delta, la transcripción como response.output_audio_transcript.delta y response.done cierra el turno.

El cliente recopila todo eso en un único bucle 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

Descodifica los deltas de audio, únelos en orden y escribe el resultado en un archivo response.wav. Para capturar las palabras del interlocutor, activa audio.input.transcription y lee conversation.item.input_audio_transcription.completed.

Construyendo el workflow del asistente de citas

Ahora las piezas se convierten en una conversación: petición de reserva, pregunta aclaratoria, comprobación de disponibilidad, franjas ofrecidas, confirmación. Para mantener el contexto entre turnos, cada turno nuevo reconecta con el id de conversación y activa la reanudación de sesión.

Añadiendo llamadas a herramientas al agente de voz

En la clínica, el agente debe comprobar disponibilidad antes de prometer una hora. Las herramientas personalizadas son cómo el modelo llega a tu código: emite una solicitud, tu aplicación ejecuta la función y envías el resultado de vuelta.

La herramienta es una función simple más un esquema JSON que se añade a la configuración de sesión. Este es el esquema de 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"],
    },
}

El bucle tiene una forma fija. Cuando el modelo quiere la herramienta, envía response.function_call_arguments.done con los argumentos. Ejecutas la función, devuelves un function_call_output y luego envías response.create para que el agente continúe. Si te olvidas de ese response.create final, el agente se queda en silencio.

El viaje de ida y vuelta de la herramienta, explicado. Imagen del autor.

Funciones personalizadas como esta se ejecutan en tu código. La demo de Streamlit registra tres más del mismo archivo: book_appointment, transfer_to_human, y end_call. Las herramientas integradas, como búsqueda web, búsqueda en X, búsqueda en collections y herramientas MCP remotas, se ejecutan en los servidores de xAI.

Gestión de fallos en herramientas

Las herramientas fallan, y un agente de voz que asume éxito puede prometer una franja que no existe. Mi ToolRegistry.execute nunca lanza excepciones: una consulta fallida vuelve como un diccionario {"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 estado de error explícito evita que el agente trate llamadas fallidas como si hubieran tenido éxito.

Control de costes

Antes de servir esto a nadie, entiende cuánto cuesta una llamada. El audio se factura a 0,05 $ por minuto, contando tanto lo que envías como lo que recibes. Los eventos de entrada de texto se facturan a 0,004 $ cada uno. Los resultados de function_call_output y los eventos response.create no se facturan.

El cliente lo va sumando sobre la marcha, así que el coste es una propiedad que puedes consultar en cualquier 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 número de telefonía aprovisionado por xAI añade un recargo de 0,01 $ por minuto, que el helper aplica cuando estableces telephony=True. Las herramientas alojadas por xAI se facturan aparte: la búsqueda web y en X rondan 5 $ por mil llamadas, y la búsqueda de archivos unos 2,50 $.

Gestión de errores y casos límite

La mayoría de fallos caen en una lista corta:

• Clave de API ausente o inválida devuelve 401 en el handshake: compruébala primero

• Un equipo bloqueado devuelve 403, y un límite de tasa devuelve 429, que debes reintentar con backoff

• Configuración de sesión mal formada devuelve 400, normalmente por una errata en un campo

• Formato de audio no soportado produce ruido, no un error, así que iguala la frecuencia de sesión

• Falta de response.create tras un resultado de herramienta deja el agente colgado

• Un intento duplicado de reserva puede causar problemas reales, así que no reintentes a ciegas

Reintentar una lectura fallida como check_availability es seguro, pero reintentar una escritura fallida como una reserva real puede duplicarla. Cualquier acción que cambie datos necesita primero una comprobación de idempotencia.

Uso de tokens efímeros para apps cliente

Hasta ahora hemos asumido que el código se ejecuta en tu servidor, donde debe estar la clave de API. Si un navegador o app móvil se conecta directamente, usa tokens efímeros.

Tu servidor llama a POST https://api.x.ai/v1/realtime/client_secrets con tu clave, recibe una respuesta con el token y pasa el valor del token al cliente. En mi ejecución, la respuesta incluía value y 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()

Los navegadores no pueden establecer cabeceras WebSocket personalizadas, así que el token viaja en la cabecera sec-websocket-protocol con el prefijo xai-client-secret..

Convertir el workflow en un endpoint de FastAPI

Un endpoint permite que un frontend u otro servicio invoque el workflow. La ruta valida el cuerpo de la solicitud con un modelo de Pydantic, acepta un mensaje de texto o una ruta a audio y devuelve la transcripción, el audio de respuesta, el log de herramientas, la latencia y el coste estimado.

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

Arráncalo con uvicorn app:app --reload y abre http://localhost:8000/docs. Lee XAI_API_KEY del entorno del servidor y nunca la aceptes desde el cuerpo de una petición.