Grok Voice Agent Builder：Python で手を動かす実践ガイド

xAI は Voice Agent Builder を公開しました。音声エージェントを作成するためのコンソールです。コールフローを記述し、ドキュメントやツールを添付し、声を選びます。

音声エージェントのコンソールを試すとき、リリースノートよりもコードに組み込む部分が重要です。WebSocket セッションの設定、音声の流れ、ツール呼び出しの場所、通話コスト、他アプリからワークフローを呼ぶ方法などです。

以下のコードは、そのフローを Voice Agent API に直接対して再構築します。具体的には、空き状況を確認し、音声で応答し、コストを追跡し、ツールの失敗を処理し、FastAPI エンドポイントを公開するクリニックの予約アシスタントを使います。

Grok Voice Agent Builder とは？

Voice Agent Builder は、Grok Voice 上で音声エージェントを作成・デプロイするための xAI のコンソールです。2026年7月1日にベータ版として開始しました。音声認識、言語モデル、音声合成を別々のサービスとして使うのではなく、単一のボイスモデル経路を使います。

このコンソールには、電話機能、ドキュメント検索、ツールとコネクタ、ガードレール、リモート MCP サーバー、録音・文字起こし・トレースつきの通話ログが含まれます。

音声は分単位で課金されます。コンソールはまだベータ版のため、ここでは API を直接使います。

Builder の下で動く Grok Voice Agent API の仕組み

コンソールの下には Voice Agent API があり、Builder と同じランタイムを公開するリアルタイム WebSocket API です。

Builder は Voice API の上に載っています。画像：筆者作成。

ここで使うモデルは grok-voice-think-fast-1.0 です。 grok-voice-latest エイリアスは最新モデルを指します。ここではそれを使いますが、デプロイするアプリではバージョン付きの名前に固定します。xAI はこのモデルの τ-voice Bench リーダーボードで 67.3% のスコアを報告しています。私はそれをあくまで一つのデータポイントとして扱い、保証とは見なしません。

互換性メモ: この API は OpenAI Realtime API と互換です。OpenAI のリアルタイムエンドポイントとやり取りするコードがある場合、主にベース URL とキーを変更するだけです。

プロジェクト概要：何を作るか

このクリニック用アシスタントは、音声入力を受け取り、合成音声で応答し、追質問を行い、枠を提示する前に空き状況を確認し、必要に応じて人間に引き継ぎます。コアの例は 1 つのツールを使い、Streamlit デモでは予約・転送・通話終了のアクションを追加します。

このチュートリアルの中核は 4 つのファイルに分かれ、それぞれが 1 つの役割を持ちます:

• voice_client.py は WebSocket クライアント、音声ヘルパー、コスト計測を保持

• tools.py は check_availability と、Streamlit が使う追加のデモ用ツールを保持

• assistant.py はシステムプロンプト、セッション設定、ワークフローを保持

• app.py は全体を FastAPI で提供

これら 4 ファイルが記事の導線です。リポジトリには視覚的デモ用の app_streamlit.py と、Windows 起動用の run.py も含まれますが、コアのフローが動いてから戻ってきます。

前提条件

コードを動かす前に、Python 3.10 以降、xAI アカウント、console.x.ai からの API キー、プリペイドのクレジット、環境変数・JSON・WebSocket に関する基本的な知識が必要です。

プロジェクトのセットアップ

フォルダと仮想環境を作成し、パッケージをインストールします:

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

新規チェックアウトでも同じ構成になるよう、これらのパッケージを requirements.txt に固定してください。

.env ファイルを Python ファイルの隣に作成します:

XAI_API_KEY=xai-your-key-here

.env を .gitignore に追加します。API キーはサーバーに留めてください。

音声エージェントの構築

さっそく作っていきましょう。

WebSocket で Grok Voice Agent API に接続する

最初のステップは接続を開くことです。モデルはクエリパラメータで、キーはハンドシェイク時のベアラートークンで渡します:

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

有効なキーで実行すると、最初に表示されるイベントは session.created です。これはソケットが開き、設定可能になったことを意味します。

session.created イベントで接続確認。画像：筆者作成。

音声セッションの設定

生のソケットは、まだエージェントとして構成されていません。 session オブジェクトを含む session.update イベントを送って形作ります。

ボイス、音声形式、指示

最も頻繁に触る 3 つの設定は、声、音声形式、システムプロンプトです。リアルタイム API は 5 つの名前付きボイス eve、 ara、 rex, sal, leo と、任意のカスタムクローンを公開します。音声はデフォルトで audio/pcm 、24000 Hz です。入力と出力は個別に設定します。

以下はアシスタントが使うセッション設定で、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],
    }

instructions フィールドはシステムプロンプトです。このクリニックのプロンプトは短く保っています。長い音声応答は聞き取りにくいからです。

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.

エスカレーションの一文は、クリニックのエージェントが医療助言をしないようにするためです。最後の 2 行はスコープを守り、発話が不明瞭なときにループしないようにします。設定では今日の日付も追記しています。実際のテストでは、「July 6th」のような日付で年を誤推測することがあったためです。

ターン検出の調整

ターン検出は、話し終えたと判断する方法です。turn_detection.type を server_vad にすると、サーバーが無音でターンを終了します。null のままにすると、オーディオバッファのコミットでターンを制御できます。ファイルベースのフローではこちらを使います。

