Deploy your agent

# pip install websockets sounddevice numpy
import asyncio
import base64
import json
import os
import signal

import numpy as np
import sounddevice as sd
import websockets

URL = "wss://agents.assemblyai.com/v1/ws"
API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
AGENT_ID = os.environ.get("AGENT_ID", "your-agent-id-here")

SAMPLE_RATE = 24000   # PCM16 mono @ 24kHz is the audio/pcm default
CHANNELS = 1
BLOCKSIZE = 1200      # frames per chunk = 50ms at 24kHz


async def main():
    # Auth is a raw API key in the Authorization header — no "Bearer " prefix.
    async with websockets.connect(
        URL, additional_headers={"Authorization": API_KEY}
    ) as ws:
        # Bind this connection to a stored agent by id (no inline prompt/voice).
        await ws.send(json.dumps(
            {"type": "session.update", "session": {"agent_id": AGENT_ID}}
        ))

        # Block until the server confirms the session before streaming audio.
        while True:
            msg = json.loads(await ws.recv())
            if msg.get("type") == "session.ready":
                print(f"Session ready: {msg.get('session_id')}")
                break
            if msg.get("type") in ("session.error", "error"):
                print("Error:", msg.get("message", msg))
                return

        loop = asyncio.get_running_loop()
        mic_queue: asyncio.Queue[bytes] = asyncio.Queue()

        # Output stream for agent audio; kept open so we can flush on barge-in.
        out_stream = sd.OutputStream(
            samplerate=SAMPLE_RATE, channels=CHANNELS, dtype="int16"
        )
        out_stream.start()

        def mic_callback(indata, frames, time_info, status):
            # Runs on sounddevice's thread — hand bytes to the event loop safely.
            data = bytes(indata)
            loop.call_soon_threadsafe(mic_queue.put_nowait, data)

        in_stream = sd.InputStream(
            samplerate=SAMPLE_RATE,
            channels=CHANNELS,
            dtype="int16",
            blocksize=BLOCKSIZE,
            callback=mic_callback,
        )

        stop = asyncio.Event()

        async def send_audio():
            while not stop.is_set():
                chunk = await mic_queue.get()
                # Each mic chunk is base64-encoded PCM16 bytes.
                await ws.send(json.dumps({
                    "type": "input.audio",
                    "audio": base64.b64encode(chunk).decode("ascii"),
                }))

        async def receive():
            async for raw in ws:
                msg = json.loads(raw)
                mtype = msg.get("type")
                if mtype == "reply.audio":
                    pcm = base64.b64decode(msg["data"])
                    out_stream.write(np.frombuffer(pcm, dtype=np.int16))
                elif mtype == "transcript.user":
                    print("You:", msg.get("text", ""))
                elif mtype == "transcript.agent":
                    print("Agent:", msg.get("text", ""))
                elif mtype == "reply.done":
                    # On barge-in, drop queued playback so stale audio stops.
                    if msg.get("status") == "interrupted":
                        out_stream.abort()
                        out_stream.start()
                elif mtype == "session.ended":
                    stop.set()
                    break
                elif mtype in ("session.error", "error"):
                    print("Error:", msg.get("message", msg))

        # Trip `stop` on Ctrl+C so we can end the session gracefully.
        loop.add_signal_handler(signal.SIGINT, stop.set)

        with in_stream:
            tasks = [asyncio.create_task(send_audio()),
                     asyncio.create_task(receive())]
            await stop.wait()
            for t in tasks:
                t.cancel()

        # Politely end the session and give the server a moment to confirm.
        try:
            await ws.send(json.dumps({"type": "session.end"}))
            await asyncio.wait_for(ws.recv(), timeout=2.0)
        except (asyncio.TimeoutError, websockets.ConnectionClosed):
            pass

        out_stream.stop()
        out_stream.close()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass