Skip to main content

Prerequisites

That’s it. No other setup required.
Real-time transcription over a persistent connection. Send audio, get partial and final transcripts back as they happen.
Install:
pip install "websockets>=14"
main.py:
import asyncio
import json
import urllib.request
import websockets

API_KEY = "YOUR_TELNYX_API_KEY"
STREAM_URL = "https://kexp-mp3-128.streamguys1.com/kexp128.mp3"

async def transcribe():
    url = (
        "wss://api.telnyx.com/v2/speech-to-text/transcription"
        "?transcription_engine=Deepgram"
        "&model=nova-3"
        "&input_format=mp3"
        "&interim_results=true"
    )
    headers = {"Authorization": f"Bearer {API_KEY}"}

    async with websockets.connect(
        url, additional_headers=headers
    ) as ws:
        # Listen for transcripts
        async def listen():
            async for message in ws:
                data = json.loads(message)
                transcript = data.get("transcript", "")
                if not transcript:
                    continue
                prefix = "FINAL" if data.get("is_final") else "partial"
                print(f"[{prefix}] {transcript}")

        listener = asyncio.create_task(listen())

        # Stream audio from KEXP Radio
        req = urllib.request.urlopen(STREAM_URL)
        try:
            while chunk := req.read(4096):
                await ws.send(chunk)
                await asyncio.sleep(0.05)
        except KeyboardInterrupt:
            pass

        await ws.send(json.dumps({"type": "CloseStream"}))
        listener.cancel()

asyncio.run(transcribe())
Run it:
python main.py

Example output

Connected. Streaming KEXP Radio...

[partial] the latest news from
[partial] the latest news from the BBC
[FINAL] The latest news from the KEXP Radio.
[partial] tensions continue
[partial] tensions continue to rise in the
[FINAL] Tensions continue to rise in the region as diplomatic talks stall.

What’s next