Raw WebSocket voice agent with AssemblyAI (Python)

Frameworks are great, but sometimes you want to see the wire. This is the simplest possible voice agent: a microphone, a raw WebSocket, and AssemblyAI's Universal-3.5 Pro Realtime streaming speech-to-text. No SDK abstractions, no framework — just the protocol. Once you've seen how the messages flow, everything the SDKs and integrations do makes sense.

If you'd rather not hand-roll the transport, skip to the end — the managed Voice Agent API bundles STT, LLM, and TTS behind one connection at a flat $4.50/hr. But if you're the kind of person who reads the RFC, keep going.

How AssemblyAI streaming WebSockets work

The whole protocol fits in your head. You open a WebSocket to the v3 streaming endpoint, send binary PCM16 audio frames, and read JSON messages back. Here's the shape of it:

• Endpoint: wss://streaming.assemblyai.com/v3/ws?sample_rate=16000&speech_model=universal-3-5-pro. The old /v2/realtime/ws returns HTTP 410 — don't use it.
• Auth: your raw API key in the Authorization header. No Bearer prefix — that's for the Voice Agent API, not streaming STT. In a browser or mobile app, never ship the key: mint a temp token from GET https://streaming.assemblyai.com/v3/token?expires_in_seconds=60 and pass ?token=.
• Audio: PCM16 mono 16kHz, sent as binary frames of 50–1000ms each, no faster than real time.
• Server messages: Begin, SpeechStarted, Turn (read transcript; end_of_turn is false for partials, true for finals), and Termination.
• Client messages: binary audio, plus {"type":"Terminate"}, {"type":"ForceEndpoint"}, and {"type":"UpdateConfiguration", ...}.

The one rule you must not skip: always send Terminate when you're done. It flushes the final transcript and closes the session cleanly.

Python: raw WebSocket streaming

This uses the websockets library — no AssemblyAI SDK at all. It opens the connection, streams audio from any async source, and reads transcripts as they arrive.

import asyncio, json, os, websockets

URL = "wss://streaming.assemblyai.com/v3/ws?sample_rate=16000&speech_model=universal-3-5-pro"

async def run(audio_source):
    async with websockets.connect(
        URL,
        additional_headers={"Authorization": os.environ["ASSEMBLYAI_API_KEY"]},
    ) as ws:
        async def send_audio():
            async for chunk in audio_source:      # 16kHz mono PCM16, 50-1000ms
                await ws.send(chunk)
            await ws.send(json.dumps({"type": "Terminate"}))  # required

        async def recv_loop():
            async for raw in ws:
                msg = json.loads(raw)
                if msg["type"] == "Turn":
                    tag = "FINAL" if msg["end_of_turn"] else "partial"
                    print(f"{tag}: {msg['transcript']}")
                elif msg["type"] == "Termination":
                    return

        await asyncio.gather(send_audio(), recv_loop())

Feed run() anything that yields PCM16 chunks — a mic capture loop, a file reader, an upstream socket. The send_audio and recv_loop coroutines run concurrently so you read transcripts while you're still talking.

Node: raw WebSocket streaming

Same protocol, the ws library instead:

import WebSocket from 'ws';

const ws = new WebSocket(
  'wss://streaming.assemblyai.com/v3/ws?sample_rate=16000&speech_model=universal-3-5-pro',
  { headers: { authorization: process.env.ASSEMBLYAI_API_KEY } });

ws.on('message', (raw) => {
  const msg = JSON.parse(raw.toString());
  if (msg.type === 'Turn') {
    console.log(msg.end_of_turn ? `FINAL: ${msg.transcript}` : `partial: ${msg.transcript}`);
  }
});

// send PCM16 mono 16kHz binary frames with ws.send(pcm16Buffer)

function stop() {
  ws.send(JSON.stringify({ type: 'Terminate' }));  // required
}

