Recipe: Speech, Transcription and Media Questions

Goal

Use the one-shot multimedia calls end to end: speak a sentence to a WAV file (or stream it as it is generated, including an agent's answer), transcribe it back (whole and streamed), ask a model a question about the recording, look at a file locally without a model, and generate and then edit an image — reading the cost of each call as you go.

Every function here lives in doc_processing_toolkit.multimedia, validates its options against the registry before sending anything, and returns a plain dict rather than raising on a provider failure. Multimedia explains the rules; this page is the walkthrough.

Prerequisites

• OPENAI_API_KEY for gpt-4o-mini-tts, gpt-transcribe and gpt-image-2 (the organisation must be verified for GPT Image models), and GOOGLE_GEMINI_API_KEY for gemini-3.5-flash-lite and the Gemini 3.8 TTS rows. Keys are read from .env at the repo root.
• Optional: the [audio] extra, for transcribing containers a provider does not accept (see Audio the provider does not accept).

The snippets run top to bottom as one script; the file written in the first step is read by the later ones.

Speech out: save a WAV

from doc_processing_toolkit.multimedia import synthesize_speech

speech = synthesize_speech(
    "gpt-4o-mini-tts",
    "The invoice total is four hundred and twelve dollars.",
    voice="coral",
    output_format="wav",
    instructions="Speak calmly and clearly.",
    output_path="invoice_total.wav",
)

print(speech["output_path"], speech["mime_type"])   # invoice_total.wav audio/wav
print(speech["cost"], speech["cost_estimated"])     # 0.0011964 False

voice and output_format are checked against the row's supported_voices and supported_output_formats, so a typo fails before the request. cost_estimated is False: the toolkit asks OpenAI for the audio as an SSE stream, whose last event carries the provider's own token counts, and joins the audio chunks back into one clip. DPT_OPENAI_TTS_SSE_USAGE=off restores the plain audio request, whose cost is estimated from the clip length (cost_estimated True). Gemini TTS rows report real token counts.

Style, dialogue and custom voices on Gemini 3.8

styled = synthesize_speech(
    "gemini-3.8-flash-lite-tts",
    "The invoice total is four hundred and twelve dollars. <short pause> Thank you.",
    voice="Kore",
    instructions="Warm and unhurried.",
)
print(styled["mime_type"], styled["cost_estimated"])   # audio/wav False

dialogue = synthesize_speech(
    "gemini-3.8-flash-tts",
    "Joe: Did the invoice arrive?\nJane: Yes, this morning.\nI filed it straight away.",
    multi_speaker_config=[{"speaker": "Joe", "voice": "Puck"}, {"speaker": "Jane", "voice": {"id": "en-us-advisor-1"}}],
    output_path="invoice_dialogue.wav",
)

The Gemini 3.8 rows speak the text exactly as written, so the toolkit sends instructions as the part's style instead of prepending them: transcribing styled returns only the sentence. <short pause> is one of the inline tags these rows understand. In the dialogue, each Name: line starts a turn, the line without a label continues Jane's turn, and the labels themselves are not spoken. A voice can be a prebuilt name or {"id": ...} from Google's Voices API. output_format="pcm" returns the raw samples (audio/l16;rate=24000) instead of a WAV.

Speak as it is generated

synthesize_speech_stream yields the audio in chunks while the rest is still being made, so playback can start in about a second instead of after the whole clip:

import wave

from doc_processing_toolkit.multimedia import synthesize_speech_stream

with wave.open("streamed.wav", "wb") as out:
    out.setnchannels(1)
    out.setsampwidth(2)
    out.setframerate(24000)
    for frame in synthesize_speech_stream(
        "gemini-3.8-flash-lite-tts", "The invoice total is four hundred and twelve dollars.", voice="Kore"
    ):
        if frame["type"] == "audio":
            out.writeframes(frame["data"])      # or hand it to a player as it arrives
        elif frame["type"] == "done":
            print(frame["cost"], frame["cost_estimated"], frame["chunks"])

A Google stream is raw 24 kHz 16-bit mono PCM (audio/l16;rate=24000); on OpenAI ask for output_format="pcm" to get the same shape. The cost arrives in the done frame. On Google a streamed clip reports more audio tokens than the same clip made with synthesize_speech (see Streaming speech), so stream for playback, not for files.

Speak an agent's answer as it streams

An agent does not stream speech itself: its multimedia tools are unary. For a voice UI, stream the agent's text, cut it into sentences, and speak each sentence as soon as it is complete. The first sentence plays while the agent is still writing the rest.

import asyncio
import re
from contextlib import aclosing

from doc_processing_toolkit import Agent
from doc_processing_toolkit.multimedia import asynthesize_speech_stream

SENTENCE_END = re.compile(r"(?<=[.!?])\s+")


async def speak(sentence: str, play) -> float:
    cost = 0.0
    async with aclosing(asynthesize_speech_stream("gpt-4o-mini-tts", sentence, output_format="pcm")) as frames:
        async for frame in frames:
            if frame["type"] == "audio":
                await play(frame["data"])
            elif frame["type"] == "done":
                cost = frame["cost"]
    return cost


async def answer_aloud(agent: Agent, question: str, play) -> float:
    buffer, total = "", 0.0
    async for event in agent.stream(question):
        if event.event_type == "text_delta" and not event.metadata:   # prose, not a JSON payload
            buffer += event.content
            *finished, buffer = SENTENCE_END.split(buffer)
            for sentence in finished:
                total += await speak(sentence, play)
        elif event.event_type == "complete":
            total += event.metadata["cost"]      # the agent run itself
    if buffer.strip():
        total += await speak(buffer, play)
    return total

Each sentence is its own TTS request with its own cost, so the function adds them to the run's cost. Here speech and text alternate; to keep reading the agent's stream while a sentence plays, put the sentences on an asyncio.Queue and speak them from a second task.

An agent with tools, like most, streams its answer as prose. An agent with no tools answers through the structured-output path, where the deltas are a JSON payload tagged metadata == {"structured_output": True}; the check above skips them, and that agent's answer is event.metadata["data"] on complete, to speak in one piece.

Look at a file without a model

from doc_processing_toolkit.multimedia import inspect_media

facts = inspect_media("invoice_total.wav")
facts["kind"], facts["mime_type"], facts["duration_s"]   # ('audio', 'audio/wav', 4.05)

models = inspect_media("invoice_total.wav", include_models=True)["accepted_by"]
# every registry row whose input_types accept audio

inspect_media makes no network call. It reads the container, the size, image dimensions or a WAV duration from the bytes themselves, and with include_models=True lists the rows that would take the file — the cheap way to answer "can I send this to that model?" before paying to find out.

Speech in: transcribe a recording

from doc_processing_toolkit.multimedia import transcribe_audio

text = transcribe_audio("gpt-transcribe", "invoice_total.wav")

print(text["text"])               # The invoice total is four hundred and twelve dollars.
print(text["duration_minutes"])   # 0.0833...
print(text["duration_source"])    # 'provider'
print(text["cost"], text["cost_estimated"])   # 0.000375 False

gpt-transcribe is billed per minute, so the clip length is the price, and duration_source says where the length came from. The precedence is fixed: "provider" (the model's own measurement), then "caller" (a duration_seconds you pass), then "wav_header", then "byte_estimate" — and only the last one sets cost_estimated.

Pass duration_seconds whenever you know the length. It does not override a provider measurement — gpt-transcribe reports its billed length, so the source stays "provider" below — but on a model or a response that reports none it replaces the byte heuristic, which under-counts a compressed browser recording about tenfold:

from pathlib import Path

raw = Path("invoice_total.wav").read_bytes()
text = transcribe_audio("gpt-transcribe", raw, duration_seconds=4)
print(text["duration_source"], text["cost_estimated"])   # provider False

See Transcription duration for the full table.

Speaker labels come from gpt-4o-transcribe-diarize. For a clip longer than 30 seconds OpenAI requires a chunking strategy. The row supplies "auto" for you, and a provider_options value replaces it:

meeting = transcribe_audio("gpt-4o-transcribe-diarize", "standup.wav", diarization=True)
# chunking_strategy="auto" went out with the request; prompt= is refused on this model

tuned = transcribe_audio(
    "gpt-4o-transcribe-diarize",
    "standup.wav",
    diarization=True,
    provider_options={"chunking_strategy": {"type": "server_vad", "silence_duration_ms": 500}},
)

Stream the transcript

import asyncio
from doc_processing_toolkit.multimedia import atranscribe_audio_stream

async def stream():
    async for frame in atranscribe_audio_stream("gpt-transcribe", raw):
        if frame["type"] == "delta":
            print(frame["text"], end="", flush=True)
        elif frame["type"] == "done":
            print()
            print(frame["cost"], frame["duration_source"])   # 0.000375 provider

asyncio.run(stream())

Frames are delta, segment, an optional error, and exactly one done that carries the same keys transcribe_audio returns. transcribe_audio(..., stream=True) consumes the same stream into one result dict instead. Streaming is about the output: the recording must already exist. For live microphone capture, open a transcription session.

Streaming is gated on the registry's supports_streaming_transcription: gpt-transcribe and gpt-4o-transcribe-diarize have it, whisper-1 and gemini-3.5-transcribe do not and are refused locally.

Audio the provider does not accept

A browser usually records WebM/Opus or MP4, and not every provider accepts those containers. When the sniffed container is not on the model's accepted list, transcribe_audio transcodes it and reports converted_from. The converter is optional, so an install stays lean:

pip install "doc-processing-toolkit[audio]"    # a bundled ffmpeg via imageio-ffmpeg

from doc_processing_toolkit.audio_convert import converter_available

converter_available()   # 'ffmpeg', 'pyav', or None

Discovery order is DPT_FFMPEG_PATH (an explicit path to an ffmpeg binary), then ffmpeg on PATH, then the binary the [audio] extra ships, then PyAV. With none of them present, an unaccepted container raises MultimediaValidationError naming the extra instead of reaching the provider as a 400. DPT_AUDIO_CONVERT_TIMEOUT and DPT_AUDIO_CONVERT_MAX_MB bound one transcode — see Configuration.

Ask a model about a file

from doc_processing_toolkit.multimedia import describe_media

answer = describe_media(
    "gemini-3.5-flash-lite",
    "invoice_total.wav",
    "What amount of money is mentioned? Answer with the number only.",
)

print(answer["text"])                        # 412
print(answer["cost"], answer["cost_estimated"])
print(answer["media"]["kind"])               # 'audio' — the inspect_media record
print(answer["duration_s"])                  # wall-clock seconds the call took, not the clip length

describe_media checks the file's kind against the model's input_types before uploading it, so a PDF sent to an image-only row fails locally instead of being billed. Google's video knobs (start_offset, end_offset, fps, media_resolution) raise on any other provider rather than being dropped.

Generate and edit an image

from doc_processing_toolkit.multimedia import edit_image, generate_image, image_size_for

size = image_size_for("gpt-image-2", aspect_ratio="3:2", target="1K")   # '1248x832', no API call

image = generate_image(
    "gpt-image-2",
    "A flat icon of a paper invoice with a green tick",
    size=size,
    quality="low",
    output_dir="images",
)
print(image["paths"])            # ['images/generated_image_YYYYMMDD_HHMMSS_ffffff_1.png']
print(image["cost"])             # 0.004105
print(image["cost_breakdown"])   # named lines that sum to cost

edited = edit_image(
    "gpt-image-2",
    "Make the tick red",
    image=image["images"][0],
    size=size,
    quality="low",
    output_dir="images",
)
print(edited["cost"])            # 0.012182 — the source image is billed as image input

image_size_for solves a size the model accepts: GPT Image 2 publishes a dimension rule (multiples of 16, ratio at most 3:1, a pixel range) rather than a preset list, and the helper returns the closest valid WIDTHxHEIGHT for a target budget of "1K", "2K" or "4K". On a preset model it returns the nearest listed preset. Gemini image rows take a resolution class (size="1K") plus aspect_ratio instead of pixels.

Read the returned paths rather than predicting file names. Every image result carries input_tokens, output_tokens and a cost_breakdown (cost_input_text_usd, cost_input_image_usd, cost_text_usd, cost_image_usd, cost_cache_discount_usd) that sums to cost.

A named parameter beats provider_options on the image calls

If you set the same key both ways, generate_image / edit_image keep the named parameter and log the discarded passthrough value; synthesize_speech / transcribe_audio let provider_options win. See A named parameter beats provider_options.

To keep editing the same picture across turns — or across processes — use an ImageSession instead of chaining edit_image calls.

Errors come back in the result

Each result carries error, error_code and retryable. A provider failure does not raise:

if image["error"]:
    print(image["error_code"], image["retryable"])

moderation_blocked is never retryable; empty_response, timeout, connection_error, a 429 and any 5xx are. Invalid options — an unknown voice, a size outside the rule, a mask on a Google model — raise MultimediaValidationError before any request.

The same calls over MCP

python -m doc_processing_toolkit.mcp_servers.multimedia_server exposes these capabilities as MCP tools, so an agent can use them without importing the toolkit. Each tool is a thin wrapper around the Python call above, with the same validation and the same cost fields.

Python call	MCP tool	What changes at the JSON boundary
synthesize_speech	synthesize_speech	multi_speaker_config arrives as multi_speaker_config_json; the audio is written to a file and the result carries output_path, never the bytes
transcribe_audio	transcribe_audio	the recording is audio_path or audio_base64; list options are *_json strings (custom_vocabulary_json, languages_json, keywords_json); the transcript is also saved as JSON, transcript_path
atranscribe_audio_stream	—	not exposed: a tool call returns once; transcribe_audio(stream=True) is the tool form
generate_image	generate_image	output_compression=-1 means "unset"; the images are written to files and the result carries paths
edit_image	edit_image	images are image_path / image_base64 (or image_paths_json / images_base64_json for several); the mask is mask_path / mask_base64
image_size_for	image_size_for	returns {size, model, error, error_code, retryable}; no API call
describe_media, inspect_media	—	Python only
ImageSession	—	Python only

Validation failures come back as {"error": ...} results rather than exceptions, so a calling agent can read the limitation and adapt. See Multimedia MCP Server.

Since 0.5.1 the tools write files and never return base64 (a model can neither hear nor see it, and it floods the context). The folder is the call's output_dir, else DPT_MULTIMEDIA_OUTPUT_DIR, else <system temp>/doc_processing_toolkit/multimedia (per-user on Linux and macOS); include_base64 is accepted and ignored. See Files, not base64.

To give an agent only some of the tools, list them in the tool entry's allowed_tools (0.5.1). For example, {"type": "fastmcp", "server": multimedia_server, "allowed_tools": ["generate_image"]} offers only generate_image. See Choosing which tools a server exposes.

from doc_processing_toolkit.mcp_servers.multimedia_server import generate_image, transcribe_audio

image = generate_image(model="gpt-image-2", prompt="A lighthouse at dusk", output_dir="out/")
image["paths"], image["images_base64"]        # ['out/generated_image_..._1.png'], []

notes = transcribe_audio(model="gpt-transcribe", audio_path="meeting.wav", output_dir="out/")
notes["text"], notes["transcript_path"]        # the text, and 'out/transcript_gpt-transcribe_....json'

What to validate

• speech["output_path"] exists and inspect_media reads a non-zero duration_s from it
• the transcript matches the spoken sentence, and duration_source is not "byte_estimate"
• the streamed transcription's done frame carries the same cost as transcribe_audio
• a streamed speech clip's done frame carries a cost. On OpenAI it matches synthesize_speech. On Google it is higher: about 25% on the Gemini 3.8 rows (see Streaming speech)
• describe_media answers from the audio, and answer["media"]["kind"] == "audio"
• image["cost_breakdown"] values sum to image["cost"], and the edit's cost_input_image_usd is non-zero

Test sources

• tests/unit/test_multimedia.py, tests/unit/test_multimedia_provider_calls.py
• tests/unit/test_transcription_duration.py, tests/unit/test_transcription_gemini_and_streaming.py
• tests/unit/test_describe_media.py, tests/unit/test_multimedia_server.py
• tests/unit/test_tts_characterization.py, tests/unit/test_tts_gemini38_and_sse.py
• tests/unit/test_tts_stream.py, tests/unit/test_multimedia_file_output.py
• tests/integration/test_multimedia_live.py, tests/integration/test_multimedia_server_live.py

Related pages

• Multimedia — validation, the size solver, duration precedence, cost schemas
• Multi-turn Image Sessions — generate and keep editing across turns
• Realtime Voice Session — live speech, captions and interpretation
• MCP — the multimedia MCP server