Server VAD には知っておくべき 3 つの設定があります。threshold は音声と認識する最小音量、 silence_duration_ms はどれくらいの無音でターンを終えるか、 prefix_padding_ms は発話開始前の少しの音声を保持します。エージェントが話を遮る場合は、まず silence_duration_ms を上げてください。

エージェントへ音声を送る

次に、発信者の音声を送ります。音声はセッション形式に一致している必要があります。モノラル 16 bit PCM、24000 Hz を base64 でエンコードし、チャンクで送信します。

クライアントはファイルをスライスでストリーミングし、バッファをコミットしてターンの終わりを示します:

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)

サンプルレートやエンコードが session.update と一致しないと、明示的なエラーではなくノイズや無音になる場合があります。音声は input_audio_buffer.append を通るため、メッセージ単位ではなく時間で課金されます。

音声応答を受け取る

応答をリクエストすると、音声は response.output_audio.delta、トランスクリプトは response.output_audio_transcript.delta として届き、 response.done がターンを閉じます。

クライアントはそれらを 1 つの 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

音声デルタをデコードして順に連結し、結果を response.wav に書き出します。発信者自身の発話を取得するには、audio.input.transcription を設定し、 conversation.item.input_audio_transcription.completed を読みます。

予約アシスタントのワークフローを構築する

ここからは会話になります。予約の要望、確認の質問、空き状況の確認、枠の提示、確定。ターン間でコンテキストを運ぶため、新しいターンごとに会話 ID で再接続し、セッション再開を有効にします。

音声エージェントにツール呼び出しを追加

クリニックでは、時間を約束する前に空き状況の確認が必須です。カスタムツールは、モデルがあなたのコードに到達する手段です。モデルがリクエストを出し、アプリケーションが関数を実行し、結果を送り返します。

ツールはプレーンな関数と、セッション設定に入れる JSON スキーマで構成します。以下は 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"],
    },
}

ループの形は決まっています。モデルがツールを必要とすると、引数付きの response.function_call_arguments.done を送ります。あなたは関数を実行し、function_call_output を返し、その後 response.create を送ってエージェントを続行させます。最後の response.create を忘れると、エージェントは沈黙します。

ツール呼び出しの往復を解説。画像：筆者作成。

このようなカスタム関数はあなたのコード内で動きます。Streamlit デモでは同じファイルからさらに 3 つ、book_appointment, transfer_to_human, end_call を登録します。組み込みツール（web 検索、X 検索、collections 検索、リモート MCP ツールなど）は xAI のサーバーで実行されます。

ツール失敗のハンドリング

ツールは失敗します。成功を前提にした音声エージェントは、存在しない枠を約束してしまう恐れがあります。私の ToolRegistry.execute は例外を投げません。失敗した検索は {"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)}

明示的なエラー状態により、失敗したツール呼び出しを成功と見なすのを防げます。

コスト計測の追加

提供前に、通話のコストを把握してください。音声は送受信の双方を分計算で $0.05/分 で課金します。テキスト入力イベントは 1 件あたり $0.004 です。function_call_output の結果や response.create イベントは課金対象外です。

クライアントは進行中に追跡するので、いつでもコストをプロパティとして参照できます:

@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

xAI が提供する電話番号を使うと 1 分あたり $0.01 のテレフォニー加算が発生します。ヘルパーは telephony=True 設定時に加味します。xAI がホストするツールは別課金です。web 検索と X 検索は約 $5/千回、ファイル検索は約 $2.50 です。

エラーとエッジケースの扱い

多くの失敗は次の短いリストに収まります:

• API キーの欠如や不正はハンドシェイク時に 401 を返すため、まずキーを確認

• ブロックされたチームは 403、レート制限は 429 を返すので、バックオフで再試行

• 不正なセッション設定は 400（多くはフィールド名のタイプミス）

• 未対応の音声形式はエラーにならずノイズになることがあるため、セッションのレートに合わせる

• ツール結果後の response.create 欠落でエージェントがハングする

• 重複予約の再試行は重大な問題を招くため、安易に再試行しない

読み取り系（check_availability など）の失敗は再試行しても安全ですが、実際の予約のような書き込み系の失敗を再試行すると二重予約になり得ます。データを変更するアクションには、まず冪等性チェックが必要です。

クライアントアプリでのエフェメラルトークン利用

これまでの説明は、API キーがあるべきサーバー側でコードが動く前提です。ブラウザやモバイルアプリが直接接続する場合は、エフェメラルトークン を使ってください。

サーバーは POST https://api.x.ai/v1/realtime/client_secrets をキー付きで呼び、トークンのレスポンスを受け取り、その値をクライアントに渡します。私の実行では、レスポンスに value と 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()

ブラウザはカスタム WebSocket ヘッダーを設定できないため、トークンは sec-websocket-protocol ヘッダーで xai-client-secret. 接頭辞とともに渡します。

ワークフローを FastAPI エンドポイントにする

エンドポイントがあれば、フロントエンドや別サービスからワークフローを呼び出せます。ルートは Pydantic モデルでリクエスト本文を検証し、テキストメッセージまたは音声パスを受け取り、トランスクリプト、応答音声、ツールログ、レイテンシ、推定コストを返します。

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

uvicorn app:app --reload で起動し、 http://localhost:8000/docs を開きます。 XAI_API_KEY はサーバー環境から読み取り、決してリクエスト本文から受け取らないでください。