Note the header is authorization with the raw key — again, no Bearer. Send your audio buffers with ws.send() and call stop() when the turn is over.

Improving accuracy mid-conversation with agent_context

Here's the feature that turns a transcriber into a voice-agent speech layer. Universal-3.5 Pro Realtime takes an agent_context value — you tell it what your agent just said, and it biases recognition toward the expected reply. Ask "what's your email address?" and the model leans toward spelling out an address instead of guessing at homophones. Across 20,000 voice-agent files, this cut WER by 10.2%. Rolling Context Carryover is on by default, so prior finalized user turns carry over automatically; you just refresh the agent side after each reply.

On a raw WebSocket you seed it as a query param at connection time, then update it mid-stream with UpdateConfiguration:

# send after each agent reply, over the same open socket:
await ws.send(json.dumps({
    "type": "UpdateConfiguration",
    "agent_context": "Sure, what date would you like to book?",
}))

This is the same capability LiveKit and Pipecat expose as Context Carryover — you're just doing it by hand. A couple of streaming notes specific to the Pro model: end_of_turn_confidence_threshold is ignored, language_code biases to a single language, and language_detection isn't supported on the Pro streaming model.

What about end-of-turn detection?

On a raw socket you decide when a turn is done — the server emits Turn messages with end_of_turn, and you can force a boundary with {"type":"ForceEndpoint"}. If you're running inside a framework like LiveKit instead, use the framework's turn-detection model rather than wiring your own — that's the supported path there. This raw approach is for when you own the whole loop.

Raw WebSocket vs the Voice Agent API

Building on raw WebSockets is the right call when you want full control of the transport, you're stitching STT into an existing system, or you're just learning how the pieces fit. But it's only the speech layer — you still have to bring an LLM, a TTS voice, turn-taking, and interruption handling, and wire them together.

If your goal is a working agent rather than a protocol tour, the Voice Agent API does all of that behind one WebSocket at a flat $4.50/hr — STT (this same Universal-3.5 Pro Realtime model), your choice of LLM, and TTS, managed. Streaming STT on its own is $0.45/hr base. Same accuracy either way: 6.99% WER on the Pipecat voice-agent benchmark, and the only model in Coval's independent Human Parity Zone. For a fuller build, see how to build with the Voice Agent API.

Build it

• Get your free API key — stream your first audio in minutes.
• Talk to a live agent — hear the managed Voice Agent API in action.
• Explore streaming speech-to-text — the full streaming reference.

Frequently asked questions

Which endpoint and model do I use for streaming?

Connect to wss://streaming.assemblyai.com/v3/ws with sample_rate=16000 and speech_model=universal-3-5-pro (Universal-3.5 Pro Realtime). Audio is PCM16 mono 16kHz. The old /v2/realtime/ws endpoint returns HTTP 410.

How do I get started with a raw WebSocket connection?

Open a WebSocket to the v3 endpoint with your raw API key in the Authorization header (no Bearer), stream 16kHz mono PCM16 binary frames of 50–1000ms, read Turn messages for transcripts, and always send {"type":"Terminate"} when you're done. No SDK required.

How does end-of-turn detection work?

The server sends Turn messages with an end_of_turn flag, and you can force a boundary with {"type":"ForceEndpoint"}. Note that end_of_turn_confidence_threshold is ignored on the Pro streaming model. Inside a framework, use the framework's turn-detection model.

Can I improve accuracy mid-conversation?

Yes. Send agent_context — seed it at connection time and update it mid-stream via {"type":"UpdateConfiguration","agent_context":"..."}. It cut WER by 10.2% across 20,000 voice-agent files, and prior user turns carry over by default.

When should I use the Voice Agent API instead of a raw WebSocket?

Use the raw WebSocket when you want control of the transport or you're integrating STT into an existing system. Use the Voice Agent API when you want a complete agent — STT, LLM, and TTS behind one connection at a flat $4.50/hr — without assembling it yourself.