Speechmatics Batch API Client

Python client for Speechmatics Batch API, with both async and blocking interfaces.

Migrating from speechmatics-python (the legacy BatchClient)? See MIGRATION.md for a full guide, including a method-by-method mapping table.

Features

• Async (AsyncClient) and blocking (Client) API clients with comprehensive error handling
• Synchronous transcription support: get a transcript in a single request, without polling
• Type hints throughout for better IDE support
• Environment variable support for credentials
• Easy-to-use interface for submitting, monitoring, and retrieving transcription jobs
• Full job configuration support with all Speechmatics features
• Intelligent transcript formatting with speaker diarization
• Support for multiple output formats (JSON, TXT, SRT)

Installation

pip install speechmatics-batch

Usage

Quick Start

from speechmatics.batch import Client

# Create a client using environment variable SPEECHMATICS_API_KEY
with Client() as client:
    # Simple transcription
    result = client.transcribe("audio.wav")
    print(result.transcript_text)

Async

AsyncClient is the async/await equivalent of Client, for code that already runs an event loop. It exposes the same methods and returns the same models:

import asyncio
from speechmatics.batch import AsyncClient

async def main():
    async with AsyncClient() as client:
        result = await client.transcribe("audio.wav")
        print(result.transcript_text)

asyncio.run(main())

Synchronous Transcription

By default a transcription job is submitted, polled until it finishes, and then its transcript is fetched. For short audio you can instead ask the server to hold the request open until the transcript is ready, so one call replaces the whole cycle. Pass wait (in seconds) to do this:

from speechmatics.batch import Client, FormatType

with Client() as client:
    # One request: submit, transcribe and return the transcript
    text = client.transcribe("audio.wav", wait=60, format_type=FormatType.TXT)
    print(text)

If the job is still running when the wait elapses, transcribe() falls back to polling automatically, so longer audio keeps working unchanged.

wait is also available on the individual operations, for full control:

from speechmatics.batch import Client, JobStatus, TranscriptNotReadyError

with Client() as client:
    job = client.submit_job("audio.wav", wait=60)

    if job.status == JobStatus.DONE:
        print(job.transcript.transcript_text)  # already available, no extra call
    else:
        # status is JobStatus.CREATED: the wait elapsed, the job is still running
        print(client.wait_for_completion(job.id).transcript_text)

Requesting a transcript before it exists raises TranscriptNotReadyError (a subclass of JobError), which is the signal to retry:

try:
    transcript = client.get_transcript(job.id, wait=30)
except TranscriptNotReadyError:
    transcript = client.wait_for_completion(job.id)

Notes:

• Synchronous transcription is available on Speechmatics SaaS only. on-premises deployments do not support it.
• The server caps how long it will wait, and intermediate proxies may close long-held connections, so treat the fallback path as the normal case for longer audio.
• The API applies a small default wait to the GET endpoints when wait is omitted. Pass wait=0 to return immediately.

Everything above works identically on AsyncClient with await:

async with AsyncClient() as client:
    text = await client.transcribe("audio.wav", wait=60, format_type=FormatType.TXT)

Polling and Timeouts

When a transcript isn't returned by wait, the client polls the job status until it finishes. Polling starts at min_polling_interval and backs off towards polling_interval, so short jobs are picked up quickly without long jobs making hundreds of requests:

result = client.wait_for_completion(
    job.id,
    min_polling_interval=0.5,  # first gap between status checks
    polling_interval=5.0,      # ceiling the backoff climbs to
    timeout=3600.0,            # give up after an hour
)

Both intervals must be greater than 0, and up to 20% jitter is applied to each wait so that concurrent clients don't synchronise into bursts.

Waiting is bounded by default: timeout is one hour unless you change it. Pass timeout=None only if a job that never reaches a terminal state should block indefinitely.

A long wait makes many status requests, so a single failed one doesn't abandon the job: connection errors, request timeouts and HTTP 408/429/5xx are retried, up to 5 consecutive failures. Failures that are an answer rather than a blip — bad credentials, an unknown job, an expired job — are raised straight away.

Note that the API may also hold each status request open briefly before answering it, and the SDK doesn't depend on how long that is. The intervals above control what the client adds on top, so the time between checks can be longer than the interval you set.

JWT Authentication

For enhanced security, use temporary JWT tokens instead of static API keys. JWTs are short-lived (60 seconds default) and automatically refreshed:

from speechmatics.batch import AsyncClient, JWTAuth

auth = JWTAuth("your-api-key", ttl=60)

async with AsyncClient(auth=auth) as client:
    # Tokens are cached and auto-refreshed automatically
    result = await client.transcribe("audio.wav")
    print(result.transcript_text)

Ideal for long-running applications or when minimizing API key exposure. See the authentication documentation for more details.

Basic Job Workflow

import asyncio
from speechmatics.batch import AsyncClient, JobConfig, JobType, TranscriptionConfig

async def main():
    # Create client with explicit API key
    async with AsyncClient(api_key="your-api-key") as client:

        # Configure transcription
        config = JobConfig(
            type=JobType.TRANSCRIPTION,
            transcription_config=TranscriptionConfig(
                language="en",
                enable_entities=True,
                diarization="speaker"
            )
        )

        # Submit job
        job = await client.submit_job("audio.wav", config=config)
        print(f"Job submitted: {job.id}")

        # Wait for completion
        result = await client.wait_for_completion(
            job.id,
            polling_interval=2.0,
            timeout=300.0
        )

        # Access results
        print(f"Transcript: {result.transcript_text}")
        print(f"Confidence: {result.confidence}")

