Qwen3.8 LiveTranslate Flash Realtime

import asyncio, json, os, base64, websockets
URL = "wss://api.empiriolabs.ai/v1/realtime?model=qwen3-8-livetranslate-flash-realtime"
async def main():
    async with websockets.connect(
        URL,
        additional_headers=[("Authorization", f"Bearer {os.environ['EMPIRIOLABS_API_KEY']}")],
        max_size=None,
    ) as ws:
        print(json.loads(await ws.recv())["type"])  # session.created
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "voice": "Tina",
                "translation": {"language": "en"},
                "modalities": ["text", "audio"],
            },
        }))
        for chunk in read_microphone_chunks():
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(chunk).decode(),
            }))
        async for raw in ws:
            event = json.loads(raw)
            if event["type"] == "response.audio.delta":
                ...  # base64 audio chunk
            elif event["type"] == "response.done":
                break
asyncio.run(main())