Realtime transcription

Realtime transcription turns live audio, such as a microphone, a phone line, or a broadcast, into text while the person is still speaking. This page explains how a session runs from start to finish, how to configure it, how to read its events, and how to trade speed for accuracy.

A session is one WebSocket connection that goes through four stages, always in the same order. Your application sends the audio, and the API sends the text back while the audio still flows.

Read from top to bottom: each arrow is one message, from your application on the left or from the API on the right.

At each stage, your application does the following:

If the API rejects a message, it sends an error event at that stage instead. See Common issues.

Open a WebSocket connection to the realtime endpoint, with the model in the query string and your API key in the Authorization header:

wss://api.mistral.ai/v1/audio/transcriptions/realtime?model=voxtral-mini-transcribe-realtime-3
Authorization: Bearer $MISTRAL_API_KEY

wss://api.mistral.ai/v1/audio/transcriptions/realtime?model=voxtral-mini-transcribe-realtime-3
Authorization: Bearer $MISTRAL_API_KEY

With the Python or TypeScript SDK, connect() opens the connection and returns once the session is created:

import asyncio
import os

from mistralai.client import Mistral

async def main():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    async with await client.audio.realtime.connect(
        model="voxtral-mini-transcribe-realtime-3",
    ) as connection:
        print(connection.session.model_dump_json(indent=2))

asyncio.run(main())

import asyncio
import os

from mistralai.client import Mistral

async def main():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    async with await client.audio.realtime.connect(
        model="voxtral-mini-transcribe-realtime-3",
    ) as connection:
        print(connection.session.model_dump_json(indent=2))

asyncio.run(main())

Send your settings in a session.update message, before the first audio chunk. The session answers with session.updated, which echoes every setting it applies:

{
  "type": "session.update",
  "session": {
    "audio_format": { "encoding": "pcm_s16le", "sample_rate": 24000 },
    "target_streaming_delay_ms": 480,
    "language": "en",
    "context_bias": ["Northwind", "Quillara", "Alder Creek Design", "Rotterdam"]
  }
}

{
  "type": "session.update",
  "session": {
    "audio_format": { "encoding": "pcm_s16le", "sample_rate": 24000 },
    "target_streaming_delay_ms": 480,
    "language": "en",
    "context_bias": ["Northwind", "Quillara", "Alder Creek Design", "Rotterdam"]
  }
}

The following script sends these settings, streams a WAV file in chunks of 100 ms, and prints the transcript as it arrives.

import asyncio
import base64
import json
import os
import wave

import websockets

MODEL = "voxtral-mini-transcribe-realtime-3"
URL = f"wss://api.mistral.ai/v1/audio/transcriptions/realtime?model={MODEL}"
HEADERS = {"Authorization": f"Bearer {os.environ['MISTRAL_API_KEY']}"}
AUDIO_FILE = "customer_call.wav"  # 16-bit mono PCM WAV

async def send_audio(ws, path):
    with wave.open(path, "rb") as wav:
        while chunk := wav.readframes(wav.getframerate() // 10):
            await ws.send(json.dumps({"type": "input_audio.append", "audio": base64.b64encode(chunk).decode()}))
    await ws.send(json.dumps({"type": "input_audio.end"}))

async def main():
    with wave.open(AUDIO_FILE, "rb") as wav:
        sample_rate = wav.getframerate()

    async with websockets.connect(URL, additional_headers=HEADERS) as ws:
        print(json.loads(await ws.recv())["type"])  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "audio_format": {"encoding": "pcm_s16le", "sample_rate": sample_rate},
                "target_streaming_delay_ms": 480,
                "language": "en",
                "context_bias": ["Northwind", "Quillara", "Alder Creek Design", "Rotterdam"],
            },
        }))
        sender = asyncio.create_task(send_audio(ws, AUDIO_FILE))

        async for message in ws:
            event = json.loads(message)
            if event["type"] == "session.updated":
                print("session.updated:", json.dumps(event["session"]))
            elif event["type"] == "transcription.text.delta":
                print(event["text"], end="", flush=True)
            elif event["type"] == "transcription.done":
                print("\ntranscription.done")
                break
            elif event["type"] == "error":
                print("error:", event["error"]["message"])
                break
        sender.cancel()

