Grok Voice Agent Builder: um guia prático em Python

A xAI lançou o Voice Agent Builder, um console para criar agentes de voz. Você descreve o fluxo da chamada, anexa documentos e ferramentas e escolhe uma voz.

Quando eu testo um console de agente de voz, me importo menos com o anúncio de lançamento e mais com o que preciso conectar no código: como a sessão WebSocket é configurada, como o áudio flui, onde ocorrem as chamadas de ferramentas, quanto a ligação custa e como outro app chamaria o workflow.

O código abaixo reconstrói esse fluxo diretamente contra a Voice Agent API. Especificamente, vamos usar um assistente de agendamento de clínica que verifica disponibilidade, responde por voz, acompanha custos, lida com falhas de ferramentas e expõe um endpoint FastAPI.

O que é o Grok Voice Agent Builder?

O Voice Agent Builder é o console da xAI para criar e implementar agentes de voz no Grok Voice. Foi lançado em beta em 1º de julho de 2026. Em vez de usar serviços separados de fala para texto, modelo de linguagem e texto para fala, ele usa um caminho único de modelo de voz.

O console inclui telefonia, recuperação de documentos, ferramentas e conectores, guardrails, servidores MCP remotos e logs de chamadas com gravações, transcrições e traces.

O áudio é cobrado por minuto. Como o console ainda está em beta, vamos usar a API diretamente.

Como a Grok Voice Agent API funciona por baixo do Builder

Por baixo do console está a Voice Agent API, uma API WebSocket em tempo real que expõe o mesmo runtime usado pelo Builder.

O Builder fica sobre a Voice API. Imagem do autor.

O modelo usado aqui é grok-voice-think-fast-1.0. O alias grok-voice-latest aponta para o modelo mais novo. Eu uso aqui, mas para um app em produção eu fixaria o nome versionado. A xAI reporta pontuação de 67,3% para este modelo no ranking τ-voice Bench; eu trato isso como um dado a mais, não uma garantia.

Nota de compatibilidade: a API é compatível com a OpenAI Realtime API. Se você já tem código que fala com o endpoint realtime da OpenAI, basicamente muda a URL base e a chave.

Visão geral do projeto: o que vamos construir

O assistente da clínica recebe entrada falada, responde com uma voz gerada, faz perguntas de seguimento, verifica a disponibilidade antes de oferecer um horário e transfere para um humano quando necessário. O exemplo central usa uma ferramenta; o demo em Streamlit adiciona ações de reservar, transferir e encerrar a chamada.

O tutorial principal se divide em quatro arquivos, cada um com uma função:

• voice_client.py guarda o cliente WebSocket, utilitários de áudio e o rastreamento de custos

• tools.py guarda check_availability, além de ferramentas extras do demo usado no Streamlit

• assistant.py guarda o prompt do sistema, a configuração da sessão e o workflow

• app.py serve tudo via FastAPI

Esses quatro arquivos são o caminho deste artigo. O repositório também inclui app_streamlit.py para o demo visual e run.py como um lançador no Windows, mas vamos voltar a eles depois que o fluxo central estiver funcionando.

Pré-requisitos

Antes de rodar o código, você precisa de Python 3.10 ou superior, uma conta na xAI, uma chave de API em console.x.ai, créditos pré-pagos e noções básicas de variáveis de ambiente, JSON e WebSockets.

Configurando o projeto

Crie uma pasta e um ambiente virtual e depois instale os pacotes:

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

Fixe esses pacotes em um requirements.txt para que um clone novo use a mesma configuração.

Crie um arquivo .env ao lado dos arquivos Python:

XAI_API_KEY=xai-your-key-here

Adicione .env ao .gitignore. A chave de API deve ficar no servidor.

Construindo o agente de voz

Vamos começar a construir.

Conectando à Grok Voice Agent API via WebSocket

O primeiro passo é abrir a conexão. Passe o modelo como parâmetro de query e sua chave como bearer token no 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())

Com uma chave ativa, o primeiro evento que você verá é session.created, o que significa que o socket abriu e está pronto para configurar.

O evento session created confirma a conexão. Imagem do autor.

Configurando a sessão de voz

Um socket ativo não é um agente configurado. Você o molda enviando um evento session.update com um objeto session.

Voz, formato de áudio e instruções

As três configurações mais usadas são a voz, o formato de áudio e o prompt do sistema. A API realtime expõe cinco vozes nomeadas, eve, ara, rex, sal, e leo, além de qualquer clone personalizado. O áudio padrão é audio/pcm a 24000 Hz, com entrada e saída configuradas separadamente.

Aqui está a configuração de sessão usada pelo assistente, montada em 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],
    }

O campo instructions é o prompt do sistema. Este prompt da clínica fica curto porque respostas longas de voz são difíceis de acompanhar:

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.

A linha de escalonamento mantém o agente da clínica fora de aconselhamento médico. As duas últimas mantêm o foco no escopo e evitam loops quando a fala do cliente não fica clara. A configuração também adiciona a data de hoje porque, nos meus testes, o modelo pode chutar o ano errado para datas como "6 de julho".

Ajustando a detecção de turnos

Detecção de turnos é como o agente decide que você parou de falar. Defina turn_detection.type como server_vad e o servidor encerra o turno no silêncio. Deixe como null e você controla os turnos comprometendo o buffer de áudio, que é o que uso para o fluxo por arquivo.

O VAD do servidor tem três ajustes úteis: threshold define quão alto o áudio precisa estar para contar como fala, silence_duration_ms define o tamanho da pausa que encerra o turno e prefix_padding_ms mantém um pouco de áudio antes do início da fala. Se o seu agente estiver interrompendo as pessoas, aumente primeiro o silence_duration_ms.

