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. The same pattern works for any live audio source: a meeting bot, a phone bridge, or a broadcast feed.
Prerequisites
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
Write a script to stream my microphone to the DeepL Voice API and print the transcript and translations as each sentence concludes.
Run the complete example
Install the dependencies:
pip install requests sounddevice websockets
live_translation.py and replace YOUR_AUTH_KEY with your DeepL API key:
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
❯ 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"
}
"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>"}}
{"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}
]
}
}
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()
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.
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": {}}))
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