asyncio.run(main())

Advanced Configuration

import asyncio
from speechmatics.batch import (
    AsyncClient,
    JobConfig,
    JobType,
    Model,
    TranscriptionConfig,
    TranslationConfig,
    SummarizationConfig
)

async def main():
    async with AsyncClient(api_key="your-api-key") as client:

        # Advanced job configuration
        config = JobConfig(
            type=JobType.TRANSCRIPTION,
            transcription_config=TranscriptionConfig(
                language="en",
                model=Model.ENHANCED,
                enable_entities=True,
                diarization="speaker",
            ),
            translation_config=TranslationConfig(target_languages=["es", "fr"]),
            summarization_config=SummarizationConfig(
                content_type="conversational", summary_length="brief"
            ),
        )

        result = await client.transcribe("audio.wav", config=config)

        # Access advanced features
        if result.summary:
            print(f"Summary: {result.summary}")
        if result.translations:
            print(f"Translations: {result.translations}")

asyncio.run(main())

Manual Job Management

import asyncio
from speechmatics.batch import AsyncClient, JobStatus

async def main():
    async with AsyncClient() as client:

        # Submit job
        job = await client.submit_job("audio.wav")

        # Check job status
        job_details = await client.get_job_info(job.id)
        print(f"Status: {job_details.status}")

        # Wait for completion manually
        while job_details.status == JobStatus.RUNNING:
            await asyncio.sleep(5)
            job_details = await client.get_job_info(job.id)

        if job_details.status == JobStatus.DONE:
            # Get transcript
            transcript = await client.get_transcript(job.id)
            print(transcript.transcript_text)
        else:
            print(f"Job failed with status: {job_details.status}")

asyncio.run(main())

Bulk and Concurrent Transcription

To transcribe many files with a concurrency cap, use asyncio.gather with a semaphore (AsyncClient). concurrency limits how many jobs are in flight at once — the example below submits 8 files with a cap of 5, so at most 5 run concurrently and the rest queue behind the semaphore:

import asyncio
from speechmatics.batch import AsyncClient, JobConfig, JobType, TranscriptionConfig

async def transcribe_all(paths, concurrency=5):
    config = JobConfig(type=JobType.TRANSCRIPTION, transcription_config=TranscriptionConfig(language="en"))
    semaphore = asyncio.Semaphore(concurrency)

    async with AsyncClient() as client:
        async def run(path):
            async with semaphore:
                job = await client.submit_job(path, config=config)
                return path, await client.wait_for_completion(job.id)

        return await asyncio.gather(*(run(path) for path in paths))

paths = [f"audio_{i}.wav" for i in range(8)]
results = asyncio.run(transcribe_all(paths))

The equivalent with the blocking Client uses a thread pool, since each transcribe() call blocks on network I/O. max_workers plays the same role as concurrency above — it caps how many requests run at once, regardless of how many paths are submitted:

from concurrent.futures import ThreadPoolExecutor
from speechmatics.batch import Client, JobConfig, JobType, TranscriptionConfig

def transcribe_all(paths, concurrency=5):
    config = JobConfig(type=JobType.TRANSCRIPTION, transcription_config=TranscriptionConfig(language="en"))

    with Client() as client, ThreadPoolExecutor(max_workers=concurrency) as pool:
        futures = {pool.submit(client.transcribe, path, config=config): path for path in paths}
        return [(futures[future], future.result()) for future in futures]

paths = [f"audio_{i}.wav" for i in range(8)]
results = transcribe_all(paths)

Different Output Formats

import asyncio
from speechmatics.batch import AsyncClient, FormatType

async def main():
    async with AsyncClient() as client:
        job = await client.submit_job("audio.wav")

        # Get JSON format (default)
        json_result = await client.get_transcript(job.id, format_type=FormatType.JSON)
        print(json_result.transcript_text)

        # Get plain text
        txt_result = await client.get_transcript(job.id, format_type=FormatType.TXT)
        print(txt_result)

        # Get SRT subtitles
        srt_result = await client.get_transcript(job.id, format_type=FormatType.SRT)
        print(srt_result)

asyncio.run(main())

Error Handling

import asyncio
from speechmatics.batch import (
    AsyncClient,
    BatchError,
    AuthenticationError,
    JobError,
    TimeoutError
)

async def main():
    try:
        async with AsyncClient() as client:
            result = await client.transcribe("audio.wav", timeout=120.0)
            print(result.transcript_text)

    except AuthenticationError:
        print("Invalid API key")
    except BatchError as e:
        print(f"Job submission failed: {e}")
    except JobError as e:
        print(f"Job processing failed: {e}")
    except TimeoutError as e:
        print(f"Job timed out: {e}")
    except FileNotFoundError:
        print("Audio file not found")

asyncio.run(main())

Connection Configuration

import asyncio
from speechmatics.batch import AsyncClient, ConnectionConfig

async def main():
    # Custom connection settings
    config = ConnectionConfig(
        url="https://asr.api.speechmatics.com/v2",
        api_key="your-api-key",
        connect_timeout=30.0,
        operation_timeout=600.0
    )

    async with AsyncClient(conn_config=config) as client:
        result = await client.transcribe("audio.wav")
        print(result.transcript_text)

asyncio.run(main())

Logging

The client supports logging with job id tracing for debugging. To increase logging verbosity, set DEBUG level in your example code:

import logging
import sys

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout)
    ]
)

Environment Variables

The client supports the following environment variables:

• SPEECHMATICS_API_KEY: Your Speechmatics API key
• SPEECHMATICS_BATCH_URL: Custom API endpoint URL (optional)