Real-Time Voice Quickstart - DeepL Documentation

In this tutorial you’ll run a small Python program that captures audio from your microphone, streams it to the DeepL Voice API, and prints the transcript plus German and French translations to your terminal, sentence by sentence, while you speak. You can then extend it to play the German translation aloud. The same pattern works for any live audio source: a meeting bot, a phone bridge, or a broadcast feed.

Prerequisites

A DeepL API account with Voice API access
Python 3.10 or later
Headphones, if you want to hear the translation spoken
A microphone (no microphone? see Simulate a live stream from a file)

Building with an AI coding agent?

Wire it up to the DeepL Docs MCP Server so it can search and read this documentation while it writes code. In Claude Code:

claude mcp add --transport http deepl-docs https://developers.deepl.com/mcp

Then describe what you want to build. To get the same result as this tutorial, paste:

Write a script to stream my microphone to the DeepL Voice API and print the transcript and translations as each sentence concludes.

Setup instructions for Claude Desktop, Cursor, VS Code, and other MCP clients are on the Docs MCP Server page.

Run the complete example

Install the dependencies:

pip install requests sounddevice websockets

Then save this as live_translation.py and replace YOUR_AUTH_KEY with your DeepL API key:

live_translation.py

import asyncio
import base64
import json
import signal

import requests
import sounddevice
import websockets

AUTH_KEY = "YOUR_AUTH_KEY"
SESSION_ENDPOINT = "https://api.deepl.com/v3/voice/realtime"
TARGET_LANGUAGES = ["de", "fr"]
SAMPLE_RATE = 16000  # Must match the rate declared in source_media_content_type
CHUNK_FRAMES = 3200  # 200 ms per chunk at 16 kHz
RECORD_SECONDS = 30  # Safety cap; Ctrl+C stops recording earlier


def create_session() -> dict:
    response = requests.post(
        SESSION_ENDPOINT,
        headers={"Authorization": f"DeepL-Auth-Key {AUTH_KEY}"},
        json={
            "source_media_content_type": "audio/pcm;encoding=s16le;rate=16000",
            "target_languages": TARGET_LANGUAGES,
        },
    )
    response.raise_for_status()
    return response.json()