asyncio.run(main())

import asyncio
import base64
import json
import os
import wave

import websockets

MODEL = "voxtral-mini-transcribe-realtime-3"
URL = f"wss://api.mistral.ai/v1/audio/transcriptions/realtime?model={MODEL}"
HEADERS = {"Authorization": f"Bearer {os.environ['MISTRAL_API_KEY']}"}
AUDIO_FILE = "customer_call.wav"  # 16-bit mono PCM WAV

async def send_audio(ws, path):
    with wave.open(path, "rb") as wav:
        while chunk := wav.readframes(wav.getframerate() // 10):
            await ws.send(json.dumps({"type": "input_audio.append", "audio": base64.b64encode(chunk).decode()}))
    await ws.send(json.dumps({"type": "input_audio.end"}))

async def main():
    with wave.open(AUDIO_FILE, "rb") as wav:
        sample_rate = wav.getframerate()

    async with websockets.connect(URL, additional_headers=HEADERS) as ws:
        print(json.loads(await ws.recv())["type"])  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "audio_format": {"encoding": "pcm_s16le", "sample_rate": sample_rate},
                "target_streaming_delay_ms": 480,
                "language": "en",
                "context_bias": ["Northwind", "Quillara", "Alder Creek Design", "Rotterdam"],
            },
        }))
        sender = asyncio.create_task(send_audio(ws, AUDIO_FILE))

        async for message in ws:
            event = json.loads(message)
            if event["type"] == "session.updated":
                print("session.updated:", json.dumps(event["session"]))
            elif event["type"] == "transcription.text.delta":
                print(event["text"], end="", flush=True)
            elif event["type"] == "transcription.done":
                print("\ntranscription.done")
                break
            elif event["type"] == "error":
                print("error:", event["error"]["message"])
                break
        sender.cancel()

asyncio.run(main())

session.update takes the following fields. Send only the ones you want to change: the others keep the default that session.created reports.

| context_bias | Array of strings | [] | Help the model spell names and terms. See Bias context and vocabulary. | | diarize | Boolean | false | Label each piece of text with its speaker. See Diarize speakers. | | speaker_references | Object: a name for each base64 audio clip | Not set | Name known speakers, up to 8 clips of 3 to 15 seconds. Requires diarize. See Diarize speakers. | | turn_detection | Object | Not set | Know when the user starts and finishes speaking. See Turn detection. |

Everything the API sends arrives as a JSON message with a type. Read the type first, then the fields of that event:

Each transcription.text.delta adds new text after the text already sent. A delta never corrects earlier text, so your application only appends: there's nothing to replace.

• Show each delta as soon as it arrives. It's final when you receive it.
• Append it as is. Deltas carry their own spaces and punctuation, such as " for" or ".", so don't add separators.
• Save the text of transcription.done as the transcript of record.

When the audio stops, send input_audio.end. The API sends the last deltas, then transcription.done with the full transcript. Close the connection once you've received it.

The model doesn't write a word as soon as it hears it: it waits to hear what follows, because the next sounds help it pick the right word. target_streaming_delay_ms sets that wait for the session.

• A shorter delay gets the text out sooner, but less accurately.
• A longer delay gives the model more context, and more accurate text, but later.

The delay takes any integer from 80 to 1200, or 2400. Any other value returns an error. The Python SDK can set it with update_session(), along with the audio format.

Readers see words at once, and the text they keep is the accurate one. Live captions and note-taking benefit most.

Most issues come back as an error event with a message. These are the ones specific to live sessions:

For other errors, see Troubleshooting.