GitHub - speechmatics/speechmatics-python-sdk: Python SDKs for Speechmatics APIs

Speechmatics Python SDK provides convenient access to enterprise-grade speech-to-text APIs from Python applications.

Fully typed with type definitions for all request params and response fields. Modern Python with async/await patterns, type hints, and context managers for production-ready code.

55+ Languages • Realtime & Batch • Custom vocabularies • Speaker diarization • Speaker ID

Get API Key • Documentation • Academy Examples

# Choose the package for your use case:

# Batch transcription
pip install speechmatics-batch

# Realtime streaming
pip install speechmatics-rt

# Voice agents
pip install speechmatics-voice

# Text-to-speech
pip install speechmatics-tts

git clone https://github.com/speechmatics/speechmatics-python-sdk.git
cd speechmatics-python-sdk

python -m venv .venv
.venv\Scripts\activate
# On Mac/Linux: source .venv/bin/activate

# Install development dependencies for all SDKs
make install-dev

# Install pre-commit hooks
pre-commit install

Note

All examples use load_dotenv() to load your API key from a .env file. Create a .env file with SPEECHMATICS_API_KEY=your_key_here.

import asyncio
import os
from dotenv import load_dotenv
from speechmatics.batch import AsyncClient

load_dotenv()

async def main():
    client = AsyncClient(api_key=os.getenv("SPEECHMATICS_API_KEY"))
    result = await client.transcribe("audio.wav")
    print(result.transcript_text)
    await client.close()

asyncio.run(main())

pip install speechmatics-batch python-dotenv

import asyncio
import os
from dotenv import load_dotenv
from speechmatics.rt import (
    AsyncClient,
    ServerMessageType,
    TranscriptionConfig,
    TranscriptResult,
    AudioFormat,
    AudioEncoding,
    Microphone,
)

load_dotenv()

CHUNK_SIZE = 4096

async def main():
    client = AsyncClient(api_key=os.getenv("SPEECHMATICS_API_KEY"))
    mic = Microphone(sample_rate=16000, chunk_size=CHUNK_SIZE)

    @client.on(ServerMessageType.ADD_TRANSCRIPT)
    def on_final(message):
        result = TranscriptResult.from_message(message)
        if result.metadata.transcript:
            print(f"[final]: {result.metadata.transcript}")

    @client.on(ServerMessageType.ADD_PARTIAL_TRANSCRIPT)
    def on_partial(message):
        result = TranscriptResult.from_message(message)
        if result.metadata.transcript:
            print(f"[partial]: {result.metadata.transcript}")

    mic.start()

    try:
        await client.start_session(
            transcription_config=TranscriptionConfig(language="en", enable_partials=True),
            audio_format=AudioFormat(encoding=AudioEncoding.PCM_S16LE, sample_rate=16000),
        )
        print("Speak now...")

        while True:
            await client.send_audio(await mic.read(CHUNK_SIZE))
    finally:
        mic.stop()
        await client.close()


asyncio.run(main())

pip install speechmatics-rt python-dotenv pyaudio

import asyncio
import os
from dotenv import load_dotenv
from speechmatics.tts import AsyncClient, Voice, OutputFormat

load_dotenv()

async def main():
    client = AsyncClient(api_key=os.getenv("SPEECHMATICS_API_KEY"))

    response = await client.generate(
        text="Hello! Welcome to Speechmatics Text-to-Speech",
        voice=Voice.SARAH,
        output_format=OutputFormat.WAV_16000
    )

    audio_data = await response.read()
    with open("output.wav", "wb") as f:
        f.write(audio_data)
        print("Audio saved to output.wav")

    await client.close()

asyncio.run(main())

pip install speechmatics-tts python-dotenv

import asyncio
import os
from dotenv import load_dotenv
from speechmatics.rt import Microphone
from speechmatics.voice import VoiceAgentClient, VoiceAgentConfigPreset, AgentServerMessageType

load_dotenv()

