Quickstart

export BOSON_API_KEY="bai-xxxx"
pip install websockets sounddevice

import asyncio
import base64
import json
import os
import sounddevice as sd
import websockets

async def voice_agent():
    # Speaker output matching the default audio format: 24 kHz, 16-bit, mono PCM
    speaker = sd.RawOutputStream(samplerate=24000, channels=1, dtype="int16")
    speaker.start()

    async with websockets.connect(
        "wss://api.boson.ai/v1/realtime",
        additional_headers={"Authorization": f"Bearer {os.environ['BOSON_API_KEY']}"}
    ) as ws:
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "model": "higgs-realtime",
                "instructions": "You are a helpful assistant.",
                "audio": {
                    "input": {"turn_detection": {"type": "server_vad"}},
                    "output": {"voice": "default"}
                }
            }
        }))

        # Send a text message
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {"type": "message", "role": "user",
                     "content": [{"type": "input_text", "text": "Hello!"}]}
        }))
        await ws.send(json.dumps({"type": "response.create"}))

        # Play audio responses as they stream in
        async for msg in ws:
            event = json.loads(msg)
            if event["type"] == "response.output_audio.delta":
                speaker.write(base64.b64decode(event["delta"]))
            elif event["type"] == "response.output_audio_transcript.done":
                print("Assistant:", event["transcript"])
            elif event["type"] == "response.done":
                break

    speaker.stop()

asyncio.run(voice_agent())