async def send_microphone_audio(ws, stop: asyncio.Event) -> None:
    stream = sounddevice.RawInputStream(
        samplerate=SAMPLE_RATE, channels=1, dtype="int16"
    )
    stream.start()
    try:
        for _ in range(RECORD_SECONDS * SAMPLE_RATE // CHUNK_FRAMES):
            if stop.is_set():
                break
            # Read in a worker thread so receiving continues while we block
            data, _overflowed = await asyncio.to_thread(stream.read, CHUNK_FRAMES)
            encoded = base64.b64encode(bytes(data)).decode("ascii")
            await ws.send(json.dumps({"source_media_chunk": {"data": encoded}}))
    finally:
        stream.stop()
        stream.close()
    print("Finalizing transcripts...")
    await ws.send(json.dumps({"end_of_source_media": {}}))


async def receive_results(ws) -> None:
    pending = {}  # Concluded text per language, buffered until a sentence ends

    def handle_update(label: str, segments: list) -> None:
        text = pending.get(label, "") + "".join(s["text"] for s in segments)
        if text.rstrip().endswith((".", "!", "?")):
            print(f"[{label}] {text.strip()}")
            text = ""
        pending[label] = text

    async for message in ws:
        data = json.loads(message)
        if "source_transcript_update" in data:
            handle_update("source", data["source_transcript_update"]["concluded"])
        elif "target_transcript_update" in data:
            update = data["target_transcript_update"]
            handle_update(update["language"], update["concluded"])
        elif "end_of_stream" in data:
            # The very last message: flush any unfinished sentences and exit
            for label, text in pending.items():
                if text.strip():
                    print(f"[{label}] {text.strip()}")
            return
        elif "error" in data:
            raise RuntimeError(f"Voice API error: {data['error']['error_message']}")


async def main() -> None:
    stop = asyncio.Event()
    try:
        asyncio.get_running_loop().add_signal_handler(signal.SIGINT, stop.set)
    except NotImplementedError:
        pass  # Windows: Ctrl+C raises KeyboardInterrupt instead

    session = create_session()
    url = f"{session['streaming_url']}?token={session['token']}"
    async with websockets.connect(url) as ws:
        print("Connected. Speak now (Ctrl+C to stop)...")
        await asyncio.gather(send_microphone_audio(ws, stop), receive_results(ws))
    print("Done. All transcripts are final.")


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

Run it and speak in any supported source language. Each sentence prints once it’s final, first the transcript, then each translation. Press Ctrl+C when you’re done: the script stops recording, waits for the remaining results, and exits cleanly.

❯ python live_translation.py
Connected. Speak now (Ctrl+C to stop)...
[source] Hello everyone, welcome to today's demo.
[de] Hallo zusammen, willkommen zur heutigen Demo.
[fr] Bonjour à tous, bienvenue à la démonstration d'aujourd'hui.
^CFinalizing transcripts...
[source] We are testing real-time voice translation.
[de] Wir testen die Sprachübersetzung in Echtzeit.
[fr] Nous testons la traduction vocale en temps réel.
Done. All transcripts are final.

How it works

The program does three things: it creates a session over HTTPS, streams audio to the WebSocket URL it gets back, and handles result messages until the server confirms everything is final.

The session request

create_session sends a POST request with the audio format and target languages. The only required field is source_media_content_type; for raw microphone audio, that’s PCM at 16 kHz. The response contains the WebSocket URL and a one-time token, and the program connects to {streaming_url}?token={token}:

{
  "session_id": "4f911080-cfe2-41d4-8269-0e6ec15a0354",
  "streaming_url": "wss://api.deepl.com/v3/voice/realtime/connect",
  "token": "VGhpcyBpcyBhIGZha2UgdG9rZW4K"
}

If you know the speaker’s language in advance, add "source_language": "en" and "source_language_mode": "fixed" to the request body. This skips language detection and reduces latency. Without them, the language is detected automatically. See the Request Session reference for all parameters, including glossary and formality options.

Sending audio

send_microphone_audio sends each 200-millisecond microphone chunk as a JSON text message with the raw audio base64-encoded:

{"source_media_chunk": {"data": "<base64-encoded audio>"}}

Keep chunks between 50 and 250 milliseconds for the best latency. When the audio ends (Ctrl+C or the safety cap), the function sends one final message that tells the API to finalize all pending results:

{"end_of_source_media": {}}

Receiving transcripts and translations

While audio is being sent, the server pushes source_transcript_update and target_transcript_update messages on the same connection:

{
  "target_transcript_update": {
    "language": "de",
    "concluded": [
      {"text": " Hallo zusammen,", "start_time": 0, "end_time": 1500}
    ],
    "tentative": [
      {"text": " willkommen zur heutigen Demo", "start_time": 1500, "end_time": 2000}
    ]
  }
}

Concluded segments are final and sent only once; tentative segments are provisional and refined by later updates. receive_results appends concluded text to a per-language buffer and prints the buffer whenever a sentence completes, which keeps the terminal readable. A UI would also render the tentative text as a live preview that updates in place; see Understanding Voice Sessions for this delivery model. After end_of_source_media, the server sends the remaining updates, then end_of_stream as the very last message, at which point it’s safe to close the connection. See the WebSocket Streaming reference for the full message schema.

Reconnecting after a drop

Networks drop, and the Voice API lets you resume a session instead of starting over. If the WebSocket closes unexpectedly, exchange your token for a fresh streaming URL and token, then reconnect. Session state, including configuration and translation context, is preserved.

def reconnect(token: str) -> dict:
    response = requests.get(
        SESSION_ENDPOINT,
        headers={"Authorization": f"DeepL-Auth-Key {AUTH_KEY}"},
        params={"token": token},
    )
    response.raise_for_status()
    return response.json()

The response has the same shape as the session response above. Always pass the token from your most recent session or reconnection response: each token is single-use, and presenting an outdated one invalidates the session (a 400-level error). In that case, create a new session with create_session(). To add this to the example, wrap the websockets.connect block in a try/except websockets.exceptions.ConnectionClosed loop that calls reconnect() and connects again with the new URL and token.

Hear the translation spoken

The Voice API can also speak the translation aloud, so listeners hear the German version while you talk. In this step you extend live_translation.py to request translated speech and play it through your headphones.

Use headphones. If the translated speech plays through your speakers, the microphone picks it up and sends it back for translation.

Request translated speech

Two session fields turn on translated speech. target_media_languages names the language to speak, and the server then streams synthesized audio of each German translation alongside the text. target_media_content_type sets the audio format: raw 24 kHz PCM plays directly through sounddevice, with no decoder. Add a constant for that playback rate next to SAMPLE_RATE:

OUTPUT_SAMPLE_RATE = 24000  # Must match the rate declared in target_media_content_type

Then add both fields to the request body in create_session:

json={
    "source_media_content_type": "audio/pcm;encoding=s16le;rate=16000",
    "target_languages": TARGET_LANGUAGES,
    "target_media_languages": ["de"],
    "target_media_content_type": "audio/pcm;encoding=s16le;rate=24000",
},

A session synthesizes speech for one language, which also appears in the text translations. To pick a voice or another audio format, see the Request Session reference.

Play the audio as it arrives

Add a player task that writes audio from a queue to your headphones. Like the microphone reader, it blocks in a worker thread so the receive loop keeps running:

async def play_translated_speech(audio: asyncio.Queue) -> None:
    stream = sounddevice.RawOutputStream(
        samplerate=OUTPUT_SAMPLE_RATE, channels=1, dtype="int16"
    )
    stream.start()
    try:
        while (pcm := await audio.get()) is not None:
            await asyncio.to_thread(stream.write, pcm)
    finally:
        stream.stop()
        stream.close()

Translated speech arrives in target_media_chunk messages, each with a data array of base64-encoded audio packets. Replace receive_results with this version, which puts every decoded packet on the queue. The finally block queues None however the loop ends, so the player finishes the remaining audio and stops instead of waiting forever:

async def receive_results(ws, audio: asyncio.Queue) -> None:
    pending = {}  # Concluded text per language, buffered until a sentence ends

    def handle_update(label: str, segments: list) -> None:
        text = pending.get(label, "") + "".join(s["text"] for s in segments)
        if text.rstrip().endswith((".", "!", "?")):
            print(f"[{label}] {text.strip()}")
            text = ""
        pending[label] = text

    try:
        async for message in ws:
            data = json.loads(message)
            if "source_transcript_update" in data:
                handle_update("source", data["source_transcript_update"]["concluded"])
            elif "target_transcript_update" in data:
                update = data["target_transcript_update"]
                handle_update(update["language"], update["concluded"])
            elif "target_media_chunk" in data:
                # Raw PCM has no header packets, so every packet is playable audio
                for packet in data["target_media_chunk"]["data"]:
                    await audio.put(base64.b64decode(packet))
            elif "end_of_stream" in data:
                # The very last message: flush any unfinished sentences and exit
                for label, text in pending.items():
                    if text.strip():
                        print(f"[{label}] {text.strip()}")
                return
            elif "error" in data:
                raise RuntimeError(f"Voice API error: {data['error']['error_message']}")
    finally:
        await audio.put(None)

Finally, create the queue in main and run the player next to the other two tasks:

audio = asyncio.Queue()
async with websockets.connect(url) as ws:
    print("Connected. Speak now (Ctrl+C to stop)...")
    await asyncio.gather(
        send_microphone_audio(ws, stop),
        receive_results(ws, audio),
        play_translated_speech(audio),
    )

Run it with translated speech

Here’s the complete script with all the changes from this section. Save it as live_speech_translation.py if you want to keep the text-only version, and replace YOUR_AUTH_KEY with your DeepL API key:

live_speech_translation.py

import asyncio
import base64
import json
import signal

import requests
import sounddevice
import websockets

AUTH_KEY = "YOUR_AUTH_KEY"
SESSION_ENDPOINT = "https://api.deepl.com/v3/voice/realtime"
TARGET_LANGUAGES = ["de", "fr"]
SAMPLE_RATE = 16000  # Must match the rate declared in source_media_content_type
OUTPUT_SAMPLE_RATE = 24000  # Must match the rate declared in target_media_content_type
CHUNK_FRAMES = 3200  # 200 ms per chunk at 16 kHz
RECORD_SECONDS = 30  # Safety cap; Ctrl+C stops recording earlier


def create_session() -> dict:
    response = requests.post(
        SESSION_ENDPOINT,
        headers={"Authorization": f"DeepL-Auth-Key {AUTH_KEY}"},
        json={
            "source_media_content_type": "audio/pcm;encoding=s16le;rate=16000",
            "target_languages": TARGET_LANGUAGES,
            "target_media_languages": ["de"],
            "target_media_content_type": "audio/pcm;encoding=s16le;rate=24000",
        },
    )
    response.raise_for_status()
    return response.json()


async def send_microphone_audio(ws, stop: asyncio.Event) -> None:
    stream = sounddevice.RawInputStream(
        samplerate=SAMPLE_RATE, channels=1, dtype="int16"
    )
    stream.start()
    try:
        for _ in range(RECORD_SECONDS * SAMPLE_RATE // CHUNK_FRAMES):
            if stop.is_set():
                break
            # Read in a worker thread so receiving continues while we block
            data, _overflowed = await asyncio.to_thread(stream.read, CHUNK_FRAMES)
            encoded = base64.b64encode(bytes(data)).decode("ascii")
            await ws.send(json.dumps({"source_media_chunk": {"data": encoded}}))
    finally:
        stream.stop()
        stream.close()
    print("Finalizing transcripts...")
    await ws.send(json.dumps({"end_of_source_media": {}}))


async def play_translated_speech(audio: asyncio.Queue) -> None:
    stream = sounddevice.RawOutputStream(
        samplerate=OUTPUT_SAMPLE_RATE, channels=1, dtype="int16"
    )
    stream.start()
    try:
        while (pcm := await audio.get()) is not None:
            await asyncio.to_thread(stream.write, pcm)
    finally:
        stream.stop()
        stream.close()


async def receive_results(ws, audio: asyncio.Queue) -> None:
    pending = {}  # Concluded text per language, buffered until a sentence ends

    def handle_update(label: str, segments: list) -> None:
        text = pending.get(label, "") + "".join(s["text"] for s in segments)
        if text.rstrip().endswith((".", "!", "?")):
            print(f"[{label}] {text.strip()}")
            text = ""
        pending[label] = text

    try:
        async for message in ws:
            data = json.loads(message)
            if "source_transcript_update" in data:
                handle_update("source", data["source_transcript_update"]["concluded"])
            elif "target_transcript_update" in data:
                update = data["target_transcript_update"]
                handle_update(update["language"], update["concluded"])
            elif "target_media_chunk" in data:
                # Raw PCM has no header packets, so every packet is playable audio
                for packet in data["target_media_chunk"]["data"]:
                    await audio.put(base64.b64decode(packet))
            elif "end_of_stream" in data:
                # The very last message: flush any unfinished sentences and exit
                for label, text in pending.items():
                    if text.strip():
                        print(f"[{label}] {text.strip()}")
                return
            elif "error" in data:
                raise RuntimeError(f"Voice API error: {data['error']['error_message']}")
    finally:
        await audio.put(None)


async def main() -> None:
    stop = asyncio.Event()
    try:
        asyncio.get_running_loop().add_signal_handler(signal.SIGINT, stop.set)
    except NotImplementedError:
        pass  # Windows: Ctrl+C raises KeyboardInterrupt instead

    session = create_session()
    url = f"{session['streaming_url']}?token={session['token']}"
    audio = asyncio.Queue()
    async with websockets.connect(url) as ws:
        print("Connected. Speak now (Ctrl+C to stop)...")
        await asyncio.gather(
            send_microphone_audio(ws, stop),
            receive_results(ws, audio),
            play_translated_speech(audio),
        )
    print("Done. All transcripts are final.")


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

Put on your headphones and run it:

python live_speech_translation.py

The terminal shows the same transcript and translations as before. A moment after each sentence, you hear it spoken in German. The audio contains only speech, so playback pauses whenever you do.

Simulate a live stream from a file

If you don’t have a microphone, or you want reproducible input while developing, stream a pre-recorded file at real-time pace instead. Set "source_media_content_type": "audio/auto" in the session request so the format is detected, and swap send_microphone_audio for a reader that paces itself:

async def send_file_audio(ws, path: str, stop: asyncio.Event) -> None:
    with open(path, "rb") as audio_file:
        while chunk := audio_file.read(6400):
            if stop.is_set():
                break
            encoded = base64.b64encode(chunk).decode("ascii")
            await ws.send(json.dumps({"source_media_chunk": {"data": encoded}}))
            # Pace the upload to simulate live audio
            await asyncio.sleep(0.2)
    print("Finalizing transcripts...")
    await ws.send(json.dumps({"end_of_source_media": {}}))

Any recording of speech in a supported format works, for example an MP3 of a podcast episode.

Don’t send audio faster than 2x real-time. Uploading a file as fast as the network allows triggers rate limits and terminates the session.

Next steps

See the WebSocket Streaming reference for the complete message schema and all event types.
To understand the session flow and token lifecycle in depth, see Understanding Voice Sessions.
To choose a voice or another audio format for translated speech, see the Request Session reference.
To reduce bandwidth with MessagePack instead of JSON, see Message Encoding.
For a fuller command-line version of this program, including glossary and formality options, see the DeepL Voice CLI example.
To apply a custom glossary to your translations, see the Glossaries API.