async def main():
    client = VoiceAgentClient(
        api_key=os.getenv("SPEECHMATICS_API_KEY"),
        config=VoiceAgentConfigPreset.load("adaptive")
    )

    @client.on(AgentServerMessageType.ADD_SEGMENT)
    def on_segment(message):
        for segment in message.get("segments", []):
            print(f"[{segment.get('speaker_id', 'S1')}]: {segment.get('text', '')}")

    @client.on(AgentServerMessageType.END_OF_TURN)
    def on_turn_end(message):
        print("[END OF TURN]")

    mic = Microphone(sample_rate=16000, chunk_size=320)
    mic.start()

    try:
        await client.connect()
        print("Voice agent ready. Speak now...")

        while True:
            await client.send_audio(await mic.read(320))
    finally:
        mic.stop()
        await client.disconnect()

asyncio.run(main())

pip install speechmatics-voice speechmatics-rt python-dotenv pyaudio

Tip

Ready for more? Explore 20+ working examples at Speechmatics Academy — voice agents, integrations, use cases, and migration guides.

from dotenv import load_dotenv
from livekit import agents
from livekit.agents import AgentSession, Agent, RoomInputOptions
from livekit.plugins import openai, silero, speechmatics, elevenlabs

load_dotenv()


class VoiceAssistant(Agent):
    """Voice assistant agent with Speechmatics STT."""

    def __init__(self) -> None:
        super().__init__(instructions="You are a helpful voice assistant. Be concise and friendly.")


async def entrypoint(ctx: agents.JobContext):
    """
    Main entrypoint for the voice assistant.

    Pipeline: LiveKit Room - Speechmatics STT - OpenAI LLM - ElevenLabs TTS - LiveKit Room
    """
    await ctx.connect()

    # Speech to text: Speechmatics with speaker diarization
    stt = speechmatics.STT(
        language="en",
        speaker_active_format="<{speaker_id}>{text}</{speaker_id}>",
        focus_speakers=["S1"],
    )

    # Language Model: OpenAI
    llm = openai.LLM(model="gpt-4o-mini")

    # Text-to-Speech: ElevenLabs
    tts = elevenlabs.TTS(voice_id="21m00Tcm4TlvDq8ikWAM")

    # Voice Activity Detection: Silero
    vad = silero.VAD.load()

    # Create and start Agent Session
    session = AgentSession(stt=stt, llm=llm, tts=tts, vad=vad)
    await session.start(
        room=ctx.room,
        agent=VoiceAssistant(),
        room_input_options=RoomInputOptions(),
    )

    # Send initial greeting
    await session.generate_reply(instructions="Say a short hello and ask how you can help.")


if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))

pip install livekit-agents livekit-plugins-speechmatics livekit-plugins-openai livekit-plugins-elevenlabs livekit-plugins-silero

import asyncio
import os
from dotenv import load_dotenv
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.services.openai.llm import OpenAILLMService, OpenAILLMContext
from pipecat.services.speechmatics.stt import SpeechmaticsSTTService, Language
from pipecat.services.speechmatics.tts import SpeechmaticsTTSService
from pipecat.transports.local.audio import LocalAudioTransport

load_dotenv()

async def main():
    # Configure Speechmatics STT with speaker diarization
    stt = SpeechmaticsSTTService(
        api_key=os.getenv("SPEECHMATICS_API_KEY"),
        params=SpeechmaticsSTTService.InputParams(
            language=Language.EN,
            speaker_active_format="@{speaker_id}: {text}"
        )
    )

    # Configure Speechmatics TTS
    tts = SpeechmaticsTTSService(
        api_key=os.getenv("SPEECHMATICS_API_KEY"),
        voice_id="sarah"
    )

    # Configure LLM (OpenAI, Anthropic, etc.)
    llm = OpenAILLMService(
        api_key=os.getenv("OPENAI_API_KEY"),
        model="gpt-4o"
    )

    # Set up conversation context
    context = OpenAILLMContext([
        {"role": "system", "content": "You are a helpful AI assistant."}
    ])
    context_aggregator = llm.create_context_aggregator(context)

    # Build pipeline: Audio Input -> STT -> LLM -> TTS -> Audio Output
    transport = LocalAudioTransport()
    pipeline = Pipeline([
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ])

    # Run the voice bot
    runner = PipelineRunner()
    task = PipelineTask(pipeline)

    print("Voice bot ready! Speak into your microphone...")
    await runner.run(task)