Enviando áudio para o agente

Agora vamos enviar a voz do cliente. O áudio deve combinar com o formato da sessão: mono PCM 16 bits a 24000 Hz, codificado em base64 e enviado em blocos.

O cliente faz streaming do arquivo em fatias e depois confirma o buffer para marcar o fim do 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 sua taxa de amostragem ou codificação não coincidir com a de session.update, você pode ouvir chiado ou silêncio em vez de um erro claro. O áudio passa por input_audio_buffer.append, então a cobrança é pela duração, não por mensagem.

Recebendo respostas em voz

Depois de solicitar uma resposta, o áudio chega como response.output_audio.delta, a transcrição chega como response.output_audio_transcript.delta e response.done fecha o turno.

O cliente reúne tudo isso em um único loop assíncrono:

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

Decodifique os deltas de áudio, junte-os em ordem e grave o resultado em um arquivo response.wav. Para capturar as palavras do próprio cliente, defina audio.input.transcription e leia conversation.item.input_audio_transcription.completed.

Construindo o workflow do assistente de agendamentos

Agora as peças viram uma conversa: pedido de agendamento, pergunta de esclarecimento, verificação de disponibilidade, horários oferecidos, confirmação. Para manter o contexto entre os turnos, cada turno novo reconecta com o id da conversa e opta por retomar a sessão.

Adicionando chamadas de ferramentas ao agente de voz

Para a clínica, o agente precisa verificar a disponibilidade antes de prometer um horário. Ferramentas personalizadas são como o modelo acessa seu código: ele emite um pedido, sua aplicação roda a função e você envia o resultado de volta.

A ferramenta é uma função simples mais um schema JSON que entra na configuração da sessão. Aqui está o schema 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"],
    },
}

O loop tem um formato fixo. Quando o modelo quer a ferramenta, ele envia response.function_call_arguments.done com os argumentos. Você roda a função, retorna um function_call_output e depois envia response.create para o agente continuar. Se esquecer esse response.create final, o agente fica em silêncio.

A viagem de ida e volta da chamada de ferramenta explicada. Imagem do autor.

Funções personalizadas como essa rodam no seu código. O demo em Streamlit registra mais três do mesmo arquivo: book_appointment, transfer_to_human, e end_call. Ferramentas nativas, como web search, X search, busca em collections e ferramentas MCP remotas (MCP), rodam nos servidores da xAI.

Lidando com falhas de ferramentas

Ferramentas falham, e um agente de voz que assume sucesso pode prometer um horário que não existe. Meu ToolRegistry.execute nunca lança exceção: uma consulta que falhou volta como um dicionário {"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)}

Um estado de erro explícito impede o agente de tratar chamadas de ferramenta com falha como sucesso.

Adicionando rastreamento de custos

Antes de servir isso para alguém, saiba quanto custa uma chamada. O áudio é cobrado a US$ 0,05 por minuto, contando tanto o que você envia quanto o que recebe. Eventos de entrada de texto são cobrados a US$ 0,004 cada. Resultados de function_call_output e eventos response.create não são cobrados.

O cliente acompanha isso em tempo real, então o custo é uma propriedade que você pode ler a qualquer 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

Um número provisionado pela xAI acrescenta a sobretaxa de telefonia de US$ 0,01 por minuto, que o helper aplica quando você define telephony=True. Ferramentas hospedadas pela xAI são cobradas separadamente: web search e X search ficam por volta de US$ 5 por mil chamadas, e busca em arquivos por cerca de US$ 2,50.

Tratando erros e casos-limite

A maioria das falhas cai em uma lista curta:

• Falta de chave de API ou inválida retorna 401 no handshake, então verifique a chave primeiro

• Time bloqueado retorna 403 e limite de taxa retorna 429 — faça retry com backoff

• Configuração de sessão malformada retorna 400, geralmente um typo no nome de algum campo

• Formato de áudio não suportado gera chiado, não erro, então combine a taxa da sessão

• Um response.create ausente após um resultado de ferramenta deixa o agente pendurado

• Uma tentativa de reserva duplicada pode causar problemas reais, então não faça retry às cegas

Repetir uma leitura que falhou, como check_availability, é seguro, mas repetir uma escrita que falhou, como uma reserva de fato, pode duplicar um agendamento. Qualquer ação que mude dados precisa de verificação de idempotência antes.

Usando tokens efêmeros em apps cliente

Até aqui assumimos que o código roda no seu servidor, onde a chave de API deve ficar. Se um app web ou mobile conectar direto, use tokens efêmeros.

Seu servidor chama POST https://api.x.ai/v1/realtime/client_secrets com sua chave, recebe uma resposta com um token e repassa o valor ao cliente. No meu teste, a resposta incluiu 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()

Navegadores não conseguem definir cabeçalhos personalizados no WebSocket, então o token vai no cabeçalho sec-websocket-protocol com o prefixo xai-client-secret..

Transformando o workflow em um endpoint FastAPI

Um endpoint permite que um frontend ou outro serviço chame o workflow. A rota valida o corpo da requisição com um modelo Pydantic, recebe uma mensagem em texto ou um caminho de áudio e retorna a transcrição, o áudio de resposta, o log de ferramentas, a latência e o custo 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,
    }

Rode com uvicorn app:app --reload e abra http://localhost:8000/docs. Leia XAI_API_KEY do ambiente do servidor e nunca aceite pelo corpo da requisição.