asyncio.run(main())

pip install pipecat-ai[speechmatics, openai] pyaudio

from speechmatics.models import BatchTranscriptionConfig
from speechmatics.batch_client import BatchClient

with BatchClient("API_KEY") as client:
    job_id = client.submit_job("audio.wav", BatchTranscriptionConfig("en"))
    transcript = client.wait_for_completion(job_id, transcription_format='txt')
    print(transcript)

import asyncio
from speechmatics.batch import AsyncClient, TranscriptionConfig, FormatType

async def main():
    client = AsyncClient(api_key="API_KEY")

    job = await client.submit_job(
        "audio.wav",
        transcription_config=TranscriptionConfig(language="en")
    )
    result = await client.wait_for_completion(job.id, format_type=FormatType.TXT)
    print(result)

    await client.close()

asyncio.run(main())

import asyncio
import os
from dotenv import load_dotenv
from speechmatics.batch import AsyncClient, TranscriptionConfig

load_dotenv()

async def main():
    api_key = os.getenv("SPEECHMATICS_API_KEY")
    client = AsyncClient(api_key=api_key)

    # Use medical domain for better accuracy with clinical terminology
    job = await client.submit_job(
        "patient_interview.wav",
        transcription_config=TranscriptionConfig(
            language="en",
            domain="medical",
            additional_vocab=[
                {"content": "hypertension"},
                {"content": "metformin"},
                {"content": "echocardiogram"},
                {"content": "MRI", "sounds_like": ["M R I"]},
                {"content": "CT scan", "sounds_like": ["C T scan"]}
            ]
        )
    )

    result = await client.wait_for_completion(job.id)
    print(f"Transcript:\n{result.transcript_text}")

    await client.close()

asyncio.run(main())

pip install speechmatics-batch python-dotenv

import asyncio
import os
from dotenv import load_dotenv
from speechmatics.rt import Microphone
from speechmatics.voice import (
    VoiceAgentClient,
    VoiceAgentConfigPreset,
    AgentServerMessageType,
)

load_dotenv()

async def main():
    api_key = os.getenv("SPEECHMATICS_API_KEY")

    # Load a preset configuration (options: adaptive, scribe, captions, external, fast)
    config = VoiceAgentConfigPreset.load("adaptive")

    # Initialize microphone
    mic = Microphone(sample_rate=16000, chunk_size=320)
    if not mic.start():
        print("PyAudio not available - install with: pip install pyaudio")
        return

    # Create voice agent client
    client = VoiceAgentClient(api_key=api_key, config=config)

    @client.on(AgentServerMessageType.ADD_SEGMENT)
    def on_segment(message):
        for segment in message.get("segments", []):
            speaker_id = segment.get("speaker_id", "S1")
            text = segment.get("text", "")
            print(f"[{speaker_id}]: {text}")

    @client.on(AgentServerMessageType.END_OF_TURN)
    def on_turn_end(message):
        print("[END OF TURN]")

    try:
        await client.connect()
        print("Voice agent started. Speak into your microphone (Ctrl+C to stop)...")

        while True:
            audio_chunk = await mic.read(320)
            await client.send_audio(audio_chunk)

    except KeyboardInterrupt:
        print("\nStopping...")
    finally:
        mic.stop()
        await client.disconnect()

asyncio.run(main())

pip install speechmatics-voice speechmatics-rt python-dotenv pyaudio

Caution

Security Best Practice: Never hardcode API keys in your source code. Always use environment variables or secure secret management systems.

export SPEECHMATICS_API_KEY="your_api_key_here"

import asyncio
import os
from dotenv import load_dotenv
from speechmatics.batch import AsyncClient

load_dotenv()

async def main():
    client = AsyncClient(api_key=os.getenv("SPEECHMATICS_API_KEY"))
    # Use client here
    # ...
    await client.close()

asyncio.run(main())

Warning

Browser Security: For browser-based transcription, always use temporary JWT tokens to avoid exposing your long-lived API key. Pass the token as a query parameter: wss://eu2.rt.speechmatics.com/v2?jwt=<token>

import asyncio
from speechmatics.batch import AsyncClient, JWTAuth

async def main():
    # Generate temporary token (expires after ttl seconds)
    auth = JWTAuth(api_key="your_api_key", ttl=3600)
    client = AsyncClient(auth=auth)
    # Use client here
    # ...
    await client.close()

asyncio.run(main())

import asyncio
from speechmatics.rt import AsyncClient, ConnectionConfig

async def main():
    # Configure WebSocket connection parameters
    conn_config = ConnectionConfig(
        ping_timeout=60.0,      # Timeout waiting for pong response (seconds)
        ping_interval=20.0,     # Interval for WebSocket ping frames (seconds)
        open_timeout=30.0,      # Timeout for establishing connection (seconds)
        close_timeout=10.0      # Timeout for closing connection (seconds)
    )

    client = AsyncClient(
        api_key="KEY",
        url="wss://eu2.rt.speechmatics.com/v2",
        conn_config=conn_config
    )
    # Use client here
    # ...
    await client.close()

asyncio.run(main())

import asyncio
from speechmatics.batch import AsyncClient, TranscriptionConfig
from speechmatics.batch import BatchError, JobError, AuthenticationError
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def transcribe_with_retry(audio_file):
    client = AsyncClient(api_key="YOUR_API_KEY")
    try:
        job = await client.submit_job(
            audio_file,
            transcription_config=TranscriptionConfig(language="en")
        )
        return await client.wait_for_completion(job.id)
    except AuthenticationError:
        print("Authentication failed")
        raise
    except (BatchError, JobError) as e:
        print(f"Transcription failed: {e}")
        raise
    finally:
        await client.close()

asyncio.run(transcribe_with_retry("audio.wav"))

import asyncio
from speechmatics.batch import AsyncClient, ConnectionConfig

async def main():
    # Configure HTTP connection settings for batch API
    conn_config = ConnectionConfig(
        connect_timeout=30.0,      # Timeout for connection establishment
        operation_timeout=300.0    # Default timeout for API operations
    )

    client = AsyncClient(
        api_key="KEY",
        conn_config=conn_config
    )
    # Use client here
    # ...
    await client.close()

asyncio.run(main())

import asyncio
from speechmatics.batch import AsyncClient

async def main():
    client = AsyncClient(api_key="YOUR_API_KEY")
    # Uses global SaaS endpoints automatically
    await client.close()

asyncio.run(main())

docker pull speechmatics/transcription-engine:latest
docker run -p 9000:9000 speechmatics/transcription-engine

import asyncio
from speechmatics.batch import AsyncClient

async def main():
    client = AsyncClient(
        api_key="YOUR_LICENSE_KEY",
        url="http://localhost:9000/v2"
    )
    # Use on-premises instance
    await client.close()

asyncio.run(main())

# Install the sm-realtime chart
helm upgrade --install speechmatics-realtime \
  oci://speechmaticspublic.azurecr.io/sm-charts/sm-realtime \
  --version 0.7.0 \
  --set proxy.ingress.url="speechmatics.example.com"

# 1. Install (30 seconds)
pip install speechmatics-batch python-dotenv

# 2. Set API key (30 seconds)
export SPEECHMATICS_API_KEY="your_key_here"

# 3. Run test (4 minutes)
python3 << 'EOF'
import asyncio
import os
from speechmatics.batch import AsyncClient, TranscriptionConfig, AuthenticationError
from dotenv import load_dotenv

load_dotenv()

async def test():
    api_key = os.getenv("SPEECHMATICS_API_KEY")

    # Replace with your audio file path
    audio_file = "your_audio_file.wav"

    client = AsyncClient(api_key=api_key)
    try:
        print("Submitting transcription job...")
        job = await client.submit_job(audio_file, transcription_config=TranscriptionConfig(language="en"))
        print(f"Job submitted: {job.id}")

        print("Waiting for completion...")
        result = await client.wait_for_completion(job.id)

        print(f"\nTranscript: {result.transcript_text}")
        print("\nTest completed successfully!")

    except AuthenticationError as e:
        print(f"\nAuthentication Error: {e}")
    finally:
        await client.close()

asyncio.run(test())
EOF