Toolkit Core API

Core Token and Cost Utilities

doc_processing_toolkit.core

Token counting and cost calculation utilities.

This module provides provider-aware token counting helpers and cost calculations used by provider call paths and orchestration utilities.

See Also

• Core API reference
• Cost tracking and reporting
• Core module overview

async_count_tokens_anthropic(provider, model_name, content_parts, client, logger) async

Estimates or calculates token count for input content (text and potentially other types) using Anthropic-specific methods.

Provides accurate token counts where possible via API methods, with appropriate fallbacks to approximation methods. Logs warnings if specific methods fail or approximation is used. For multimodal input, local counting is often an estimate; API metadata (usage_metadata or usage) after a call provides the most accurate counts.

Parameters:

Name	Type	Description	Default
model_name	str	The specific model name (e.g. "claude-4-5-sonnet-latest").	required
content_parts	list or str	List containing strings (for text) and/or dictionaries for media. Tool Definitions and Output Response Schema's should be passed as strings. str: Text content (e.g., "Hello, world!"). Supported dictionary formats: dict: Structured content with 'type' field: base64 Image: {'type': 'base64', 'media_type': 'image/png', 'data': str} (media_type fixed to 'image/png') Document: {'type': 'base64', 'media_type': 'application/pdf', 'data': str} (media_type fixed to 'application/pdf')	required
client	object	The initialized async API client (needed for Anthropic).	required
logger	Logger	Logger instance for logging messages and warnings.	required

Returns:

Name	Type	Description
int	int	The estimated or calculated token count for the input content. Returns 0 if content is empty or invalid.

Notes

• Anthropic: Uses client.messages.count_tokens() API when available, falls back to text approximation.

Warning: Pre-call token counting for multimodal content (text + images) is approximate and may differ from the actual token count reported after the API call, particularly for images which often use variable token counts based on resolution and complexity.

async_count_tokens_openai(provider, model_name, content_parts, client, logger, image_detail='high') async

Count tokens with the OpenAI Token Counting API, discarding the source label.

The count-only view of :func:async_count_tokens_openai_with_source, which documents the fallback chain. Use this when the count is all you need; use the _with_source form when the count will be reported, so a tiktoken estimate is not presented as an API figure.

Parameters:

Name	Type	Description	Default
provider	str	Must be "openai".	required
model_name	str	OpenAI model name (e.g. "gpt-5.4-mini").	required
content_parts	Union[List[Any], str]	List of content parts (strings, dicts, etc.).	required
client	Any	AsyncOpenAI client instance.	required
logger	Logger	Logger instance.	required
image_detail	str	Image detail level passed to the canonical content-part builder. Defaults to "high".	'high'

Returns:

Name	Type	Description
int	int	Estimated token count.

async_count_tokens_openai_with_source(provider, model_name, content_parts, client, logger, image_detail='high') async

Count tokens using the OpenAI Token Counting API (async version).

This async version is for use in streaming/retry code paths where the async OpenAI client is available. Uses the same Token Counting API as the sync version, and falls back to tiktoken — with the encoding resolved from the registry — then to a text approximation.

Parameters:

Name	Type	Description	Default
provider	str	Must be "openai".	required
model_name	str	OpenAI model name (e.g. "gpt-5.4-mini").	required
content_parts	Union[List[Any], str]	List of content parts (strings, dicts, etc.).	required
client	Any	AsyncOpenAI client instance.	required
logger	Logger	Logger instance.	required
image_detail	str	Image detail level passed to the canonical content-part builder. Must match the value used by the real API call so the pre-flight estimate is apples-to-apples with billing. Default "high" matches the toolkit's default for OpenAI calls.	'high'

Returns:

Type	Description
int	Tuple[int, str]: The token count and the token_source label naming
str	how it was obtained — "openai_token_counting_api" for the API,
Tuple[int, str]	"openai_tiktoken_fallback_*" for a local tiktoken estimate, or
Tuple[int, str]	"text_approximation_tiktoken_failed" as a last resort.

calculate_cost(model_name, input_tokens, output_tokens, built_in_tool_usage, model_info_config, logger, input_token_breakdown=None, output_token_breakdown=None, pricing_scheme=None, cached_input_tokens=0, cached_token_breakdown=None, cache_write_tokens=0, cache_write_ttl='5m', service_tier=None, audio_minutes=0.0, cost_components_out=None, tier_prompt_tokens=None)

Calculates the estimated cost based on token counts, tool usage, and loaded pricing information.

Handles tiered pricing (the registry's tiered_200k Google rows, tiered_272k OpenAI rows and tiered_100k Anthropic rows), provider-specific billing rules (e.g., Anthropic 5-minute minimum billing), and tool costs for accurate billing calculations.

Parameters:

Name	Type	Description	Default
model_name	str	The specific model name used (e.g., "gpt-4o").	required
input_tokens	int	The number of input tokens.	required
output_tokens	int	The number of output tokens.	required
built_in_tool_usage	dict	Dictionary containing built-in tool usage information with keys like: 'tools_used': List of tools that were used 'search_count': Number of searches performed (for web search tools) 'code_sessions': Number of code execution sessions 'search_context_size': Context size for search ("low", "medium", "high") 'provider': Provider name ("google", "anthropic", "openai", "fireworks")	required
model_info_config	dict	The complete model information config loaded from model_information.json.	required
logger	Logger	Logger instance for logging messages.	required
input_token_breakdown	dict	A dictionary containing the breakdown of input tokens by type. Defaults to None. Supported token types: "MODALITY_UNSPECIFIED", "TEXT", "IMAGE", "VIDEO", "AUDIO", "DOCUMENT", "UNKNOWN". The input side splits three ways — AUDIO at input_audio, IMAGE at input_image and everything left over at the text input rate — whenever the price row publishes rates for the modalities present. A row that publishes only one of them takes the same two-way split it always did.	None
output_token_breakdown	dict	A dictionary containing the breakdown of output tokens by type. Defaults to None. Supported token types: "MODALITY_UNSPECIFIED", "TEXT", "IMAGE", "VIDEO", "AUDIO", "DOCUMENT", "UNKNOWN".	None
pricing_scheme	str	Pricing scheme identifier for provider-specific billing variations for built-in tools. Defaults to None. Used primarily for Google search grounding to differentiate between: "pricing_scheme_gemini_2": Gemini 2.x models (billed per prompt) "pricing_scheme_gemini_3": Gemini 3.x models (billed per search) When None, uses provider defaults or legacy pricing.	None
cached_input_tokens	int	Number of prompt tokens served from the provider's prompt cache. Convention: this count is a subset of input_tokens (OpenAI, Google, and Fireworks all report cached tokens inside the prompt total; Anthropic call sites must pass input_tokens as uncached + cache reads to honor the same convention). The cached share is re-priced at the model's cached_input registry rate via subtraction — never added on top. On a tiered_200k / tiered_272k / tiered_100k row the tier-specific cached_input_less_than_<N>_prompt / cached_input_greater_than_<N>_prompt key is preferred when present, matching the tier this request landed in, with cached_input as the fallback. If the model has no cached rate at all, the cached share stays at the full input rate and a warning is logged (a silent 0.0 default would make cached tokens free). Defaults to 0.	0
cached_token_breakdown	dict	Modality breakdown of the cached tokens (e.g. {"AUDIO": 120, "TEXT": 900}, Google's usage_metadata.cache_tokens_details). Used only when the price row carries a cached_input_audio rate AND the input side was billed with an audio split: the cached AUDIO share is then re-priced against the audio input rate at cached_input_audio and the rest against the text input rate at the cached rate above (Gemini 2.5 Flash models publish a higher cached-audio rate). An IMAGE share works the same way against cached_input_image, and a realtime prompt carrying both is discounted three ways in one pass. Without this breakdown, or without a published per-modality cached rate, the whole cached share uses the flat cached rate as before. Defaults to None.	None
cache_write_tokens	int	Prompt-cache write tokens. Two readings, chosen by the price row. Anthropic (usage.cache_creation_input_tokens): billed in addition to input_tokens at the model's cache_write_5m / cache_write_1h registry price per cache_write_ttl — on a tiered_100k row at the cache_write_5m_<tier> / cache_write_1h_<tier> price of the tier the request landed in, the flat key being the fallback. OpenAI GPT-5.6+ (usage.input_tokens_details.cache_write_tokens, a row carrying cache_write_input_multiplier): a subset of input_tokens, disjoint from cached_input_tokens, re-priced at multiplier x the uncached input rate of the tier and service tier the request landed in — only the difference is added, in cache_write_usd. Defaults to 0.	0
cache_write_ttl	str	Which cache-write price applies: "5m" or "1h". Defaults to "5m".	'5m'
service_tier	str	Service tier to price at (e.g. "priority"). When the model's pricing entry carries a matching service_tiers row, that row overlays the base prices for token pricing (keys the tier row does not define — e.g. an unpublished cached rate — fall back to the base row). A requested tier with no registry row logs a warning and bills at standard rates. Defaults to None (standard).	None
audio_minutes	float	Measured audio duration for models with pricing_schema: "per_minute" (STT, and duration-billed realtime rows). Ignored for token-priced models. When a per_minute row is priced with no duration, the call bills the row's own minimum_billable_minutes — :data:MINIMUM_BILLABLE_MINUTES for every row that omits the key — and returns cost_estimated True rather than a confident $0.00. A row that publishes minimum_billable_minutes: 0 opts out of both, because on a session model a zero duration is measured rather than unknown. Defaults to 0.0.	0.0
tier_prompt_tokens	int	The prompt length a tiered_100k row picks its tier on, when this call prices only part of a request. The Anthropic paths bill a cache write split across the 5m and 1h TTLs as one calculate_cost call per share, and each share must land in the tier of the whole request, not of its own few tokens. Defaults to input_tokens + cache_write_tokens. Ignored on every other schema.	None
cost_components_out	dict	When a dict is supplied it is cleared and filled with this call's per-component costs in USD: input_text_usd, input_image_usd, input_audio_usd, cache_discount_usd (negative, the cached re-pricing), cache_write_usd, output_text_usd, output_image_usd, output_audio_usd, duration_usd, tool_usd and total_usd. Every entry but total_usd sums to total_usd on every pricing schema, so a caller can publish a cost breakdown without re-deriving any rate. duration_usd carries a per_minute row's charge (billed minutes x price_per_minute, including a minimum-billable floor) and is 0.0 on every token-priced row; on a per_minute row every token component is 0.0. A call that prices nothing (invalid token counts, an unknown model, an unusable price row) fills every entry with 0.0 rather than leaving a reused dict holding the previous call's numbers. Defaults to None.	None

Returns:

Name	Type	Description
tuple	Tuple[float, bool, Dict[str, bool]]	A 3-element tuple containing: - float: The estimated total cost for the API call (tokens + tools). Returns 0.0 if pricing is not found or inputs are invalid. - bool: Whether the overall cost calculation includes any estimates (True if any tool costs are estimated) - dict: Per-tool cost estimation flags (e.g., {"web_search": False, "code_execution": True}) Keys correspond to tools used, values indicate if that tool's cost is estimated

Notes

Provider-Specific Billing Rules:

• Anthropic: Applies 5-minute minimum billing per code execution session when calculated cost is below minimum
• OpenAI: Reasoning models with web search include content tokens in input_tokens at model rate
• Google: Tiered pricing based on input token count for specific models
• OpenAI: Tiered pricing (272K threshold) for every model whose registry row carries pricing_schema: "tiered_272k" (currently the GPT-5.4, 5.5, 5.6 and 6 families)
• Anthropic: Tiered pricing (100K threshold) for every row carrying pricing_schema: "tiered_100k" (Claude Haiku 5.5). The prompt length that picks the tier is input_tokens + cache_write_tokens — the call sites fold cache reads into input_tokens and pass writes separately, and Anthropic counts both — so a 60k-token prompt that writes 50k tokens to the cache is a 110k-token prompt and bills the high tier on every key, writes included, while a 60k-token prompt whose 50k are cache reads stays in the low tier. A cache-only request (input_tokens equal to the reads) decides its tier on the reads alone, which is the same rule. A caller that prices one request in several calls — the Anthropic paths price a cache write split across the 5m and 1h TTLs as one call per share — passes the whole prompt length as tier_prompt_tokens so every share lands in the same tier as the request.
• Fireworks: Standard per-token pricing

Tier boundaries are inclusive on the low side on all three vendors: a prompt of exactly 200,000 tokens bills at the Google *_less_than_200k_prompt rates, a prompt of exactly 272,000 tokens at the OpenAI *_less_than_272k_prompt rates and a prompt of exactly 100,000 tokens, reads and writes included, at the Anthropic *_less_than_100k_prompt rates. See :data:TIERED_PRICING_THRESHOLDS.

count_tokens(provider, model_name, content_parts, client, logger, image_detail='high')

Count input tokens, discarding the source label.

The count-only view of :func:count_tokens_with_source, which documents the per-provider counting strategy, the accepted content_parts shapes and the accuracy caveats for multimodal input. Use this when the count is all you need; use :func:count_tokens_with_source when the count will be reported, so an estimate is not presented as an API figure.

Parameters:

Name	Type	Description	Default
provider	str	The API provider name ('google', 'anthropic', 'openai', 'fireworks').	required
model_name	str	The specific model name (e.g. "gpt-5.6-luna").	required
content_parts	Union[List[Any], str]	Text and/or media parts, in the shapes listed on :func:count_tokens_with_source.	required
client	Any	The initialized API client for provider.	required
logger	Logger	Logger instance for messages and warnings.	required
image_detail	str	Image detail level applied to every image part when the content is rendered for the OpenAI Token Counting API. Ignored by the other providers. Defaults to "high".	'high'

Returns:

Name	Type	Description
int	int	The estimated or calculated token count. Returns 0 if content is
	int	empty or invalid.

Example

import logging

from doc_processing_toolkit.core import count_tokens
from doc_processing_toolkit.utils import initialize_clients

clients = initialize_clients(provider="openai")
tokens = count_tokens("openai", "gpt-5.6-luna", ["Hello, world!"], clients["openai"], logging.getLogger())

count_tokens_with_source(provider, model_name, content_parts, client, logger, image_detail='high')

Estimates or calculates token count for input content (text and potentially other types) using provider-specific methods.

Provides accurate token counts where possible via API methods, with appropriate fallbacks to approximation methods. Logs warnings if specific methods fail or approximation is used. For multimodal input, local counting is often an estimate; API metadata (usage_metadata or usage) after a call provides the most accurate counts.

Parameters:

Name	Type	Description	Default
provider	str	The API provider name ('google', 'anthropic', 'openai', 'fireworks').	required
model_name	str	The specific model name (e.g., "gpt-4o", "claude-3-7-sonnet-latest").	required
content_parts	list or str	List containing strings (for text) and/or dictionaries for media. Tool Definitions and Output Response Schema's should be passed as strings. str: Text content (e.g., "Hello, world!"). Supported dictionary formats: dict: Structured content with 'type' field: Google: - Base64 Image: {'type': 'base64', 'media_type': 'image/png', 'data': str} (media_type fixed to 'image/png') - Video: {'type': 'video', 'media_type': str, 'data': bytes} (media_type='video/mp4' | 'video/mpeg' | 'video/mov' | 'video/avi' | 'video/x-flv' | 'video/mpg' | 'video/webm' | 'video/wmv' | 'video/3gpp') - Audio: {'type': 'audio', 'media_type': str, 'data': bytes} (media_type='audio/wav' | 'audio/mp3', | 'audio/aiff' | 'audio/aac' | 'audio/ogg' | 'audio/flac') - Document: {'type': 'document', 'media_type': str, 'data': bytes} (media_type='application/pdf' | 'application/x-javascript' | 'text/javascript' | 'application/x-python' | 'text/x-python' | 'text/plain' | 'text/html' | 'text/css' | 'text/md' | 'text/csv' | 'text/xml' | 'text/rtf') - YouTube: {'type': 'youtube', 'url': str} Anthropic, OpenAI, Fireworks: - base64 Image: {'type': 'base64', 'media_type': 'image/png', 'data': str} (media_type fixed to 'image/png') - Document: {'type': 'base64', 'media_type': 'application/pdf', 'data': str} (media_type fixed to 'application/pdf')	required
client	object	The initialized API client (needed for Google, Anthropic, and some other providers).	required
logger	Logger	Logger instance for logging messages and warnings.	required
image_detail	str	Image detail level applied to every image part when the content is rendered for the OpenAI Token Counting API, so the pre-flight estimate matches the real call. Ignored by the other providers. Defaults to "high".	'high'

Returns:

Type	Description
int	Tuple[int, str]: The estimated or calculated token count for the input
str	content (0 if content is empty or invalid), and the token_source
Tuple[int, str]	label naming how it was obtained — an API source such as
Tuple[int, str]	"openai_token_counting_api", a local estimate such as
Tuple[int, str]	"openai_tiktoken_fallback_registry", or a
Tuple[int, str]	"text_approximation_*" last resort. Callers that report token
Tuple[int, str]	counts use the label so an estimate is never presented as an API
Tuple[int, str]	figure; :func:count_tokens is the count-only wrapper.

Notes

• Google: Uses client.models.count_tokens() API when available, falls back to text approximation. Media parts are uploaded to the Files API first; an upload that is not yet ACTIVE at count time (video, typically) is excluded from the API call and approximated locally, so one still-processing file no longer degrades the whole count to text approximation.
• OpenAI: Uses the Token Counting API (client.responses.input_tokens.count()), which counts image and file parts as well as text; falls back to tiktoken (text only) and then to text approximation.
• Anthropic: Uses client.messages.count_tokens() API when available, falls back to text approximation.
• Fireworks: Uses the GPT-4 tiktoken encoding for text plus a flat per-image estimate (FIREWORKS_IMAGE_TOKEN_ESTIMATE); falls back to text approximation.
• Others: Uses text approximation (4 chars per token).

Warning: Pre-call token counting for multimodal content (text + images) is approximate and may differ from the actual token count reported after the API call, particularly for images which often use variable token counts based on resolution and complexity.

Utility Functions

doc_processing_toolkit.utils

Shared utility functions for provider validation, initialization, and helpers.

This module contains cross-provider validation logic, environment-driven client initialization, citation cleanup, and helper functions used by agent/provider execution paths.

See Also

• Core API reference

IterationEventUsage

The usage one tool loop's iteration_complete events report.

Every model request of the loop is added where the loop prices it; each event takes what accumulated since the previous one. So the events of a run sum to its requests up to the last event: the first event of a tool run carries the request that asked for the tools as well as its follow-up. A request after the run's last event (an empty-turn re-request after the final round), or in a run that ends without one (a forced tool_choice exit, a Path B answer on the first request, an early error), is on the run's result only. input_tokens is billed input summed across requests, not the latest prompt's size.

Attributes:

Name	Type	Description
input_tokens	int	Input tokens billed since the previous event.
output_tokens	int	Output tokens (reasoning included) since then.
cost	float	Their cost in USD.

Example

usage = IterationEventUsage()
usage.add(1_200, 80, 0.0021)  # the request that asked for the tools
usage.add(1_450, 60, 0.0024)  # its follow-up
usage.take()  # {"input_tokens": 2650, "output_tokens": 140, "cost": 0.0045}

add(input_tokens, output_tokens, cost)

Add one billed request.

Parameters:

Name	Type	Description	Default
input_tokens	int	The request's billed input tokens.	required
output_tokens	int	Its billed output tokens.	required
cost	float	Its cost in USD.	required

take()

The input_tokens / output_tokens / cost of the next event.

Returns what accumulated since the previous event, and resets.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"input_tokens", "output_tokens", "cost"}, in
Dict[str, Any]	that order.

NonRetryableStreamEventError

Bases: ValueError

A stream failure event that the same request would run into again.

Raised by a provider's stream consumer for a failure event it has already classified as deterministic: a 4xx status, a validation failure, a policy stop. should_retry_api_call never retries it, whatever its message says; without the class, the predicate's keyword fallback turned a message such as "try again later" back into five attempts. It subclasses ValueError, so a caller that catches ValueError still sees it.

NormalizedStreamError

Bases: Exception

Internal stream error wrapper for retry-aware propagation.

Carries provider-independent fields (status_code, code) so the existing retry and billing helpers can classify stream-level failures without relying on provider SDK exception constructors.

StreamEventBuffer

Buffer consecutive stream deltas of the same type for grouped output.

flush()

Flush the current buffered content, if any.

push(event_type, content)

Add content and return a flushed block when the event type changes.

TruncatedResultStore

The full text of tool results the size cap cut, for read_truncated_result.

One store serves a run, or an AgentSession for its lifetime. Only truncated results are kept, in memory, keyed by the id the marker names: the provider's tool-call id, made unique with a -2, -3… suffix when a provider reuses one (Kimi's functions.<name>:0 repeats across turns) or when there is none (Gemini models before 3.5). Nothing here is exported with a session; a restored session retires the ids its history names (:meth:retire), so an old marker can never read a newer result. Several text items are kept joined by a newline, as the model was shown them.

Each result has an allowance of read_chars characters the model may read back, beyond what it was first shown. Every character a read returns counts, a part read twice included. A read returns at most the cap the result was cut at, and always ends with a note of where it stopped, so a slice is never bare JSON (read_tool_reported_spend cannot count a tool's spend twice).

Attributes:

Name	Type	Description
read_chars	int	The allowance given to each result remembered from now on (Agent(truncated_result_read_chars=)).

Example

store = TruncatedResultStore(read_chars=8000)
marker = store.remember("call_1", "x" * 10000, 4000, 4000)  # appended after the 4,000 shown
text, is_error = store.read("call_1", 4000)  # the next 4,000 characters

result_count property

int: How many truncated results the store holds.

__init__(read_chars=0)

Create an empty store.

Parameters:

Name	Type	Description	Default
read_chars	int	The per-result read allowance, in characters.	0

clear()

Forget every stored result (retired ids stay retired).

read(tool_call_id, offset)

Return the next slice of a stored result.

Parameters:

Name	Type	Description	Default
tool_call_id	Any	The id from the marker.	required
offset	Any	Where to start, in characters (an int, or digits).	required

Returns:

Name	Type	Description
	str	Tuple[str, bool]: The text for the model and whether it is an
error	bool	an unknown id ("This result was not truncated"), an id a
	Tuple[str, bool]	restored session retired, a bad offset, or a used-up allowance.

remember(tool_call_id, text, shown, cap)

Keep a truncated result's full text and return the marker for it.

Parameters:

Name	Type	Description	Default
tool_call_id	Optional[str]	The provider's id for the call.	required
text	str	The result's full text (its text items joined by a newline).	required
shown	int	Where the cut falls in text: the offset to read on from.	required
cap	int	The cap it was cut at: the characters the model was shown, and the most one read returns.	required

Returns:

Name	Type	Description
str	str	The marker appended where the text was cut.

retire(ids)

Mark ids an earlier store gave out, so they are never reused.

Called when a session is restored: its history still holds markers naming those ids, but their full text was not saved. A read of one answers that the result is no longer available, and a new result that would take one gets a suffix instead.

Parameters:

Name	Type	Description	Default
ids	Any	The ids, e.g. from :func:truncated_result_ids.	required

UnsupportedPartError

Bases: ValueError

Raised when a prompt part is not supported and the policy says "raise".

Only raised under on_unsupported_part="raise". It subclasses ValueError so a caller that already narrows on ValueError around a provider call keeps working.

Attributes:

Name	Type	Description
dropped_parts	List[Dict[str, Any]]	The full drop records for this call, in prompt order — the same dicts that reach AgentResult.dropped_parts. Present even though the run was refused, so the caller can report every offending part, not only the first.

Example

try:
    agent.run_sync("Summarise this", files=[pdf])
except UnsupportedPartError as err:
    for record in err.dropped_parts:
        print(record["kind"], record["reason"])

__init__(message, dropped_parts=None)

Build the error.

Parameters:

Name	Type	Description	Default
message	str	The human-readable reason, naming the part and model.	required
dropped_parts	list or None	The drop records collected so far.	None

anthropic_thinking_is_off(thinking_config)

Whether an Anthropic thinking config asks for thinking to be switched off.

Parameters:

Name	Type	Description	Default
thinking_config	Any	The caller's thinking value, in the unified or the provider shape. Anything that is not a dict is not an "off" request.	required

Returns:

Name	Type	Description
bool	bool	True when thinking_config["type"] is one of
	bool	data:ANTHROPIC_THINKING_OFF_TYPES.

Example

anthropic_thinking_is_off({"type": "disabled"})       # True
anthropic_thinking_is_off({"type": "between_tools"})  # True
anthropic_thinking_is_off({"type": "adaptive"})       # False
anthropic_thinking_is_off(None)                       # False

apply_defer_loading(tools_list, predicate, server_map, logger)

Stamp defer_loading: true on the client tools the predicate selects.

The stamp is applied in place, exactly once per request, to the array that is about to be sent. Order is never changed: a deferred tool keeps its position, so the frozen Anthropic tool order (P2b-2) and the goldens are unaffected by which tools are deferred.

Three rules are enforced here rather than left to the provider:

• Server tools are never deferred. Only a client tool entry — one that carries an input_schema (Anthropic) or is {"type": "function"} (OpenAI) — is eligible. The tool-search tool itself, web_search, code_interpreter, mcp_toolset and friends are skipped, which is also why the tool-search tool can never be deferred by accident.
• A deferred entry never carries cache_control. Anthropic rejects that pairing with a 400, and a breakpoint on a deferred tool would be meaningless anyway (deferred tools are excluded from the cached prefix). Any existing breakpoint on a selected entry is removed with a warning.
• A named tool that matches nothing warns once, so a typo is visible instead of silently loading every tool eagerly.

Parameters:

Name	Type	Description	Default
tools_list	List[Dict[str, Any]]	The wire tools array, mutated in place.	required
predicate	Optional[Callable[[str, Optional[str]], bool]]	The two-argument predicate from :func:resolve_defer_tools, or None to do nothing.	required
server_map	Optional[Dict[str, Any]]	tool_name -> {"server_name": ...} routing table from the provider's MCP extraction, used to give the predicate its second argument. May be None.	required
logger	Logger	Logger for the warnings described above.	required

Returns:

Type	Description
List[str]	List[str]: The names actually stamped, in wire order. Empty when
List[str]	predicate is None or nothing matched.

Example

deferred = apply_defer_loading(tools, predicate, server_map, logger)
# ['lookup_sku', 'search_inventory']

attach_retry_summary(result, accumulator)

Attach RetryAccumulator totals to a provider result dict.

The accumulator is populated by the _on_retry_attempt tenacity before_sleep hook, which fires once per failed retry attempt (but NOT for the final successful attempt or the final exhaustion attempt). So accumulator.attempts contains only the failed-and-retried attempts; the final attempt's cost is already carried in result["input_tokens"] / result["output_tokens"] / result["cost"] (from the provider's success path, or from _on_retry_exhausted when the final attempt also failed).

This helper adds the retry totals onto the top-level cost/token fields so that AgentResult.cost reflects everything the provider billed for — including the failed-and-retried attempts. It also sets the diagnostic retry_* fields consumed by AgentResult and IterationMetrics.

When the accumulator is None or empty, the helper only ensures the retry keys exist with zero defaults so downstream code can rely on .get(key, default).

Parameters:

Name	Type	Description	Default
result	Dict[str, Any]	Provider function's return dict. Mutated in place and returned.	required
accumulator	Optional[Any]	RetryAccumulator from the caller's scope, passed to the provider function via the _retry_accumulator kwarg and populated by tenacity's retry hooks.	required

Returns:

Type	Description
Dict[str, Any]	The same result dict, with retry_count, retry_cost,
Dict[str, Any]	retry_input_tokens, retry_output_tokens, retry_details
Dict[str, Any]	keys set, and the top-level cost/token fields rolled up.

Notes

result["cached_input_tokens"] is rolled up with the rest. It has to be: input_tokens gains the retried attempts' prompts, and a cached share that did not move with it would make "uncached = input − cached" wrong for every retried run.

base64_encode_if_needed(data)

Ensure data is base64 encoded string.

If data is already a string, assume it's base64 and return as-is. If data is bytes, base64 encode and return string.

Parameters:

Name	Type	Description	Default
data	Any	Data to encode (string or bytes)	required

Returns:

Name	Type	Description
str	str	Base64 encoded string

build_omission_notice(model_name, omitted_descriptions)

Build a text notice describing what content was omitted.

Parameters:

Name	Type	Description	Default
model_name	str	Model name	required
omitted_descriptions	list	List of omitted content descriptions	required

Returns:

Name	Type	Description
str	str	Notice text to append to function response (empty if nothing omitted)

cap_tool_result(tool_result, max_chars, tool_name, logger, tool_call_id=None, truncation_store=None)

Truncate an MCP-shaped tool result's text to max_chars characters.

Applied by all four tool loops at the point a result is fed back to the model, to every result the toolkit itself executes — local functions and every client-side MCP format, mcp_remote_client included. The text items of {"content": [...], "isError": bool} are counted together; when their total exceeds max_chars, text is kept up to the cap, the item that crosses it is cut and ends with … [truncated: N of M characters shown; raise max_tool_result_chars to keep more], and later text items are dropped. Non-text items (images, audio, resources) pass through untouched. One WARNING is logged per truncated result. The input is never mutated, so the raw result the loop records keeps its full text.

With a truncation_store (Agent(truncated_result_read_chars=)) the full text is kept in the store and the marker tells the model how to read on: … [truncated: N of M characters shown. Call read_truncated_result(tool_call_id="…", offset=N) to read up to K more.]. The built-in tool's own results are not cut again: each slice is already within the cap.

Parameters:

Name	Type	Description	Default
tool_result	Any	The result from execute_mcp_tool.	required
max_chars	int	The cap from :func:resolve_max_tool_result_chars; 0 or less disables it.	required
tool_name	str	Tool name, for the warning.	required
logger	Logger	Logger for the warning.	required
tool_call_id	Optional[str]	The provider's id for the call, which the marker names so the model can read on. Only used with a store.	None
truncation_store	Optional[TruncatedResultStore]	Where a truncated result's full text is kept for read_truncated_result; None (the default) keeps the original marker.	None

Returns:

Name	Type	Description
Any	Any	tool_result itself when nothing is cut (or it is not
	Any	MCP-shaped), otherwise a truncated shallow copy.

Example

capped = cap_tool_result({"content": [{"type": "text", "text": "x" * 10}]}, 4, "t", logger)
capped["content"][0]["text"].startswith("xxxx\n\n… [truncated: 4 of 10 characters shown")  # True

carry_tool_reported_spend(exc, records)

Attach a tool loop's spend records to an exception leaving the loop.

Parameters:

Name	Type	Description	Default
exc	Optional[BaseException]	The exception being re-raised.	required
records	Optional[List[Dict[str, Any]]]	The loop's tool_reported_costs.	required

convert_user_prompt_to_openai_input(user_prompt_parts, logger, image_detail='high', *, silent=False)

Convert user_prompt_parts to OpenAI Responses API input format.

This function converts the toolkit's standardized user_prompt_parts format into OpenAI's Responses API input format. It consolidates all content parts into a single user message with proper content types.

Parameters:

Name	Type	Description	Default
user_prompt_parts	list	List containing strings (for text) and/or dictionaries for media. The order of the list should match the desired input order for the model. Supported dictionary formats: base64 Image: {'type': 'base64', 'media_type': str, 'data': str} (any image/* MIME type, e.g. 'image/png', 'image/jpeg', 'image/webp') Document: {'type': 'base64', 'media_type': 'application/pdf', 'data': str, 'filename': str}	required
logger	Logger	Logger instance for debug output	required
image_detail	str	Image detail level applied uniformly to all images in the request. Supported values are model-dependent (see supported_image_detail_values in model_information.json): "low", "high" (default), "auto", or "original" (gpt-5.4+ only). This is a request-level setting — it is not configurable per image.	'high'
silent	(bool, keyword - only)	When True, suppresses the per-image and per-document INFO log lines. Used by the pre-flight token-counting path to avoid duplicating these messages alongside the real API call.	False

Returns:

Name	Type	Description
list	list	List containing a single user message with OpenAI Responses API format. Format: [{"role": "user", "content": [...content parts...]}]

Notes

The output format matches the structure used in call_openai_llm_structured_output_text:

• Uses "input_text" type for text content
• Uses "input_image" type with the specified image_detail for images
• Uses "input_file" type for PDFs with filename

Raises:

Type	Description
ValueError	If no content parts are generated or user_prompt_parts is empty

Examples:

parts = ["Hello", {"type": "base64", "media_type": "image/png", "data": "..."}]
result = convert_user_prompt_to_openai_input(parts, logger)
# Returns: [{"role": "user", "content": [{"type": "input_text", "text": "Hello"}, ...]}]

create_final_response_tool(response_schema, provider, strict=False)

Creates the 'submit_final_response' tool definition for Path B structured outputs.

This function generates a tool definition that instructs the model to call it when ready to provide a final structured response. The tool's input schema is derived from the provided Pydantic response_schema.

Parameters:

Name	Type	Description	Default
response_schema	Any	Pydantic model class defining the expected response structure.	required
provider	str	Provider name ("anthropic", "openai", "google", "fireworks"). Determines the format of the returned tool definition.	required
strict	bool	Whether to enable strict mode for schema validation. OpenAI: Adds "strict": true to tool definition (supported) Anthropic: IGNORED. This tool is only used for Path B, which is for models that do NOT support native structured outputs. Strict mode requires supports_function_calling_and_structured_outputs=True, so Path B models cannot use strict mode. Fireworks: Not supported, parameter is ignored Google: Not compatible with Path B (uses tool_choice "any"). Parameter is ignored.	False

Returns:

Type	Description
Any	dict or FunctionDeclaration: Tool definition in provider-specific format. Anthropic: {"name": str, "description": str, "input_schema": dict} OpenAI: {"type": "function", "name": str, "description": str, "parameters": dict, "strict": bool} Google: google_types.FunctionDeclaration object Fireworks: {"type": "function", "function": {"name": str, "description": str, "parameters": dict}}

Examples:

from pydantic import BaseModel, Field

class MyResponse(BaseModel):
    answer: str = Field(description="The final answer")
    confidence: float = Field(description="Confidence score 0-1")

tool = create_final_response_tool(MyResponse, "openai", strict=True)
print(tool["name"])
'submit_final_response'

default_code_execution_output_dir(provider)

Create a fresh per-run folder for code-execution output files and return its path.

./<provider>_code_execution_files_<YYYYMMDD_HHMMSS>_<8 hex>/ in the current working directory, used by every provider when no code_execution_output_dir is given and a run has files to save (0.5.3). Until then the default was the timestamp alone, so two runs that started in the same second in the same directory shared a folder and a cited name both produced was silently overwritten, both records reporting success. The random suffix and the exclusive os.mkdir (retried on a clash) mean two runs never share a folder; the timestamp keeps the folders sortable, and a glob on <provider>_code_execution_files_* still matches. The folder keeps the process's default permissions (tempfile.mkdtemp would make it private to the user, which a client reading the files as another user could not open).

Parameters:

Name	Type	Description	Default
provider	str	The folder prefix: "openai", "google" or "anthropic".	required

Returns:

Name	Type	Description
str	str	The absolute path of the folder, which exists on return.

Raises:

Type	Description
OSError	If the folder cannot be created (a read-only working directory).

Example:

```python
folder = default_code_execution_output_dir("openai")
# 'C:/work/openai_code_execution_files_20261004_143012_3f9a1c2e'
```

execute_mcp_tool(tool_name, tool_input, server_map, logger) async

Execute a tool via its MCP server (provider-agnostic).

Returns MCP's standard content format - NO custom intermediate format. Each provider's translation function handles conversion to provider-specific format.

This function is shared across all provider implementations.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool to execute	required
tool_input	dict	Dictionary of tool arguments	required
server_map	dict	Mapping of tool names to server info dicts containing: 'session': MCP ClientSession instance 'server_name': Name of the MCP server	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
dict	dict	Standard MCP content structure (per MCP specification):
	dict	{ "content": [ # List of content items {"type": "text", "text": "..."}, {"type": "image", "data": "base64...", "mimeType": "image/png"}, {"type": "resource", "uri": "file://...", "mimeType": "..."}, # ... other MCP content types ], "isError": False # True if tool execution failed
	dict	}

Notes

MCP Content Types (from MCP specification):

• text: {"type": "text", "text": "content"}
• image: {"type": "image", "data": "base64...", "mimeType": "image/png"}
• audio: {"type": "audio", "data": "base64...", "mimeType": "audio/wav"}
• video: {"type": "video", "data": "base64...", "mimeType": "video/mp4"} OR resource URI
• resource: {"type": "resource", "uri": "file://...", "mimeType": "..."}
• embedded_resource: {"type": "embedded_resource", "data": "base64...", "mimeType": "..."}

extract_final_response_tool_call(response, provider, logger)

Extract the 'submit_final_response' tool call arguments from a response.

This function searches through the provider's response for a tool call to 'submit_final_response' and extracts the arguments (the structured response data).

Parameters:

Name	Type	Description	Default
response	Any	The API response object from the provider.	required
provider	str	Provider name ("anthropic", "openai", "google", "fireworks").	required
logger	Logger	Logger instance for logging.	required

Returns:

Type	Description
Optional[dict]	dict or None: The parsed arguments from the tool call if found, None otherwise.

Notes

This function handles the different response formats from each provider:

• Anthropic: response.content contains tool_use blocks
• OpenAI: response.output contains function_call items
• Google: response.candidates[0].content.parts contains function_call parts
• Fireworks: response.choices[0].message.tool_calls contains tool calls

extract_partial_stream_cost(exception, *, model_name, logger, fallback_input_tokens=0, fallback_output_tokens=0, fallback_cost=0.0, fallback_token_source='unknown', tool_usage=None, model_info_config=None)

Extract partial-cost data from a stream exception for error-path accounting.

Stream consumers in each provider attach a _partial_stream_data attribute to exceptions so that tokens produced before the failure are not lost. This helper centralises the lookup + cost-recomputation step used by every provider's top-level except Exception handler, and its cost is what the failed run reports.

When _partial_stream_data is present this helper recomputes the cost via calculate_cost (so pricing reflects the current model info), taking the larger of fallback_input_tokens / fallback_output_tokens and the values carried by the exception — that preserves any usage metadata the caller has already captured from a prior iteration's response.

Every per-request billing input the payload carries is forwarded, because a request that failed was billed by the same rules as one that succeeded:

• cached_input_tokens — a SUBSET of input_tokens on the toolkit's usual convention (subtracted out and re-priced, never added). Absent means 0, which prices the whole prompt at the uncached rate as before.
• served_tier — the tier the provider actually SERVED. Reduced through the registry (:func:_registry_priced_service_tier), so an absent, "standard" or unrecognised value bills the base row exactly as it did before this parameter was read.
• input_token_breakdown / output_token_breakdown — the modality splits. Absent leaves calculate_cost's own no-breakdown handling in place, which is what flags the cost estimated on a multimodal row.
• request_cost_records — the tool loops' already-priced records of the attempt's completed requests. The cost is their sum plus the in-flight request's remainder, never the run-to-date token sum priced as one request (which over-bills a tiered_* row), and their output tokens join the reported output_tokens.

When _partial_stream_data is absent the provided fallback values are returned unchanged, so callers can call this helper unconditionally.

A request the provider rejected before streaming (the exception carries an HTTP error response, or the connection never reached the server — see :data:REJECTED_BEFORE_STREAM) was billed nothing, so its pre-flight estimate is not priced: with no completed records the result is the fallback values (zero on a first request), token_source and reason are "rejected_before_stream" and cost_estimated is False; with records only the records are billed. A stream that failed after its first chunk keeps the estimated partial billing.

Parameters:

Name	Type	Description	Default
exception	Exception	The exception caught by the provider's top-level handler.	required
model_name	str	Model identifier for pricing lookup.	required
logger	Logger	Logger for diagnostic messages.	required
fallback_input_tokens	int	Input tokens to use if partial data is absent (typically the handler's accumulated input_token_count).	0
fallback_output_tokens	int	Output tokens to use if partial data is absent.	0
fallback_cost	float	Cost to use if partial data is absent.	0.0
fallback_token_source	str	Token-source label to use if partial data is absent.	'unknown'
tool_usage	Optional[Dict[str, Any]]	Provider tool-usage dict; required for calculate_cost.	None
model_info_config	Optional[Dict[str, Any]]	Pricing config dict; required for calculate_cost.	None

Returns:

Type	Description
Dict[str, Any]	Dict with keys input_tokens, output_tokens, cost,
Dict[str, Any]	token_source, cached_input_tokens, served_tier,
Dict[str, Any]	cost_estimated and reason. token_source is
Dict[str, Any]	"stream_partial" when partial data was recovered,
Dict[str, Any]	"rejected_before_stream" when the only request was rejected before
Dict[str, Any]	streaming, otherwise fallback_token_source; cost_estimated is
Dict[str, Any]	what calculate_cost reported, so a caller can publish an estimate as
Dict[str, Any]	an estimate instead of an exact figure; reason is
Dict[str, Any]	"rejected_before_stream" when the in-flight request was rejected
Dict[str, Any]	before streaming and None otherwise.

Example

partial_cost = extract_partial_stream_cost(
    exc, model_name=model_name, logger=logger,
    tool_usage=tool_usage, model_info_config=model_info_config,
)
result["cost"] = partial_cost["cost"]
result["cost_estimated"] = partial_cost["cost_estimated"]

fetch_mcp_resource(uri, logger) async

Fetch resource from URI (for resource-type MCP content).

Notes

Handles the following URI schemes

• file:// URIs → Read from filesystem
• http:// / https:// URIs → Fetch via HTTP
• data: URIs → Decode data URI (base64)

Parameters:

Name	Type	Description	Default
uri	str	Resource URI to fetch	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
bytes	bytes	Raw bytes of the resource

Raises:

Type	Description
ValueError	If URI scheme is not supported
FileNotFoundError	If file:// resource doesn't exist
ClientError	If HTTP fetch fails

fix_json_braces(json_str)

Attempts to fix common JSON formatting issues, like extra braces or markdown fences.

Parameters:

Name	Type	Description	Default
json_str	str	The potentially malformed JSON string.	required

Returns:

Name	Type	Description
str	str	The cleaned JSON string.

get_cached_client(provider, kind)

Return the process-cached client for (provider, kind), if valid.

Async entries are only returned to callers running on the same event loop that created the client; anything else is a miss.

get_client_init_error(provider, kind)

Return the last recorded construction failure for (provider, kind).

Parameters:

Name	Type	Description	Default
provider	str	Provider name ("google", "anthropic", "openai", "fireworks", "llmwhisperer").	required
kind	str	"sync" or "async".	required

Returns:

Type	Description
Optional[BaseException]	The recorded exception, or None if the last construction succeeded
Optional[BaseException]	(or was never attempted).

get_directory_size_mb(directory_path)

Calculate the total size of a directory in megabytes.

Parameters:

Name	Type	Description	Default
directory_path	str	Path to the directory	required

Returns:

Name	Type	Description
float	float	Total size in megabytes

get_model_limits()

Loads model information and extracts the context window and max output tokens for all models.

Returns:

Name	Type	Description
dict	dict	A dictionary mapping model names to their limits. Example: {'gemini-1.5-pro': {'context_window_tokens': 1048576, 'max_output_tokens_request': 8192}}

get_providers_for_models(model_names, logger=None)

Gets a unique list of providers for a given list of model names based on pricing info.

Parameters:

Name	Type	Description	Default
model_names	List[str]	A list of model names.	required

Returns:

Type	Description
List[str]	List[str]: A unique list of provider names corresponding to the models.

get_session_kind_capabilities()

Build the (provider, session_kind) capability record from the registry.

Realtime routing is a provider -> adapter map today, which cannot express that one provider serves several session families or that one model serves more than one surface. This record is the data-only answer to "which session kinds does this provider support, and with which models", so discovery and the dedicated-session work read it instead of hardcoding a table.

Returns:

Type	Description
Dict[Tuple[str, str], Tuple[str, ...]]	Dict[Tuple[str, str], Tuple[str, ...]]: (provider, session_kind)
Dict[Tuple[str, str], Tuple[str, ...]]	mapped to the sorted model IDs serving it. A pair with no models is
Dict[Tuple[str, str], Tuple[str, ...]]	absent rather than empty.

Example

from doc_processing_toolkit.utils import get_session_kind_capabilities

record = get_session_kind_capabilities()
sorted(kind for provider, kind in record if provider == "openai")
# ['image', 'realtime', 'speech', 'text', 'transcription', 'translation']

initialize_async_clients(provider=None, logger=None, api_key=None)

Initializes and returns asynchronous API clients for all supported providers based on .env variables.

Can return either a dictionary of all available clients or a single client instance.

Parameters:

Name	Type	Description	Default
provider	str	If specified, only that provider's client is constructed and returned — the other providers' clients are never built. If None, constructs and returns a dictionary of all initialized clients.	None
logger	Optional[Logger]	Optional logger instance for logging initialization status.	None
api_key	Optional[str]	Optional explicit key for the requested provider. Overrides the environment variable, and the resulting client bypasses the shared process cache.	None

Returns:

Type	Description
Any	dict or client instance: A dictionary of async clients or a single async client object.

Notes

Construction is gated on provider (not just the return value) so unrequested clients are never created and orphaned to garbage collection — orphaned SDK clients cost file descriptors and can emit teardown noise (e.g. the Fireworks SDK eagerly opens aiohttp sessions per client when an event loop is running).

The process cache serves both call forms; cached async clients are only reused on the event loop that created them, so the all-providers form on a fresh loop still constructs, caches and (at that loop's shutdown) closes them.

The Anthropic and OpenAI clients are built with max_retries=0, so tenacity (retry_on_api_error) is the only retry layer. DPT_SDK_RETRIES=default restores the SDK's own default (2), and a non-negative integer is passed through; any other value logs one WARNING and keeps 0. The variable is read when a client is built, so set it before the first request. Fireworks 1.x clients always use max_retries=0.

initialize_clients(provider=None, logger=None, api_key=None)

Initializes and returns synchronous API clients for all supported providers based on .env variables.

Can return either a dictionary of all available clients or a single client instance.

Parameters:

Name	Type	Description	Default
provider	str	If specified, only that provider's client is constructed and returned — the other providers' clients are never built. If None, constructs and returns a dictionary of all initialized clients.	None
logger	Optional[Logger]	Optional logger instance for logging initialization status.	None
api_key	Optional[str]	Optional explicit key for the requested provider. Overrides the environment variable, and the resulting client bypasses the shared process cache.	None

Returns:

Type	Description
Any	dict or client instance: A dictionary of clients or a single client object.

Notes

Construction is gated on provider (not just the return value) so unrequested clients are never created and orphaned to garbage collection — orphaned SDK clients cost file descriptors and can emit teardown noise (e.g. the Fireworks SDK eagerly opens aiohttp sessions per client when an event loop is running).

The process cache serves both call forms. The all-providers form used to bypass it and rebuild every client on each call, which is what made a repeated initialize_clients() multiply live clients.

The Anthropic and OpenAI clients are built with max_retries=0, so tenacity (retry_on_api_error) is the only retry layer. DPT_SDK_RETRIES=default restores the SDK's own default (2), and a non-negative integer is passed through; any other value logs one WARNING and keeps 0. The variable is read when a client is built, so set it before the first request. Fireworks 1.x clients always use max_retries=0.

is_billable_error(exception)

Classify whether an API error likely incurred token charges.

When an API request fails, the provider may or may not charge for tokens consumed before the failure. This function classifies errors to determine whether cost estimation should be attempted.

Parameters:

Name	Type	Description	Default
exception	Exception	The exception raised by the provider SDK or HTTP client.	required

Returns:

Type	Description
bool	True if the error likely incurred token charges, False otherwise.

Notes

Billable errors (tokens likely consumed before failure):

• Timeout errors (APITimeoutError, DeadlineExceeded) — generation ran until timeout; input + partial output tokens were consumed.
• 500 Internal Server Error — server may have processed tokens before crash.
• 529 Overloaded — request may have been partially processed.

Non-billable errors (tokens NOT consumed):

• 400 Bad Request — request rejected before processing.
• 401 Unauthorized — auth failure before processing.
• 403 Forbidden — permission failure before processing.
• 404 Not Found — model or endpoint not found.
• 429 Rate Limit — request rejected before processing.

An SDK-wrapped transport failure is classified from its chained cause (openai >= 3.14.0 raises APIConnectionError / APITimeoutError from the real httpx2 error), so a mid-stream drop dressed up as APIConnectionError is billable while a connect failure dressed up as APITimeoutError is not.

load_model_information()

Loads model information from the JSON configuration file bundled with the package.

The file is read once per process and cached (it is immutable bundled package data); call _load_model_information_cached.cache_clear() in tests that need a fresh read. This function itself is a thin uncached wrapper so test fixtures can keep patching the module attribute.

Returns:

Name	Type	Description
dict	dict	The model information configuration.

Raises:

Type	Description
RuntimeError	If the file cannot be read or parsed — chained from the original OSError/ValueError so the real cause (e.g. EMFILE) is visible at the failure site instead of a downstream 'NoneType' crash.

map_tool_name(standard_name, provider, reverse=False)

Map between standard and provider-specific tool names.

Parameters:

Name	Type	Description	Default
standard_name	str	The standardized tool name to map	required
provider	str	The provider name (e.g., "google", "openai")	required
reverse	bool	If True, map from provider-specific to standard name; otherwise, map from standard to provider-specific	False

Returns:

Type	Description
str	The mapped tool name, or the original name if no mapping exists

mcp_content_to_text(content_items, logger)

Convert MCP content to plain text representation.

Useful for providers with limited multimodal support (e.g., Fireworks).

Notes

Converts all content types to text

• text → as-is
• image → [Image: mime_type]
• audio → [Audio: mime_type]
• video → [Video: mime_type]
• resource → [Resource: uri]
• embedded_resource → [Embedded Resource: mime_type]

Parameters:

Name	Type	Description	Default
content_items	list	List of MCP content items	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
str	str	Plain text representation of all content

merge_reasoning_into_schema_enabled(logger=None)

Whether reasoning text is copied into a reasoning-named schema field.

When a caller's output_schema has a field named internal_reasoning, thoughts or reasoning (case-insensitive), every provider appends the model's reasoning text to that field of the parsed data under a ***Internal Reasoning:*** heading. The same text is always on AgentResult.reasoning_content, which is the canonical location; the merge stays the default for clients that read it from data and is scheduled to be retired in 0.5.

This is the ONE predicate all fourteen merge sites consult (they never read the environment themselves). DPT_MERGE_REASONING_INTO_SCHEMA controls it, case-insensitive with surrounding whitespace ignored:

• unset, empty, on, 1, true or yes: merge (today's behaviour);
• off, 0, false or no: do not merge — the schema field keeps exactly what the model wrote, and the reasoning is on reasoning_content only;
• anything else: ignored with one WARNING per distinct value; merge.

The variable is read on every call, so it takes effect without a restart.

Parameters:

Name	Type	Description	Default
logger	Optional[Logger]	Logger for the unrecognised-value warning; the module logger when None.	None

Returns:

Name	Type	Description
bool	bool	True to merge (the default), False when switched off.

Example

import os

os.environ["DPT_MERGE_REASONING_INTO_SCHEMA"] = "off"
merge_reasoning_into_schema_enabled()  # False

nearest_built_in_tool(tool_name, provider, supported_tools)

Name the nearest equivalent built-in tool a model actually supports.

Consulted only when a requested built-in tool has already been rejected. The suggestion is checked against this model's own supported_built_in_tools list, so the message can never point a caller at a second tool the model also lacks, and it is returned in the spelling the caller should pass — the standardized name, which :func:map_tool_name translates to the provider's own.

Nothing is aliased or substituted: the pairs in :data:BUILT_IN_TOOL_NEAREST_EQUIVALENT are tools that serve a similar intent, not the same tool under two names, so swapping one for the other silently would change what the model comes back with.

Parameters:

Name	Type	Description	Default
tool_name	str	The rejected tool name, in the standardized spelling or in any provider's own spelling — google_url_context handed to Anthropic resolves to url_context and is answered, because copying a name across providers is how callers arrive here.	required
provider	str	Provider name ("anthropic", "openai", "google", "fireworks").	required
supported_tools	List[str]	The model row's supported_built_in_tools list.	required

Returns:

Type	Description
Optional[str]	Optional[str]: The standardized name of the nearest equivalent the
Optional[str]	model supports, or None when the provider has no counterpart (or
Optional[str]	has one this particular model does not support either).

Example

nearest_built_in_tool("web_fetch", "google", ["google_url_context"])
# 'url_context'
nearest_built_in_tool("web_search", "fireworks", ["remote_mcp"])
# None

normalize_retryable_stream_exception(provider, exception, *, raw_event_type=None)

Convert retry-worthy generic stream exceptions into a normalized wrapper.

Provider-native API errors should usually be re-raised directly by the caller. This helper is for transport/runtime stream failures that should be visible to @retry_on_api_error without widening the public API surface.

parse_skill_md_frontmatter(skill_path, logger)

Parse SKILL.md file and extract YAML frontmatter.

Parameters:

Name	Type	Description	Default
skill_path	str	Path to the skill directory	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
dict	dict	{"name": str, "description": str, "valid": bool, "error": str}

read_tool_reported_spend(tool_result, tool_name, tool_call_id, iteration, logger)

Read the spend a tool reported in its result.

Called by all four tool loops on the raw result, before :func:cap_tool_result, so truncation cannot hide the marker. The marker is the _dpt_spend key at the top level of the tool's returned object: a dict result, or a text content item holding that object as JSON (how every MCP transport and local @tool function carries a dict). A malformed value is ignored with a WARNING. Returns None when DPT_AGENT_TOOL_REPORTED_COST=off.

Parameters:

Name	Type	Description	Default
tool_result	Any	The result the loop received (MCP-shaped for every client-executed tool).	required
tool_name	str	The tool's name.	required
tool_call_id	Optional[str]	The provider's id for the call.	required
iteration	int	The loop iteration (1-based) the call belongs to.	required
logger	Logger	Logger for a malformed marker.	required

Returns:

Type	Description
Optional[Dict[str, Any]]	Optional[Dict[str, Any]]: ``{"tool_name", "tool_call_id", "usd",
Optional[Dict[str, Any]]	"estimated", "iteration"}, orNone`` when the result reports no
Optional[Dict[str, Any]]	spend.

Example

result = {"content": [{"type": "text", "text": '{"ok": true, "_dpt_spend": {"usd": 0.01}}'}]}
read_tool_reported_spend(result, "speak", "call_1", 1, logger)["usd"]  # 0.01

release_output_name(taken, path)

Give back the name :func:unique_output_path chose for path after its write failed.

A name counts as taken once it is chosen, so without this a failed write would push the next file of that name to (3) with no (2) on disk. None (no name was chosen before the failure) is a no-op.

Parameters:

Name	Type	Description	Default
taken	Set[str]	The run's set of written names.	required
path	Optional[str]	The path :func:unique_output_path returned, or None.	required

remove_inline_citations(text, provider='auto')

Remove inline citations from text generated by LLM APIs.

Parameters:

Name	Type	Description	Default
text	str	The text containing citations to remove	required
provider	str	The provider name ("openai", "google", "anthropic", or "auto")	'auto'

Returns:

Type	Description
str	Cleaned text with citations removed

report_dropped_parts(func)

Put dropped_parts on a provider entry function's result dict (B12).

The four provider entry functions return their result dict from a dozen places each — success, every recovery branch, every error shape. Rather than touch every one, this decorator owns the single guarantee: the caller's dropped_parts_sink (created here when the caller passed none) is copied onto whatever dict the function returns, so dropped_parts is present on every run regardless of policy or outcome.

Applied OUTSIDE @retry_on_api_error, so it sees the value tenacity finally produced. Each tenacity attempt re-validates the same prompt, which is why the entry functions clear the sink on the way in — the list reflects the attempt that actually ran, never a per-retry pile-up.

A raised :class:UnsupportedPartError passes straight through; it carries its own dropped_parts.

Parameters:

Name	Type	Description	Default
func	Callable	A provider entry function (sync or async) that takes a dropped_parts_sink keyword.	required

Returns:

Name	Type	Description
Callable	Callable	The wrapped function, same signature.

Example

@report_dropped_parts
@retry_on_api_error
def call_x_llm_structured_output_text(..., dropped_parts_sink=None): ...

resolve_defer_tools(defer_tools)

Normalize defer_tools into a two-argument predicate.

Agent(defer_tools=...) accepts a set (or any iterable) of tool names, or a callable. The callable form is always invoked with two positional arguments, (tool_name, server_name) — server_name is the MCP server the tool came from, or None for a tool that has no server. A one-argument callable is therefore rejected here rather than at the first tool, because a TypeError raised mid-extraction would look like a provider failure.

Parameters:

Name	Type	Description	Default
defer_tools	Any	None, an iterable of tool-name strings, or a callable taking (tool_name, server_name) and returning a truthy value for the tools that should be deferred.	required

Returns:

Type	Description
Optional[Callable[[str, Optional[str]], bool]]	Optional[Callable[[str, Optional[str]], bool]]: A predicate, or None
Optional[Callable[[str, Optional[str]], bool]]	when defer_tools is None or an empty collection (nothing to
Optional[Callable[[str, Optional[str]], bool]]	defer, so no tool-search machinery is engaged at all).

Raises:

Type	Description
TypeError	If defer_tools is neither a callable nor an iterable of strings, or a callable that cannot accept two positional arguments.

Example

predicate = resolve_defer_tools({"search_inventory", "lookup_sku"})
predicate("search_inventory", "catalog")  # True
predicate("send_email", "mail")  # False

resolve_max_tool_result_chars(max_tool_result_chars, model_name, provider, logger)

Resolve the per-result character cap applied to tool results.

None derives the default from the registry: context_window_tokens * TOOL_RESULT_CAP_CHARS_PER_TOKEN * TOOL_RESULT_CAP_CONTEXT_FRACTION — a quarter of the window at four characters per token, numerically one character per token of window: 200,000 characters on a 200k-token model, 1,000,000 on a 1M-token model. The window comes from agent.compaction.resolve_context_window, which warns and floors at 128,000 tokens when the registry has no value. An explicit integer is used as given; 0 (or a negative value) turns the cap off, and so does any value larger than a result can be.

Parameters:

Name	Type	Description	Default
max_tool_result_chars	Optional[int]	The caller's setting.	required
model_name	str	Registry model id, for the default.	required
provider	str	Provider name, for the default.	required
logger	Logger	Logger for the context-window lookup.	required

Returns:

Name	Type	Description
int	int	The cap in characters; 0 means no cap.

Example

resolve_max_tool_result_chars(None, "claude-sonnet-5", "anthropic", logger)  # 1 char per window token
resolve_max_tool_result_chars(0, "claude-sonnet-5", "anthropic", logger)  # 0 -> off

resolve_session_kinds(model_name)

Return the session kinds a model can serve, derived from the registry.

The answer comes from descriptions.<model>.modality (absent means a text model) plus the capability keys already on the row: a row declaring supports_translation adds "translation", and a realtime row that accepts audio input also serves "transcription" on the same socket. Nothing here is hardcoded per provider — a new model reaches the right session family by registry row alone, exactly like routing and pricing do.

A row may also state its surfaces outright with session_kinds, which is read verbatim instead of derived. That exists for the rows derivation cannot reach: gpt-live-transcribe is a modality: "realtime" row (it is a socket model, not a file one) that serves only transcription, and no combination of modality and input types distinguishes it from a conversational row. Values outside :data:SESSION_KINDS are dropped with a warning rather than invented into the vocabulary.

Parameters:

Name	Type	Description	Default
model_name	str	Registry model ID.	required

Returns:

Type	Description
str	Tuple[str, ...]: The supported session kinds, ordered as in
...	data:SESSION_KINDS. Empty when the model is not in the registry.

Example

from doc_processing_toolkit.utils import resolve_session_kinds

resolve_session_kinds("gpt-5.6-luna")
# ('text',)

resolve_tokenizer_encoding(model_name, provider, logger)

Resolve a model's tiktoken encoding from the registry, warning loudly on a miss.

This replaces the model_name.lower().startswith("gpt-5") rescue that used to sit inside the OpenAI tiktoken fallback. That rule broke the repo's "model additions are JSON-only" rule: a future gpt-6 row matched no prefix, so tiktoken.encoding_for_model raised and every pre-flight estimate for it silently degraded to the ~4-chars-per-token text approximation — quietly wrong compaction thresholds and quietly wrong estimate_tokens answers, with nothing in the logs naming the cause.

Resolution order:

1. descriptions.<model>.tokenizer_encoding — the registry's answer.
2. :data:DEFAULT_TOKENIZER_ENCODING, with a warning naming the exact JSON path to add, mirroring :func:doc_processing_toolkit.agent.compaction.resolve_context_window.

Warn-and-default rather than raise: token counting is an estimate feeding optional machinery (compaction thresholds, failure-path cost estimates), so a missing registry key must not abort a run. One call emits at most one warning.

Parameters:

Name	Type	Description	Default
model_name	str	Registry model ID (e.g. "gpt-5.6-luna").	required
provider	str	Provider key the model belongs to, used only to make the warning actionable (e.g. "openai").	required
logger	Logger	Logger for the loud-miss warning.	required

Returns:

Type	Description
str	Tuple[str, bool]: The tiktoken encoding name, and True when it came
bool	from the registry or False when the default had to be used. Callers
Tuple[str, bool]	use the flag to label the count honestly rather than to change it.

Example

import logging

from doc_processing_toolkit.utils import resolve_tokenizer_encoding

encoding, from_registry = resolve_tokenizer_encoding("gpt-5.6-luna", "openai", logging.getLogger())
# ("o200k_base", True)

robust_ocr_text_reader(input_data)

Reads text from a file path or uses text directly, handling potential encoding issues.

Also cleans the text by removing non-standard characters and escaping braces.

Parameters:

Name	Type	Description	Default
input_data	str or file path	Either a file path string or a string containing the text.	required

Returns:

Name	Type	Description
str	str	Cleaned and processed text. Returns an empty string if input is invalid or reading fails.

safe_output_filename(name, fallback)

Reduce a provider-supplied filename to a name that stays inside the output folder.

A code-execution download names its file after what the model or the provider reports (container_file_citation.filename on OpenAI, the Files API metadata on Anthropic, a blob's display_name on Google). Joined as given, ../../outside.txt or an absolute path would write outside code_execution_output_dir. The name is cut to its last path component (either separator), every character in :data:_OUTPUT_FILENAME_BAD_CHARS becomes _, trailing dots and spaces go (Windows drops them), a Windows device name (nul, con.txt, COM1) is prefixed with _, and a name that ends up empty becomes fallback, cleaned the same way (file when nothing is left of it either).

Parameters:

Name	Type	Description	Default
name	Any	The reported filename; anything but a non-empty string yields fallback.	required
fallback	str	The name to use when nothing usable is left, such as file_<id>.	required

Returns:

Name	Type	Description
str	str	A single path component with no separator, reserved character or
	str	device name.

Example:

```python
safe_output_filename("../../etc/passwd", "file_1")  # "passwd"
safe_output_filename("report 12:30.csv", "file_1")  # "report 12_30.csv"
safe_output_filename("nul", "file_1")  # "_nul"
safe_output_filename("..", "file_1")  # "file_1"
```

setup_logger(log_file, level=logging.DEBUG, name='doc_processing')

Set up a logger instance to write logs to a specified file.

Ensures that handlers are not duplicated if called multiple times for the same log file name (though typically used with unique names like page identifiers).

Parameters:

Name	Type	Description	Default
log_file	str	The full path to the log file. If None, logs to console only.	required
level	int	Logging level (e.g., logging.DEBUG, logging.INFO).	DEBUG
name	str	Base name for the logger.	'doc_processing'

Returns:

Type	Description
Logger	logging.Logger: Configured logger instance.

should_retry_api_call(exception)

Custom retry condition function for Tenacity.

Checks for known rate limit exceptions, standard HTTP status codes, specific Google API retryable errors, and keywords in error messages to determine if an API call should be retried.

Tenacity is the toolkit's only retry layer: the Anthropic and OpenAI SDK clients are built with max_retries=0 (see DPT_SDK_RETRIES), so this predicate covers everything those SDKs used to retry on their own. The retryable set is:

• NormalizedStreamError (retryable by construction), and never a NonRetryableStreamEventError (a failure event its stream consumer classified as deterministic), whatever the rules below would say;
• rate-limit classes: openai.RateLimitError, anthropic.RateLimitError, the Fireworks rate-limit class, and Google ResourceExhausted / ServiceUnavailable / InternalServerError / DeadlineExceeded;
• transport faults: openai.APIConnectionError / openai.APITimeoutError, anthropic.APIConnectionError / anthropic.APITimeoutError, and bare httpx / httpx2 transport errors;
• HTTP status 408, 409, 429, and every status >= 500 except 501 (Not Implemented) — so 500, 502–504, 529 and gateway statuses such as Cloudflare's 520–524. The status is read from status_code (OpenAI, Anthropic, Fireworks) and, for the google.genai.errors.ClientError / ServerError that every Gemini call raises, from .code (falling back to status_code) — only an HTTP-range code counts, so a Gemini Live WebSocket close code such as 1011 is not treated as a 5xx;
• a small set of rate-limit / overload keywords in the error message, only when the status is not a known non-retryable one (400, 401, 403, 404, 413, 501).

The server's x-should-retry response header overrides all of the above, exactly as the Anthropic and OpenAI SDKs treat it: true retries whatever the status, false prevents a retry even on a retryable status or class; when it is absent (or any other value) the rules above apply. The header is read off any exception that carries a response with headers, google.genai.errors.APIError included.

Parameters:

Name	Type	Description	Default
exception	Exception	The exception object caught by tenacity.	required

Returns:

Name	Type	Description
bool	bool	True if the call should be retried, False otherwise.

shutdown_clients()

Close and clear every process-cached client (embedder clean-exit hook).

Sync clients are closed directly. Async clients are closed on the event loop that created them when it is still alive (from inside that loop the close completes after this call returns; from another thread it is waited for briefly); one whose loop is already closed is dropped for garbage collection, which closes its sockets. Never raises.

Warning

Called from a thread that is already running an event loop, the coroutine closes are only scheduled — this function returns before they finish. Await :func:shutdown_clients_async there instead.

shutdown_clients_async() async

Awaitable counterpart of :func:shutdown_clients for async callers.

:func:shutdown_clients can only schedule a coroutine close when it is called from a thread that is already running an event loop — an ASGI shutdown handler, for example — and returns before the close completes. Awaiting this function instead drains the closes on the calling loop, so the connection pools are really released before the process continues.

What it does, in one pass over the cache (which is cleared first, under the lock, so a concurrent caller cannot hand out a client that is being closed):

• Async clients bound to the calling loop — close() (or aclose()) is awaited, all of them together, under a shared _CLIENT_CLOSE_TIMEOUT_SECONDS budget.
• Async clients bound to another or a closed loop — discarded without being awaited. Their transports can only be driven by their own loop; CPython closes the sockets when the client is collected.
• Sync clients — closed inline, exactly as :func:shutdown_clients does.

Clients belonging to an Agent(api_key=...) are never touched: a per-key Agent bypasses the shared cache entirely and closes its own clients in Agent.close().

The call is idempotent and safe when nothing is cached, and it never raises — a failing close is swallowed rather than propagated into shutdown.

Example

from contextlib import asynccontextmanager

from fastapi import FastAPI

from doc_processing_toolkit.utils import shutdown_clients_async


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await shutdown_clients_async()


app = FastAPI(lifespan=lifespan)

strip_registry_metadata(row)

Drop reserved _-prefixed metadata keys from a registry row.

Registry rows may carry provenance keys — _source, _verified, _tier — recording where a price or limit came from, when it was last checked, and which account tier it describes. They are documentation for whoever edits model_information.json; no consumer reads them, and none may re-emit them. Readers that select individual keys by name ignore them for free, but any reader that hands a whole row onward — a discovery listing, a prompt built from the catalogue — has to filter, or the provenance note becomes part of a public payload.

Non-dict input is returned unchanged, so this is safe to apply to a value that may be a missing row ({}) or a scalar.

Parameters:

Name	Type	Description	Default
row	Any	A registry row (pricing/limits/descriptions entry), or any value.	required

Returns:

Name	Type	Description
Any	Any	A shallow copy of row without its _-prefixed keys when
	Any	row is a dict; otherwise row itself.

Example

from doc_processing_toolkit.utils import strip_registry_metadata

row = {"input": 2.0, "output": 12.0, "_source": "vendor page 2026-09-19"}
strip_registry_metadata(row)
# {'input': 2.0, 'output': 12.0}

tool_reported_cost_enabled()

Whether an agent counts the spend its tools report (DPT_AGENT_TOOL_REPORTED_COST).

Unset, empty or on (the default) counts it; off restores the totals from before 0.5.1 exactly: nothing is read and the new result fields keep their empty defaults. Any other value logs one WARNING and stays on.

Returns:

Type	Description
bool	True unless the variable is off.

Example

import os

os.environ["DPT_AGENT_TOOL_REPORTED_COST"] = "off"
tool_reported_cost_enabled()  # False

tool_reported_spend_on(exc)

The tool-spend records an exception carries out of a tool loop.

Parameters:

Name	Type	Description	Default
exc	Optional[BaseException]	The exception, or None.	required

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: The records :func:carry_tool_reported_spend
List[Dict[str, Any]]	attached, or [].

tool_result_data(tool_result)

A tool result as JSON-clean data, for the full-result event field.

The event's result string is Python's str() of the result, which a client cannot parse reliably. This is the same result as data:

• one text item (how a local @tool function's return value and most MCP results arrive): that text, decoded when it holds a JSON object or array, so a tool that returned a dict gives the dict back;
• any other MCP result: its list of content items, each text item's text decoded the same way in place (image and audio items keep their base64);
• anything else: the value itself.

Every value is passed through one JSON round trip, so the data is JSON-clean (a non-JSON value becomes its str(), a NaN or infinite number becomes None) and each call returns a fresh copy.

Parameters:

Name	Type	Description	Default
tool_result	Any	The raw result the loop received, before the size cap.	required

Returns:

Name	Type	Description
Any	Any	The result as JSON data.

Raises:

Type	Description
TypeError	For a structure JSON cannot hold at all (a dict key that is not a string or number); :func:tool_result_event_fields catches it.
ValueError	For a circular reference; caught the same way.
RecursionError	For nesting deeper than the interpreter allows; caught the same way.

Example

tool_result_data({"content": [{"type": "text", "text": '{"path": "a.wav"}'}], "isError": False})
# {'path': 'a.wav'}

tool_result_event_fields(tool_result, preview_chars=DEFAULT_TOOL_RESULT_PREVIEW_CHARS)

The result fields of a tool_call_result stream or hook event.

The event carries the client's copy of the result, which is independent of the model's copy (the one max_tool_result_chars cuts): this never changes what the model is sent.

Parameters:

Name	Type	Description	Default
tool_result	Any	The raw result the loop received, before the size cap.	required
preview_chars	Optional[int]	Agent(tool_result_preview_chars=): how many characters of the result's str() the event carries. 0 means all of it, plus result_data. None or a negative value means the default, 500.	DEFAULT_TOOL_RESULT_PREVIEW_CHARS

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"result", "result_truncated"}, and result_data
Dict[str, Any]	(:func:tool_result_data) when preview_chars is 0.
Dict[str, Any]	result_truncated says whether result is shorter than the whole
Dict[str, Any]	result, so a client knows whether it holds everything. A result JSON
Dict[str, Any]	cannot represent at all (a tuple dict key, a circular reference) gets
Dict[str, Any]	its str() as result_data: building the event must never fail
Dict[str, Any]	the run, whose model copy does not depend on it.

Example

tool_result_event_fields({"content": []}, 5)
# {'result': "{'con", 'result_truncated': True}

tool_spend_event_fields(spend)

The tool_call_result event fields for a call's reported spend.

Parameters:

Name	Type	Description	Default
spend	Optional[Dict[str, Any]]	The :func:read_tool_reported_spend record, or None.	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"tool_reported_cost", "tool_reported_cost_estimated"}
Dict[str, Any]	when the tool reported spend, else {}, so events from tools that
Dict[str, Any]	report nothing keep their existing keys.

tool_spend_marker(usd, estimated=False)

The marker a tool puts in its result to report what it spent.

Merge it into the dict a tool returns; the agent that called the tool adds usd to its own cost (see :func:read_tool_reported_spend).

Parameters:

Name	Type	Description	Default
usd	float	What the tool's own provider calls cost, in US dollars.	required
estimated	bool	Whether that figure is an estimate rather than billed usage.	False

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"_dpt_spend": {"usd": ..., "estimated": ...}}.

Example

from doc_processing_toolkit.utils import tool_spend_marker

@tool
def translate(text: str) -> dict:
    result = my_translation_call(text)
    return {"text": result.text, **tool_spend_marker(result.cost)}

truncated_result_ids(value)

The ids named by read_truncated_result markers anywhere in value.

Used when a session is restored: its history keeps the markers but not the text they refer to, so the new store retires them (:meth:TruncatedResultStore.retire). value is serialised to JSON first, so markers inside nested provider messages are found whatever their escaping.

Parameters:

Name	Type	Description	Default
value	Any	Messages, or any JSON-serialisable structure.	required

Returns:

Name	Type	Description
set	set	The ids found (empty when there are none).

unique_output_path(output_dir, filename, taken, logger=None)

A path for filename under output_dir that no earlier write of this run used.

taken holds the names this run has written so far — one set per run (the Google tool-combination loop carries it across iterations) or per download call where that call is the run's only writer. A name in it gets (2), (3), ... before its extension (report.csv → report (2).csv; a name without an extension or a dotfile gets the suffix at its end) and the chosen name is added. Names compare as os.path.normcase of themselves: on Windows A.csv and a.csv are one name, elsewhere two (on a case-insensitive macOS volume the second then overwrites the first, as it did before). A write that fails gives its name back through :func:release_output_name. Nothing on disk is consulted: a file an earlier run left in a reused explicit folder is overwritten as it always was, so a client that reruns a job into its own folder still finds report.csv where it looks. Every code-execution write site goes through it (0.5.3): until then a second file of one run with a name already written — a later iteration's generated_code_1.py on Google, two names that :func:safe_output_filename cleans to one, a model saving output.csv twice — overwrote the first, both records reporting success.

Parameters:

Name	Type	Description	Default
output_dir	str	The folder the file is written to.	required
filename	str	A single path component (already cleaned by :func:safe_output_filename where the name comes from a provider).	required
taken	Set[str]	The names this run has written, updated in place.	required
logger	Optional[Logger]	Where one INFO line goes when the name had to change; nothing is logged for an unused name.	None

Returns:

Name	Type	Description
str	str	The path to write to; its basename is the name to record.

Example:

```python
taken: set = set()
unique_output_path(folder, "report.csv", taken, logger)   # .../report.csv
unique_output_path(folder, "report.csv", taken, logger)   # .../report (2).csv
```

validate_and_parse_structured_response(tool_args, response_schema, logger)

Validate and parse structured response data against a Pydantic schema.

Parameters:

Name	Type	Description	Default
tool_args	dict	The tool call arguments to validate.	required
response_schema	Any	Pydantic model class to validate against.	required
logger	Logger	Logger instance for logging.	required

Returns:

Name	Type	Description
tuple	Tuple[Optional[dict], Optional[str]]	(parsed_data, error_message) parsed_data: Validated dict if successful, None if validation failed error_message: Error description if validation failed, None if successful

validate_built_in_tools(provider, model_name, built_in_tools, model_info_config, logger, filter_for_structured_output=None)

Validate tools against model capabilities from model_information.json.

Parameters:

Name	Type	Description	Default
provider	str	Provider name (anthropic, openai, google)	required
model_name	str	The model name to validate against	required
built_in_tools	Dict[str, BuiltInToolConfig]	Dictionary of tool configurations	required
model_info_config	dict	The model information configuration	required
logger	Logger	Logger instance	required
filter_for_structured_output	List[str]	Optional list of tool names to filter for concurrent use with structured outputs. If provided, only tools in this list will be validated and returned. Tools not in this list will be added to disabled_tools with a warning.	None

Returns:

Type	Description
Tuple[Dict[str, BuiltInToolConfig], Dict[str, BuiltInToolConfig], List[str]]	Tuple of (validated_tools, disabled_tools, warnings)

validate_builtin_tools_concurrency(model_name, mcp_servers, built_in_tools, model_info_config, logger)

Validate that built-in tools and custom MCP tools can be used concurrently.

Some providers/models do NOT support combining built-in tools (web_search, code_execution, etc.) with custom function declarations: - Google Gemini: supported on the Gemini 3 text rows (Preview); false on the Gemini 3 image rows and on every Gemini 2.5 row - Anthropic: DOES support concurrent use - OpenAI: DOES support concurrent use - Fireworks: Varies by model

This function checks model-by-model using the 'supports_concurrent_built_in_and_custom_tools' key in model_information.json.

Parameters:

Name	Type	Description	Default
model_name	str	Model name to validate	required
mcp_servers	Optional[list]	List of MCP server connections (or None)	required
built_in_tools	Optional[dict]	Dict of built-in tool configurations (or None)	required
model_info_config	dict	Model information configuration	required
logger	Logger	Logger instance	required

Raises:

Type	Description
ValueError	If both MCP tools and built-in tools are provided but model doesn't support concurrent use

validate_effort_config_anthropic(effort_config, model_name, model_info_config, logger)

Validates effort configuration for Anthropic models.

The effort parameter is generally available (no beta header required) and controls how eager Claude is about spending tokens. It affects text responses, tool calls, and extended thinking depth.

For Claude Opus 4.6 with adaptive thinking, effort is the recommended way to control thinking depth instead of budget_tokens.

Parameters:

Name	Type	Description	Default
effort_config	str	Effort level. Supported values depend on the model: "low": Most efficient, significant token savings "medium": Balanced approach with moderate savings "high": High capability (equivalent to not setting the parameter) "max": Absolute maximum capability (Opus 4.6 only - returns error on other models)	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
str	Optional[str]	Validated effort configuration or None if not supported or invalid

validate_extended_thinking_config_anthropic(thinking_config, model_name, model_info_config, logger)

Validates extended thinking configuration for Anthropic models.

Notes

Supports two thinking modes

• Adaptive thinking (recommended for models that support it, e.g., claude-opus-4-6): Claude dynamically decides when and how much to think. No beta header required. Automatically enables interleaved thinking. Use the effort parameter to control thinking depth instead of budget_tokens.
• Legacy/manual thinking (for older models): Uses explicit budget_tokens and interleaved_thinking flags.

When a model supports adaptive thinking (supports_adaptive_thinking=True in model_information.json) and a legacy thinking config is provided (with budget_tokens), the config is automatically converted to adaptive mode with a warning.

Parameters:

Name	Type	Description	Default
thinking_config	Any	Dict with thinking configuration. Supported formats: Adaptive: {"type": "adaptive"} Legacy: {"budget_tokens": 10000, "interleaved_thinking": True} Off: {"type": "disabled"} or {"type": "between_tools"}. Returns None here: what an "off" request sends depends on the registry row and is resolved by the provider entry points.	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
dict	Optional[dict]	Validated thinking configuration or None if not supported, invalid, or an "off" request. Adaptive mode returns: {"type": "adaptive", "display": "summarized"} Legacy mode returns: {"budget_tokens": int, "interleaved_thinking": bool} Plus every key the caller supplied that the toolkit does not interpret, preserved verbatim.

Notes

Only type, display, budget_tokens and the two accepted spellings of the interleaved flag are normalised. Every other key is preserved and reaches the API unchanged, so a caller can use a thinking.* field this toolkit has never heard of — for example thinking={"type": "adaptive", "block_binding": {...}} — without waiting for a release. An unrecognised key is logged at WARNING, since the commonest cause is a misspelling, but it is still forwarded.

validate_function_calling_structured_output_mode(model_name, response_schema, thinking_config, model_info_config, provider, logger)

Determines structured output mode for function calling and validates compatibility.

This function checks if a model supports native structured outputs with function calling (Path A) or requires the tool-based fallback approach (Path B). For Anthropic models using Path B, it also checks thinking configuration compatibility.

Parameters:

Name	Type	Description	Default
model_name	str	The model name to validate.	required
response_schema	Any	Pydantic model class for structured output, or None if not using.	required
thinking_config	dict	Extended thinking configuration (Anthropic-specific), or None.	required
model_info_config	dict	Model information configuration from model_information.json.	required
provider	str	Provider name ("anthropic", "openai", "google", "fireworks").	required
logger	Logger	Logger instance for logging messages.	required

Returns:

Name	Type	Description
tuple	Tuple[bool, bool]	(use_native_mode: bool, disable_thinking: bool) use_native_mode: True if model supports native structured outputs with function calling (Path A). False if tool-based approach should be used (Path B). disable_thinking: True if thinking_config should be ignored due to Path B incompatibility (Anthropic-specific). Always False for other providers.

Notes

Path A (native mode): Uses provider's output_format/response_format parameter.

- Anthropic: Uses output_format parameter (requires structured-outputs beta)
- OpenAI: Uses text.format in Responses API
- Google: Uses response_schema in generation_config
- Fireworks: Uses response_format with JSON schema

Path B (tool-based): Creates a 'submit_final_response' tool with the schema.

- Requires tool_choice="any" (or equivalent) to force tool use
- Anthropic: Incompatible with extended thinking (thinking will be disabled)
- Model calls the tool when ready to provide final structured response

validate_function_calling_support(model_name, model_info_config, logger)

Validate that a model supports function calling (provider-agnostic).

This should be called at the START of any function calling implementation for any provider (Anthropic, Google, OpenAI, Fireworks).

Parameters:

Name	Type	Description	Default
model_name	str	Model name to validate	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Raises:

Type	Description
ValueError	If model doesn't support function calling

Notes

Uses "supports_function_calling" key from model_information.json descriptions.

validate_image_detail_openai(image_detail, model_name, model_info_config, logger)

Validates image detail configuration for OpenAI models.

Parameters:

Name	Type	Description	Default
image_detail	str	Image detail level - "low", "high", "auto", or "original"	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
str	Optional[str]	Validated image detail value or None if not supported or invalid

validate_media_resolution_google(media_resolution, model_name, model_info_config, logger)

Validates global media resolution configuration for Google Gemini models.

Parameters:

Name	Type	Description	Default
media_resolution	str or None	Media resolution value String: one of MEDIA_RESOLUTION_UNSPECIFIED, MEDIA_RESOLUTION_LOW, MEDIA_RESOLUTION_MEDIUM, MEDIA_RESOLUTION_HIGH None: use default (MEDIA_RESOLUTION_UNSPECIFIED)	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Type	Description
Optional[str]	str or None: Validated media resolution value or None if not supported/invalid

validate_models(model_name, allowed_models_config)

Validates the provided generator and evaluator model names against allowed lists derived from the config.

Logs progress at DEBUG level. Raises ValueError if validation fails.

Parameters:

Name	Type	Description	Default
model_name	str	The name of the model to validate.	required
allowed_models_config	dict	The structure containing allowed models, loaded from model_information.json['pricing'] or model_information.json['limits']. Example: {'anthropic': {'models': {...}}, 'google': {'models': {...}}}	required

Returns:

Name	Type	Description
str	str	A string containing the provider name for the supplied model name.

Raises:

Type	Description
ValueError	If the model name is not found within the provided configuration or its provider cannot be determined.

validate_multimodal_function_response(model_name, tool_result, model_info_config, logger)

Validate and potentially filter multimodal content from function responses.

Parameters:

Name	Type	Description	Default
model_name	str	Model name to check capabilities	required
tool_result	dict	MCP tool result dict with 'content' and 'isError' keys	required
model_info_config	dict	Model information configuration	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
tuple	Tuple[list, list]	(filtered_content_items, omitted_items_descriptions) filtered_content_items: List of content items (filtered if needed) omitted_items_descriptions: List of strings describing omitted content (empty if nothing was omitted)

Notes

Some models do NOT support multimodal content in function responses

• Gemini 3: DOES support multimodal function responses
• Gemini 2.5: Does NOT support multimodal function responses
• Anthropic Claude: DOES support multimodal function responses
• OpenAI: Limited support (images only, no video/audio)
• Fireworks: Very limited support

This function checks model capability and performs graceful degradation

1. If multimodal supported: returns content as-is
2. If NOT supported: strips multimodal, adds notice, logs, returns filtered content

Uses "supports_multimodal_function_responses" key from model_information.json.

validate_per_part_media_resolution_google(media_resolution_list, user_prompt_parts, model_name, model_info_config, logger)

Validates per-part media resolution configuration for Google Gemini models.

Parameters:

Name	Type	Description	Default
media_resolution_list	list or None	List of media resolution values, one per part in user_prompt_parts. Each element should be a valid resolution string or None (to use default/global media resolution for that part). valid resolution strings: one of MEDIA_RESOLUTION_UNSPECIFIED, MEDIA_RESOLUTION_LOW, MEDIA_RESOLUTION_MEDIUM, MEDIA_RESOLUTION_HIGH or MEDIA_RESOLUTION_ULTRA_HIGH	required
user_prompt_parts	list	User prompt parts list (matching the format required for the call_google_llm_structured_output_text function)	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Type	Description
Optional[dict]	dict or None: Dictionary mapping media part index to resolution value, or None if not supported/invalid Example: {0: "MEDIA_RESOLUTION_HIGH", 2: "MEDIA_RESOLUTION_LOW"} (Only includes entries where resolution is specified - text parts and None values are omitted)

validate_reasoning_config_openai(reasoning_config, model_name, model_info_config, logger)

Validates reasoning configuration for OpenAI models.

Parameters:

Name	Type	Description	Default
reasoning_config	dict	Reasoning configuration with optional effort, summary and mode keys Example: {"effort": "low", "summary": "auto"} Example (pro mode, GPT-5.6+ models only): {"effort": "high", "mode": "pro"}	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
dict	Optional[dict]	Validated reasoning configuration or None if not supported or invalid

Notes

The mode key (e.g. GPT-5.6 pro mode) is validated against the model's supported_reasoning_modes list in model_information.json; models without that field do not support the parameter and the key is dropped with a warning. In "pro" mode the allowed effort values come from the model's supported_reasoning_values_pro_mode list (pro mode restricts effort — e.g. GPT-5.6 pro requires at least "medium"); an effort outside that list is dropped so the API default for pro mode applies, rather than failing the request.

validate_reasoning_effort_fireworks(reasoning_effort, model_name, model_info_config, logger)

Validates reasoning effort configuration for Fireworks models.

Parameters:

Name	Type	Description	Default
reasoning_effort	str | None	Requested reasoning effort level. Accepted values are per-model and read from the model's supported_reasoning_values list — commonly "none", "low", "medium" and "high", with "xhigh" on newer models. Some models omit "none" because thinking is always on.	required
model_name	str	Model name to validate against.	required
model_info_config	dict	Model information configuration from model_information.json.	required
logger	Logger	Logger instance.	required

Returns:

Type	Description
Optional[str]	str | None: The validated reasoning effort value, or None when the value was dropped.

Notes

An unsupported value is dropped rather than raised, so the run still completes — but the drop is logged at WARNING level. Proceeding without the requested effort means the model's own default effort applies, which changes behavior, latency and cost.

Two distinct cases both return None:

• The model lists no supported_reasoning_values at all, so it does not support reasoning and the parameter is ignored.
• The model supports reasoning but not the specific value requested.

validate_skill_description_anthropic(description, logger)

Validates custom skill description meets Anthropic API requirements.

Parameters:

Name	Type	Description	Default
description	str	The skill description to validate	required
logger	Logger	Logger instance	required

Returns:

Type	Description
Tuple[bool, str]	Tuple[bool, str]: (is_valid, error_message)

validate_skill_name_anthropic(name, logger)

Validates custom skill name meets Anthropic API requirements.

Parameters:

Name	Type	Description	Default
name	str	The skill name to validate	required
logger	Logger	Logger instance	required

Returns:

Type	Description
Tuple[bool, str]	Tuple[bool, str]: (is_valid, error_message)

validate_skills_config_anthropic(skills, model_name, model_info_config, built_in_tools, logger)

Validates skills configuration for Anthropic models.

Parameters:

Name	Type	Description	Default
skills	Any	Skills configuration - can be: Single built-in skill name (str): "xlsx" Multiple built-in skills (list): ["xlsx", "pptx"] Custom skill directory path (str): "/path/to/skill/" Mixed skills (list): ["/path/to/skill/", "xlsx"] Dict format: {"type": "anthropic", "skill_id": "xlsx", "version": "latest"}	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration	required
built_in_tools	dict or None	Built-in tools configuration	required
logger	Logger	Logger instance	required

Returns:

Type	Description
Tuple[List[dict], List[str], List[str]]	Tuple[List[dict], List[str], List[str]]: (validated_skills, custom_skill_paths, warnings) - validated_skills: List of skill dicts ready for API - custom_skill_paths: List of custom skill directory paths to upload - warnings: List of warning messages

validate_structured_output_and_tools(response_schema, built_in_tools, model_info_config, provider, model_name, logger)

Validates structured outputs and built-in tools configuration, checking for concurrent support.

Parameters:

Name	Type	Description	Default
response_schema	Any	Pydantic schema or None	required
built_in_tools	Any	Dict of tool configs or None	required
model_info_config	dict	Model information configuration from model_information.json	required
provider	str	Provider name for logging	required
model_name	str	Model name to check for concurrent support	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
tuple	tuple	(use_structured_output: bool, use_tools: bool, compatible_tools: list | None) - compatible_tools: List of tool names that support concurrent use with structured outputs. Empty list if model doesn't support concurrent use. None if only one mode is being used (not concurrent).

Raises:

Type	Description
ValueError	If neither structured outputs nor built-in tools is provided, or if both are provided but the model doesn't support concurrent use.

validate_thinking_budget_google(thinking_budget, model_name, model_info_config, logger)

Validates thinking budget configuration for Google Gemini models.

The per-model budget rules are data-driven: they come from descriptions.<model>.thinking_budget_rules in model_information.json (keys min, max, default, allow_zero). A model without that registry entry does not take thinking_budget — Gemini 3 models take thinking_level instead (see validate_thinking_level_google) — and the parameter is dropped with a warning.

Parameters:

Name	Type	Description	Default
thinking_budget	int or None	Thinking budget value Integer: specific token budget None: use the model's registry default (-1 dynamic, or null to leave thinking unset) 0: disable thinking (models whose rules set allow_zero) -1: dynamic thinking	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Type	Description
Optional[int]	int or None: Validated thinking budget or None if not supported or invalid

validate_thinking_level_google(thinking_level, model_name, model_info_config, logger)

Validates thinking level configuration for Google Gemini models.

Parameters:

Name	Type	Description	Default
thinking_level	str	Thinking level - model-specific values ("minimal", "low", "medium" or "high")	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
str	Optional[str]	Validated thinking level ("minimal", "low", "medium" or "high") or None if not supported or invalid

validate_user_prompt_parts(user_prompt_parts, model_name, model_info_config, logger, *, on_unsupported_part='drop', dropped_parts=None)

Validates user_prompt_parts against supported input types for the given model.

This function checks each part in user_prompt_parts to ensure it's a supported input type for the specified model. Text strings are always allowed. Dictionary parts are validated by constructing their "type-media_type" format (e.g., "base64-image/png", "base64-application/pdf") and checking against the model's input_types list in model_information.json.

Notes

Supported Input Types by Provider:

- Google: "base64-image/{png,jpeg,jpg,gif,webp,bmp}", "video", "audio", "document", "youtube"
- OpenAI: "base64-image/{png,jpeg,jpg,gif,webp,bmp}", "base64-application/pdf"
- Anthropic: "base64-image/{png,jpeg,jpg,gif,webp,bmp}", "base64-application/pdf"
- Fireworks: "base64-image/{png,jpeg,jpg,gif,webp,bmp}"

Input Type Format: For dictionary parts, the input type is constructed as:

- "{part['type']}-{part['media_type']}" for base64 content (images, PDFs)
- "{part['type']}" for other types (video, audio, document, youtube)

Parameters:

Name	Type	Description	Default
user_prompt_parts	list	List of user prompt parts. Can contain: Strings (text content, always allowed) Dicts with 'type' and optionally 'media_type' keys	required
model_name	str	Model name to look up in the descriptions section	required
model_info_config	dict	Model information configuration containing input_types for each model in the descriptions section	required
logger	Logger	Logger instance for warnings	required
on_unsupported_part	str	Policy for a part the model cannot accept — "drop" (default, today's behaviour: warn and filter it out) or "raise" (refuse the run with :class:UnsupportedPartError before any request is sent). Anything else is itself a ValueError.	'drop'
dropped_parts	list or None	Optional caller-supplied list that collects one record per dropped part, extended in place so the caller sees the drops whichever way the call returns. Each record is {"kind", "index", "reason", "provider", "model"} — kind is the constructed input-type key ("video", "base64-application/pdf", or "<python type>" for a part that is neither a string nor a dict). Records are appended under BOTH policies, so a run that drops parts is detectable without scraping logs.	None

Returns:

Name	Type	Description
list	list	Filtered list of valid user_prompt_parts. Invalid parts are logged as warnings and excluded from the returned list.

Raises:

Type	Description
ValueError	If on_unsupported_part is not "drop" or "raise".
UnsupportedPartError	If on_unsupported_part="raise" and any part is not supported by the model.

Examples:

parts = ["Hello", {"type": "base64", "media_type": "image/png", "data": "..."}]
valid_parts = validate_user_prompt_parts(parts, "gpt-4o", config, logger)
# Returns all parts if image/png is supported by gpt-4o

parts = [{"type": "video", "uri": "..."}]  # video not supported by gpt-4o
valid_parts = validate_user_prompt_parts(parts, "gpt-4o", config, logger)
# Returns empty list, logs warning about unsupported video type

drops = []
validate_user_prompt_parts(parts, "gpt-4o", config, logger, dropped_parts=drops)
# drops == [{"kind": "video", "index": 0, "reason": "...", ...}]

validate_verbosity_config_openai(verbosity_config, model_name, model_info_config, logger)

Validates verbosity configuration for OpenAI models.

Parameters:

Name	Type	Description	Default
verbosity_config	str	Verbosity level - "low", "medium", or "high"	required
model_name	str	Model name to validate against	required
model_info_config	dict	Model information configuration from model_information.json	required
logger	Logger	Logger instance	required

Returns:

Name	Type	Description
str	Optional[str]	Validated verbosity configuration ("low", "medium", or "high") or None if not supported or invalid

wait_retry_after_or_exponential(retry_state)

Tenacity wait strategy: honour retry-after, else back off exponentially.

When the failed attempt's exception carries an HTTP response with a retry-after-ms / retry-after header (seconds or HTTP-date), wait that long, capped at RETRY_AFTER_MAX_SECONDS (60 s). Otherwise use the toolkit's exponential back-off: wait_exponential(multiplier=1, min=4, max=60), i.e. 4 s, 4 s, 4 s, 8 s between the five attempts.

Parameters:

Name	Type	Description	Default
retry_state	RetryCallState	The tenacity call state for the failed attempt.	required

Returns:

Type	Description
float	The number of seconds to sleep before the next attempt.

Example

from tenacity import retry, stop_after_attempt
from doc_processing_toolkit.utils import wait_retry_after_or_exponential

@retry(wait=wait_retry_after_or_exponential, stop=stop_after_attempt(5))
def call_provider():
    ...

Shared Schemas

doc_processing_toolkit.schemas

Schema definitions for structured output from LLM calls.

This module defines Pydantic models and related classes for structured output from LLM calls.

See Also

• Core module overview
• Core API reference

SimpleTextResponse

Bases: BaseModel

Simple response schema for plain text mode.

This schema is auto-injected by structured output functions when neither a response_schema nor built_in_tools are specified. It enables a simplified "plain text mode" that returns just the model's text response without requiring users to define a schema.

The structured output functions will automatically extract the 'text' field and return it directly in the 'data' key of the response dictionary.

This schema is not intended to be used directly by end users - it is an internal implementation detail to support the plain text operation mode.

Example usage (internal): # When user calls: call_anthropic_llm_structured_output_text(..., response_schema=None, built_in_tools=None) # The function auto-injects: response_schema = SimpleTextResponse # And returns: {'data': response.text, 'input_tokens': ..., ...}

PDF Processing Utilities

doc_processing_toolkit.pdf_processor

PDF processing and image/OCR preparation utilities.

This module provides helper functions to convert PDFs to images, run recursive PDF conversion across directories, and call LLM Whisperer extraction flows.

See Also

• Core API reference
• PDF processing recipes
• Core module overview

pdf_to_images(pdf_path, output_dir=None, dpi=600, disable_decompression_bomb=True, return_images=False, poppler_path='C:/Program Files/Poppler/poppler-24.08.0/Library/bin')

Convert a PDF file to high-quality images suitable for OCR. Each page of the PDF will be saved as a PNG image or returned as PIL Image objects.

Parameters:

Name	Type	Description	Default
pdf_path	str	Path to the PDF file	required
output_dir	str	Directory where the images will be saved. Required when return_images is False.	None
dpi	int	DPI for the output images (default: 600)	600
disable_decompression_bomb	bool	Whether to disable decompression bomb protection (default: True)	True
return_images	bool	If True, return a list of PIL Image objects instead of saving to disk (default: False)	False
poppler_path	str	Path to the Poppler binaries directory. Required for PDF-to-image conversion. On Windows, defaults to "C:/Program Files/Poppler/poppler-24.08.0/Library/bin". Override this for other OS platforms or custom Poppler installations.	'C:/Program Files/Poppler/poppler-24.08.0/Library/bin'

Returns:

Name	Type	Description
list	list	If return_images is False, returns a list of dictionaries with keys 'image_path'. If return_images is True, returns a list of dictionaries with keys 'page' and 'Image'.

Raises:

Type	Description
ValueError	If return_images is False and output_dir is not provided

preprocess_image(image, provider_list, logger)

Resizes, converts, and formats a PIL image based on the strictest requirements of the specified generator and evaluator model providers.

Handles pixel count, file size, max edge dimensions, color mode conversion (if needed), and output format (PIL or Base64).

Parameters:

Name	Type	Description	Default
image	Image	The input PIL image object.	required
provider_list	list	A unique list of provider names for the generator and evaluator models.	required
logger	Logger	Logger instance for logging preprocessing steps.	required

Returns:

Name	Type	Description
dict	Optional[dict]	A dictionary containing the preprocessed image data: {'type': 'pil' or 'base64', 'data': processed_image_data, 'media_type': str (if base64)} Returns None if preprocessing fails critically.

process_all_pdfs(root_directory, dpi=600, return_images=False)

Recursively search for all PDF files in the given directory and its subdirectories, and convert each one to high-quality images.

Parameters:

Name	Type	Description	Default
root_directory	str	The root directory to search for PDF files	required
dpi	int	DPI for the output images (default: 600)	600
return_images	bool	If True, return a dictionary mapping PDF paths to lists of PIL Image objects instead of saving images to disk (default: False)	False

Returns:

Type	Description
Union[int, dict]	When return_images is False: int: The number of PDF files processed.
Union[int, dict]	When return_images is True: dict: Dictionary where keys are PDF paths and values are lists of image objects.

process_all_pdfs_with_llm_whisperer(root_directory, return_extracted_text=False)

Process all PDF files in a directory using the LLM Whisperer API.

Parameters:

Name	Type	Description	Default
root_directory	str	Root directory to search for PDF files.	required
return_extracted_text	bool	If True, return a dictionary mapping PDF paths to lists of extracted text for each page instead of writing to files (default: False)	False

Returns:

Type	Description
Union[int, dict]	When return_extracted_text is False: int: The number of PDF files processed.
Union[int, dict]	When return_extracted_text is True: dict: Dictionary where keys are PDF paths and values are lists of dictionaries with extracted text for each page.

process_pdf_with_llm_whisperer(pdf_path, output_dir=None, return_extracted_text=False)

Process a PDF file using the LLM Whisperer API and save the raw extracted text for each page to .txt files or return it as a structured list.

Parameters:

Name	Type	Description	Default
pdf_path	str	Path to the PDF file.	required
output_dir	str	Directory where the raw extracted text .txt files should be saved. Required when return_extracted_text is False.	None
return_extracted_text	bool	If True, return a list of dictionaries containing the extracted text for each page instead of writing to files (default: False)	False

Returns:

Type	Description
Optional[list]	When return_extracted_text is False: None.
Optional[list]	When return_extracted_text is True: list: A list of dictionaries with keys 'page' and 'Extracted Text' for each page in the PDF.

Raises:

Type	Description
ValueError	If return_extracted_text is False and output_dir is not provided

Multimedia Utilities

doc_processing_toolkit.multimedia

Native multimedia capabilities: text-to-speech, speech-to-text, and image generation/editing. OpenAI + Google only.

Design: every capability keeps the MVP path mandatory — model + input — with first-class optional parameters for the high-value knobs, each validated against the registry's capability metadata (descriptions.<model>), plus a provider_options passthrough dict for anything provider-specific that is not normalized. Voice names are opaque provider strings (the registry lists them; there is no cross-provider enum).

Cost reporting uses the registry's multimedia pricing schemas: audio_output (TTS: text tokens in, audio tokens out), per_minute (STT: measured audio duration), and multimodal_output (images).

Realtime voice (Phase C) lives in :mod:doc_processing_toolkit.realtime.

See Also

• Multimedia concepts

MultimediaValidationError

Bases: ValueError

Raised when a requested option is unsupported by the target model.

The message names the provider limitation (unsupported combos fail at validation instead of at the API).

__getattr__(name)

Resolve the image-session names from :mod:doc_processing_toolkit.image_session.

Parameters:

Name	Type	Description	Default
name	str	The attribute being looked up on this module.	required

Returns:

Type	Description
Any	The requested object.

Raises:

Type	Description
AttributeError	For any name this module does not define.

asynthesize_speech_stream(model, text, voice=None, output_format=None, speed=None, instructions=None, multi_speaker_config=None, provider_options=None, logger=None) async

Stream speech audio as it is generated, from async code.

The async form of :func:synthesize_speech_stream: the same arguments, validation, request and frames, on the toolkit's async clients.

Parameters:

Name	Type	Description	Default
model	str	A registered TTS model whose row carries supports_streaming: true.	required
text	str	The text to speak.	required
voice	Optional[Union[str, Dict[str, str]]]	Voice name or custom-voice object {"id": ...}.	None
output_format	Optional[str]	OpenAI: any listed format. Google: omit it or "pcm".	None
speed	Optional[float]	OpenAI playback speed (0.25-4.0).	None
instructions	Optional[str]	Delivery or style instructions.	None
multi_speaker_config	Optional[List[Dict[str, Any]]]	Google only: {"speaker", "voice"} entries.	None
provider_options	Optional[Dict[str, Any]]	Passthrough request options.	None
logger	Optional[Logger]	Optional logger (module logger by default).	None

Yields:

Name	Type	Description
dict	AsyncIterator[Dict[str, Any]]	The frames of :func:synthesize_speech_stream: audio per
	AsyncIterator[Dict[str, Any]]	chunk, an optional error, then exactly one done.

Raises:

Type	Description
MultimediaValidationError	On the first iteration, as for :func:synthesize_speech_stream.

Example

from contextlib import aclosing

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 websocket.send_bytes(frame["data"])

Notes

Stopping early: wrap the generator in contextlib.aclosing (or call its aclose()) so the HTTP stream closes at once rather than when the event loop finalises the generator. Cancelling the consuming task closes it too. Costs and errors behave as in :func:synthesize_speech_stream.

atranscribe_audio_stream(model, audio, *, language=None, prompt=None, keywords=None, languages=None, duration_seconds=None, provider_options=None, logger=None) async

Stream the transcript of a completed recording, frame by frame.

The audio must already exist — this streams the model's output, not its input. For live microphone capture use the realtime surface (:func:doc_processing_toolkit.realtime.open_realtime_session).

Parameters:

Name	Type	Description	Default
model	str	A registered STT model whose row carries supports_streaming_transcription. whisper-1 does not: the API silently ignores stream on it, so it is rejected here.	required
audio	Any	Bytes, a file path, or {"data": bytes|base64, "filename": str}.	required
language	Optional[str]	Optional single ISO-639-1 hint.	None
prompt	Optional[str]	Optional context/style prompt.	None
keywords	Optional[List[str]]	Words or phrases guiding transcription. Gated on supports_keywords.	None
languages	Optional[List[str]]	Possible languages, ISO-639-1. Gated on supports_language_hints.	None
duration_seconds	Optional[float]	The clip's real length, used for the per-minute cost in the final frame exactly as in :func:transcribe_audio.	None
provider_options	Optional[Dict[str, Any]]	Passthrough kwargs. stream is set by this function and must not be supplied.	None
logger	Optional[Logger]	Optional logger.	None

Yields:

Name	Type	Description
dict	Any	{"type": "delta", "text": ...} for each incremental chunk,
	Any	{"type": "segment", "segment": {...}} for each finalized speaker
	Any	segment, an optional {"type": "error", ...} if the stream failed
	Any	part-way, and exactly one final {"type": "done", ...} frame carrying
	Any	the same keys :func:transcribe_audio returns.

Raises:

Type	Description
MultimediaValidationError	If the model is not a streaming-capable OpenAI STT row, an option is unsupported, the container is not accepted and cannot be converted, or provider_options carries stream.

Example

async for frame in atranscribe_audio_stream("gpt-transcribe", clip_bytes):
    if frame["type"] == "delta":
        print(frame["text"], end="")
    elif frame["type"] == "done":
        print(frame["cost"], frame["duration_source"])

Notes

The done frame's text is the provider's own final transcript when a transcript.text.done event arrived, and the concatenated deltas otherwise (with a warning). A stream that yields neither is reported as an empty_response error rather than as a successful, billed, empty transcription — the defect this function exists to close.

classify_media_error(exc)

Classify a provider SDK exception raised by a multimedia call.

OpenAI carries a string code/type on its API errors; google-genai carries an integer HTTP code. Both are read, with the HTTP status as the fallback and the exception class name as the last resort.

Parameters:

Name	Type	Description	Default
exc	BaseException	The exception raised by the provider SDK.	required

Returns:

Name	Type	Description
tuple	tuple	(error_code, retryable) — error_code is a stable
	tuple	lower-case string and retryable says whether re-sending the same
	tuple	request could succeed.

Example

try:
    result = generate_image("gpt-image-2", "a cat")
except Exception as exc:
    code, retryable = classify_media_error(exc)

describe_media(model, media, question, *, start_offset=None, end_offset=None, fps=None, media_resolution=None, instructions='', max_tokens=None, logger=None)

Ask a multimodal model a question about one media file (C8).

The tool-less-Agent pattern, packaged: the modality is checked against the registry before anything is uploaded, the Google-only video knobs are rejected on other providers instead of being dropped, and the result keeps the module's returns-a-dict contract — a provider failure comes back as error / error_code / retryable rather than an exception.

Parameters:

Name	Type	Description	Default
model	str	Any registered model whose input_types accept the media.	required
media	Any	Bytes, a file path, or {"data": bytes|base64, "filename": str}.	required
question	str	What to ask about the file.	required
start_offset	Optional[str]	Google video clipping start, e.g. "10s".	None
end_offset	Optional[str]	Google video clipping end, e.g. "20s".	None
fps	Optional[float]	Google video sampling rate in frames per second.	None
media_resolution	Optional[str]	Google MEDIA_RESOLUTION_* value, validated against the row's supported_media_resolution_values.	None
instructions	str	Optional system prompt.	''
max_tokens	Optional[int]	Optional output cap.	None
logger	Optional[Logger]	Optional logger.	None

Returns:

Name	Type	Description
dict	Dict[str, Any]	``{"text", "model", "input_tokens", "output_tokens", "cost",
	Dict[str, Any]	"cost_estimated", "duration_s", "media", "error", "error_code",
	Dict[str, Any]	"retryable"}.media` is the :func:inspect_media` record, and
	Dict[str, Any]	duration_s is the wall-clock time the call took.

Raises:

Type	Description
MultimediaValidationError	If the model does not accept this media kind, or a Google-only option was passed to another provider.

Example

from doc_processing_toolkit.multimedia import describe_media

result = describe_media(
    "gemini-3.5-flash-lite",
    "clip.mp4",
    "What happens in this clip?",
    start_offset="0s",
    end_offset="8s",
    fps=1.0,
)
print(result["text"], result["cost"])

edit_image(model, prompt, image, mask=None, size=None, provider_options=None, output_dir=None, logger=None, *, quality=None, n=1, aspect_ratio=None, background=None, output_format=None, output_compression=None, input_fidelity=None, timeout=None, thinking_level=None)

Edit an image with a text instruction.

OpenAI image models support pixel-mask editing; Google image models are instruction-driven only — passing a mask there raises a validation error naming the limitation.

Parameters:

Name	Type	Description	Default
model	str	A registered image model.	required
prompt	str	The edit instruction.	required
image	Any	Bytes, a file path, {"data": ..., "filename": ...}, or a list of any of those. OpenAI accepts up to 16 reference images on one edit; Google takes them as parts ahead of the prompt and documents up to 14 (up to 4 characters, up to 10 objects). On a Google row whose registry input_types list "video" or "document" a video clip or a PDF is accepted as a reference too, declared as what it is; every other row refuses it before any request. A scalar is still accepted unchanged.	required
mask	Any	Optional mask (OpenAI only): transparent regions are replaced. With several images the API applies it to the first.	None
size	Optional[str]	Output size, now honoured on both providers. Validated against the registry's presets, or against size_constraints where the row publishes a rule.	None
provider_options	Optional[Dict[str, Any]]	Passthrough kwargs for anything not named here. A key a named parameter already set is dropped with a warning.	None
output_dir	Optional[str]	When given, edited images are written there.	None
logger	Optional[Logger]	Optional logger.	None
quality	Optional[str]	Rendering quality, validated against supported_quality (OpenAI).	None
n	int	Number of images to return. Gemini returns one per call, so n > 1 is rejected there rather than silently ignored.	1
aspect_ratio	Optional[str]	"W:H" (Google only), validated against supported_aspect_ratios.	None
background	Optional[str]	"transparent", "opaque" or "auto" (OpenAI); "transparent" requires the row's supports_transparent_background and a png/webp output_format.	None
output_format	Optional[str]	"png", "jpeg" or "webp" (OpenAI).	None
output_compression	Optional[int]	0-100 for webp/jpeg output (OpenAI).	None
input_fidelity	Optional[str]	"low" or "high" — how hard the model works to preserve the input's features, especially faces (OpenAI images.edit, on rows whose supports_input_fidelity is true).	None
timeout	Optional[float]	Per-request timeout in seconds.	None
thinking_level	Optional[str]	How much the model thinks before editing (Google only): one of the row's supported_thinking_levels. Thinking tokens are billed at the row's text rate and reported in output_tokens.	None

Returns:

Name	Type	Description
dict	Dict[str, Any]	same shape as :func:generate_image, including cost_breakdown, error_code and retryable.

Raises:

Type	Description
MultimediaValidationError	If the model is not an image model, a mask is passed to a model without supports_mask_editing, or an option is unsupported by the model or the target provider.

Notes

The upload filename for image and mask is derived from the bytes themselves (image.png, mask.jpg, ...) because OpenAI reads each uploaded part's content type off its name; a caller-supplied filename that already names an image type is kept as-is.

size used to be validated on every provider but forwarded on the OpenAI branch only — a silent quality regression on Google, where an accepted "2K" still produced a 1K image. It now goes through the same version-aware ImageConfig helper :func:generate_image uses.

generate_image(model, prompt, size=None, quality=None, n=1, provider_options=None, output_dir=None, logger=None, *, aspect_ratio=None, background=None, output_format=None, output_compression=None, timeout=None, thinking_level=None)

Generate images from a text prompt.

Parameters:

Name	Type	Description	Default
model	str	gpt-image-2 / gpt-image-2.5-flare / gpt-image-2.5-sunburst (OpenAI) or a registered Gemini image model (gemini-3.1-flash-image, gemini-3-pro-image, gemini-3.1-flash-lite-image).	required
prompt	str	The image description.	required
size	Optional[str]	Output size, validated against the registry. Provider-shaped: OpenAI takes pixel dimensions ("1024x1024"), Gemini a resolution class ("1K" / "2K" / "4K", and gemini-3.1-flash-lite-image accepts "1K" only). A row that publishes size_constraints also accepts any other "WIDTHxHEIGHT" satisfying the rule — :func:image_size_for solves for one.	None
quality	Optional[str]	Validated against supported_quality (OpenAI).	None
n	int	Number of images. Gemini returns one per call, so n > 1 is rejected there rather than silently ignored.	1
provider_options	Optional[Dict[str, Any]]	Passthrough kwargs for anything not named here. A key a named parameter already set is dropped with a warning — see Notes.	None
output_dir	Optional[str]	When given, images are also written there, under the extension matching output_format.	None
logger	Optional[Logger]	Optional logger.	None
aspect_ratio	Optional[str]	"W:H", validated against supported_aspect_ratios (Google only — the OpenAI Images API sizes by pixels, so use size=image_size_for(model, aspect_ratio=...) there).	None
background	Optional[str]	"transparent", "opaque" or "auto" (OpenAI). "transparent" is allowed only on a row whose supports_transparent_background is true, and requires a png/webp output_format.	None
output_format	Optional[str]	"png", "jpeg" or "webp" (OpenAI), validated against supported_output_formats.	None
output_compression	Optional[int]	0-100 compression level for webp/jpeg output (OpenAI).	None
thinking_level	Optional[str]	How much the model thinks before drawing (Google only): one of the row's supported_thinking_levels, e.g. "minimal", "medium" or "high". Thinking tokens are billed at the row's text rate and reported in output_tokens; rows that think by default do so without this parameter.	None
timeout	Optional[float]	Per-request timeout in seconds. Forwarded to the OpenAI SDK as timeout, and to Google as http_options.timeout (milliseconds).	None

Returns:

Name	Type	Description
dict	Dict[str, Any]	{"images": [bytes, ...], "paths", "model", "input_tokens", "output_tokens", "cost", "cost_estimated", "cost_breakdown", "error", "error_code", "retryable"}.
	Dict[str, Any]	cost_breakdown carries the named cost lines a client-facing report
	Dict[str, Any]	needs — cost_input_text_usd, cost_input_image_usd,
	Dict[str, Any]	cost_text_usd, cost_image_usd, cost_cache_discount_usd —
	Dict[str, Any]	and sums to cost. error_code is a stable string
	Dict[str, Any]	("moderation_blocked", "empty_response", "max_output_tokens")
	Dict[str, Any]	and retryable says whether re-sending the identical request could
succeed	Dict[str, Any]	a moderation block never can.

Raises:

Type	Description
MultimediaValidationError	If the model is not an image model, or an option is unsupported by it or by the target provider. Nothing is dropped silently.

Notes

Named parameters beat provider_options. Both are merged into one request, and on a key both set, the registry-validated named value wins and the passthrough value is dropped with a warning. Leave the named argument unset to send a passthrough value instead. On Google, size and aspect_ratio are carried by an ImageConfig built from only the fields the installed google-genai declares.

Example

from doc_processing_toolkit.multimedia import generate_image, image_size_for

result = generate_image(
    "gpt-image-2.5-flare",
    "a flat-design coffee cup",
    size=image_size_for("gpt-image-2.5-flare", aspect_ratio="3:4", target="1K"),
    quality="low",
    background="transparent",
    output_format="png",
)
print(result["cost"], result["cost_breakdown"])

google_tts_raw_pcm_enabled()

Whether output_format="pcm" returns raw samples on a Google TTS row.

Controlled by DPT_GOOGLE_TTS_RAW_PCM (default on). off restores the behaviour before 0.5.1, when a Google reply was always returned as a WAV container whatever output_format asked for. Any other value logs one WARNING and stays on.

Returns:

Type	Description
bool	True unless the variable is off.

Example

import os

os.environ["DPT_GOOGLE_TTS_RAW_PCM"] = "off"
google_tts_raw_pcm_enabled()  # False: "pcm" returns a WAV again

image_size_for(model, width=None, height=None, *, aspect_ratio=None, target=None, logger=None)

Solve for a size value the model will accept.

A pure function: it reads the registry and returns a string, making no network call and mutating nothing. On a row that publishes size_constraints it searches the legal grid — edges on the published multiple, aspect ratio inside the published range, edge and pixel counts inside the published bounds — and returns the nearest valid "WIDTHxHEIGHT", preferring (in order) the closest aspect ratio, then the closest pixel count, then the smaller image. On a row that publishes only supported_sizes it returns the closest listed preset, so callers can use one helper for every model.

A result above the row's experimental_above_pixels threshold is returned with a warning, not an error — the provider accepts it.

Parameters:

Name	Type	Description	Default
model	str	A registered image model.	required
width	Optional[int]	Desired width in pixels. Supply with height.	None
height	Optional[int]	Desired height in pixels. Supply with width.	None
aspect_ratio	Optional[str]	"W:H" (for example "3:4"). Overrides the ratio implied by width/height, and may be used on its own.	None
target	Optional[str]	One of "1K", "2K", "4K" — a total-pixel budget that overrides the one implied by width/height. Defaults to "2K" when only aspect_ratio is given.	None
logger	Optional[Logger]	Optional logger for the experimental-resolution warning.	None

Returns:

Name	Type	Description
str	str	A value ready to pass as size to :func:generate_image or
	str	func:edit_image.

Raises:

Type	Description
MultimediaValidationError	If model is not an image model, the inputs are incomplete or non-positive, the requested aspect ratio is outside the model's published range, or no size on the grid satisfies every rule. The message names the violated rule.

Example

from doc_processing_toolkit.multimedia import image_size_for

# A 3:4 page at ~2K, snapped onto the multiple-of-16 grid.
image_size_for("gpt-image-2", aspect_ratio="3:4", target="2K")
# '1664x2224'

# A Gemini row publishes classes, not pixels.
image_size_for("gemini-3.1-flash-image", 2048, 2048)
# '2K'

inspect_media(media, *, include_models=False)

Describe a media part locally — no network call, no provider client (C8).

Answers the questions a caller needs before paying for an upload: what kind of part this is, its MIME type and size, its dimensions or duration where those are cheaply knowable from the header, and which registry rows accept that kind at all.

Parameters:

Name	Type	Description	Default
media	Any	Bytes, a file path, or {"data": bytes|base64, "filename": str}.	required
include_models	bool	When True, also list every registry model whose input_types accept this kind. Off by default because the sweep is only interesting when a caller is choosing a model.	False

Returns:

Name	Type	Description
dict	Dict[str, Any]	``{"kind", "mime_type", "size_bytes", "filename", "width",
	Dict[str, Any]	"height", "duration_s", "accepted_by"}.width/height`` are
	Dict[str, Any]	None for anything but a recognised image header, duration_s is
	Dict[str, Any]	None for anything but a RIFF/WAVE payload, and accepted_by is
	Dict[str, Any]	None unless include_models was set.

Raises:

Type	Description
MultimediaValidationError	If media is none of the accepted shapes.

Example

from doc_processing_toolkit.multimedia import inspect_media

info = inspect_media("scan.png", include_models=True)
print(info["kind"], info["mime_type"], info["width"], info["height"])

openai_tts_sse_usage_enabled()

Whether OpenAI TTS is requested as an SSE stream so the reply carries usage.

Controlled by DPT_OPENAI_TTS_SSE_USAGE (default on). It applies to rows with supports_sse_usage: true. off restores the request before 0.5.1 (raw audio, no stream_format) and its estimated cost. Any other value logs one WARNING and stays on.

Returns:

Type	Description
bool	True unless the variable is off.

Example

import os

os.environ["DPT_OPENAI_TTS_SSE_USAGE"] = "off"
openai_tts_sse_usage_enabled()  # False: raw audio, estimated cost

synthesize_speech(model, text, voice=None, output_format=None, speed=None, instructions=None, multi_speaker_config=None, provider_options=None, output_path=None, logger=None)

Generate speech audio from text (Phase A TTS).

Parameters:

Name	Type	Description	Default
model	str	A registered TTS model (gpt-4o-mini-tts, gemini-3.8-flash-tts, gemini-3.8-flash-lite-tts, gemini-3.1-flash-tts-preview, gemini-2.5-flash-preview-tts).	required
text	str	The text to speak. For multi-speaker input, write the dialogue as Name: line turns with names matching multi_speaker_config. The Gemini 3.8 rows read the text as a verbatim transcript, so anything in it is spoken; their inline tags (<laugh>, <sigh>, <short pause>) still work.	required
voice	Optional[Union[str, Dict[str, str]]]	Voice name, validated against the registry's supported_voices. Defaults to the row's first listed voice. A custom-voice object {"id": "voice_..."} is accepted on OpenAI and on Google rows with supports_custom_voices (designed, replicated or Extended Voice Library ids): it bypasses the list and only its shape is checked.	None
output_format	Optional[str]	Audio format, validated against supported_output_formats. OpenAI: mp3/opus/aac/flac/wav/pcm. Google: "wav" (the default) returns a WAV container; "pcm" returns the raw 16-bit samples with mime_type audio/l16;rate=24000 (DPT_GOOGLE_TTS_RAW_PCM=off returns the WAV container instead, as before 0.5.1).	None
speed	Optional[float]	Playback speed multiplier (OpenAI only; validated via the registry's supports_speed and the documented 0.25-4.0 range).	None
instructions	Optional[str]	Delivery or style instructions, refused on a row with supports_instructions: false. OpenAI sends them as instructions. On a Google row with supports_speech_metadata (the Gemini 3.8 rows) they become speech_metadata.style on each text part and are never spoken; on the preview rows they are prepended to the text as a natural-language style prompt.	None
multi_speaker_config	Optional[List[Dict[str, Any]]]	Google only: a list of {"speaker": name, "voice": voice_name_or_custom_voice} entries (validated via supports_multi_speaker). On the Gemini 3.8 rows the transcript is split into one text part per turn with speech_metadata.speaker (also when the speakers are configured only through provider_options["speech_config"]). A line opens a turn when it starts with a configured name, in any case, and a colon; a line without one continues the previous turn. Text before the first speaker line, or a line labelled with an unconfigured name, fails locally.	None
provider_options	Optional[Dict[str, Any]]	Passthrough kwargs merged into the provider request. On Google a speech_config here is deep-merged over the one the named parameters build (the passthrough wins leaf by leaf) instead of colliding with it. On OpenAI stream_format="audio" keeps the raw-audio request and its estimated cost for that call; any other stream_format is rejected, because an SSE event stream cannot be returned as audio bytes.	None
output_path	Optional[str]	When given, the audio is also written to this path.	None
logger	Optional[Logger]	Optional logger (module logger by default).	None

Returns:

Name	Type	Description
dict	Dict[str, Any]	{"audio_bytes", "mime_type", "model", "voice", "input_tokens", "output_audio_tokens", "cost", "cost_estimated", "output_path", "error", "error_code", "retryable"}.
	Dict[str, Any]	On an OpenAI row with supports_sse_usage the request asks for an
	Dict[str, Any]	SSE stream and reads the provider's own usage, so cost_estimated is
	Dict[str, Any]	False; DPT_OPENAI_TTS_SSE_USAGE=off restores the raw-audio
	Dict[str, Any]	request and the estimate. A stream whose audio arrived without a usage
	Dict[str, Any]	event keeps its audio and estimates the cost, with a WARNING. A stream
	Dict[str, Any]	with no usable audio (none at all, an undecodable chunk, an error
	Dict[str, Any]	event) is repeated once as raw audio, with a WARNING; both requests are
	Dict[str, Any]	billed and the result's tokens and cost are their sum.
	Dict[str, Any]	A blocked Google candidate returns audio_bytes == b"" with error
	Dict[str, Any]	naming the finish_reason/block_reason, error_code
	Dict[str, Any]	("moderation_blocked", "empty_response",
	Dict[str, Any]	"max_output_tokens") and retryable. cost is still the real
	Dict[str, Any]	cost of the request: the prompt tokens were billed even though no audio
	Dict[str, Any]	came back. Nothing is written to output_path in that case and it
	Dict[str, Any]	comes back None.

Raises:

Type	Description
MultimediaValidationError	An option the row does not support, or a malformed multi-speaker transcript on a Gemini 3.8 row.

Example

from doc_processing_toolkit.multimedia import synthesize_speech

result = synthesize_speech(
    "gemini-3.8-flash-lite-tts",
    "Welcome back. <laugh> It has been a while.",
    voice="Kore",
    instructions="Warm and unhurried.",
)
open("welcome.wav", "wb").write(result["audio_bytes"])

synthesize_speech_stream(model, text, voice=None, output_format=None, speed=None, instructions=None, multi_speaker_config=None, provider_options=None, logger=None)

Stream speech audio as it is generated, frame by frame.

Takes the arguments of :func:synthesize_speech except output_path and sends the same request, so the Gemini 3.8 style, dialogue and custom-voice handling all apply. Use it to start playback before the clip is finished; :func:asynthesize_speech_stream is the async form.

Parameters:

Name	Type	Description	Default
model	str	A registered TTS model whose row carries supports_streaming: true (gpt-4o-mini-tts, the Gemini 3.8 rows, gemini-3.1-flash-tts-preview).	required
text	str	The text to speak, as for :func:synthesize_speech.	required
voice	Optional[Union[str, Dict[str, str]]]	Voice name or custom-voice object {"id": ...}.	None
output_format	Optional[str]	OpenAI: any listed format; pcm (24 kHz 16-bit mono) or wav play most easily as they arrive. Google streams raw 16-bit PCM only: omit it or pass "pcm".	None
speed	Optional[float]	OpenAI playback speed (0.25-4.0).	None
instructions	Optional[str]	Delivery or style instructions.	None
multi_speaker_config	Optional[List[Dict[str, Any]]]	Google only: {"speaker", "voice"} entries.	None
provider_options	Optional[Dict[str, Any]]	Passthrough request options, as for :func:synthesize_speech.	None
logger	Optional[Logger]	Optional logger (module logger by default).	None

Yields:

Name	Type	Description
dict	Dict[str, Any]	{"type": "audio", "data": bytes, "mime_type", "index"} per
	Dict[str, Any]	chunk; an optional ``{"type": "error", "error", "error_code",
	Dict[str, Any]	"retryable"}`` when the stream fails part-way; then exactly one
	Dict[str, Any]	{"type": "done", ...} frame with the :func:synthesize_speech
	Dict[str, Any]	result keys except audio_bytes and output_path, plus
	Dict[str, Any]	total_bytes and chunks.

Raises:

Type	Description
MultimediaValidationError	On the first iteration, before any request: an option the row does not support, a row without supports_streaming, or a Google output_format other than "pcm".

Example

from doc_processing_toolkit.multimedia import synthesize_speech_stream

for frame in synthesize_speech_stream("gemini-3.8-flash-lite-tts", "Welcome back.", voice="Kore"):
    if frame["type"] == "audio":
        player.write(frame["data"])  # 24 kHz 16-bit mono PCM
    elif frame["type"] == "done":
        print(frame["cost"], frame["total_bytes"])

Notes

The cost is in the done frame: both providers report usage only at the end (Google on the last chunk, OpenAI in the speech.audio.done event on a supports_sse_usage row). A caller that stops iterating early closes the HTTP stream and gets no done frame and no cost; one INFO line records the bytes received. A stream cannot be repeated, so a missing usage event is estimated (cost_estimated) and a bad chunk is reported as an error frame instead of the unary function's raw-audio retry. A request that fails before anything arrives (a 429, a refused connection) reports a cost of 0.0; one that drops part-way is billed from what was received and flagged cost_estimated. An argument the SDK rejects raises, as in :func:synthesize_speech, instead of becoming an error frame.

transcribe_audio(model, audio, language=None, prompt=None, diarization=False, timestamps=False, provider_options=None, logger=None, *, duration_seconds=None, custom_vocabulary=None, mode=None, languages=None, keywords=None, stream=False)

Transcribe a completed recording to text.

Three surfaces, chosen from the registry rather than from the model name:

• OpenAI per-minute STT (gpt-transcribe, gpt-4o-transcribe-diarize, whisper-1) — billed by measured duration.
• Gemini file transcription (gemini-3.5-transcribe, a row whose descriptions.modality is "stt") — generateContent with audio_transcription_config, billed by audio-input and text-output tokens.
• Gemini audio understanding — any audio-capable Gemini text model, asked in a prompt to transcribe. Token-billed.

Parameters:

Name	Type	Description	Default
model	str	A registered STT model or an audio-capable Gemini text model.	required
audio	Any	Bytes, a file path, or {"data": bytes|base64, "filename": str}.	required
language	Optional[str]	Optional single language hint (ISO-639-1 on OpenAI). Prefer languages where the model supports it.	None
prompt	Optional[str]	Optional context/style prompt. On Gemini audio understanding it replaces the default "transcribe this verbatim" instruction; it is not accepted by gemini-3.5-transcribe, which has no text input, nor by a row publishing supports_prompt: false (gpt-4o-transcribe-diarize).	None
diarization	bool	Speaker diarization — gated on supports_diarization.	False
timestamps	bool	Word timestamps — gated on supports_word_timestamps.	False
provider_options	Optional[Dict[str, Any]]	Passthrough kwargs merged into the provider request. stream is rejected here; see stream below. A row with default_chunking_strategy (the diarize row: "auto", which the API requires for clips over 30 s) sends that chunking_strategy unless this dict carries one.	None
logger	Optional[Logger]	Optional logger.	None
duration_seconds	Optional[float]	Keyword-only. The clip's real length in seconds, when the caller knows it. Must be a finite, non-negative number; 0 is accepted and treated as "unknown". It is used when the provider reports no duration of its own, and it takes precedence over both the WAV header and the byte heuristic — which assumes uncompressed 16 kHz/16-bit/mono PCM and under-counts a browser's WebM/Opus recording by up to ~10x. Supplying it makes a per-minute cost exact and lets the per-request duration limits be checked before the upload.	None
custom_vocabulary	Optional[List[str]]	Keyword-only. Up to 1,000 phrases biasing recognition toward domain terms, brand names and proper nouns. Gated on supports_custom_vocabulary; incompatible with diarization and timestamps, which the API rejects outright.	None
mode	Optional[str]	Keyword-only. "VERBATIM" (default — word-for-word, keeps fillers) or "SMART" (removes disfluencies, formats the result). "SMART" is incompatible with diarization and timestamps. Gated on supported_transcription_modes.	None
languages	Optional[List[str]]	Keyword-only. Language hints — BCP-47 on Gemini (["es-ES"]; an empty list means "auto-detect"), ISO-639-1 on OpenAI. Gated on supports_language_hints.	None
keywords	Optional[List[str]]	Keyword-only. OpenAI's keywords — words or phrases guiding transcription. Gated on supports_keywords; the Gemini equivalent is custom_vocabulary.	None
stream	bool	Keyword-only. Consume the provider's streamed transcript into this same result instead of waiting for one JSON body. Gated on supports_streaming_transcription. For the deltas themselves use :func:atranscribe_audio_stream.	False

Returns:

Name	Type	Description
dict	Dict[str, Any]	{"text", "segments", "words", "duration_minutes", "duration_source", "converted_from", "input_tokens", "output_tokens", "model", "cost", "cost_estimated", "error", "error_code", "retryable"}.
	Dict[str, Any]	duration_source is one of "provider", "caller",
	Dict[str, Any]	"wav_header", "byte_estimate" or "unavailable", in that
	Dict[str, Any]	order of precedence. Only "byte_estimate" sets cost_estimated
	Dict[str, Any]	on a duration-billed model; on a token-billed one cost_estimated
	Dict[str, Any]	means the token counts themselves were not the provider's.
	Dict[str, Any]	A per-minute model never reports a confident $0.00: an unmeasurable clip
	Dict[str, Any]	is billed at the 0.02-minute floor and flagged as an estimate.
	Dict[str, Any]	converted_from names the original container when the payload had to
	Dict[str, Any]	be transcoded to one the model accepts, else None.

Raises:

Type	Description
MultimediaValidationError	If duration_seconds is not a finite, non-negative number; if the model does not support a requested option or an incompatible combination was requested; if the audio container is not accepted and no converter is installed; or if provider_options carries stream.

Example

# A 60-second WebM/Opus recording from the browser.
result = transcribe_audio("gpt-transcribe", webm_bytes, duration_seconds=60)
result["duration_minutes"]  # 1.0
result["duration_source"]   # "caller"
result["cost_estimated"]    # False

# Speaker-attributed, word-timestamped Gemini transcription.
result = transcribe_audio(
    "gemini-3.5-transcribe", meeting_bytes, diarization=True, timestamps=True
)
result["segments"][0]["speaker"]  # 'spk_1'
result["words"][0]["start"]       # 0.1

Notes

The upload filename is derived from the audio container sniffed off the first bytes (audio.wav, audio.mp3, audio.ogg, ...) because OpenAI's transcription endpoints accept only known audio extensions and read the part's type off its name. A caller-supplied filename with an audio extension is kept as-is.

The OpenAI verbose_json response format is requested for every model the registry flags supports_word_timestamps (today whisper-1), whether or not timestamps was asked for: that format also carries the provider-measured clip duration. A plain response can carry one too — a usage of type duration, which is how gpt-transcribe reports its billed seconds (observed live) — and either is the most accurate source for duration_minutes and the per-minute cost. The gate is the registry flag, never a model name, so a new STT model needs only its registry row.

Recordings over a model's per-request limit are rejected with a named error. The toolkit never splits a recording and stitches the pieces back together: timestamps would need re-basing and speaker labels do not survive a split.

Image Sessions

doc_processing_toolkit.image_session

Multi-turn image generation and editing sessions.

ImageSession keeps a conversation with an image model across several turns and — crucially — survives the process that started it: :meth:ImageSession.to_dict returns a JSON-serialisable envelope that :meth:ImageSession.from_dict restores byte-for-byte, so a host can persist a session to object storage between turns.

Only OpenAI and Google generate images. Anthropic states that Claude "cannot generate, produce, edit, manipulate, or create images", and Fireworks documents no image-generation endpoint, so a model from either raises :class:~doc_processing_toolkit.multimedia.MultimediaValidationError naming the two providers that do.

Persistence modes:

• "replay" — the session holds the whole conversation itself and replays it on every turn. Portable, self-contained, exactly costed. This is what "auto" resolves to on both providers.
• "server" — OpenAI only, and an explicit opt-in: the Responses API holds the chain (previous_response_id) and a mainline response_model hosts the image_generation tool. Two models are billed per step, each from its own reported usage (the image model's arrives in response.tool_usage["image_gen"]; the cost is an estimate only when a response lacks it), and the chain expires after 30 days.

The Google replay rule is strict. Every model part that carries a thought_signature — the first part after the thoughts, and every inline_data part — must be sent back unchanged or the API answers 400. The session therefore keeps the provider's own types.Content objects and serialises them with model_dump(mode="json"); it never re-encodes, resizes or re-saves the bytes a signature is bound to.

See Also

• Multimedia concepts

ImageSession

A serialisable multi-turn image generation / editing session.

The session runs one step per call to :meth:start / :meth:step, records an :class:ImageTurn for each, and can be round-tripped through :meth:to_dict / :meth:from_dict between steps — including from another process, because the envelope carries the provider's replay state, not a server-side chain id.

Parameters:

Name	Type	Description	Default
model	str	The image model: gpt-image-2, gpt-image-2.5-flare, gpt-image-2.5-sunburst, gemini-3-pro-image, gemini-3.1-flash-image or gemini-3.1-flash-lite-image.	required
response_model	Optional[str]	OpenAI persistence="server" only — the mainline model that hosts the image_generation tool and is billed alongside the image model.	None
persistence	str	"auto" (the default, resolving to "replay" on both providers), "replay", or "server".	'auto'
aspect_ratio	Optional[str]	"W:H" (Google), validated against the registry.	None
size	Optional[str]	Output size, validated against the registry exactly as :func:~doc_processing_toolkit.multimedia.generate_image does.	None
quality	Optional[str]	Rendering quality (OpenAI).	None
output_format	Optional[str]	"png"/"jpeg"/"webp" (OpenAI).	None
output_compression	Optional[int]	0-100 for webp/jpeg (OpenAI).	None
background	Optional[str]	"transparent"/"opaque"/"auto" (OpenAI).	None
moderation	Optional[str]	"auto"/"low" (OpenAI).	None
input_fidelity	Optional[str]	"low"/"high" on edits (OpenAI).	None
partial_images	Optional[int]	Streaming partials (OpenAI server path). Each one costs an extra 100 image output tokens.	None
timeout	Optional[float]	Per-request timeout in seconds. Google receives it as http_options.timeout, not as a hand-rolled retry loop.	None
thinking_level	Optional[str]	Gemini thinking depth for every turn, a level from the row's supported_thinking_levels ("minimal", "medium" or "high" on gemini-nano-banana-2.1), sent as ThinkingConfig(thinking_level=...) and kept in the envelope's settings. Refused at construction on a row that declares no levels or on an OpenAI model. None leaves the row's default.	None
max_turns	Optional[int]	Refuse to grow past this many turns. None disables the cap. Defaults to :data:DEFAULT_MAX_TURNS.	DEFAULT_MAX_TURNS
max_history_bytes	Optional[int]	Refuse to grow past this much replay history. None disables the cap. Defaults to :data:DEFAULT_MAX_HISTORY_BYTES.	DEFAULT_MAX_HISTORY_BYTES
image_store	Any	Optional sink for image bytes so an envelope need not embed them: a callable (data, mime_type, sha256) -> reference or a directory path. Used by to_dict(externalize=True).	None
image_loader	Any	The inverse of image_store — reference -> bytes, or a truthy value meaning "the references are file paths".	None
provider_options	Optional[Dict[str, Any]]	Passthrough kwargs. A key a named parameter already set is dropped with a warning, exactly as on the one-shot image facades.	None
logger	Optional[Logger]	Optional logger.	None

Raises:

Type	Description
MultimediaValidationError	If model is not an OpenAI or Google image model, if persistence is unknown, if persistence="server" is asked for on Google, or if any option is unsupported by the model.

Example

from doc_processing_toolkit.multimedia import ImageSession

session = ImageSession("gemini-3.1-flash-lite-image", size="1K")
first = session.start("a red cube on a white background")

envelope = session.to_dict()           # JSON-serialisable
restored = ImageSession.from_dict(envelope)

second = restored.step("make the cube blue")
print(second.cost, restored.total_cost)

cost_breakdown property

Named USD lines summed across every request, matching :attr:total_cost.

cost_estimated property

True when any request's cost carried an estimated component.

cost_records property

Every per-request cost record the session produced, in order.

history_bytes property

Approximate size of the replay history the next request would send.

total_cost property

Session cost in USD — the sum of the per-request records.

Never computed by summing tokens across requests: tiered and per-image pricing make that wrong (costs are billed per request).

total_input_tokens property

Prompt tokens across every request the session made.

total_output_tokens property

Completion tokens across every request the session made.

turns property

Every recorded turn, failed ones included, in order.

astep(prompt, *, image=None) async

Async twin of :meth:step, for a server that must not block its loop.

Same arguments, same returned :class:ImageTurn, and the same session state afterwards — turns, cost records, replay history and the to_dict() envelope are exactly what :meth:step would leave. The provider SDK calls the session makes are synchronous, so the whole step runs on the event loop's default executor (the pool asyncio.to_thread uses, with the caller's context copied in); the loop stays free for the render's duration. On a session with no turns it is the async form of :meth:start.

One step at a time. A second astep (or step) on the same session while one is running is refused with :class:~doc_processing_toolkit.multimedia.MultimediaValidationError, not queued: each step edits the image the previous one produced, so two concurrent steps would race for the same history. Await one before starting the next, or use one session per conversation.

Cancellation does not stop a render already sent. Cancelling the awaiting task raises CancelledError in the caller, but the step in the worker thread runs to completion and records its turn (the provider bills it either way), and the session accepts the next step only after that.

Parameters:

Name	Type	Description	Default
prompt	str	The instruction for this step.	required
image	Any	Optional extra reference image(s) for this step — or, on a Google row that accepts them, a video clip or a PDF (see :meth:start).	None

Returns:

Name	Type	Description
ImageTurn	ImageTurn	The recorded turn, status "completed" or
	ImageTurn	"failed".

Raises:

Type	Description
MultimediaValidationError	If prompt is empty, a configured bound has been reached, or another step is already running on this session.

Example

session = ImageSession("gpt-image-2", quality="low")
first = await session.astep("a red cube on a white background")
second = await session.astep("make the cube blue")

from_dict(payload, *, logger=None, image_store=None, image_loader=None) classmethod

Restore a session from :meth:to_dict output.

Parameters:

Name	Type	Description	Default
payload	Dict[str, Any]	The envelope.	required
logger	Optional[Logger]	Optional logger for the restored session.	None
image_store	Any	Sink for a later to_dict(externalize=True).	None
image_loader	Any	reference -> bytes resolver, required when the envelope was written with externalize=True.	None

Returns:

Name	Type	Description
ImageSession	'ImageSession'	A session that continues exactly where the saved one
	'ImageSession'	stopped — on Google, with every thought signature intact. A session
	'ImageSession'	the toolkit downgraded to persistence='replay' mid-run comes
	'ImageSession'	back in that mode, without the response_model and
	'ImageSession'	partial_images that only persistence='server' accepts.

Raises:

Type	Description
MultimediaValidationError	If the envelope is not a mapping, was written by a newer schema version, or holds external references with no image_loader.

Example

restored = ImageSession.from_dict(json.loads(blob))
turn = restored.step("make the cube blue")

latest_image()

The most recent successfully produced image, or None.

start(prompt, *, image=None)

Run the session's first step.

Parameters:

Name	Type	Description	Default
prompt	str	What to generate (or, with image, how to edit it).	required
image	Any	Optional reference image(s) to start from — bytes, a path, a {"data", "filename"} dict, or a list of those. On a Google row whose registry input_types list "video" or "document", a video clip or a PDF is accepted too; it stays in the replayed history, so every later turn re-sends it and is billed for it again.	None

Returns:

Name	Type	Description
ImageTurn	ImageTurn	The recorded turn.

Raises:

Type	Description
MultimediaValidationError	If the session already has turns.

step(prompt, *, image=None)

Run one generation / edit step and record it.

The conversation advances only on success. A failed step still returns (and stores) an :class:ImageTurn carrying the observed usage, cost, error_code and retryable, so a caller's accounting never loses a request it was billed for.

Parameters:

Name	Type	Description	Default
prompt	str	The instruction for this step.	required
image	Any	Optional extra reference image(s) for this step — or, on a Google row that accepts them, a video clip or a PDF (see :meth:start).	None

Returns:

Name	Type	Description
ImageTurn	ImageTurn	The recorded turn, status "completed" or
	ImageTurn	"failed".

Raises:

Type

Description

MultimediaValidationError

If prompt is empty, a configured bound has been reached, or another step (sync or :meth:astep) is already running on this session. On a Google row, also when Google could not process a reference it was given (the upload ended FAILED or never went ACTIVE), or a Files API upload the history names is dead and its bytes were not kept. A transport failure on an upload or a refresh is a failed turn, not an exception, like one on the request.

to_dict(*, externalize=False)

Serialise the whole session to a JSON-serialisable envelope.

The envelope is the load-bearing feature of B7: it carries the provider's replay state, not a server-side chain id, so another process — or another host — can continue the conversation. On Google that state is the native Content list dumped with model_dump(mode="json"), which round-trips thought signatures and inline image bytes byte-for-byte. Hand-rolled base64 of that structure is not supported, because a signature is bound to the exact bytes.

Parameters:

Name	Type	Description	Default
externalize	bool	When True, every image payload (the delivered artifacts and the inline bytes inside the replay state) is handed to the session's image_store and replaced by the reference it returns, so the envelope stays small enough for a database column or a message body. Requires image_store.	False

Returns:

Name	Type	Description
dict	Dict[str, Any]	``{"schema_version", "provider", "model", "response_model",
	Dict[str, Any]	"persistence", "settings", "limits", "sdk", "externalized",
	Dict[str, Any]	"turns", "provider_state"}. On Googleprovider_state`` holds
	Dict[str, Any]	history and, since schema version 2, media: one entry per
	Dict[str, Any]	Files API upload the history names, with its bytes inline (base64
	Dict[str, Any]	under bytes) or, when externalised, the store reference under
	Dict[str, Any]	ref, so a restored session can re-upload it after the 48-hour
	Dict[str, Any]	expiry.

Raises:

Type	Description
MultimediaValidationError	If externalize=True without an image_store.

Example

envelope = session.to_dict(externalize=True)   # refs, not bytes
blob = json.dumps(envelope)

ImageTurn dataclass

One completed step of an :class:ImageSession.

A turn is returned by :meth:ImageSession.start and :meth:ImageSession.step and is also what to_dict()["turns"] stores. A failed turn is recorded too — its observed usage and cost are real — but it does not advance the conversation.

Attributes:

Name	Type	Description
index	int	Zero-based position in the session.
prompt	str	The instruction the caller gave for this step.
revised_prompt	Optional[str]	The prompt the provider says it actually used, when it reports one (OpenAI image_generation_call.revised_prompt).
image_bytes	Optional[bytes]	The delivered artifact — the provider's original bytes, never a re-encode. None on a failed turn.
mime_type	Optional[str]	MIME type of image_bytes.
text	Optional[str]	Any text the model returned alongside the image.
image_ref	Dict[str, Any]	{"kind", "value", "mime_type", "sha256", "size_bytes"} describing where the artifact lives in a serialised envelope. kind is "inline_b64", "external" or "absent".
provider_ref	Dict[str, Any]	Provider identifiers for the turn — response_id and image_generation_call_id on OpenAI.
input_tokens	int	Prompt tokens the provider reported.
output_tokens	int	Completion tokens the provider reported.
cost	float	Total USD for the step, the sum of its per-request records.
cost_estimated	bool	True when any component was derived rather than reported.
cost_breakdown	Dict[str, float]	Named USD lines that sum to cost.
cost_records	List[Dict[str, Any]]	One entry per API request the step made. OpenAI persistence="server" emits two — the mainline model and the image model — because costs are billed per request, never by summing tokens across requests.
status	str	"completed" or "failed".
error	Optional[str]	Human-readable failure description, or None.
error_code	Optional[str]	Stable machine-readable code from :func:~doc_processing_toolkit.multimedia.classify_media_error.
retryable	bool	Whether re-sending the identical request could succeed.
finish_reason	Optional[str]	The provider's finish reason, when it gives one.
recovered_from	Optional[str]	Which recovery rung produced this turn, or None.
signatures_lost	bool	True only when this turn's rebuild replaced the session's signed Google history — the rebuild produced an image, so the thought signatures from earlier turns are gone for good. A rebuild that raised or came back blocked leaves the signed history (and this flag) untouched; recovered_from still names the rung that ran.
size_downgraded	bool	True when the step was retried at a smaller resolution than the one requested.
media_refreshed	int	How many expired Files API uploads in the replayed history were re-uploaded from their kept bytes before (or during) this step. 0 on every step that needed none.
model	str	The image model.
response_model	Optional[str]	The mainline model on the OpenAI server path.
persistence	str	The mode the step actually ran in.

from_dict(payload, image_bytes=None) classmethod

Rebuild a turn from :meth:to_dict output.

Parameters:

Name	Type	Description	Default
payload	Dict[str, Any]	A mapping produced by :meth:to_dict.	required
image_bytes	Optional[bytes]	The artifact, already resolved by the session (from the inline payload or through an image_loader).	None

Returns:

Name	Type	Description
ImageTurn	'ImageTurn'	The restored turn.

to_dict()

Serialise the turn, with image_bytes carried by image_ref.

Returns:

Name	Type	Description
dict	Dict[str, Any]	A JSON-serialisable mapping. The raw image_bytes field is
	Dict[str, Any]	not included — image_ref holds either the base64 payload or an
	Dict[str, Any]	external reference, so an envelope can be stored without embedding
	Dict[str, Any]	megabytes.

Audio Conversion

doc_processing_toolkit.audio_convert

Optional audio container conversion for speech-to-text.

Speech-to-text providers accept a fixed list of audio containers, and a browser recording is usually WebM/Opus or MP4 — neither of which every registered STT model takes. :mod:doc_processing_toolkit.multimedia compares the container it sniffs off the payload against the accepted list in the registry's audio_requirements section and calls :func:convert_audio when they disagree.

The converter is an optional dependency, deliberately. A bundled ffmpeg is a ~25 MB wheel and PyAV is ~36 MB; forcing either on every installing application to serve an STT path most of them never call is exactly the kind of constraint the dependency rule in the repository guide rejects. So this module discovers a converter instead of requiring one, and when there is none the caller gets a named error telling it which extra to install rather than a provider 400.

Discovery order, decided once per process (:func:converter_available):

1. DPT_FFMPEG_PATH — an explicit ffmpeg executable, which always wins.
2. ffmpeg on PATH.
3. The ffmpeg binary bundled with imageio-ffmpeg, if that package is installed.
4. av (PyAV), which decodes in-process with no subprocess.
5. Nothing — :func:converter_available returns None.

Environment

DPT_FFMPEG_PATH: Absolute path to an ffmpeg executable. Checked first. DPT_AUDIO_CONVERT_TIMEOUT: Seconds to allow one ffmpeg subprocess (default 60). A transcode that outruns it is killed and raises. DPT_AUDIO_CONVERT_MAX_MB: Largest input accepted for conversion (default 200). Guards against handing a multi-gigabyte recording to a subprocess and buffering the result in memory.

Example

from doc_processing_toolkit.audio_convert import convert_audio, converter_available

if converter_available():
    wav_bytes = convert_audio(webm_bytes, source_ext="webm")

See Also

• Multimedia concepts

AudioConversionError

Bases: RuntimeError

Raised when audio conversion is needed but cannot be performed.

Carries the reason in its message: either no converter is installed (in which case the message names the extra) or the converter itself failed.

convert_audio(data, *, source_ext, container='wav', sample_rate=16000, channels=1, sample_width=2, logger=None)

Transcode an audio payload into a container the target model accepts.

Parameters:

Name	Type	Description	Default
data	bytes	The raw audio bytes.	required
source_ext	str	The sniffed container extension of data ("webm", "m4a", ...). It is a decoder hint, not a trusted label.	required
container	str	Target container. "wav" on every path; other values need an ffmpeg backend.	'wav'
sample_rate	int	Target sample rate in Hz.	16000
channels	int	Target channel count (1 is mono).	1
sample_width	int	Target sample width in bytes (2 is 16-bit PCM).	2
logger	Optional[Logger]	Optional logger.	None

Returns:

Name	Type	Description
bytes	bytes	The converted payload.

Raises:

Type	Description
AudioConversionError	If no converter is installed (the message names the extra to install), the payload is larger than DPT_AUDIO_CONVERT_MAX_MB, the conversion times out, or the converter rejects the input.

Example

wav_bytes = convert_audio(webm_bytes, source_ext="webm")

converter_available(logger=None)

Name the audio converter this process can use, if any.

The verdict is computed once and cached: discovery touches the filesystem and imports an optional package, and the answer cannot change inside a running process except through :func:reset_converter_cache.

Parameters:

Name	Type	Description	Default
logger	Optional[Logger]	Optional logger for the one-time discovery line.	None

Returns:

Type	Description
Optional[str]	"ffmpeg", "pyav", or None when no converter is installed.

Example

converter_available()  # 'ffmpeg'

pcm_to_wav(pcm, *, sample_rate=16000, channels=1, sample_width=2)

Wrap raw PCM frames in a RIFF/WAVE header.

Available without any converter, because the standard library's :mod:wave module writes the header. Useful for callers holding raw PCM (a realtime capture buffer) that need a container an STT model accepts.

Parameters:

Name	Type	Description	Default
pcm	bytes	Raw little-endian PCM frames.	required
sample_rate	int	Frame rate in Hz.	16000
channels	int	Channel count.	1
sample_width	int	Bytes per sample.	2

Returns:

Name	Type	Description
bytes	bytes	A complete WAV payload.

Example

wav_bytes = pcm_to_wav(pcm_frames, sample_rate=24000)

reset_converter_cache()

Forget the cached discovery verdict.

Only tests and a process that has just installed a converter need this; ordinary callers never do.

Realtime Voice Sessions

doc_processing_toolkit.realtime

Neutral realtime-voice sessions over OpenAI Realtime and Gemini Live.

Design: the public surface is provider-neutral, not a mirror of any one vendor's wire protocol. A session is connect -> configure -> send frames -> consume events -> close, and each adapter maps its provider's protocol down to that shape honestly:

• OpenAI (:class:OpenAIRealtimeAdapter) speaks the GA Realtime WebSocket protocol through the openai SDK's client.realtime.connect() surface: session.update / input_audio_buffer.append (base64 PCM16) / response.create out, response.output_audio.delta / response.output_audio_transcript.delta / response.done back.
• Google (:class:GeminiLiveAdapter) speaks Gemini Live through the google-genai SDK's client.aio.live.connect() surface: raw 16 kHz PCM in, 24 kHz PCM out, setup/serverContent message shapes, and a server-enforced session-duration cap. The cap is invisible to callers: the adapter enables session resumption, watches for the server's goAway warning, and reconnects transparently, emitting a single session_resumed event so an application can observe the seam if it wants to. A reconnect before the server has issued any resumption handle cannot resume: it emits session_restarted instead, because the new socket has no memory of the conversation.

Neither provider's event names, message shapes, nor reconnect semantics leak into :class:RealtimeEvent.

Tools are dispatched by the host, bounded and hook-controlled. Neither realtime protocol answers a tool call for you, so :class:RealtimeToolDispatcher loads the same tool formats :class:~doc_processing_toolkit.agent.core.Agent accepts, executes each call through the shared MCP path, and emits the text loops' hook events (tool_call_start, the before_tool_call veto, and tool_call_result). Google's Live API additionally offers asynchronous function calling (behavior=NON_BLOCKING), which the adapter uses when the registry row says the model supports it: calls are dispatched without awaiting, tracked by id, answered with a scheduling hint as they finish, and cancelled on close or reconnect.

A session has a kind, not just a provider. The adapter is chosen from (provider, session_kind): realtime is the voice conversation above, transcription is a dedicated speech-to-text socket with no model turn at all — OpenAI's session.type='transcription' and Gemini Live captions — and translation is a dedicated interpreter socket, whose adapters live in :mod:doc_processing_toolkit.realtime_translation. live is OpenAI's GPT-Live family, a third protocol on /v1/live/* with its own event vocabulary and duration billing, which lives in :mod:doc_processing_toolkit.realtime_gpt_live and registers itself into the routing table at the bottom of this module. open_realtime_session(model, session_kind=...) routes between them from the registry and rejects a kind a model does not serve before the socket opens; open_transcription_session, open_translation_session and :func:~doc_processing_toolkit.realtime_gpt_live.open_live_session are the same call under their own names. Transcripts correlate by :attr:RealtimeEvent.item_id, because a transcription session completes items out of order — except on GPT-Live, which issues no item ids at all and correlates by role and timeline instead.

Visual input is capability-gated and rate-gated. send_image attaches a still to the conversation; send_video_frame offers a frame from a continuous capture and is deliberately lossy — frames arriving faster than max_video_fps are coalesced, so a 30 fps producer neither grows memory nor multiplies a bill in which every frame is billable image tokens. Audio is never gated and never dropped. Image tokens are tracked (:attr:RealtimeUsage.input_image_tokens) and priced at the row's input_image rate.

Every model-specific behaviour is registry-driven, never name-matched. The descriptions.<model> row decides whether a text turn goes out as send_realtime_input(text=...) or send_client_content, whether response_complete follows interaction_status == IDLE or turn_complete, which function-calling behaviours and response schedulings are legal, and whether proactive_audio / enable_affective_dialog / thinking_level may be sent at all. :func:realtime_capabilities reports the provider's advertised capabilities and the toolkit features actually implemented for a model, as two separate maps.

xAI escape hatch: xAI's voice API is deliberately OpenAI-Realtime compatible, so :class:OpenAIRealtimeAdapter accepts base_url (and websocket_base_url / api_key) overrides through provider_options::

session = await open_realtime_session(
    "gpt-realtime",
    provider_options={"base_url": "https://api.x.ai/v1", "api_key": XAI_KEY},
)

Usage is tracked in both directions (text and audio tokens, including the cached share where the provider reports it) and priced through :func:doc_processing_toolkit.core.calculate_cost under the registry's audio_realtime schema. session.usage and session.cost are live throughout the session and final after :meth:RealtimeSession.close.

Every adapter in this module goes through its vendor SDK's own realtime client, so no raw WebSocket code and no new dependency is involved: the websockets package both SDKs use underneath is already a hard requirement of google-genai, which the toolkit pins.

The one documented exception to that convention is OpenAI translation. gpt-realtime-translate is served from /v1/realtime/translations, a different endpoint with a disjoint session.* event vocabulary; the installed openai SDK (3.16.2) hardcodes /realtime in AsyncRealtime._prepare_url and declares no Literal["translation"] session type anywhere in openai.types.realtime, so that socket is unreachable through it. :class:~doc_processing_toolkit.realtime_translation.OpenAITranslationAdapter therefore speaks the protocol over a raw websockets client — the exception is confined to that one class, in its own module, and adds no dependency. Gemini live translation needs no exception: it is the same BidiGenerateContent socket and the same SDK client as :class:GeminiLiveAdapter, plus one translation_config field.

Speech synthesis, transcription, and image generation live in :mod:doc_processing_toolkit.multimedia.

See Also

• Multimedia concepts

GeminiLiveAdapter

Bases: RealtimeSession

Gemini Live adapter over the google-genai SDK's Live surface.

Uses client.aio.live.connect(model=..., config=...) rather than a hand-rolled WebSocket, so the SDK owns the BidiGenerateContent handshake, the setup/serverContent framing, and message parsing.

The session-duration cap is hidden from callers. Gemini Live sessions are capped server-side; the adapter enables session_resumption at connect, records every resumption handle the server issues, reacts to the server's goAway warning, and reconnects with the stored handle when the connection drops — replaying the same configuration. Callers see one uninterrupted :meth:events stream with a single session_resumed event marking the seam. max_reconnects in provider_options bounds the retries (default 10). When the connection drops before the server has issued any handle, the reconnect opens a fresh session with no memory of the conversation; the seam is then a session_restarted event plus a WARNING, never session_resumed.

provider_options also accepts any :class:google.genai.types.LiveConnectConfig field verbatim (temperature, context_window_compression, proactivity, tools, ...). A key the adapter builds itself (see :attr:passthrough_overrides) replaces the adapter's object and logs one WARNING naming the parameter it overrides.

billed_audio_seconds property

The audio duration a per_minute row is billed on.

Precedence, the same rule file transcription follows: the provider's own reported seconds when it reported any (a transcription session reports them per completed item), otherwise the session's connected wall clock. Provider first because a socket that was open for two minutes while thirty seconds of speech passed through it is billed for the speech, not the silence — and only the provider knows which.

Returns:

Name	Type	Description
float	float	Seconds.

capabilities property

Provider capabilities and toolkit features for this model.

See :func:realtime_capabilities — this is the same report, for the session's own model.

compression_policy property

What extends this session, stated without conflating two mechanisms.

resumption reconnects a dropped transport and replays the handle; it does not enlarge the context window. compression is the separate context-window policy. A provider's session-duration caps are reported alongside so a caller can see which limit it is up against.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"resumption", "compression", "limits"}.

connected property writable

Whether a transport is currently established.

Assigning to this attribute also drives :attr:session_duration_seconds: setting it True starts a billing segment and setting it False closes one. Adapters keep assigning it exactly as before.

emitted_audio_ms property

Milliseconds of assistant audio this session has emitted for the current item.

This is the only playback figure the toolkit can know: it counts the PCM the adapter handed to the caller, not what the caller's speaker actually played. A caller that buffers should pass its own figure to :meth:truncate_response.

manual_turn_control property

Whether automatic activity detection is off for this session.

True when turn_detection was {} or {"disabled": True} (the neutral spelling of manual turn control). The client then marks every user turn itself: :meth:send_audio opens it with activity_start and :meth:commit_audio closes it with activity_end.

max_video_fps property

The video-frame ceiling in force, in frames per second.

Comes from configure(max_video_fps=...), else the registry row's max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS. 0 means the gate is off.

session_duration_seconds property

Billable wall-clock seconds this session has been connected.

Measured from the transport, not from the audio frames: a realtime provider that bills per minute bills connection time, and a frame-derived figure would need a per-provider sample-rate assumption and would under-count the silence the provider still charges for. Gemini Live's reconnects (the hidden session-duration cap) close one segment and open another, so the gap between them is excluded rather than billed.

Returns:

Name	Type	Description
float	float	Seconds, accumulated across every connected segment and
	float	including the segment currently open.

Example

async with await open_realtime_session("gpt-realtime") as session:
    ...
print(session.session_duration_seconds, session.cost)

tool_call_behavior property

The function-calling behaviour in force: BLOCKING or NON_BLOCKING.

tool_dispatcher property

The session's tool dispatcher, or None when no tools were given.

__aenter__() async

Connect if not already connected.

__aexit__(exc_type, exc, tb) async

Always close the transport, even on an exception.

__init__(model, **kwargs)

See :class:RealtimeSession for the shared arguments.

cancel_response() async

Interrupt generation with an empty completed client turn.

Gemini Live has no cancel message; the documented control is that turn_complete=true unconditionally interrupts generation, which is only available on rows that accept send_client_content mid-session.

Raises:

Type	Description
RealtimeValidationError	The model has no explicit interrupt (send audio, or let voice-activity detection barge in, instead).

capability(name)

Read one registry capability, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:REALTIME_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The registry row's value, or the default for an absent key.

close() async

Close the Live session and finalize the session cost.

commit_audio(create_response=True) async

Close the current input-audio turn, by the route the session is in.

The name matches :meth:OpenAIRealtimeAdapter.commit_audio so a caller can drive either provider the same way; the wire message depends on the turn-detection setting:

• Manual turn control (turn_detection={"disabled": True} or {}): sends activity_end, closing the activity :meth:send_audio opened. The model answers the turn. The next :meth:send_audio opens a new activity. A commit with no open activity (nothing sent since the last commit) sends nothing.
• Automatic activity detection (the default): sends audio_stream_end, telling the server the microphone paused so it flushes any buffered audio instead of waiting for more.

Parameters:

Name	Type	Description	Default
create_response	bool	Accepted for parity with the OpenAI adapter. On Gemini Live the end of a turn is itself what triggers the response, so there is no separate request to make.	True

Raises:

Type	Description
RealtimeConnectionError	The session is not connected.

Example

session = await open_realtime_session(
    "gemini-3.1-flash-live-preview", turn_detection={"disabled": True}
)
await session.send_audio(pcm_16k)   # activity_start + audio
await session.commit_audio()        # activity_end -> the model answers

configure(*, voice=None, instructions=None, output_format=None, turn_detection=None, modalities=None, input_transcription=None, thinking_level=None, proactive_audio=None, enable_affective_dialog=None, context_compression=None, tool_call_behavior=None, tool_response_scheduling=None, max_video_fps=None, provider_options=None)

Validate and stage session options.

Every argument is optional and only the ones supplied are changed, so this can be called repeatedly to adjust one knob at a time. Options are checked against the registry's descriptions.<model> row, so an unsupported voice or format raises :class:RealtimeValidationError naming what the model does support rather than failing mid-session.

Before :meth:connect the staged options form the initial session payload. After it, OpenAI applies them to the live session at the next send or event poll; Gemini Live fixes its configuration at connect time, so a post-connect change there logs a warning and takes effect on the next reconnect.

Parameters:

Name	Type	Description	Default
voice	Optional[Union[str, Dict[str, str]]]	Voice name (validated against supported_voices), or on OpenAI a custom-voice object {"id": "voice_..."}, which bypasses the list and is shape-checked only.	None
instructions	Optional[str]	System instructions for the model.	None
output_format	Optional[str]	Audio output format (validated against supported_output_formats).	None
turn_detection	Optional[Dict[str, Any]]	Voice-activity/turn-detection configuration in the neutral vocabulary: None keeps the provider default; {} or {"disabled": True} requests manual turn control, which on Gemini switches :meth:GeminiLiveAdapter.send_audio / :meth:GeminiLiveAdapter.commit_audio to the activity_start / activity_end route. Translated per provider by :func:translate_turn_detection, never forwarded verbatim, and refused on a row publishing supports_turn_detection: false.	None
modalities	Optional[List[str]]	Output modalities, e.g. ["audio"] or ["text"].	None
input_transcription	Optional[Union[bool, Dict[str, Any]]]	Whether to ask the provider to transcribe the caller's own audio (surfaced as transcript_delta events with role="user"). OpenAI conversational sessions also take a dict of transcription guidance fields - see :meth:OpenAIRealtimeAdapter.configure; every other adapter rejects a dict.	None
thinking_level	Optional[str]	Background-reasoning level, validated against the row's supported_thinking_levels. A model whose row omits that key (gemini-3.8-live, where the vendor says to omit thinking_level entirely) drops the value with a WARNING. Gemini sends it as thinking_config.thinking_level; the OpenAI reasoning rows (gpt-realtime-2 family) send it as session.reasoning.effort (kill switch DPT_OPENAI_REALTIME_REASONING=off).	None
proactive_audio	Optional[bool]	Whether the model may speak unprompted. A row with proactive_audio_always_on rejects False locally, because the API returns an error for it.	None
enable_affective_dialog	Optional[bool]	Affective-dialogue toggle. Rejected locally on a row with supports_affective_dialog: false (the 3.8 Live models removed the feature from the API).	None
context_compression	Optional[Any]	Context-window compression policy — False to disable, True for the provider default, or {"trigger_tokens": int, "sliding_window_target_tokens": int}. This is not session resumption: resumption reconnects a dropped transport, compression extends how long the context lasts. See :attr:compression_policy.	None
tool_call_behavior	Optional[str]	"BLOCKING" or "NON_BLOCKING"; None takes the registry row's function_calling_behavior_default.	None
tool_response_scheduling	Optional[str]	"SILENT" / "WHEN_IDLE" / "INTERRUPT" for asynchronous function responses. Only legal where the row sets supports_function_response_scheduling.	None
max_video_fps	Optional[float]	Ceiling on video frames put on the wire, in frames per second (registry max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS). 0 disables the gate and sends every frame. Applies to :meth:send_video_frame only.	None
provider_options	Optional[Dict[str, Any]]	Extra provider-specific options merged into the existing passthrough dict. A key the adapter builds itself is replaced by the passthrough value with one WARNING per key per session (:attr:GeminiLiveAdapter.passthrough_overrides; the OpenAI tools key); the OpenAI audio object is merged one level deep instead.	None

Raises:

Type	Description
RealtimeValidationError	An option the registry row says the model cannot accept.

connect() async

Open the Live session with the staged configuration.

events() async

Yield neutral events, reconnecting transparently across the session cap.

Yields:

Name	Type	Description
RealtimeEvent	AsyncIterator[RealtimeEvent]	Translated Live server messages. A session_resumed event marks each transparent reconnect that sent a resumption handle, and session_restarted one that had no handle to send (the new session does not remember the conversation); the stream itself never breaks for either.

export()

This session's record as JSON-clean data (session-portability slice D).

Read-only and available at any time; final after :meth:close. The transcript is the text of every completed turn, in completion order, and is the shape provider_options={"initial_items": ...} (an OpenAI Realtime or Gemini Live session) and GPT-Live's provider_options={"initial_input": ...} take to seed the next session: [{"role": e["role"], "text": e["text"]} for e in exported["transcript"]] (content in place of text for GPT-Live). A transcription session exports its completed captions the same way. A translation session and a GPT-Live session retain no transcript by design (an interpreter's two transcripts are on source_transcript / translated_transcript and its events; GPT-Live's text is on its own event stream) and export usage and cost with an empty list. closed_at and close_reason are set by every family's :meth:close; on GPT-Live close_reason is the provider's session.closed reason when one arrived. TTS and STT are not sessions.

Returns:

Type	Description
Dict[str, Any]	``{"schema_version": 1, "provider", "model", "session_kind",
Dict[str, Any]	"started_at", "closed_at", "close_reason", "duration_seconds",
Dict[str, Any]	"usage", "cost", "cost_estimated", "cost_components",
Dict[str, Any]	"transcript"}``.

Example

record = session.export()
json.dumps(record)                       # always JSON-clean
seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
next_session = open_realtime_session(model, provider_options={"initial_items": seed})

flush_pending_frame() async

Send the held video frame if the rate gate now allows it.

A capture loop that stops pushing would otherwise leave its last frame held forever, so the adapters also call this from :meth:send_audio — a live session sends audio continuously, which makes it a free heartbeat.

Returns:

Name	Type	Description
bool	bool	Whether a frame was actually sent.

send_audio(pcm_bytes) async

Stream raw 16 kHz PCM16 audio to the Live session.

Also releases a video frame the rate gate is holding (see :meth:RealtimeSession.flush_pending_frame). Under manual turn control (:attr:manual_turn_control) the first chunk of a turn is preceded by activity_start, because with automatic activity detection disabled the server treats audio outside an activity as no speech at all.

Parameters:

Name	Type	Description	Default
pcm_bytes	bytes	Raw PCM16 mono audio at 16 kHz.	required

send_image(image, *, media_type=None, detail=None) async

Send one still image as user input.

A still is a deliberate, discrete act — a photographed page, a chart, a screenshot to ask about — so it is never rate-gated: it goes on the wire immediately, like :meth:send_text. Use :meth:send_video_frame for a continuous capture.

The model must declare image (or video) among its registry input_types; a row that declares neither raises before anything is sent.

Parameters:

Name	Type	Description	Default
image	Any	Encoded image bytes (PNG/JPEG/GIF/WebP/BMP) or a base64 string of them. Raw pixel buffers are not accepted — encode the frame first.	required
media_type	Optional[str]	MIME type. Sniffed from the payload when omitted; a payload the toolkit cannot identify is rejected rather than guessed at.	None
detail	Optional[str]	OpenAI's input_image detail hint ("auto", "low", "high"). Ignored on Gemini Live, which has no equivalent.	None

Raises:

Type	Description
RealtimeValidationError	The model takes no visual input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

await session.send_image(png_bytes, detail="low")
await session.send_text("What does this chart show?")

send_text(text) async

Send a text turn, by the route this model actually supports.

send_client_content is documented as seeding initial context only on gemini-3.1-flash-live-preview — after the first model turn it is silently ineffective there, which made a second text turn a no-op. The 3.8 Live rows re-introduced it for the whole session lifecycle. The route is chosen from the registry key supports_client_content_mid_session, never from the model name.

Parameters:

Name	Type	Description	Default
text	str	The user turn.	required

send_video_frame(frame, *, media_type=None) async

Offer one video frame to the model, subject to the rate gate.

Unlike :meth:send_image this is lossy on purpose. A capture loop pushing 30 fps into a model that re-reads every frame as billable image tokens would multiply the bill roughly thirtyfold for a picture that barely changed, so frames arriving faster than :attr:max_video_fps are coalesced: the newest frame is held and the one it superseded is counted in :attr:frames_dropped. Nothing is queued, so memory does not grow with the frame rate, and no frame ever delays audio.

Parameters:

Name	Type	Description	Default
frame	Any	Encoded image bytes, or a base64 string of them.	required
media_type	Optional[str]	MIME type; sniffed when omitted.	None

Returns:

Name	Type	Description
bool	bool	Whether this frame went on the wire now. False means it is
	bool	held and will be sent by the next :meth:send_audio,
	bool	meth:send_video_frame or :meth:flush_pending_frame once the
	bool	gate opens — or superseded by a newer frame first.

Raises:

Type	Description
RealtimeValidationError	The model takes no video input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

async for jpeg in camera:                  # 30 fps
    await session.send_video_frame(jpeg)   # ~1 fps reaches the model
print(session.frames_sent, session.frames_dropped)

truncate_response(*, item_id=None, played_ms=None) async

Tell the provider how much of its last answer the caller actually heard.

Barge-in leaves the provider believing it spoke its whole answer while the caller stopped the playback early. Truncation rewrites the stored turn to the audible prefix so the next turn is grounded in what was really heard.

Parameters:

Name	Type	Description	Default
item_id	Optional[str]	The item to truncate; the adapter's last assistant item by default.	None
played_ms	Optional[int]	Milliseconds actually played; :attr:emitted_audio_ms by default.	None

Raises:

Type	Description
RealtimeValidationError	The provider has no truncation control.

GeminiTranscriptionAdapter

Bases: _TranscriptionSessionMixin, GeminiLiveAdapter

Gemini Live captions (gemini-3.5-transcribe-live).

The same BidiGenerateContent socket as the conversational adapter, configured for captions rather than conversation: response_modalities is TEXT, input_audio_transcription is on and output transcription is off, and no voice, tool, thinking or proactivity setting is sent.

The protocol as observed live (2026-09-21), which is not what the type hints suggest. While audio streams, interim_input_transcription carries a cumulative snapshot of the utterance so far ("This", "This is a", "This is a test"), not a fragment, so each neutral transcript_delta is the text that snapshot ADDED. The settled caption arrives once, on input_transcription, after the client marks the end of the audio stretch with audio_stream_end (:meth:commit_audio, and :meth:close sends it for you) - and Transcription.finished was None on it, so the settled channel itself is the completion signal; finished=False is still honoured as "more to come". generation_complete closes whatever is left. There are no item ids on this protocol — Gemini Live does not label conversation items — so completed captions are keyed by their arrival order instead, and :attr:RealtimeEvent.item_id stays None on a delta.

The provider reports no usage for this model - no usage_metadata message arrived at any point of a live session. A token-priced session with no token report would close at a confident $0.00, so the adapter estimates instead: input audio tokens from the seconds of audio it sent at the row's input_audio_tokens_per_second, output text tokens from the captions themselves, :attr:usage_estimated and cost_estimated both True, one WARNING. A real usage report, should one ever arrive, replaces the estimate.

The ten-minute cap is real and is not hidden. This row publishes no session resumption, so a reconnect would be an unrelated session with no memory of the audio already transcribed. The adapter therefore ends the stream with an error event naming the cap instead of silently restarting; session.compression_policy["limits"]["max_session_minutes"] reports the figure, and a caller transcribing a long meeting should open a fresh session before it and join the transcripts itself.

billed_audio_seconds property

The audio duration a per_minute row is billed on.

Precedence, the same rule file transcription follows: the provider's own reported seconds when it reported any (a transcription session reports them per completed item), otherwise the session's connected wall clock. Provider first because a socket that was open for two minutes while thirty seconds of speech passed through it is billed for the speech, not the silence — and only the provider knows which.

Returns:

Name	Type	Description
float	float	Seconds.

capabilities property

Provider capabilities and toolkit features for this model.

See :func:realtime_capabilities — this is the same report, for the session's own model.

compression_policy property

What extends this session, stated without conflating two mechanisms.

resumption reconnects a dropped transport and replays the handle; it does not enlarge the context window. compression is the separate context-window policy. A provider's session-duration caps are reported alongside so a caller can see which limit it is up against.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"resumption", "compression", "limits"}.

connected property writable

Whether a transport is currently established.

Assigning to this attribute also drives :attr:session_duration_seconds: setting it True starts a billing segment and setting it False closes one. Adapters keep assigning it exactly as before.

emitted_audio_ms property

Milliseconds of assistant audio this session has emitted for the current item.

This is the only playback figure the toolkit can know: it counts the PCM the adapter handed to the caller, not what the caller's speaker actually played. A caller that buffers should pass its own figure to :meth:truncate_response.

manual_turn_control property

Whether automatic activity detection is off for this session.

True when turn_detection was {} or {"disabled": True} (the neutral spelling of manual turn control). The client then marks every user turn itself: :meth:send_audio opens it with activity_start and :meth:commit_audio closes it with activity_end.

max_video_fps property

The video-frame ceiling in force, in frames per second.

Comes from configure(max_video_fps=...), else the registry row's max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS. 0 means the gate is off.

session_duration_seconds property

Billable wall-clock seconds this session has been connected.

Measured from the transport, not from the audio frames: a realtime provider that bills per minute bills connection time, and a frame-derived figure would need a per-provider sample-rate assumption and would under-count the silence the provider still charges for. Gemini Live's reconnects (the hidden session-duration cap) close one segment and open another, so the gap between them is excluded rather than billed.

Returns:

Name	Type	Description
float	float	Seconds, accumulated across every connected segment and
	float	including the segment currently open.

Example

async with await open_realtime_session("gpt-realtime") as session:
    ...
print(session.session_duration_seconds, session.cost)

tool_call_behavior property

The function-calling behaviour in force: BLOCKING or NON_BLOCKING.

tool_dispatcher property

The session's tool dispatcher, or None when no tools were given.

transcript property

Every completed transcript of this session, in completion order.

Completion order, not item order: a transcription socket finishes whichever item its workers finish first. A caller that needs the spoken order should sort :attr:transcripts by the order the transcript_delta events for each item_id first appeared.

__aenter__() async

Connect if not already connected.

__aexit__(exc_type, exc, tb) async

Always close the transport, even on an exception.

__init__(model, **kwargs)

See :class:RealtimeSession for the shared arguments.

cancel_response() async

Interrupt generation with an empty completed client turn.

Gemini Live has no cancel message; the documented control is that turn_complete=true unconditionally interrupts generation, which is only available on rows that accept send_client_content mid-session.

Raises:

Type	Description
RealtimeValidationError	The model has no explicit interrupt (send audio, or let voice-activity detection barge in, instead).

capability(name)

Read one registry capability, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:REALTIME_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The registry row's value, or the default for an absent key.

close() async

Settle the caption in progress, then close the socket.

commit_audio(create_response=True) async

Mark the end of the current stretch of audio.

With automatic activity detection (the default) this sends audio_stream_end, which is what makes the server settle the caption in progress: observed live, the final input_transcription arrived about half a second after it and never without it. Under manual turn control (turn_detection={"disabled": True}) it sends activity_end instead, closing the activity :meth:send_audio opened with activity_start - see :meth:GeminiLiveAdapter.commit_audio. Call it when the speaker stops (a pause in a meeting, the end of a dictation); :meth:close calls it for you when audio is still unflushed.

Parameters:

Name	Type	Description	Default
create_response	bool	Accepted for parity with the OpenAI transcription session and ignored - there is no response to create.	True

configure(*, languages=None, custom_vocabulary=None, mode=None, **kwargs)

Stage the caption options, then the shared ones.

The three fields of input_audio_transcription Google documents for live transcription, each gated on the registry row and validated with the same rules as file transcription (:func:~doc_processing_toolkit.multimedia.transcribe_audio):

Parameters:

Name	Type	Description	Default
languages	Optional[Sequence[str]]	BCP-47 language hints (["es-ES"]), sent as language_codes. An empty list asks for automatic language identification explicitly. Gated on supports_language_hints.	None
custom_vocabulary	Optional[Sequence[str]]	Phrases biasing recognition toward domain terms, at most max_custom_vocabulary_terms (a WARNING above recommended_custom_vocabulary_terms). Gated on supports_custom_vocabulary.	None
mode	Optional[str]	"VERBATIM" (the default) or "SMART" (removes disfluencies and formats). Gated on supported_transcription_modes.	None
**kwargs	Any	Any :meth:RealtimeSession.configure option.	{}

Raises:

Type	Description
RealtimeValidationError	A row without the capability key, a value of the wrong shape, too many vocabulary phrases, or a mode the row does not list.

Example

session = await open_transcription_session(
    "gemini-3.5-transcribe-live",
    languages=["en-US"],
    custom_vocabulary=["doc-processing-toolkit", "Gemini Live"],
    mode="SMART",
)

connect() async

Open the Live session with the staged configuration.

events() async

Yield neutral events, reconnecting transparently across the session cap.

Yields:

Name	Type	Description
RealtimeEvent	AsyncIterator[RealtimeEvent]	Translated Live server messages. A session_resumed event marks each transparent reconnect that sent a resumption handle, and session_restarted one that had no handle to send (the new session does not remember the conversation); the stream itself never breaks for either.

export()

This session's record as JSON-clean data (session-portability slice D).

Read-only and available at any time; final after :meth:close. The transcript is the text of every completed turn, in completion order, and is the shape provider_options={"initial_items": ...} (an OpenAI Realtime or Gemini Live session) and GPT-Live's provider_options={"initial_input": ...} take to seed the next session: [{"role": e["role"], "text": e["text"]} for e in exported["transcript"]] (content in place of text for GPT-Live). A transcription session exports its completed captions the same way. A translation session and a GPT-Live session retain no transcript by design (an interpreter's two transcripts are on source_transcript / translated_transcript and its events; GPT-Live's text is on its own event stream) and export usage and cost with an empty list. closed_at and close_reason are set by every family's :meth:close; on GPT-Live close_reason is the provider's session.closed reason when one arrived. TTS and STT are not sessions.

Returns:

Type	Description
Dict[str, Any]	``{"schema_version": 1, "provider", "model", "session_kind",
Dict[str, Any]	"started_at", "closed_at", "close_reason", "duration_seconds",
Dict[str, Any]	"usage", "cost", "cost_estimated", "cost_components",
Dict[str, Any]	"transcript"}``.

Example

record = session.export()
json.dumps(record)                       # always JSON-clean
seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
next_session = open_realtime_session(model, provider_options={"initial_items": seed})

flush_pending_frame() async

Send the held video frame if the rate gate now allows it.

A capture loop that stops pushing would otherwise leave its last frame held forever, so the adapters also call this from :meth:send_audio — a live session sends audio continuously, which makes it a free heartbeat.

Returns:

Name	Type	Description
bool	bool	Whether a frame was actually sent.

send_audio(pcm_bytes) async

Stream 16 kHz PCM16 audio to be captioned.

Parameters:

Name	Type	Description	Default
pcm_bytes	bytes	Raw PCM16 mono audio at 16 kHz; counted towards the session's audio cap and the flush cadence.	required

send_image(image, *, media_type=None, detail=None) async

Send one still image as user input.

A still is a deliberate, discrete act — a photographed page, a chart, a screenshot to ask about — so it is never rate-gated: it goes on the wire immediately, like :meth:send_text. Use :meth:send_video_frame for a continuous capture.

The model must declare image (or video) among its registry input_types; a row that declares neither raises before anything is sent.

Parameters:

Name	Type	Description	Default
image	Any	Encoded image bytes (PNG/JPEG/GIF/WebP/BMP) or a base64 string of them. Raw pixel buffers are not accepted — encode the frame first.	required
media_type	Optional[str]	MIME type. Sniffed from the payload when omitted; a payload the toolkit cannot identify is rejected rather than guessed at.	None
detail	Optional[str]	OpenAI's input_image detail hint ("auto", "low", "high"). Ignored on Gemini Live, which has no equivalent.	None

Raises:

Type	Description
RealtimeValidationError	The model takes no visual input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

await session.send_image(png_bytes, detail="low")
await session.send_text("What does this chart show?")

send_text(text) async

Not available: a transcription session has no text turn.

Raises:

Type	Description
RealtimeValidationError	Always. A transcription socket accepts audio and returns transcripts; there is no model to answer.

send_video_frame(frame, *, media_type=None) async

Offer one video frame to the model, subject to the rate gate.

Unlike :meth:send_image this is lossy on purpose. A capture loop pushing 30 fps into a model that re-reads every frame as billable image tokens would multiply the bill roughly thirtyfold for a picture that barely changed, so frames arriving faster than :attr:max_video_fps are coalesced: the newest frame is held and the one it superseded is counted in :attr:frames_dropped. Nothing is queued, so memory does not grow with the frame rate, and no frame ever delays audio.

Parameters:

Name	Type	Description	Default
frame	Any	Encoded image bytes, or a base64 string of them.	required
media_type	Optional[str]	MIME type; sniffed when omitted.	None

Returns:

Name	Type	Description
bool	bool	Whether this frame went on the wire now. False means it is
	bool	held and will be sent by the next :meth:send_audio,
	bool	meth:send_video_frame or :meth:flush_pending_frame once the
	bool	gate opens — or superseded by a newer frame first.

Raises:

Type	Description
RealtimeValidationError	The model takes no video input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

async for jpeg in camera:                  # 30 fps
    await session.send_video_frame(jpeg)   # ~1 fps reaches the model
print(session.frames_sent, session.frames_dropped)

truncate_response(*, item_id=None, played_ms=None) async

Tell the provider how much of its last answer the caller actually heard.

Barge-in leaves the provider believing it spoke its whole answer while the caller stopped the playback early. Truncation rewrites the stored turn to the audible prefix so the next turn is grounded in what was really heard.

Parameters:

Name	Type	Description	Default
item_id	Optional[str]	The item to truncate; the adapter's last assistant item by default.	None
played_ms	Optional[int]	Milliseconds actually played; :attr:emitted_audio_ms by default.	None

Raises:

Type	Description
RealtimeValidationError	The provider has no truncation control.

OpenAIRealtimeAdapter

Bases: RealtimeSession

OpenAI Realtime adapter over the openai SDK's GA WebSocket surface.

Uses AsyncOpenAI().realtime.connect(model=...), which owns the wss://api.openai.com/v1/realtime?model=... URL, the Authorization header, and the protocol handshake — so no beta header is hand-set and no raw websockets dependency is needed. The SDK's own reconnect budget is left enabled, giving connection-level resilience underneath the neutral event stream.

provider_options recognizes:

• base_url / websocket_base_url / api_key — the xAI escape hatch: xAI's voice API is OpenAI-Realtime compatible, so pointing the client at https://api.x.ai/v1 with an xAI key reuses this adapter wholesale. Overriding only the endpoint is fine: with api_key omitted the client falls back to OPENAI_API_KEY, read from the environment or the repo-root .env.
• max_retries — the SDK's transport-level reconnect budget (default 5).
• audio — merged into the adapter's own audio object one level deep, and audio.input.transcription two levels deep, so a passthrough transcription field never discards the registry's transcription model (:func:_merge_audio_passthrough).
• tools — replaces the tools the toolkit would dispatch, with one WARNING; pass tools= instead so the dispatcher stays in charge.
• any other key is merged into the session.update payload verbatim.

billed_audio_seconds property

The audio duration a per_minute row is billed on.

Precedence, the same rule file transcription follows: the provider's own reported seconds when it reported any (a transcription session reports them per completed item), otherwise the session's connected wall clock. Provider first because a socket that was open for two minutes while thirty seconds of speech passed through it is billed for the speech, not the silence — and only the provider knows which.

Returns:

Name	Type	Description
float	float	Seconds.

capabilities property

Provider capabilities and toolkit features for this model.

See :func:realtime_capabilities — this is the same report, for the session's own model.

compression_policy property

What extends this session, stated without conflating two mechanisms.

resumption reconnects a dropped transport and replays the handle; it does not enlarge the context window. compression is the separate context-window policy. A provider's session-duration caps are reported alongside so a caller can see which limit it is up against.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"resumption", "compression", "limits"}.

connected property writable

Whether a transport is currently established.

Assigning to this attribute also drives :attr:session_duration_seconds: setting it True starts a billing segment and setting it False closes one. Adapters keep assigning it exactly as before.

emitted_audio_ms property

Milliseconds of assistant audio this session has emitted for the current item.

This is the only playback figure the toolkit can know: it counts the PCM the adapter handed to the caller, not what the caller's speaker actually played. A caller that buffers should pass its own figure to :meth:truncate_response.

max_video_fps property

The video-frame ceiling in force, in frames per second.

Comes from configure(max_video_fps=...), else the registry row's max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS. 0 means the gate is off.

session_duration_seconds property

Billable wall-clock seconds this session has been connected.

Measured from the transport, not from the audio frames: a realtime provider that bills per minute bills connection time, and a frame-derived figure would need a per-provider sample-rate assumption and would under-count the silence the provider still charges for. Gemini Live's reconnects (the hidden session-duration cap) close one segment and open another, so the gap between them is excluded rather than billed.

Returns:

Name	Type	Description
float	float	Seconds, accumulated across every connected segment and
	float	including the segment currently open.

Example

async with await open_realtime_session("gpt-realtime") as session:
    ...
print(session.session_duration_seconds, session.cost)

tool_call_behavior property

The function-calling behaviour in force: BLOCKING or NON_BLOCKING.

tool_dispatcher property

The session's tool dispatcher, or None when no tools were given.

__aenter__() async

Connect if not already connected.

__aexit__(exc_type, exc, tb) async

Always close the transport, even on an exception.

__init__(model, **kwargs)

See :class:RealtimeSession for the shared arguments.

cancel_response() async

Cancel the in-flight response (response.cancel).

Raises:

Type	Description
RealtimeValidationError	The registry row says the model cannot be interrupted.

capability(name)

Read one registry capability, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:REALTIME_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The registry row's value, or the default for an absent key.

close() async

Close the WebSocket and finalize the session cost.

commit_audio(create_response=True) async

Close the current input-audio turn manually.

Only needed when turn detection is disabled — with server VAD the provider commits turns on its own.

Parameters:

Name	Type	Description	Default
create_response	bool	Also ask the model to respond immediately.	True

configure(*, noise_reduction=None, input_transcription=None, **kwargs)

Stage the OpenAI-only audio.input options, then the shared ones.

Parameters:

Name	Type	Description	Default
noise_reduction	Optional[Any]	"near_field" (headsets and other close-talking microphones), "far_field" (laptop and conference-room microphones), or False to switch it off explicitly (sent as null). None leaves it unchanged - the provider default until set.	None
input_transcription	Optional[Union[bool, Dict[str, Any]]]	True transcribes the caller's audio with the registry's default_input_transcription_model; False leaves it off. A dict also turns it on and carries the audio.input.transcription guidance fields - {"model", "language", "languages", "prompt", "keywords", "delay"} - merged over the registry model (a "model" key replaces it). language and languages are alternatives; supplying both is rejected. Which fields a given transcription model honours is OpenAI's to say (keywords / languages are documented for gpt-transcribe and gpt-live-transcribe, delay for gpt-realtime-whisper).	None
**kwargs	Any	Any :meth:RealtimeSession.configure option. On a row that publishes supported_thinking_levels, thinking_level is sent as session.reasoning.effort unless DPT_OPENAI_REALTIME_REASONING=off, which restores the earlier drop-with-a-warning.	{}

Raises:

Type	Description
RealtimeValidationError	An unrecognized noise-reduction profile, a malformed input_transcription dict, or any rejection the shared implementation makes.

Example

session = await open_realtime_session(
    "gpt-realtime-2.1-mini",
    noise_reduction="far_field",
    input_transcription={"language": "de"},
    thinking_level="low",
)

connect() async

Open the WebSocket and send the initial session.update.

events() async

Yield neutral events until the session closes.

Yields:

Name	Type	Description
RealtimeEvent	AsyncIterator[RealtimeEvent]	Translated OpenAI Realtime server events; provider events the neutral layer does not model are dropped.

export()

This session's record as JSON-clean data (session-portability slice D).

Read-only and available at any time; final after :meth:close. The transcript is the text of every completed turn, in completion order, and is the shape provider_options={"initial_items": ...} (an OpenAI Realtime or Gemini Live session) and GPT-Live's provider_options={"initial_input": ...} take to seed the next session: [{"role": e["role"], "text": e["text"]} for e in exported["transcript"]] (content in place of text for GPT-Live). A transcription session exports its completed captions the same way. A translation session and a GPT-Live session retain no transcript by design (an interpreter's two transcripts are on source_transcript / translated_transcript and its events; GPT-Live's text is on its own event stream) and export usage and cost with an empty list. closed_at and close_reason are set by every family's :meth:close; on GPT-Live close_reason is the provider's session.closed reason when one arrived. TTS and STT are not sessions.

Returns:

Type	Description
Dict[str, Any]	``{"schema_version": 1, "provider", "model", "session_kind",
Dict[str, Any]	"started_at", "closed_at", "close_reason", "duration_seconds",
Dict[str, Any]	"usage", "cost", "cost_estimated", "cost_components",
Dict[str, Any]	"transcript"}``.

Example

record = session.export()
json.dumps(record)                       # always JSON-clean
seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
next_session = open_realtime_session(model, provider_options={"initial_items": seed})

flush_pending_frame() async

Send the held video frame if the rate gate now allows it.

A capture loop that stops pushing would otherwise leave its last frame held forever, so the adapters also call this from :meth:send_audio — a live session sends audio continuously, which makes it a free heartbeat.

Returns:

Name	Type	Description
bool	bool	Whether a frame was actually sent.

send_audio(pcm_bytes) async

Append PCM16 audio (24 kHz) to the input buffer as base64.

Also releases a video frame the rate gate is holding: a live session sends audio continuously, which makes this the cheapest heartbeat for a capture loop that has gone still.

Parameters:

Name	Type	Description	Default
pcm_bytes	bytes	Raw PCM16 mono audio at 24 kHz.	required

send_image(image, *, media_type=None, detail=None) async

Send one still image as user input.

A still is a deliberate, discrete act — a photographed page, a chart, a screenshot to ask about — so it is never rate-gated: it goes on the wire immediately, like :meth:send_text. Use :meth:send_video_frame for a continuous capture.

The model must declare image (or video) among its registry input_types; a row that declares neither raises before anything is sent.

Parameters:

Name	Type	Description	Default
image	Any	Encoded image bytes (PNG/JPEG/GIF/WebP/BMP) or a base64 string of them. Raw pixel buffers are not accepted — encode the frame first.	required
media_type	Optional[str]	MIME type. Sniffed from the payload when omitted; a payload the toolkit cannot identify is rejected rather than guessed at.	None
detail	Optional[str]	OpenAI's input_image detail hint ("auto", "low", "high"). Ignored on Gemini Live, which has no equivalent.	None

Raises:

Type	Description
RealtimeValidationError	The model takes no visual input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

await session.send_image(png_bytes, detail="low")
await session.send_text("What does this chart show?")

send_text(text) async

Append a text user turn and request a response.

Parameters:

Name	Type	Description	Default
text	str	The user turn, sent as one input_text item.	required

send_video_frame(frame, *, media_type=None) async

Offer one video frame to the model, subject to the rate gate.

Unlike :meth:send_image this is lossy on purpose. A capture loop pushing 30 fps into a model that re-reads every frame as billable image tokens would multiply the bill roughly thirtyfold for a picture that barely changed, so frames arriving faster than :attr:max_video_fps are coalesced: the newest frame is held and the one it superseded is counted in :attr:frames_dropped. Nothing is queued, so memory does not grow with the frame rate, and no frame ever delays audio.

Parameters:

Name	Type	Description	Default
frame	Any	Encoded image bytes, or a base64 string of them.	required
media_type	Optional[str]	MIME type; sniffed when omitted.	None

Returns:

Name	Type	Description
bool	bool	Whether this frame went on the wire now. False means it is
	bool	held and will be sent by the next :meth:send_audio,
	bool	meth:send_video_frame or :meth:flush_pending_frame once the
	bool	gate opens — or superseded by a newer frame first.

Raises:

Type	Description
RealtimeValidationError	The model takes no video input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

async for jpeg in camera:                  # 30 fps
    await session.send_video_frame(jpeg)   # ~1 fps reaches the model
print(session.frames_sent, session.frames_dropped)

truncate_response(*, item_id=None, played_ms=None) async

Truncate an assistant item to the audio the caller actually heard.

Parameters:

Name	Type	Description	Default
item_id	Optional[str]	Item to truncate; the last assistant audio item by default.	None
played_ms	Optional[int]	Milliseconds played; :attr:emitted_audio_ms by default.	None

Raises:

Type	Description
RealtimeValidationError	No item is known to truncate.

OpenAITranscriptionAdapter

Bases: _TranscriptionSessionMixin, OpenAIRealtimeAdapter

OpenAI dedicated transcription session (session.type='transcription').

Same /v1/realtime socket and the same SDK client as the conversational adapter, opened with ?intent=transcription and configured with a transcription session object instead of a realtime one. The model is named inside that object (audio.input.transcription.model), not on the URL — which is why the two connect details are registry keys (realtime_connect_query, realtime_model_in_url) rather than a branch on the session type.

Three typed server events drive it, and they are the reason item_id exists on :class:RealtimeEvent:

• conversation.item.input_audio_transcription.delta -> transcript_delta
• ...completed -> transcript_complete (with the provider's own duration, where it reports one)
• ...failed -> transcript_failed

No tool loop runs here and no response.create is ever sent: there is no response. :meth:close drains what is still in flight first, because a committed turn is transcribed asynchronously and closing the moment the caller stops reading loses the last utterance of every session.

Example

session = await open_transcription_session("gpt-transcribe")
try:
    await session.send_audio(pcm_24k)
    await session.commit_audio(create_response=False)
    async for event in session.events():
        if event.type == "transcript_complete":
            print(event.item_id, event.text)
            break
finally:
    await session.close()
print(session.cost, session.billed_audio_seconds)

billed_audio_seconds property

The audio duration a per_minute row is billed on.

Precedence, the same rule file transcription follows: the provider's own reported seconds when it reported any (a transcription session reports them per completed item), otherwise the session's connected wall clock. Provider first because a socket that was open for two minutes while thirty seconds of speech passed through it is billed for the speech, not the silence — and only the provider knows which.

Returns:

Name	Type	Description
float	float	Seconds.

capabilities property

Provider capabilities and toolkit features for this model.

See :func:realtime_capabilities — this is the same report, for the session's own model.

compression_policy property

What extends this session, stated without conflating two mechanisms.

resumption reconnects a dropped transport and replays the handle; it does not enlarge the context window. compression is the separate context-window policy. A provider's session-duration caps are reported alongside so a caller can see which limit it is up against.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"resumption", "compression", "limits"}.

connected property writable

Whether a transport is currently established.

Assigning to this attribute also drives :attr:session_duration_seconds: setting it True starts a billing segment and setting it False closes one. Adapters keep assigning it exactly as before.

emitted_audio_ms property

Milliseconds of assistant audio this session has emitted for the current item.

This is the only playback figure the toolkit can know: it counts the PCM the adapter handed to the caller, not what the caller's speaker actually played. A caller that buffers should pass its own figure to :meth:truncate_response.

max_video_fps property

The video-frame ceiling in force, in frames per second.

Comes from configure(max_video_fps=...), else the registry row's max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS. 0 means the gate is off.

session_duration_seconds property

Billable wall-clock seconds this session has been connected.

Measured from the transport, not from the audio frames: a realtime provider that bills per minute bills connection time, and a frame-derived figure would need a per-provider sample-rate assumption and would under-count the silence the provider still charges for. Gemini Live's reconnects (the hidden session-duration cap) close one segment and open another, so the gap between them is excluded rather than billed.

Returns:

Name	Type	Description
float	float	Seconds, accumulated across every connected segment and
	float	including the segment currently open.

Example

async with await open_realtime_session("gpt-realtime") as session:
    ...
print(session.session_duration_seconds, session.cost)

tool_call_behavior property

The function-calling behaviour in force: BLOCKING or NON_BLOCKING.

tool_dispatcher property

The session's tool dispatcher, or None when no tools were given.

transcript property

Every completed transcript of this session, in completion order.

Completion order, not item order: a transcription socket finishes whichever item its workers finish first. A caller that needs the spoken order should sort :attr:transcripts by the order the transcript_delta events for each item_id first appeared.

__aenter__() async

Connect if not already connected.

__aexit__(exc_type, exc, tb) async

Always close the transport, even on an exception.

__init__(model, **kwargs)

See :class:RealtimeSession for the shared arguments.

cancel_response() async

Cancel the in-flight response (response.cancel).

Raises:

Type	Description
RealtimeValidationError	The registry row says the model cannot be interrupted.

capability(name)

Read one registry capability, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:REALTIME_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The registry row's value, or the default for an absent key.

close() async

Drain transcripts still in flight, then close the socket.

commit_audio(create_response=True) async

Close the current input-audio turn.

Parameters:

Name	Type	Description	Default
create_response	bool	Ignored and warned about. A transcription session has no response to create; the commit itself is what starts the transcription on a transcription_input_mode: "committed" row.	True

configure(*, language=None, languages=None, prompt=None, keywords=None, delay=None, **kwargs)

Stage the transcription guidance fields, then the shared options.

These five are the whole of OpenAI's AudioTranscription object besides model, and :meth:_session_payload has always put them on the wire — but :meth:RealtimeSession.configure is a closed keyword list, so nothing could put them in the config and open_transcription_session("gpt-transcribe", language="en") raised a bare TypeError instead of reaching the socket.

Only the shape is checked here. Which of them a given transcription model honours is the vendor's business (keywords and languages are documented for gpt-transcribe and gpt-live-transcribe, delay for gpt-realtime-whisper), and inventing a local name-matched gate is exactly what this codebase forbids.

Parameters:

Name	Type	Description	Default
language	Optional[str]	ISO-639-1 code of the input audio ("en"). Supplying it improves accuracy and latency.	None
languages	Optional[Sequence[str]]	Candidate ISO-639-1 codes when the language is not known in advance. Pass a list, not a string.	None
prompt	Optional[str]	Free text steering style or continuing a previous segment.	None
keywords	Optional[Sequence[str]]	Words or phrases to bias the transcription towards. Pass a list, not a string.	None
delay	Optional[str]	How long to wait before emitting text, trading latency for accuracy. One of :data:TRANSCRIPTION_DELAY_LEVELS.	None
**kwargs	Any	Any :meth:RealtimeSession.configure option, plus noise_reduction (:meth:OpenAIRealtimeAdapter.configure).	{}

Raises:

Type	Description
RealtimeValidationError	A field of the wrong shape, an empty list, or a delay outside the published set.

Example

session = await open_transcription_session(
    "gpt-transcribe", language="en", keywords=["doc-processing-toolkit"]
)

connect() async

Open the WebSocket and send the initial session.update.

events() async

Yield neutral events until the session closes.

Yields:

Name	Type	Description
RealtimeEvent	AsyncIterator[RealtimeEvent]	Translated OpenAI Realtime server events; provider events the neutral layer does not model are dropped.

export()

This session's record as JSON-clean data (session-portability slice D).

Read-only and available at any time; final after :meth:close. The transcript is the text of every completed turn, in completion order, and is the shape provider_options={"initial_items": ...} (an OpenAI Realtime or Gemini Live session) and GPT-Live's provider_options={"initial_input": ...} take to seed the next session: [{"role": e["role"], "text": e["text"]} for e in exported["transcript"]] (content in place of text for GPT-Live). A transcription session exports its completed captions the same way. A translation session and a GPT-Live session retain no transcript by design (an interpreter's two transcripts are on source_transcript / translated_transcript and its events; GPT-Live's text is on its own event stream) and export usage and cost with an empty list. closed_at and close_reason are set by every family's :meth:close; on GPT-Live close_reason is the provider's session.closed reason when one arrived. TTS and STT are not sessions.

Returns:

Type	Description
Dict[str, Any]	``{"schema_version": 1, "provider", "model", "session_kind",
Dict[str, Any]	"started_at", "closed_at", "close_reason", "duration_seconds",
Dict[str, Any]	"usage", "cost", "cost_estimated", "cost_components",
Dict[str, Any]	"transcript"}``.

Example

record = session.export()
json.dumps(record)                       # always JSON-clean
seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
next_session = open_realtime_session(model, provider_options={"initial_items": seed})

flush_pending_frame() async

Send the held video frame if the rate gate now allows it.

A capture loop that stops pushing would otherwise leave its last frame held forever, so the adapters also call this from :meth:send_audio — a live session sends audio continuously, which makes it a free heartbeat.

Returns:

Name	Type	Description
bool	bool	Whether a frame was actually sent.

send_audio(pcm_bytes) async

Append PCM16 audio (24 kHz) to the input buffer as base64.

Also releases a video frame the rate gate is holding: a live session sends audio continuously, which makes this the cheapest heartbeat for a capture loop that has gone still.

Parameters:

Name	Type	Description	Default
pcm_bytes	bytes	Raw PCM16 mono audio at 24 kHz.	required

send_image(image, *, media_type=None, detail=None) async

Send one still image as user input.

A still is a deliberate, discrete act — a photographed page, a chart, a screenshot to ask about — so it is never rate-gated: it goes on the wire immediately, like :meth:send_text. Use :meth:send_video_frame for a continuous capture.

The model must declare image (or video) among its registry input_types; a row that declares neither raises before anything is sent.

Parameters:

Name	Type	Description	Default
image	Any	Encoded image bytes (PNG/JPEG/GIF/WebP/BMP) or a base64 string of them. Raw pixel buffers are not accepted — encode the frame first.	required
media_type	Optional[str]	MIME type. Sniffed from the payload when omitted; a payload the toolkit cannot identify is rejected rather than guessed at.	None
detail	Optional[str]	OpenAI's input_image detail hint ("auto", "low", "high"). Ignored on Gemini Live, which has no equivalent.	None

Raises:

Type	Description
RealtimeValidationError	The model takes no visual input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

await session.send_image(png_bytes, detail="low")
await session.send_text("What does this chart show?")

send_text(text) async

Not available: a transcription session has no text turn.

Raises:

Type	Description
RealtimeValidationError	Always. A transcription socket accepts audio and returns transcripts; there is no model to answer.

send_video_frame(frame, *, media_type=None) async

Offer one video frame to the model, subject to the rate gate.

Unlike :meth:send_image this is lossy on purpose. A capture loop pushing 30 fps into a model that re-reads every frame as billable image tokens would multiply the bill roughly thirtyfold for a picture that barely changed, so frames arriving faster than :attr:max_video_fps are coalesced: the newest frame is held and the one it superseded is counted in :attr:frames_dropped. Nothing is queued, so memory does not grow with the frame rate, and no frame ever delays audio.

Parameters:

Name	Type	Description	Default
frame	Any	Encoded image bytes, or a base64 string of them.	required
media_type	Optional[str]	MIME type; sniffed when omitted.	None

Returns:

Name	Type	Description
bool	bool	Whether this frame went on the wire now. False means it is
	bool	held and will be sent by the next :meth:send_audio,
	bool	meth:send_video_frame or :meth:flush_pending_frame once the
	bool	gate opens — or superseded by a newer frame first.

Raises:

Type	Description
RealtimeValidationError	The model takes no video input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

async for jpeg in camera:                  # 30 fps
    await session.send_video_frame(jpeg)   # ~1 fps reaches the model
print(session.frames_sent, session.frames_dropped)

truncate_response(*, item_id=None, played_ms=None) async

Truncate an assistant item to the audio the caller actually heard.

Parameters:

Name	Type	Description	Default
item_id	Optional[str]	Item to truncate; the last assistant audio item by default.	None
played_ms	Optional[int]	Milliseconds played; :attr:emitted_audio_ms by default.	None

Raises:

Type	Description
RealtimeValidationError	No item is known to truncate.

RealtimeConnectionError

Bases: RuntimeError

Raised when a realtime transport cannot be established or has been lost beyond the adapter's reconnect budget.

RealtimeEvent dataclass

One neutral event from a realtime session.

Attributes:

Name	Type	Description
type	str	One of audio_delta, transcript_delta, transcript_complete, transcript_failed, text_delta, response_complete, turn_complete, interrupted, session_resumed, session_restarted, tool_call, tool_result, tool_call_cancelled, error. session_resumed marks a reconnect that sent a resumption handle (the conversation continues); session_restarted marks one that had no handle to send (a fresh session: earlier turns are gone from the model's context).
audio	Optional[bytes]	Raw PCM16 bytes for audio_delta (already base64-decoded).
text	Optional[str]	Transcript or text fragment for transcript_delta / text_delta, or the whole final transcript of one input item for transcript_complete.
role	str	"assistant" for model output, "user" for transcripts of the caller's own audio.
usage	Optional[Dict[str, int]]	Cumulative session usage snapshot, attached to response_complete.
cost	Optional[float]	Cumulative session cost in USD, attached to response_complete.
error	Optional[str]	Provider error message for error events, the tool error text on a failed tool_result, or the provider's reason on a transcript_failed.
item_id	Optional[str]	Provider item identifier where one exists — the correlation key a truncation needs on OpenAI, and the only key that pairs a transcript with its input item, because a transcription session completes turns out of order. None on Gemini Live, which does not label conversation items.
duration_seconds	Optional[float]	Audio seconds the provider billed for this item, where it reports them (transcript_complete on a duration-billed transcription session). None when the provider reported tokens instead, or nothing.
language	Optional[str]	BCP-47 code the provider attached to this transcript fragment, where it reports one. On a translation session it is how the detected source language is surfaced — Gemini Live auto-detects it and reports it per transcript, and there is no field to request one. None everywhere the provider labels nothing.
tool_name	Optional[str]	Tool name for tool_call / tool_result / tool_call_cancelled.
tool_call_id	Optional[str]	Provider call id, the key that pairs a result with its call. Under Google's asynchronous function calling several calls are outstanding at once, so this is the only correlation key.
tool_arguments	Optional[Dict[str, Any]]	Decoded tool arguments for tool_call.
tool_result	Optional[str]	The text the toolkit returned to the model for tool_result.
raw	Any	The untranslated provider event, for callers that need something the neutral layer does not model.

RealtimeSession

A provider-neutral realtime voice session.

Subclasses supply the transport; everything callers touch — configuration validation, the neutral event stream, usage accounting, and cost — lives here. Instances are async context managers::

async with await open_realtime_session("gpt-realtime", voice="marin") as session:
    await session.send_text("Say hello in five words.")
    async for event in session.events():
        if event.type == "audio_delta":
            speaker.write(event.audio)
        elif event.type == "response_complete":
            break
print(session.cost, session.usage.as_dict())

Attributes:

Name	Type	Description
model		The registry model name driving the session.
provider		"openai" or "google", resolved from the registry.
usage		Live :class:RealtimeUsage totals for the session.
cost		Live session cost in USD, recomputed whenever usage changes.
cost_components	Dict[str, float]	:attr:cost split per component (the COST_COMPONENT_KEYS of :func:~doc_processing_toolkit.core.calculate_cost), refreshed with it. A duration-billed session's charge is in duration_usd.
connected	bool	Whether a transport is currently established. Assigning it also drives :attr:session_duration_seconds.
session_duration_seconds	float	Billable wall-clock seconds connected, summed across reconnects.
conversation	List[Dict[str, Any]]	Completed transcripts of this session, text only, in completion order: {"role": "user" | "assistant", "text", "item_id", "completed_at"} (an answer cut short by a barge-in also carries "interrupted": True). Audio is never retained; :meth:export carries the list as transcript.
started_at	Optional[str]	UTC timestamp (ISO 8601) of the first connect, or None.
closed_at	Optional[str]	UTC timestamp of :meth:close, or None while open.
close_reason	Optional[str]	"closed" after a caller's close, "error" after a transport failure, or None while open.

billed_audio_seconds property

The audio duration a per_minute row is billed on.

Precedence, the same rule file transcription follows: the provider's own reported seconds when it reported any (a transcription session reports them per completed item), otherwise the session's connected wall clock. Provider first because a socket that was open for two minutes while thirty seconds of speech passed through it is billed for the speech, not the silence — and only the provider knows which.

Returns:

Name	Type	Description
float	float	Seconds.

capabilities property

Provider capabilities and toolkit features for this model.

See :func:realtime_capabilities — this is the same report, for the session's own model.

compression_policy property

What extends this session, stated without conflating two mechanisms.

resumption reconnects a dropped transport and replays the handle; it does not enlarge the context window. compression is the separate context-window policy. A provider's session-duration caps are reported alongside so a caller can see which limit it is up against.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"resumption", "compression", "limits"}.

connected property writable

Whether a transport is currently established.

Assigning to this attribute also drives :attr:session_duration_seconds: setting it True starts a billing segment and setting it False closes one. Adapters keep assigning it exactly as before.

emitted_audio_ms property

Milliseconds of assistant audio this session has emitted for the current item.

This is the only playback figure the toolkit can know: it counts the PCM the adapter handed to the caller, not what the caller's speaker actually played. A caller that buffers should pass its own figure to :meth:truncate_response.

max_video_fps property

The video-frame ceiling in force, in frames per second.

Comes from configure(max_video_fps=...), else the registry row's max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS. 0 means the gate is off.

session_duration_seconds property

Billable wall-clock seconds this session has been connected.

Measured from the transport, not from the audio frames: a realtime provider that bills per minute bills connection time, and a frame-derived figure would need a per-provider sample-rate assumption and would under-count the silence the provider still charges for. Gemini Live's reconnects (the hidden session-duration cap) close one segment and open another, so the gap between them is excluded rather than billed.

Returns:

Name	Type	Description
float	float	Seconds, accumulated across every connected segment and
	float	including the segment currently open.

Example

async with await open_realtime_session("gpt-realtime") as session:
    ...
print(session.session_duration_seconds, session.cost)

tool_call_behavior property

The function-calling behaviour in force: BLOCKING or NON_BLOCKING.

tool_dispatcher property

The session's tool dispatcher, or None when no tools were given.

__aenter__() async

Connect if not already connected.

__aexit__(exc_type, exc, tb) async

Always close the transport, even on an exception.

__init__(model, *, voice=None, instructions=None, output_format=None, turn_detection=None, provider_options=None, tools=None, hooks=None, max_tool_rounds=DEFAULT_MAX_TOOL_ROUNDS, tool_timeout_seconds=DEFAULT_TOOL_TIMEOUT_SECONDS, logger=None, **configure_kwargs)

Build a session object. No network activity happens until :meth:connect.

Parameters:

Name	Type	Description	Default
model	str	A registered realtime model.	required
voice	Optional[Union[str, Dict[str, str]]]	Voice name, validated against the registry's supported_voices, or an OpenAI custom-voice object {"id": "voice_..."} (shape-checked, bypasses the list).	None
instructions	Optional[str]	System instructions steering the model's persona.	None
output_format	Optional[str]	Audio format, validated against supported_output_formats.	None
turn_detection	Optional[Dict[str, Any]]	Voice-activity/turn-detection configuration in the neutral vocabulary (None keeps the provider default; {} or {"disabled": True} requests manual turn control). It is translated per provider by :func:translate_turn_detection, never forwarded verbatim.	None
provider_options	Optional[Dict[str, Any]]	Passthrough options merged into the provider's connect/session payload. A key the adapter builds itself is replaced by the passthrough value with one WARNING per key per session (:attr:GeminiLiveAdapter.passthrough_overrides; the OpenAI tools key), and the OpenAI audio object is merged one level deep rather than replaced.	None
tools	Optional[Sequence[Any]]	Tools for the model to call, in any format :class:~doc_processing_toolkit.agent.loader.ToolLoader accepts. Requires supports_tool_calls on the registry row.	None
hooks	Any	An AgentHooks object or callable receiving tool_call_start / before_tool_call / tool_call_result.	None
max_tool_rounds	int	Ceiling on tool calls dispatched in this session.	DEFAULT_MAX_TOOL_ROUNDS
tool_timeout_seconds	float	Per-call wall-clock ceiling in seconds.	DEFAULT_TOOL_TIMEOUT_SECONDS
logger	Optional[Logger]	Optional logger (module logger by default).	None
**configure_kwargs	Any	Any other :meth:configure option (modalities, input_transcription, thinking_level, proactive_audio, enable_affective_dialog, context_compression, tool_call_behavior, tool_response_scheduling, max_video_fps, and noise_reduction on the OpenAI conversational, transcription and translation adapters), staged at construction.	{}

cancel_response() async

Interrupt the model's in-flight response.

Raises:

Type	Description
RealtimeValidationError	The model or adapter cannot interrupt explicitly (the message names what it can do instead).

capability(name)

Read one registry capability, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:REALTIME_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The registry row's value, or the default for an absent key.

close() async

Close the transport and finalize :attr:cost.

configure(*, voice=None, instructions=None, output_format=None, turn_detection=None, modalities=None, input_transcription=None, thinking_level=None, proactive_audio=None, enable_affective_dialog=None, context_compression=None, tool_call_behavior=None, tool_response_scheduling=None, max_video_fps=None, provider_options=None)

Validate and stage session options.

Every argument is optional and only the ones supplied are changed, so this can be called repeatedly to adjust one knob at a time. Options are checked against the registry's descriptions.<model> row, so an unsupported voice or format raises :class:RealtimeValidationError naming what the model does support rather than failing mid-session.

Before :meth:connect the staged options form the initial session payload. After it, OpenAI applies them to the live session at the next send or event poll; Gemini Live fixes its configuration at connect time, so a post-connect change there logs a warning and takes effect on the next reconnect.

Parameters:

Name	Type	Description	Default
voice	Optional[Union[str, Dict[str, str]]]	Voice name (validated against supported_voices), or on OpenAI a custom-voice object {"id": "voice_..."}, which bypasses the list and is shape-checked only.	None
instructions	Optional[str]	System instructions for the model.	None
output_format	Optional[str]	Audio output format (validated against supported_output_formats).	None
turn_detection	Optional[Dict[str, Any]]	Voice-activity/turn-detection configuration in the neutral vocabulary: None keeps the provider default; {} or {"disabled": True} requests manual turn control, which on Gemini switches :meth:GeminiLiveAdapter.send_audio / :meth:GeminiLiveAdapter.commit_audio to the activity_start / activity_end route. Translated per provider by :func:translate_turn_detection, never forwarded verbatim, and refused on a row publishing supports_turn_detection: false.	None
modalities	Optional[List[str]]	Output modalities, e.g. ["audio"] or ["text"].	None
input_transcription	Optional[Union[bool, Dict[str, Any]]]	Whether to ask the provider to transcribe the caller's own audio (surfaced as transcript_delta events with role="user"). OpenAI conversational sessions also take a dict of transcription guidance fields - see :meth:OpenAIRealtimeAdapter.configure; every other adapter rejects a dict.	None
thinking_level	Optional[str]	Background-reasoning level, validated against the row's supported_thinking_levels. A model whose row omits that key (gemini-3.8-live, where the vendor says to omit thinking_level entirely) drops the value with a WARNING. Gemini sends it as thinking_config.thinking_level; the OpenAI reasoning rows (gpt-realtime-2 family) send it as session.reasoning.effort (kill switch DPT_OPENAI_REALTIME_REASONING=off).	None
proactive_audio	Optional[bool]	Whether the model may speak unprompted. A row with proactive_audio_always_on rejects False locally, because the API returns an error for it.	None
enable_affective_dialog	Optional[bool]	Affective-dialogue toggle. Rejected locally on a row with supports_affective_dialog: false (the 3.8 Live models removed the feature from the API).	None
context_compression	Optional[Any]	Context-window compression policy — False to disable, True for the provider default, or {"trigger_tokens": int, "sliding_window_target_tokens": int}. This is not session resumption: resumption reconnects a dropped transport, compression extends how long the context lasts. See :attr:compression_policy.	None
tool_call_behavior	Optional[str]	"BLOCKING" or "NON_BLOCKING"; None takes the registry row's function_calling_behavior_default.	None
tool_response_scheduling	Optional[str]	"SILENT" / "WHEN_IDLE" / "INTERRUPT" for asynchronous function responses. Only legal where the row sets supports_function_response_scheduling.	None
max_video_fps	Optional[float]	Ceiling on video frames put on the wire, in frames per second (registry max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS). 0 disables the gate and sends every frame. Applies to :meth:send_video_frame only.	None
provider_options	Optional[Dict[str, Any]]	Extra provider-specific options merged into the existing passthrough dict. A key the adapter builds itself is replaced by the passthrough value with one WARNING per key per session (:attr:GeminiLiveAdapter.passthrough_overrides; the OpenAI tools key); the OpenAI audio object is merged one level deep instead.	None

Raises:

Type	Description
RealtimeValidationError	An option the registry row says the model cannot accept.

connect() async

Establish the transport and apply the staged configuration.

events()

Async-iterate the session's neutral events until it closes.

Yields:

Name	Type	Description
RealtimeEvent	AsyncIterator[RealtimeEvent]	Provider events translated to the neutral vocabulary.

export()

This session's record as JSON-clean data (session-portability slice D).

Read-only and available at any time; final after :meth:close. The transcript is the text of every completed turn, in completion order, and is the shape provider_options={"initial_items": ...} (an OpenAI Realtime or Gemini Live session) and GPT-Live's provider_options={"initial_input": ...} take to seed the next session: [{"role": e["role"], "text": e["text"]} for e in exported["transcript"]] (content in place of text for GPT-Live). A transcription session exports its completed captions the same way. A translation session and a GPT-Live session retain no transcript by design (an interpreter's two transcripts are on source_transcript / translated_transcript and its events; GPT-Live's text is on its own event stream) and export usage and cost with an empty list. closed_at and close_reason are set by every family's :meth:close; on GPT-Live close_reason is the provider's session.closed reason when one arrived. TTS and STT are not sessions.

Returns:

Type	Description
Dict[str, Any]	``{"schema_version": 1, "provider", "model", "session_kind",
Dict[str, Any]	"started_at", "closed_at", "close_reason", "duration_seconds",
Dict[str, Any]	"usage", "cost", "cost_estimated", "cost_components",
Dict[str, Any]	"transcript"}``.

Example

record = session.export()
json.dumps(record)                       # always JSON-clean
seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
next_session = open_realtime_session(model, provider_options={"initial_items": seed})

flush_pending_frame() async

Send the held video frame if the rate gate now allows it.

A capture loop that stops pushing would otherwise leave its last frame held forever, so the adapters also call this from :meth:send_audio — a live session sends audio continuously, which makes it a free heartbeat.

Returns:

Name	Type	Description
bool	bool	Whether a frame was actually sent.

send_audio(pcm_bytes) async

Send a chunk of raw PCM16 microphone audio to the model.

Parameters:

Name	Type	Description	Default
pcm_bytes	bytes	Little-endian 16-bit mono PCM. OpenAI expects 24 kHz, Gemini Live expects 16 kHz — see :data:OPENAI_REALTIME_SAMPLE_RATE and :data:GOOGLE_LIVE_INPUT_SAMPLE_RATE.	required

send_image(image, *, media_type=None, detail=None) async

Send one still image as user input.

A still is a deliberate, discrete act — a photographed page, a chart, a screenshot to ask about — so it is never rate-gated: it goes on the wire immediately, like :meth:send_text. Use :meth:send_video_frame for a continuous capture.

The model must declare image (or video) among its registry input_types; a row that declares neither raises before anything is sent.

Parameters:

Name	Type	Description	Default
image	Any	Encoded image bytes (PNG/JPEG/GIF/WebP/BMP) or a base64 string of them. Raw pixel buffers are not accepted — encode the frame first.	required
media_type	Optional[str]	MIME type. Sniffed from the payload when omitted; a payload the toolkit cannot identify is rejected rather than guessed at.	None
detail	Optional[str]	OpenAI's input_image detail hint ("auto", "low", "high"). Ignored on Gemini Live, which has no equivalent.	None

Raises:

Type	Description
RealtimeValidationError	The model takes no visual input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

await session.send_image(png_bytes, detail="low")
await session.send_text("What does this chart show?")

send_text(text) async

Send a text turn and ask the model to respond.

Parameters:

Name	Type	Description	Default
text	str	The user message.	required

send_video_frame(frame, *, media_type=None) async

Offer one video frame to the model, subject to the rate gate.

Unlike :meth:send_image this is lossy on purpose. A capture loop pushing 30 fps into a model that re-reads every frame as billable image tokens would multiply the bill roughly thirtyfold for a picture that barely changed, so frames arriving faster than :attr:max_video_fps are coalesced: the newest frame is held and the one it superseded is counted in :attr:frames_dropped. Nothing is queued, so memory does not grow with the frame rate, and no frame ever delays audio.

Parameters:

Name	Type	Description	Default
frame	Any	Encoded image bytes, or a base64 string of them.	required
media_type	Optional[str]	MIME type; sniffed when omitted.	None

Returns:

Name	Type	Description
bool	bool	Whether this frame went on the wire now. False means it is
	bool	held and will be sent by the next :meth:send_audio,
	bool	meth:send_video_frame or :meth:flush_pending_frame once the
	bool	gate opens — or superseded by a newer frame first.

Raises:

Type	Description
RealtimeValidationError	The model takes no video input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

async for jpeg in camera:                  # 30 fps
    await session.send_video_frame(jpeg)   # ~1 fps reaches the model
print(session.frames_sent, session.frames_dropped)

truncate_response(*, item_id=None, played_ms=None) async

Tell the provider how much of its last answer the caller actually heard.

Barge-in leaves the provider believing it spoke its whole answer while the caller stopped the playback early. Truncation rewrites the stored turn to the audible prefix so the next turn is grounded in what was really heard.

Parameters:

Name	Type	Description	Default
item_id	Optional[str]	The item to truncate; the adapter's last assistant item by default.	None
played_ms	Optional[int]	Milliseconds actually played; :attr:emitted_audio_ms by default.	None

Raises:

Type	Description
RealtimeValidationError	The provider has no truncation control.

RealtimeToolCall dataclass

One tool call the model asked the host to run.

Attributes:

Name	Type	Description
name	str	The tool name as the model called it.
arguments	Dict[str, Any]	Decoded arguments.
call_id	Optional[str]	The provider's correlation id (None where a provider omits one).

RealtimeToolDispatcher

Bounded host-side tool dispatch for a realtime session.

Neither realtime protocol answers a tool call on the server: the host runs the tool and sends the result back. This class owns that half of the loop, reusing the pieces the text loops already use — the :class:~doc_processing_toolkit.agent.loader.ToolLoader (all nine tool formats), :func:~doc_processing_toolkit.utils.execute_mcp_tool, and the hook names tool_call_start, before_tool_call (the veto) and tool_call_result.

It is bounded in both directions a realtime session can run away: a whole session may dispatch at most max_tool_rounds calls, and one call may run for at most tool_timeout_seconds. Either ceiling produces an error result the model receives, never a silently unanswered tool slot.

Example

session = await open_realtime_session(
    "gpt-realtime-2.1-mini",
    tools=[look_up_passage],
    hooks=my_hooks,
)
async for event in session.events():
    if event.type == "tool_result":
        print(event.tool_name, event.tool_result)

tool_schemas property

Neutral {name, description, parameters} schemas, after :meth:prepare.

__init__(tools=None, *, hooks=None, max_tool_rounds=DEFAULT_MAX_TOOL_ROUNDS, tool_timeout_seconds=DEFAULT_TOOL_TIMEOUT_SECONDS, logger=None)

Build a dispatcher. No tool is loaded until :meth:prepare.

Parameters:

Name	Type	Description	Default
tools	Optional[Sequence[Any]]	Tools in any format :meth:~doc_processing_toolkit.agent.loader.ToolLoader.load accepts.	None
hooks	Any	An AgentHooks object or a callable, driven exactly as in the text loops.	None
max_tool_rounds	int	Maximum tool calls dispatched in one session.	DEFAULT_MAX_TOOL_ROUNDS
tool_timeout_seconds	float	Per-call wall-clock ceiling in seconds.	DEFAULT_TOOL_TIMEOUT_SECONDS
logger	Optional[Logger]	Optional logger (module logger by default).	None

aclose() async

Release the MCP sessions this dispatcher opened.

dispatch(call) async

Run one tool call under the hooks, the round budget and the timeout.

Parameters:

Name	Type	Description	Default
call	RealtimeToolCall	The model's call.	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"output": str, "is_error": bool}. The output is
Dict[str, Any]	always a string the adapter can hand straight back to the model —
Dict[str, Any]	a refusal, a timeout notice and a tool exception all come back this
Dict[str, Any]	way rather than raising into the event stream.

prepare() async

Load the tools, open their MCP sessions and build the routing map.

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: The neutral tool schemas, ready for an
List[Dict[str, Any]]	adapter to shape into its provider's declaration format.

RealtimeUsage dataclass

Token usage accumulated across a realtime session, split by modality.

Cached counts are a subset of their non-cached counterparts, matching the convention :func:doc_processing_toolkit.core.calculate_cost expects. :attr:output_thinking_tokens is a subset in the same sense — thinking is text-rate output, so it is inside :attr:output_text_tokens and is reported separately only so a caller can see how much of the answer was reasoning.

cached_input_tokens property

The cached share of :attr:total_input_tokens.

total_input_tokens property

Every prompt token billed this session, cached share included.

total_output_tokens property

Every completion token billed this session.

total_tokens property

Input plus output tokens.

add(other)

Accumulate another usage record into this one, field by field.

Parameters:

Name	Type	Description	Default
other	'RealtimeUsage'	The usage delta to fold in; each of its counters is added to the matching counter here, nothing is replaced.	required

as_dict()

Flat dictionary of every counter plus the derived totals.

RealtimeValidationError

Bases: ValueError

Raised when a requested session option is unsupported by the target model.

The message names the provider limitation, so unsupported combinations fail at validation rather than mid-session (the convention shared with :mod:doc_processing_toolkit.multimedia).

open_realtime_session(model, voice=None, instructions=None, provider_options=None, tools=None, hooks=None, max_tool_rounds=DEFAULT_MAX_TOOL_ROUNDS, tool_timeout_seconds=DEFAULT_TOOL_TIMEOUT_SECONDS, logger=None, session_kind=None, **configure_kwargs) async

Open a connected realtime session for a registered realtime model.

The adapter is chosen from (provider, session_kind) in model_information.json, so callers never name one. session_kind defaults to the conversational realtime kind wherever the model serves it, which is every model that worked before this parameter existed.

Parameters:

Name	Type	Description	Default
model	str	A registered realtime model (gpt-realtime, gemini-3.1-flash-live-preview), or a model reachable on the realtime socket through its api_surfaces list (gpt-transcribe).	required
voice	Optional[Union[str, Dict[str, str]]]	Voice name, validated against the registry's supported_voices, or an OpenAI custom-voice object {"id": "voice_..."}.	None
instructions	Optional[str]	System instructions steering the model's persona.	None
provider_options	Optional[Dict[str, Any]]	Provider-specific passthrough options. For OpenAI this is also where the xAI escape hatch lives (base_url, websocket_base_url, api_key); for Google it accepts any LiveConnectConfig field plus max_reconnects.	None
tools	Optional[Sequence[Any]]	Tools the model may call, in any format the :class:~doc_processing_toolkit.agent.loader.ToolLoader accepts.	None
hooks	Any	An AgentHooks object or callable receiving the tool hook events (tool_call_start, before_tool_call, tool_call_result).	None
max_tool_rounds	int	Ceiling on tool calls dispatched in the session.	DEFAULT_MAX_TOOL_ROUNDS
tool_timeout_seconds	float	Per-call wall-clock ceiling in seconds.	DEFAULT_TOOL_TIMEOUT_SECONDS
logger	Optional[Logger]	Optional logger.	None
session_kind	Optional[str]	Which session family to open — "realtime" (a voice conversation), "transcription" (captions only), "translation" (an interpreter socket; see :func:open_translation_session) or "live" (OpenAI GPT-Live; see :func:~doc_processing_toolkit.realtime_gpt_live.open_live_session). None takes the model's default. An unsupported kind is rejected before the socket opens, naming the kinds the model serves.	None
**configure_kwargs	Any	Any other :meth:RealtimeSession.configure option (output_format, turn_detection, modalities, input_transcription, thinking_level, proactive_audio, enable_affective_dialog, context_compression, tool_call_behavior, tool_response_scheduling, max_video_fps), plus noise_reduction on the OpenAI conversational, transcription and translation sessions (GPT-Live refuses it by name).	{}

Returns:

Name	Type	Description
RealtimeSession	RealtimeSession	A connected session, ready for :meth:~RealtimeSession.send_audio / :meth:~RealtimeSession.send_text.

Raises:

Type	Description
RealtimeValidationError	The model is not a registered realtime model, an option is unsupported by it, or it does not serve the requested session kind.
RealtimeConnectionError	No client could be built for the provider.

Example

talk = await open_realtime_session("gpt-realtime-2.1-mini", voice="marin")
captions = await open_realtime_session("gpt-transcribe", session_kind="transcription")

open_transcription_session(model, **kwargs) async

Open a dedicated live-transcription session.

Exactly open_realtime_session(model, session_kind="transcription", ...). It exists as its own name because a transcription session is a different object to use — no voice, no tools, no model turn, transcripts correlated by item_id — and burying that behind a keyword makes it look like a mode of the conversational session rather than a separate protocol.

Parameters:

Name	Type	Description	Default
model	str	A model that serves a transcription session (gpt-transcribe, gpt-live-transcribe, gpt-realtime-whisper, gemini-3.5-transcribe-live).	required
**kwargs	Any	Any :func:open_realtime_session argument except session_kind, plus the OpenAI transcription guidance fields (:meth:OpenAITranscriptionAdapter.configure: language, languages, prompt, keywords, delay) or the Gemini caption options (:meth:GeminiTranscriptionAdapter.configure: languages, custom_vocabulary, mode), and noise_reduction on the OpenAI rows.	{}

Returns:

Name	Type	Description
RealtimeSession	RealtimeSession	A connected transcription session.

Raises:

Type	Description
RealtimeValidationError	The model does not serve a transcription session, or an option is unsupported by it.

Example

session = await open_transcription_session("gemini-3.5-transcribe-live")
await session.send_audio(pcm_16k)

open_translation_session(model, *, target_language, **kwargs) async

Open a dedicated live-translation session.

Exactly open_realtime_session(model, session_kind="translation", ...), under its own name for the same reason :func:open_transcription_session has one: an interpreter session is a different object to use. There is no model turn, no tool, no voice and no turn-taking — audio goes in continuously and translated audio plus both transcripts come back while the speaker is still talking. The session ends when you close it; see :mod:doc_processing_toolkit.realtime_translation for the close-and-drain contract.

Parameters:

Name	Type	Description	Default
model	str	A model that serves a translation session (gemini-3.5-live-translate-preview, gpt-realtime-translate).	required
target_language	str	The BCP-47 code to translate into, validated against the row's supported_target_languages before the socket opens. Required: a session that silently interpreted into the vendor's default language would be a wrong answer the caller pays for by the minute.	required
**kwargs	Any	Any :func:open_realtime_session argument except session_kind, plus the translation options (target_languages, source_language, echo_target_language, input_transcription, and noise_reduction on OpenAI).	{}

Returns:

Name	Type	Description
RealtimeSession	RealtimeSession	A connected translation session.

Raises:

Type	Description
RealtimeValidationError	The model does not serve a translation session, the target language is unsupported, or an option the interpreter protocol has no place for was supplied.

Example

session = await open_translation_session(
    "gemini-3.5-live-translate-preview", target_language="en"
)
try:
    await session.send_audio(japanese_pcm_16k)
finally:
    await session.close()
print(session.source_transcript, "->", session.translated_transcript)

openai_realtime_reasoning_enabled(logger=None)

Read DPT_OPENAI_REALTIME_REASONING.

Parameters:

Name	Type	Description	Default
logger	Optional[Logger]	Logger for the one-time warning about an unrecognised value.	None

Returns:

Name	Type	Description
bool	bool	False only for off (case-insensitive). Unset, empty,
	bool	on and any unrecognised value (warned about once) return True.

Example

os.environ["DPT_OPENAI_REALTIME_REASONING"] = "off"
openai_realtime_reasoning_enabled()  # False

realtime_capabilities(model)

Report a realtime model's provider capabilities and toolkit features.

The two are deliberately separate maps. provider is what the registry row advertises about the vendor API; toolkit is what this module's adapter actually implements for that provider. A row may say supports_tool_calls: true while an adapter dispatches nothing, and conflating the two is how that gap stayed invisible.

toolkit_features is resolved through the adapter that serves this model's (provider, session_kind), not the provider alone: a translation or GPT-Live session refuses controls the conversational adapter on the same provider implements.

Parameters:

Name	Type	Description	Default
model	str	A registered realtime model.	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: ``{"model", "provider", "provider_capabilities",
Dict[str, Any]	"toolkit_features"}.provider_capabilities`` carries every key in
Dict[str, Any]	data:REALTIME_CAPABILITY_DEFAULTS, resolved against the registry row
Dict[str, Any]	(absent keys take the documented default).

Raises:

Type	Description
RealtimeValidationError	The model is not a registered realtime model.

Example

caps = realtime_capabilities("gemini-3.8-live")
caps["provider_capabilities"]["function_calling_behavior_default"]
# 'NON_BLOCKING'
caps["toolkit_features"]["async_tool_dispatch"]
# True

resolve_realtime_session_kind(model, session_kind=None)

Pick and validate the session kind a model will be opened with.

Parameters:

Name	Type	Description	Default
model	str	A registered realtime model.	required
session_kind	Optional[str]	The requested kind, or None to take the model's default — the conversational realtime kind where the model serves it, otherwise its only kind. A model serving several kinds and none of them conversational must name one.	None

Returns:

Name	Type	Description
str	str	The resolved session kind.

Raises:

Type	Description
RealtimeValidationError	The kind is not in the vocabulary, the model does not serve it, the model serves it only inside its conversation, or no adapter implements it for that provider. All are raised before any socket is opened.

A conversational row is reported as serving transcription too, because its own session transcribes the audio it hears (input_transcription). That is a derived statement about the conversational socket, not a second session family: the dedicated transcription adapters configure a session a conversational model rejects (Gemini's native-audio Live models close with 1007 on the text-only response a captions session asks for) or price from the wrong row. A dedicated kind is therefore accepted on a conversational row only when the row declares it in session_kinds.

Example

resolve_realtime_session_kind("gpt-realtime")
# 'realtime'
resolve_realtime_session_kind("gpt-live-transcribe")
# 'transcription'

sniff_image_media_type(data)

Identify an encoded image from its magic bytes.

Parameters:

Name	Type	Description	Default
data	bytes	The first bytes of an encoded image.	required

Returns:

Type	Description
Optional[str]	Optional[str]: The MIME type, or None for a payload the toolkit
Optional[str]	cannot identify. None is deliberate: labelling an unknown payload
Optional[str]	with a guess turns a local validation failure into a provider error the
Optional[str]	caller has already paid to send.

Example

sniff_image_media_type(png_bytes)
# 'image/png'

translate_turn_detection(turn_detection, provider, *, logger=None, model='')

Translate the neutral turn-detection dict into one provider's shape.

The neutral vocabulary is the union of both providers' voice-activity controls. A key the target provider cannot express is dropped with a WARNING naming the nearest control it does have, rather than forwarded — forwarding is what made threshold=0.5 raise a pydantic ValidationError on Google, whose AutomaticActivityDetection is extra='forbid'.

None keeps the provider default. An empty dict or {"disabled": True} asks for manual turn control.

Parameters:

Name	Type	Description	Default
turn_detection	Optional[Dict[str, Any]]	The neutral settings, or None.	required
provider	str	"openai" or "google".	required
logger	Optional[Logger]	Logger for the drop warnings (module logger by default).	None
model	str	Model name, used only in log messages.	''

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: For "openai", {"turn_detection": <dict or None>}
Dict[str, Any]	— the value that belongs under session.audio.input. For
Dict[str, Any]	"google", ``{"automatic_activity_detection": {...},
Dict[str, Any]	"realtime_input_config": {...}}`` — the two objects Gemini Live splits
Dict[str, Any]	the same settings across. Empty when turn_detection is None.

Example

translate_turn_detection({"threshold": 0.5}, "openai")
# {'turn_detection': {'type': 'server_vad', 'threshold': 0.5}}
translate_turn_detection({"threshold": 0.5}, "google")
# {'automatic_activity_detection': {'disabled': False},
#  'realtime_input_config': {}}   # plus a WARNING naming
#                                 # start_of_speech_sensitivity

Live Translation Sessions

doc_processing_toolkit.realtime_translation

Live translation sessions: speech in, translated speech and both transcripts out.

A translation session is not a conversation with a translating system prompt. There is no model turn, no tool, no voice to pick and no turn-taking: audio goes in continuously and translated audio comes back while the speaker is still talking. Both vendors say so in their own words — Google's guide contrasts "the model acts as an assistant" with "the model acts as an interpreter … uses continuous stream processing, translates as the speaker talks without waiting for turns", and OpenAI's contrasts "uses a conversation and response lifecycle" with "streams continuously from incoming audio … you don't call response.create".

That is why this is its own session kind rather than a preset:

session = await open_translation_session(
    "gemini-3.5-live-translate-preview", target_language="en"
)
try:
    async for chunk in japanese_microphone:      # 16 kHz PCM16
        await session.send_audio(chunk)
finally:
    await session.close()                        # flushes what is still in flight
print(session.source_transcript, "->", session.translated_transcript)

Termination is caller-driven, and that is the whole lifecycle contract. Neither protocol has a reliable end-of-stream event, so response_complete is emitted opportunistically where the provider happens to send something that means it (Gemini's turnComplete, OpenAI's session.closed) and must never be the only way a caller stops. You close the session; each adapter's close() then drains what is still arriving — OpenAI documents this explicitly ("closing the socket immediately can drop translated output still draining from the session") and Gemini needs it for the same reason. The drain is bounded by a wall-clock budget, so a socket that has gone silent cannot hang close().

Two transports, and only one of them is an exception to house style.

• Google (:class:GeminiLiveTranslateAdapter) is the same BidiGenerateContent socket and the same google-genai client as :class:~doc_processing_toolkit.realtime.GeminiLiveAdapter, with one extra config field (translation_config) and a much smaller config: no voice, no tools, no system instruction, no turn detection. It is token-priced under the existing audio_realtime schema, so its cost is exact.
• OpenAI (:class:OpenAITranslationAdapter) speaks /v1/realtime/translations over a raw WebSocket. This is the documented exception to the realtime module's "always go through the vendor SDK" rule (recorded in :mod:doc_processing_toolkit.realtime's own docstring): the installed openai SDK hardcodes /realtime in AsyncRealtime._prepare_url and declares no translation session type, so the endpoint is simply unreachable through it. No dependency is added — websockets is already a hard requirement of google-genai, which the toolkit pins. It is duration-priced, and every cost it produces is flagged estimated (see :class:OpenAITranslationAdapter).

Everything else is shared and neutral. target_language (validated locally against the registry row's own published list, before any socket opens), source_language accepted-but-ignored with one warning, the existing audio_delta / transcript_delta / error events, and the rule that the role distinguishes the two transcripts: role="user" is what the speaker said, role="assistant" is the translation. No new event type was needed for either provider.

See Also

• Multimedia concepts

GeminiLiveTranslateAdapter

Bases: _TranslationSessionMixin, GeminiLiveAdapter

Gemini live translation (gemini-3.5-live-translate-preview).

The same BidiGenerateContent socket, SDK client, 16 kHz PCM input and 24 kHz PCM output as :class:~doc_processing_toolkit.realtime.GeminiLiveAdapter, configured as an interpreter instead of an assistant. The whole transport difference is one config object::

types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    input_audio_transcription=types.AudioTranscriptionConfig(),
    output_audio_transcription=types.AudioTranscriptionConfig(),
    translation_config=types.TranslationConfig(
        target_language_code="en", echo_target_language=False
    ),
)

types.TranslationConfig has exactly two fields in google-genai 2.24.0 (target_language_code, echo_target_language) — there is no source language field, which is why source_language is accepted-but-ignored. The detected source language is reported back instead, on Transcription.language_code, and reaches callers as :attr:~doc_processing_toolkit.realtime.RealtimeEvent.language.

Both transcripts are neutral transcript_delta events told apart by role: role="user" is the source transcript, role="assistant" the translation. Translated speech arrives as audio_delta, exactly as in a voice conversation. There is no transcript_complete: a continuous interpreter has no items to complete, so the settled text is :attr:~_TranslationSessionMixin.source_transcript / :attr:~_TranslationSessionMixin.translated_transcript at any moment.

Four things live runs on the Live sockets proved, all handled here.

1. A native-audio Live model rejects response_modalities=["TEXT"] outright (close 1007). This session is AUDIO-only by construction and takes its text from the transcription channels, and modalities= is refused locally rather than sent.
2. The interim transcription channel carries a cumulative snapshot of the current utterance, not a fragment, and Transcription.finished was never True. Both shapes are handled by :meth:~_TranslationSessionMixin._advance_channel, and nothing waits for finished.
3. Usage arrives as one usage_metadata per second of input audio, each carrying that second's increment — 25 audio tokens in and 25 out, Google's published conversion — so the reports are summed. The first one lands three to four seconds into the stream and the last second or two before :meth:close may never be reported, so a session's reported usage trails the audio slightly; the provider's numbers are billed as they stand rather than topped up. prompt_tokens_details also lists a growing TEXT share (535, 553, 577 … in the live run) that is not part of prompt_token_count and is not billed. Until the first report arrives — and for a session that closes before any does, which a sub-second stream will — usage is estimated from the audio sent and received at the row's input_audio_tokens_per_second / output_audio_tokens_per_second, with :attr:~_TranslationSessionMixin.usage_estimated and cost_estimated set. The estimate is silent while the session runs and warns once at close if no report ever came. A real report replaces it outright.
4. Transcripts advance only while input audio keeps arriving. A caller that streams a clip and then stops sending loses the tail of both transcripts (the translated audio still completes) and the last usage report. Follow a clip with a second or two of silence, exactly as a microphone would.

Preview, and never a default. The registry row is labelled release_stage: "preview", publishes no session cap because Google publishes none for this model, and declares no session resumption — so the inherited reconnect refuses rather than silently opening an unrelated session mid-sentence. (The live server does send resumable session_resumption_update handles unasked; the row stays false until a resumed interpreter session has actually been exercised.)

Example

session = await open_translation_session(
    "gemini-3.5-live-translate-preview", target_language="en"
)
try:
    await session.send_audio(japanese_pcm_16k)
    async for event in session.events():
        if event.type == "audio_delta":
            speaker.write(event.audio)
finally:
    await session.close()

billed_audio_seconds property

The audio duration a per_minute row is billed on.

Precedence, the same rule file transcription follows: the provider's own reported seconds when it reported any (a transcription session reports them per completed item), otherwise the session's connected wall clock. Provider first because a socket that was open for two minutes while thirty seconds of speech passed through it is billed for the speech, not the silence — and only the provider knows which.

Returns:

Name	Type	Description
float	float	Seconds.

capabilities property

Provider capabilities and toolkit features for this model.

See :func:realtime_capabilities — this is the same report, for the session's own model.

compression_policy property

What extends this session, stated without conflating two mechanisms.

resumption reconnects a dropped transport and replays the handle; it does not enlarge the context window. compression is the separate context-window policy. A provider's session-duration caps are reported alongside so a caller can see which limit it is up against.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"resumption", "compression", "limits"}.

connected property writable

Whether a transport is currently established.

Assigning to this attribute also drives :attr:session_duration_seconds: setting it True starts a billing segment and setting it False closes one. Adapters keep assigning it exactly as before.

emitted_audio_ms property

Milliseconds of assistant audio this session has emitted for the current item.

This is the only playback figure the toolkit can know: it counts the PCM the adapter handed to the caller, not what the caller's speaker actually played. A caller that buffers should pass its own figure to :meth:truncate_response.

input_sample_rate property

Sample rate the session expects, from the registry row (16 kHz).

manual_turn_control property

Whether automatic activity detection is off for this session.

True when turn_detection was {} or {"disabled": True} (the neutral spelling of manual turn control). The client then marks every user turn itself: :meth:send_audio opens it with activity_start and :meth:commit_audio closes it with activity_end.

max_video_fps property

The video-frame ceiling in force, in frames per second.

Comes from configure(max_video_fps=...), else the registry row's max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS. 0 means the gate is off.

output_sample_rate property

Sample rate the translated audio arrives at, from the row (24 kHz).

session_duration_seconds property

Billable wall-clock seconds this session has been connected.

Measured from the transport, not from the audio frames: a realtime provider that bills per minute bills connection time, and a frame-derived figure would need a per-provider sample-rate assumption and would under-count the silence the provider still charges for. Gemini Live's reconnects (the hidden session-duration cap) close one segment and open another, so the gap between them is excluded rather than billed.

Returns:

Name	Type	Description
float	float	Seconds, accumulated across every connected segment and
	float	including the segment currently open.

Example

async with await open_realtime_session("gpt-realtime") as session:
    ...
print(session.session_duration_seconds, session.cost)

target_language property

The BCP-47 code this session translates into, in the row's spelling.

tool_call_behavior property

The function-calling behaviour in force: BLOCKING or NON_BLOCKING.

tool_dispatcher property

The session's tool dispatcher, or None when no tools were given.

__aenter__() async

Connect if not already connected.

__aexit__(exc_type, exc, tb) async

Always close the transport, even on an exception.

__init__(model, **kwargs)

See :class:~doc_processing_toolkit.realtime.RealtimeSession for the shared arguments.

cancel_response() async

Interrupt generation with an empty completed client turn.

Gemini Live has no cancel message; the documented control is that turn_complete=true unconditionally interrupts generation, which is only available on rows that accept send_client_content mid-session.

Raises:

Type	Description
RealtimeValidationError	The model has no explicit interrupt (send audio, or let voice-activity detection barge in, instead).

capability(name)

Read one registry capability, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:REALTIME_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The registry row's value, or the default for an absent key.

close() async

Flush the audio stream, drain what it releases, then close.

Stop iterating :meth:events before calling this: the drain reads the same socket, and two consumers would race for the last frames.

commit_audio(create_response=True) async

Tell the server the input audio stream has ended.

send_realtime_input(audio_stream_end=True) is what makes Gemini Live settle the last utterance: a live run on the sibling captions model showed the settled transcript arriving only after this signal, with Transcription.finished never set. :meth:close sends it for you, so this is only needed to settle a stretch mid-session.

Parameters:

Name	Type	Description	Default
create_response	bool	Ignored and logged. A translation session has no response to create.	True

configure(*, target_language=None, target_languages=None, source_language=None, echo_target_language=None, **kwargs)

Validate and stage the translation options, then the shared ones.

Parameters:

Name	Type	Description	Default
target_language	Optional[str]	BCP-47 code to translate into, checked against the registry row's supported_target_languages. Matching is case-insensitive and the row's own spelling is what goes on the wire.	None
target_languages	Optional[Sequence[str]]	The plural form, accepted only where the row publishes supports_multi_target_translation. A one-element list is always accepted as a synonym for target_language; more than one is refused naming the one-session-per-language rule both vendors document.	None
source_language	Optional[str]	Accepted and ignored, with one warning. Neither provider exposes a source-language field: Google's TranslationConfig has exactly two fields and OpenAI's session.update schema declares none. The source language is auto-detected.	None
echo_target_language	Optional[bool]	Google only. True parrots input audio already in the target language; False (the API default) stays silent for it. Refused on a row that does not publish supports_echo_target_language.	None
**kwargs	Any	Any :meth:~doc_processing_toolkit.realtime.RealtimeSession.configure option the interpreter protocol has a place for.	{}

Raises:

Type	Description
RealtimeValidationError	An unsupported language, more than one target language, or a conversational option (:data:TRANSLATION_REJECTED_OPTIONS).

connect() async

Open the interpreter socket, target language validated first.

Returns:

Name	Type	Description
GeminiLiveTranslateAdapter	'GeminiLiveTranslateAdapter'	This session, connected.

Raises:

Type	Description
RealtimeValidationError	No target language was named, or the one named is not in the row's published list. Raised before the socket opens.

events() async

Yield neutral events, reconnecting transparently across the session cap.

Yields:

Name	Type	Description
RealtimeEvent	AsyncIterator[RealtimeEvent]	Translated Live server messages. A session_resumed event marks each transparent reconnect that sent a resumption handle, and session_restarted one that had no handle to send (the new session does not remember the conversation); the stream itself never breaks for either.

export()

This session's record as JSON-clean data (session-portability slice D).

Read-only and available at any time; final after :meth:close. The transcript is the text of every completed turn, in completion order, and is the shape provider_options={"initial_items": ...} (an OpenAI Realtime or Gemini Live session) and GPT-Live's provider_options={"initial_input": ...} take to seed the next session: [{"role": e["role"], "text": e["text"]} for e in exported["transcript"]] (content in place of text for GPT-Live). A transcription session exports its completed captions the same way. A translation session and a GPT-Live session retain no transcript by design (an interpreter's two transcripts are on source_transcript / translated_transcript and its events; GPT-Live's text is on its own event stream) and export usage and cost with an empty list. closed_at and close_reason are set by every family's :meth:close; on GPT-Live close_reason is the provider's session.closed reason when one arrived. TTS and STT are not sessions.

Returns:

Type	Description
Dict[str, Any]	``{"schema_version": 1, "provider", "model", "session_kind",
Dict[str, Any]	"started_at", "closed_at", "close_reason", "duration_seconds",
Dict[str, Any]	"usage", "cost", "cost_estimated", "cost_components",
Dict[str, Any]	"transcript"}``.

Example

record = session.export()
json.dumps(record)                       # always JSON-clean
seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
next_session = open_realtime_session(model, provider_options={"initial_items": seed})

flush_pending_frame() async

Send the held video frame if the rate gate now allows it.

A capture loop that stops pushing would otherwise leave its last frame held forever, so the adapters also call this from :meth:send_audio — a live session sends audio continuously, which makes it a free heartbeat.

Returns:

Name	Type	Description
bool	bool	Whether a frame was actually sent.

refresh_cost()

Recompute :attr:RealtimeSession.cost now, estimating usage if needed.

A mid-session estimate is provisional and silent: a provider that reports usage on a lag (Gemini's first report arrives three to four seconds into the stream) has not failed to report it. Only :meth:close decides that no report ever came, and warns.

send_audio(pcm_bytes) async

Stream raw 16 kHz PCM16 audio to be interpreted.

Keep calling this continuously, including the silence between phrases: a translation session has no turns.

Parameters:

Name	Type	Description	Default
pcm_bytes	bytes	Little-endian 16-bit mono PCM at :attr:input_sample_rate.	required

send_image(image, *, media_type=None, detail=None) async

Send one still image as user input.

A still is a deliberate, discrete act — a photographed page, a chart, a screenshot to ask about — so it is never rate-gated: it goes on the wire immediately, like :meth:send_text. Use :meth:send_video_frame for a continuous capture.

The model must declare image (or video) among its registry input_types; a row that declares neither raises before anything is sent.

Parameters:

Name	Type	Description	Default
image	Any	Encoded image bytes (PNG/JPEG/GIF/WebP/BMP) or a base64 string of them. Raw pixel buffers are not accepted — encode the frame first.	required
media_type	Optional[str]	MIME type. Sniffed from the payload when omitted; a payload the toolkit cannot identify is rejected rather than guessed at.	None
detail	Optional[str]	OpenAI's input_image detail hint ("auto", "low", "high"). Ignored on Gemini Live, which has no equivalent.	None

Raises:

Type	Description
RealtimeValidationError	The model takes no visual input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

await session.send_image(png_bytes, detail="low")
await session.send_text("What does this chart show?")

send_text(text) async

Not available: a translation session has no text turn.

Raises:

Type	Description
RealtimeValidationError	Always. Both vendors state that only audio input is accepted for translation.

send_video_frame(frame, *, media_type=None) async

Offer one video frame to the model, subject to the rate gate.

Unlike :meth:send_image this is lossy on purpose. A capture loop pushing 30 fps into a model that re-reads every frame as billable image tokens would multiply the bill roughly thirtyfold for a picture that barely changed, so frames arriving faster than :attr:max_video_fps are coalesced: the newest frame is held and the one it superseded is counted in :attr:frames_dropped. Nothing is queued, so memory does not grow with the frame rate, and no frame ever delays audio.

Parameters:

Name	Type	Description	Default
frame	Any	Encoded image bytes, or a base64 string of them.	required
media_type	Optional[str]	MIME type; sniffed when omitted.	None

Returns:

Name	Type	Description
bool	bool	Whether this frame went on the wire now. False means it is
	bool	held and will be sent by the next :meth:send_audio,
	bool	meth:send_video_frame or :meth:flush_pending_frame once the
	bool	gate opens — or superseded by a newer frame first.

Raises:

Type	Description
RealtimeValidationError	The model takes no video input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

async for jpeg in camera:                  # 30 fps
    await session.send_video_frame(jpeg)   # ~1 fps reaches the model
print(session.frames_sent, session.frames_dropped)

truncate_response(*, item_id=None, played_ms=None) async

Tell the provider how much of its last answer the caller actually heard.

Barge-in leaves the provider believing it spoke its whole answer while the caller stopped the playback early. Truncation rewrites the stored turn to the audible prefix so the next turn is grounded in what was really heard.

Parameters:

Name	Type	Description	Default
item_id	Optional[str]	The item to truncate; the adapter's last assistant item by default.	None
played_ms	Optional[int]	Milliseconds actually played; :attr:emitted_audio_ms by default.	None

Raises:

Type	Description
RealtimeValidationError	The provider has no truncation control.

OpenAITranslationAdapter

Bases: _TranslationSessionMixin, RealtimeSession

OpenAI live translation over a raw WebSocket to /v1/realtime/translations.

This class is the realtime module's one documented exception to "always go through the vendor SDK". The endpoint is not the conversational /v1/realtime socket, the event names share no prefix with the conversational protocol, and the installed openai SDK (3.16.2) hardcodes /realtime in AsyncRealtime._prepare_url and defines no Literal["translation"] session type anywhere in openai.types.realtime. The exception is confined to this class and adds no dependency: websockets is already a hard requirement of google-genai.

The protocol, in full:

Direction	Event	What it carries
out	session.update	audio.output.language, optional audio.input.transcription / noise_reduction
out	session.input_audio_buffer.append	base64 24 kHz PCM16 mono
out	session.close	flush and drain, then session.closed
in	session.created / session.updated	the resolved session config
in	session.output_audio.delta	base64 PCM16 translated speech
in	session.output_transcript.delta	append-only target-language text
in	session.input_transcript.delta	append-only source text (opt-in)
in	error	recoverable by default — the session stays open
in	session.closed	the drain is finished

There is no response.create, no conversation and no turn.

Cost is an estimate, and says so. The model page states the row is "priced by audio duration rather than text tokens" but no OpenAI page says which duration — input audio, output audio or session wall clock. The toolkit bills the input audio it actually appended (:attr:appended_audio_seconds, the only duration measurable client-side), falls back to connected wall clock when nothing was appended, and sets cost_estimated on every cost this row produces through the registry's cost_basis_estimated flag.

Source transcripts are opt-in and may be billed twice. Passing input_transcription=True configures a second model (gpt-realtime-whisper, itself a $0.017/minute row) as audio.input.transcription.model. No OpenAI page says whether that is billed on top of the translation minute, so the toolkit warns once and does not add a second cost line it cannot source.

Example

session = await open_translation_session(
    "gpt-realtime-translate", target_language="es", input_transcription=True
)
try:
    async for chunk in pcm_24k_chunks:
        await session.send_audio(chunk)
finally:
    await session.close()
print(session.translated_transcript, session.cost, session.cost_estimated)

billed_audio_seconds property

The audio duration this per_minute row is billed on.

Precedence: the provider's own reported seconds if it ever reports any (this protocol never does), then the input audio actually appended, then connected wall clock. Appended audio beats wall clock because a caller feeding a recorded clip as fast as the socket accepts it has a wall clock of two seconds and thirty seconds of audio, and the provider bills the audio.

Returns:

Name	Type	Description
float	float	Seconds.

capabilities property

Provider capabilities and toolkit features for this model.

See :func:realtime_capabilities — this is the same report, for the session's own model.

compression_policy property

What extends this session, stated without conflating two mechanisms.

resumption reconnects a dropped transport and replays the handle; it does not enlarge the context window. compression is the separate context-window policy. A provider's session-duration caps are reported alongside so a caller can see which limit it is up against.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"resumption", "compression", "limits"}.

connected property writable

Whether a transport is currently established.

Assigning to this attribute also drives :attr:session_duration_seconds: setting it True starts a billing segment and setting it False closes one. Adapters keep assigning it exactly as before.

emitted_audio_ms property

Milliseconds of assistant audio this session has emitted for the current item.

This is the only playback figure the toolkit can know: it counts the PCM the adapter handed to the caller, not what the caller's speaker actually played. A caller that buffers should pass its own figure to :meth:truncate_response.

input_sample_rate property

Sample rate the session expects, from the registry row.

max_video_fps property

The video-frame ceiling in force, in frames per second.

Comes from configure(max_video_fps=...), else the registry row's max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS. 0 means the gate is off.

session_duration_seconds property

Billable wall-clock seconds this session has been connected.

Measured from the transport, not from the audio frames: a realtime provider that bills per minute bills connection time, and a frame-derived figure would need a per-provider sample-rate assumption and would under-count the silence the provider still charges for. Gemini Live's reconnects (the hidden session-duration cap) close one segment and open another, so the gap between them is excluded rather than billed.

Returns:

Name	Type	Description
float	float	Seconds, accumulated across every connected segment and
	float	including the segment currently open.

Example

async with await open_realtime_session("gpt-realtime") as session:
    ...
print(session.session_duration_seconds, session.cost)

target_language property

The BCP-47 code this session translates into, in the row's spelling.

tool_call_behavior property

The function-calling behaviour in force: BLOCKING or NON_BLOCKING.

tool_dispatcher property

The session's tool dispatcher, or None when no tools were given.

__aenter__() async

Connect if not already connected.

__aexit__(exc_type, exc, tb) async

Always close the transport, even on an exception.

__init__(model, **kwargs)

See :class:~doc_processing_toolkit.realtime.RealtimeSession for the shared arguments.

cancel_response() async

Interrupt the model's in-flight response.

Raises:

Type	Description
RealtimeValidationError	The model or adapter cannot interrupt explicitly (the message names what it can do instead).

capability(name)

Read one registry capability, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:REALTIME_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The registry row's value, or the default for an absent key.

close() async

Flush, drain, then close the socket exactly once and finalize cost.

Stop iterating :meth:events before calling this: the drain reads the same socket, and two consumers would race for the last frames.

configure(*, noise_reduction=None, **kwargs)

Stage the translation options, plus OpenAI's input noise reduction.

Parameters:

Name	Type	Description	Default
noise_reduction	Optional[Any]	"near_field" (close-talking microphones), "far_field" (laptop / conference microphones), or False to disable it explicitly (sent as null). None leaves the provider default. The same vocabulary and validation as the OpenAI conversational and transcription sessions (:data:~doc_processing_toolkit.realtime.NOISE_REDUCTION_TYPES).	None
**kwargs	Any	Any :meth:_TranslationSessionMixin.configure option.	{}

Raises:

Type	Description
RealtimeValidationError	An unrecognized noise-reduction profile, or any option the interpreter protocol has no place for.

connect() async

Open the socket and send the initial session.update.

The target language is required and is resolved before the socket opens, so an unsupported language never costs a connection.

events() async

Yield neutral events until the session closes.

Yields:

Name	Type	Description
RealtimeEvent	AsyncIterator[RealtimeEvent]	audio_delta for translated speech, transcript_delta (role="assistant" translated, role="user" source), error for a recoverable provider error, and one response_complete when the server confirms session.closed.

export()

This session's record as JSON-clean data (session-portability slice D).

Read-only and available at any time; final after :meth:close. The transcript is the text of every completed turn, in completion order, and is the shape provider_options={"initial_items": ...} (an OpenAI Realtime or Gemini Live session) and GPT-Live's provider_options={"initial_input": ...} take to seed the next session: [{"role": e["role"], "text": e["text"]} for e in exported["transcript"]] (content in place of text for GPT-Live). A transcription session exports its completed captions the same way. A translation session and a GPT-Live session retain no transcript by design (an interpreter's two transcripts are on source_transcript / translated_transcript and its events; GPT-Live's text is on its own event stream) and export usage and cost with an empty list. closed_at and close_reason are set by every family's :meth:close; on GPT-Live close_reason is the provider's session.closed reason when one arrived. TTS and STT are not sessions.

Returns:

Type	Description
Dict[str, Any]	``{"schema_version": 1, "provider", "model", "session_kind",
Dict[str, Any]	"started_at", "closed_at", "close_reason", "duration_seconds",
Dict[str, Any]	"usage", "cost", "cost_estimated", "cost_components",
Dict[str, Any]	"transcript"}``.

Example

record = session.export()
json.dumps(record)                       # always JSON-clean
seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
next_session = open_realtime_session(model, provider_options={"initial_items": seed})

flush_pending_frame() async

Send the held video frame if the rate gate now allows it.

A capture loop that stops pushing would otherwise leave its last frame held forever, so the adapters also call this from :meth:send_audio — a live session sends audio continuously, which makes it a free heartbeat.

Returns:

Name	Type	Description
bool	bool	Whether a frame was actually sent.

refresh_cost()

Recompute :attr:RealtimeSession.cost now, estimating usage if needed.

A mid-session estimate is provisional and silent: a provider that reports usage on a lag (Gemini's first report arrives three to four seconds into the stream) has not failed to report it. Only :meth:close decides that no report ever came, and warns.

send_audio(pcm_bytes) async

Append 24 kHz PCM16 mono audio to the translation input buffer.

Keep calling this continuously, including the silence between phrases: a translation session has no turns, and OpenAI documents that a client which stops and later resumes has the resumed audio treated as contiguous with what came before rather than as a real pause. Chunks of 200 ms match the engine's frame size; a shorter chunk is buffered until a frame is full, a longer one is split into frames back to back.

Parameters:

Name	Type	Description	Default
pcm_bytes	bytes	Little-endian 16-bit mono PCM at :data:OPENAI_TRANSLATION_SAMPLE_RATE.	required

send_image(image, *, media_type=None, detail=None) async

Send one still image as user input.

A still is a deliberate, discrete act — a photographed page, a chart, a screenshot to ask about — so it is never rate-gated: it goes on the wire immediately, like :meth:send_text. Use :meth:send_video_frame for a continuous capture.

The model must declare image (or video) among its registry input_types; a row that declares neither raises before anything is sent.

Parameters:

Name	Type	Description	Default
image	Any	Encoded image bytes (PNG/JPEG/GIF/WebP/BMP) or a base64 string of them. Raw pixel buffers are not accepted — encode the frame first.	required
media_type	Optional[str]	MIME type. Sniffed from the payload when omitted; a payload the toolkit cannot identify is rejected rather than guessed at.	None
detail	Optional[str]	OpenAI's input_image detail hint ("auto", "low", "high"). Ignored on Gemini Live, which has no equivalent.	None

Raises:

Type	Description
RealtimeValidationError	The model takes no visual input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

await session.send_image(png_bytes, detail="low")
await session.send_text("What does this chart show?")

send_text(text) async

Not available: a translation session has no text turn.

Raises:

Type	Description
RealtimeValidationError	Always. Both vendors state that only audio input is accepted for translation.

send_video_frame(frame, *, media_type=None) async

Offer one video frame to the model, subject to the rate gate.

Unlike :meth:send_image this is lossy on purpose. A capture loop pushing 30 fps into a model that re-reads every frame as billable image tokens would multiply the bill roughly thirtyfold for a picture that barely changed, so frames arriving faster than :attr:max_video_fps are coalesced: the newest frame is held and the one it superseded is counted in :attr:frames_dropped. Nothing is queued, so memory does not grow with the frame rate, and no frame ever delays audio.

Parameters:

Name	Type	Description	Default
frame	Any	Encoded image bytes, or a base64 string of them.	required
media_type	Optional[str]	MIME type; sniffed when omitted.	None

Returns:

Name	Type	Description
bool	bool	Whether this frame went on the wire now. False means it is
	bool	held and will be sent by the next :meth:send_audio,
	bool	meth:send_video_frame or :meth:flush_pending_frame once the
	bool	gate opens — or superseded by a newer frame first.

Raises:

Type	Description
RealtimeValidationError	The model takes no video input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

async for jpeg in camera:                  # 30 fps
    await session.send_video_frame(jpeg)   # ~1 fps reaches the model
print(session.frames_sent, session.frames_dropped)

truncate_response(*, item_id=None, played_ms=None) async

Tell the provider how much of its last answer the caller actually heard.

Barge-in leaves the provider believing it spoke its whole answer while the caller stopped the playback early. Truncation rewrites the stored turn to the audible prefix so the next turn is grounded in what was really heard.

Parameters:

Name	Type	Description	Default
item_id	Optional[str]	The item to truncate; the adapter's last assistant item by default.	None
played_ms	Optional[int]	Milliseconds actually played; :attr:emitted_audio_ms by default.	None

Raises:

Type	Description
RealtimeValidationError	The provider has no truncation control.

normalize_target_language(code, supported)

Match a requested language code against a row's published list.

BCP-47 tags are case-insensitive, so "PT-br" is the same tag as "pt-BR"; the registry's own spelling is returned so the wire always carries the vendor's casing.

Parameters:

Name	Type	Description	Default
code	str	The requested BCP-47 code.	required
supported	Sequence[str]	The codes the registry row publishes.	required

Returns:

Type	Description
Optional[str]	Optional[str]: The row's spelling of the code, or None when the row
Optional[str]	does not publish it.

Example

normalize_target_language("PT-br", ["pt-BR", "pt-PT"])
# 'pt-BR'

GPT-Live Sessions

doc_processing_toolkit.realtime_gpt_live

GPT-Live sessions: OpenAI's /v1/live/* family, duration-billed.

This is a third session family, not a second OpenAI realtime adapter. GPT-Live shares no endpoint, no event name and no billing unit with the GA Realtime protocol :mod:doc_processing_toolkit.realtime speaks:

	Realtime GA	GPT-Live
socket	wss://api.openai.com/v1/realtime?model=…	wss://api.openai.com/v1/live/sessions, no query parameters
handshake	session.update whenever you like	session.start first, then wait for session.started
audio in	input_audio_buffer.append	session.input_audio.append
audio out	response.output_audio.delta	session.output_audio.delta
turn end	response.done	nothing — "GPT-Live has no corresponding event marking the end of each spoken response"
tools	conversation.item.create on the same model	response.item.create + response.create against a delegated Responses backend
billing	input/output tokens by modality	$0.05 per minute of ACTIVE session time as the provider reports it, per second, no token price at all

The adapter therefore subclasses :class:~doc_processing_toolkit.realtime.RealtimeSession directly rather than :class:~doc_processing_toolkit.realtime.OpenAIRealtimeAdapter, and everything it speaks goes through the vendor SDK's own Live client (client.live.connect(), connection.session.start(...), connection.session.input_audio.append(...), connection.response.item.create(...)), so no raw WebSocket code and no new dependency is involved.

Cost is duration, and the toolkit says which duration. The billed duration is the session's active time, which is the provider's number and not the socket's: a live run on 2026-09-21 held the socket open for 15.5 s and was billed 9.0 s. session.usage.updated and session.closed carry a cumulative usage.seconds snapshot — the SDK's own :class:openai.types.live.SessionUsage docstring says "Do not sum this value across usage events", so :attr:GPTLiveAdapter.provider_audio_seconds is assigned, never accumulated. Where the provider reported nothing the session falls back to its own connected wall clock and sets :attr:~doc_processing_toolkit.realtime.RealtimeSession.cost_estimated, because the cost guide says to "use the duration reported by the API instead of timing only the audio you play". A session that never got past session.start bills $0.00 exactly, with no floor and no warning — see :data:LIVE_MINIMUM_BILLABLE_MINUTES_KEY.

The backend is billed separately and this session cannot see it. With Responses delegation configured, "backend model and tool usage is charged separately"; :attr:GPTLiveAdapter.cost covers the voice duration only, the session reports cost_estimated=True for that reason, and :attr:GPTLiveAdapter.cost_excludes names it in words. The backend responses that actually ran are listed in :attr:GPTLiveAdapter.delegated_usage, one entry per response ID, so a caller can price them against that model's own rates.

Transcripts have no item IDs. session.input_transcript.delta and session.output_transcript.delta carry delta, start_ms and end_ms and nothing else — "Transcript deltas have no item ID or event that marks a completed conversational turn". :attr:RealtimeEvent.item_id is therefore None on every GPT-Live transcript event, deliberately: inventing one would look provider-issued. Correlation is role plus the timeline, which the session keeps for you in :attr:GPTLiveAdapter.transcript_fragments, :attr:GPTLiveAdapter.input_transcript and :attr:GPTLiveAdapter.output_transcript.

Out of this first slice: stored sessions and forking, recording download (GET /v1/live/sessions/{id}/content), SIP call control, the sideband /attach connection, and the WebRTC transport with its credited 15-second initialization charge.

Example

from doc_processing_toolkit.realtime_gpt_live import open_live_session

session = await open_live_session("gpt-live-1", voice="marin")
try:
    await session.send_audio(pcm_24k)
    async for event in session.events():
        if event.type == "audio_delta":
            speaker.write(event.audio)
finally:
    await session.close()
print(session.billed_audio_seconds, session.cost, session.cost_estimated)

See Also

• Realtime concepts

GPTLiveAdapter

Bases: RealtimeSession

A GPT-Live session over the openai SDK's client.live.connect().

The lifecycle is the one the WebSockets guide prescribes and nothing else: connect with no query parameters, send session.start as the first message, wait for session.started, stream audio continuously, then send session.close and keep receiving until session.closed arrives before releasing the socket. Every step is bounded, and the socket is closed in a finally whether or not the drain succeeds.

provider_options recognizes, in addition to any key merged into the session.start payload verbatim:

Key	Type	Default	Effect
api_key / base_url / websocket_base_url	str	—	Build a dedicated client instead of the process-cached one
max_retries	int	5	The SDK's transport-level reconnect budget
sample_rate	int	registry → 24000	audio/pcm rate; one format serves both directions and is immutable after startup
handshake_timeout_seconds	float	15.0	Ceiling on the wait for session.started
close_timeout_seconds	float	15.0	Ceiling on the drain to session.closed
initial_input	list	—	Startup text history (≤ 128 messages, roles developer / user / assistant)
backend_model	str	—	Enables Responses delegation. Required before tools= can be used
backend_instructions	str	—	The backend prompt, separate from the Live instructions
backend_tools	list	—	Extra delegation tools; each type is checked against the row's live_delegation_tool_types
web_search	bool	False	Append {"type": "web_search"} to the delegation tools
backend_tool_choice	str/dict	—	"auto" / "none" / "required" / a named function
backend_parallel_tool_calls	bool	—	Whether the backend may request several calls at once
backend_max_output_tokens	int	—	At least 16 when set
backend_reasoning	dict	—	{"effort": ..., "summary": ...} for the backend model
backend_text	dict	—	{"verbosity": ...} for the backend model
backend_service_tier	str	—	auto / default / flex / priority; changes backend billing only

store is refused: it turns on 30-day stored recordings and forking, and this slice implements neither, so accepting it would retain a caller's audio for a feature the toolkit cannot use.

Attributes:

Name	Type	Description
session_id	Optional[str]	The server-assigned Live session id, after the handshake.
close_reason	Optional[str]	The provider's session.closed reason when one arrived (close_requested / expired / content / remote_hangup / connection_lost), "closed" once :meth:close has run without one, None while the session is open.
finalization_confirmed		Whether session.closed actually arrived. A connection that closes without it "does not confirm successful finalization", so the final duration is the local wall clock and :attr:cost_estimated is set.
cost_excludes	List[str]	Human-readable reasons :attr:cost is not the whole bill. Empty on a session with no Responses delegation.
delegated_usage	List[Dict[str, Any]]	One raw backend usage payload per backend response id, in arrival order, deduplicated per the cost guide's "count each backend response once, using its response ID".
input_transcript		The caller's own speech, fragments joined in delivery order.
output_transcript		The model's speech, likewise.
transcript_fragments	List[LiveTranscriptFragment]	Every fragment with its role and timeline.

audio_format property

The audio encoding in force for both directions.

billed_audio_seconds property

The duration this session is billed on.

The provider's cumulative figure whenever it has reported one - including a reported zero. GPT-Live bills active session time, so an idle session legitimately reports 0 and costs nothing; reading that zero as "not reported" billed the socket's wall clock instead (found live). Only a session the provider never reported on falls back to its own connected wall clock, flagged estimated.

capabilities property

Provider capabilities and toolkit features for this model.

See :func:realtime_capabilities — this is the same report, for the session's own model.

compression_policy property

What extends this session, stated without conflating two mechanisms.

resumption reconnects a dropped transport and replays the handle; it does not enlarge the context window. compression is the separate context-window policy. A provider's session-duration caps are reported alongside so a caller can see which limit it is up against.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: {"resumption", "compression", "limits"}.

connected property writable

Whether a transport is currently established.

Assigning to this attribute also drives :attr:session_duration_seconds: setting it True starts a billing segment and setting it False closes one. Adapters keep assigning it exactly as before.

delegated_response_ids property

Every backend response this session saw, in arrival order.

delegation_mode property

"responses" when a backend model is configured, else "client".

Client delegation is the default and costs nothing beyond the voice duration: the Live model asks the application to do the work, which is what :attr:cost can then account for in full.

emitted_audio_ms property

Milliseconds of assistant audio this session has emitted for the current item.

This is the only playback figure the toolkit can know: it counts the PCM the adapter handed to the caller, not what the caller's speaker actually played. A caller that buffers should pass its own figure to :meth:truncate_response.

max_video_fps property

The video-frame ceiling in force, in frames per second.

Comes from configure(max_video_fps=...), else the registry row's max_video_fps, else :data:DEFAULT_MAX_VIDEO_FPS. 0 means the gate is off.

sample_rate property

The sample rate in force for both directions, in hertz.

session_duration_seconds property

Billable wall-clock seconds this session has been connected.

Measured from the transport, not from the audio frames: a realtime provider that bills per minute bills connection time, and a frame-derived figure would need a per-provider sample-rate assumption and would under-count the silence the provider still charges for. Gemini Live's reconnects (the hidden session-duration cap) close one segment and open another, so the gap between them is excluded rather than billed.

Returns:

Name	Type	Description
float	float	Seconds, accumulated across every connected segment and
	float	including the segment currently open.

Example

async with await open_realtime_session("gpt-realtime") as session:
    ...
print(session.session_duration_seconds, session.cost)

tool_call_behavior property

The function-calling behaviour in force: BLOCKING or NON_BLOCKING.

tool_dispatcher property

The session's tool dispatcher, or None when no tools were given.

__aenter__() async

Connect if not already connected.

__aexit__(exc_type, exc, tb) async

Always close the transport, even on an exception.

__init__(model, **kwargs)

See :class:~doc_processing_toolkit.realtime.RealtimeSession.

Parameters:

Name	Type	Description	Default
model	str	A registered GPT-Live model (gpt-live-1).	required
**kwargs	Any	Every :class:~doc_processing_toolkit.realtime.RealtimeSession argument. Options GPT-Live cannot express are refused here, not at the API — see :data:LIVE_UNSUPPORTED_OPTIONS.	{}

append_commentary(content, *, delegation_id=None) async

Give the model text it should paraphrase aloud.

This is how a client delegation is answered: run the work in your own application, then send the verified result back under the same delegation_id.

Parameters:

Name	Type	Description	Default
content	str	Plain text, capped at live_max_append_tokens tokens.	required
delegation_id	Optional[str]	The client delegation this answers, or None.	None

Raises:

Type	Description
RealtimeValidationError	content is empty.
RealtimeConnectionError	The session is not connected.

Example

async for event in session.events():
    if event.type == "delegation_created" and event.text == "client":
        await session.append_commentary(
            "The order shipped today.", delegation_id=event.tool_call_id
        )

append_instructions(content, *, delegation_id=None) async

Add conversation instructions to the running session.

Parameters:

Name	Type	Description	Default
content	str	Plain text, capped by the API at live_max_append_tokens (500) tokens per append.	required
delegation_id	Optional[str]	A known client delegation to attach this to, or None for general session context.	None

Raises:

Type	Description
RealtimeValidationError	content is empty.
RealtimeConnectionError	The session is not connected.

append_thinking(content, *, delegation_id=None) async

Give the model a fact or progress note it can use but must not speak.

Parameters:

Name	Type	Description	Default
content	str	Plain text, capped at live_max_append_tokens tokens.	required
delegation_id	Optional[str]	The client delegation this updates, or None.	None

Raises:

Type	Description
RealtimeValidationError	content is empty.
RealtimeConnectionError	The session is not connected.

cancel_response() async

Not available: GPT-Live interrupts itself.

Raises:

Type	Description
RealtimeValidationError	Always. The model is full duplex — it listens while it speaks and yields on its own — so there is no cancel command. To stop it from the application side, send append_instructions("Stop speaking …").

capability(name)

Read one registry capability, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:REALTIME_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The registry row's value, or the default for an absent key.

close() async

Ask the session to close, drain to session.closed, then let go.

The order is the documented one and every step is bounded: send session.close, keep receiving until the terminal event arrives (up to close_timeout_seconds), and only then release the transport. A socket that answers nothing cannot hang this method — the drain runs under asyncio.wait_for and the transport is torn down in a finally either way. If the terminal event never arrives the session says so: :attr:finalization_confirmed stays False, :attr:cost_excludes names it and :attr:cost_estimated is set.

commit_audio(create_response=True) async

Not available: GPT-Live commits nothing and is never asked to speak.

Parameters:

Name	Type	Description	Default
create_response	bool	Ignored.	True

Raises:

Type	Description
RealtimeValidationError	Always.

configure(**kwargs)

Stage session options, refusing the ones GPT-Live cannot express.

Identical to :meth:~doc_processing_toolkit.realtime.RealtimeSession.configure except that the options in :data:LIVE_UNSUPPORTED_OPTIONS are rejected at the call, before the shared implementation can drop one with a warning or reject it with a Realtime-shaped message. A caller migrating a conversational session gets one sentence naming what to do instead.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Any :meth:~doc_processing_toolkit.realtime.RealtimeSession.configure keyword.	{}

Raises:

Type	Description
RealtimeValidationError	An option this family cannot express, or any rejection the shared implementation makes.

Example

session.configure(instructions="Be concise.", voice="marin")

connect() async

Open the socket, send session.start and wait for session.started.

Returns:

Name	Type	Description
GPTLiveAdapter	'GPTLiveAdapter'	self, connected.

Raises:

Type	Description
RealtimeConnectionError	The socket could not be opened, the server answered session.start with an error (the provider's own message is quoted verbatim — this is what an account without gpt-live-1 entitlement looks like), or session.started did not arrive inside the handshake budget. In every case the socket is closed and the session is never marked connected, so it bills nothing.

events() async

Yield neutral events until the session closes.

There is no response_complete on this protocol. GPT-Live emits nothing marking the end of a spoken response, so the if event.type == "response_complete": break idiom that works on the conversational adapters never fires here. The terminal event is :data:EVENT_SESSION_CLOSED, which only arrives after :meth:close asks for it — a GPT-Live session ends when the application ends it.

Yields:

Name	Type	Description
RealtimeEvent	AsyncIterator[RealtimeEvent]	audio_delta, transcript_delta (item_id always None), text_delta, delegation_created, tool_call / tool_result, usage_updated, error and finally session_closed.

export()

This session's record as JSON-clean data (session-portability slice D).

Read-only and available at any time; final after :meth:close. The transcript is the text of every completed turn, in completion order, and is the shape provider_options={"initial_items": ...} (an OpenAI Realtime or Gemini Live session) and GPT-Live's provider_options={"initial_input": ...} take to seed the next session: [{"role": e["role"], "text": e["text"]} for e in exported["transcript"]] (content in place of text for GPT-Live). A transcription session exports its completed captions the same way. A translation session and a GPT-Live session retain no transcript by design (an interpreter's two transcripts are on source_transcript / translated_transcript and its events; GPT-Live's text is on its own event stream) and export usage and cost with an empty list. closed_at and close_reason are set by every family's :meth:close; on GPT-Live close_reason is the provider's session.closed reason when one arrived. TTS and STT are not sessions.

Returns:

Type	Description
Dict[str, Any]	``{"schema_version": 1, "provider", "model", "session_kind",
Dict[str, Any]	"started_at", "closed_at", "close_reason", "duration_seconds",
Dict[str, Any]	"usage", "cost", "cost_estimated", "cost_components",
Dict[str, Any]	"transcript"}``.

Example

record = session.export()
json.dumps(record)                       # always JSON-clean
seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
next_session = open_realtime_session(model, provider_options={"initial_items": seed})

flush_pending_frame() async

Send the held video frame if the rate gate now allows it.

A capture loop that stops pushing would otherwise leave its last frame held forever, so the adapters also call this from :meth:send_audio — a live session sends audio continuously, which makes it a free heartbeat.

Returns:

Name	Type	Description
bool	bool	Whether a frame was actually sent.

live_capability(name)

Read one GPT-Live registry key, falling back to its documented default.

Parameters:

Name	Type	Description	Default
name	str	A key of :data:LIVE_CAPABILITY_DEFAULTS.	required

Returns:

Name	Type	Description
Any	Any	The descriptions.<model> value, or the default.

mute_input() async

Stop sending the caller's audio to the model.

Muting does not pause billing: active session time includes time when the user speaks, the assistant speaks, both are silent, or the backend is working.

Raises:

Type	Description
RealtimeConnectionError	The session is not connected.

send_audio(pcm_bytes) async

Append raw audio to the session, base64-encoded.

For audio/pcm a trailing odd byte is carried into the next chunk rather than rejected — "PCM chunks must contain complete 16-bit samples, so their byte length must be even", while chunk boundaries are otherwise arbitrary, so splitting a sample across two appends is the caller's buffering, not an error. G.711 is one byte per sample and needs no alignment.

Parameters:

Name	Type	Description	Default
pcm_bytes	bytes	Raw audio in the session's configured format at :attr:sample_rate — not a WAV or any other container.	required

Raises:

Type	Description
RealtimeConnectionError	The session is not connected.

send_image(image, *, media_type=None, detail=None) async

Send one still image as user input.

A still is a deliberate, discrete act — a photographed page, a chart, a screenshot to ask about — so it is never rate-gated: it goes on the wire immediately, like :meth:send_text. Use :meth:send_video_frame for a continuous capture.

The model must declare image (or video) among its registry input_types; a row that declares neither raises before anything is sent.

Parameters:

Name	Type	Description	Default
image	Any	Encoded image bytes (PNG/JPEG/GIF/WebP/BMP) or a base64 string of them. Raw pixel buffers are not accepted — encode the frame first.	required
media_type	Optional[str]	MIME type. Sniffed from the payload when omitted; a payload the toolkit cannot identify is rejected rather than guessed at.	None
detail	Optional[str]	OpenAI's input_image detail hint ("auto", "low", "high"). Ignored on Gemini Live, which has no equivalent.	None

Raises:

Type	Description
RealtimeValidationError	The model takes no visual input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

await session.send_image(png_bytes, detail="low")
await session.send_text("What does this chart show?")

send_text(text) async

Queue typed text for the delegated backend, not for the voice model.

GPT-Live's frontend has no user-text event: text reaches a session as startup input history, as an append (:meth:append_instructions / :meth:append_thinking / :meth:append_commentary), or — with Responses delegation — as a user message queued for the backend with response.item.create, which is what this does. The continuing response.create is deferred while a delegated response is still waiting for tool results, because "if it is waiting for function results, return all required results first".

Parameters:

Name	Type	Description	Default
text	str	The typed value, used by the backend as the exact text.	required

Raises:

Type	Description
RealtimeValidationError	The session uses client delegation, where there is no backend to queue for. The message names the three routes that do work.
RealtimeConnectionError	The session is not connected.

Example

await session.send_text("My order number is A0042.")

send_video_frame(frame, *, media_type=None) async

Offer one video frame to the model, subject to the rate gate.

Unlike :meth:send_image this is lossy on purpose. A capture loop pushing 30 fps into a model that re-reads every frame as billable image tokens would multiply the bill roughly thirtyfold for a picture that barely changed, so frames arriving faster than :attr:max_video_fps are coalesced: the newest frame is held and the one it superseded is counted in :attr:frames_dropped. Nothing is queued, so memory does not grow with the frame rate, and no frame ever delays audio.

Parameters:

Name	Type	Description	Default
frame	Any	Encoded image bytes, or a base64 string of them.	required
media_type	Optional[str]	MIME type; sniffed when omitted.	None

Returns:

Name	Type	Description
bool	bool	Whether this frame went on the wire now. False means it is
	bool	held and will be sent by the next :meth:send_audio,
	bool	meth:send_video_frame or :meth:flush_pending_frame once the
	bool	gate opens — or superseded by a newer frame first.

Raises:

Type	Description
RealtimeValidationError	The model takes no video input, or the payload is unidentifiable, oversized or the wrong type.
RealtimeConnectionError	The session is not connected.

Example

async for jpeg in camera:                  # 30 fps
    await session.send_video_frame(jpeg)   # ~1 fps reaches the model
print(session.frames_sent, session.frames_dropped)

truncate_response(*, item_id=None, played_ms=None) async

Tell the provider how much of its last answer the caller actually heard.

Barge-in leaves the provider believing it spoke its whole answer while the caller stopped the playback early. Truncation rewrites the stored turn to the audible prefix so the next turn is grounded in what was really heard.

Parameters:

Name	Type	Description	Default
item_id	Optional[str]	The item to truncate; the adapter's last assistant item by default.	None
played_ms	Optional[int]	Milliseconds actually played; :attr:emitted_audio_ms by default.	None

Raises:

Type	Description
RealtimeValidationError	The provider has no truncation control.

unmute_input() async

Resume sending the caller's audio to the model.

Raises:

Type	Description
RealtimeConnectionError	The session is not connected.

LiveTranscriptFragment dataclass

One transcript fragment, with the only correlation key GPT-Live gives.

A GPT-Live transcript delta carries no item ID and no turn-completed event, so a fragment is identified by who spoke and when, not by an item. The timeline is the session timeline: milliseconds from the start of the session, not from the start of a turn.

Attributes:

Name	Type	Description
role	str	"user" for the caller's own speech (session.input_transcript.delta), "assistant" for the model's (session.output_transcript.delta).
text	str	The fragment, exactly as the provider sent it. Fragments are appended in delivery order; a fragment is not a complete turn.
start_ms	int	Inclusive start on the session timeline.
end_ms	int	Exclusive end on the session timeline. Equal to start_ms for appended context.
event_id	str	The provider's event id for this delta, which is what makes a replayed fragment detectable.

Example

for fragment in session.transcript_fragments:
    print(fragment.role, fragment.start_ms, fragment.text)

open_live_session(model, **kwargs) async

Open a GPT-Live voice session.

Exactly open_realtime_session(model, session_kind="live", ...). It has its own name because a GPT-Live session is a different object to use — no audio commit, no response-complete event, tools that run against a delegated backend, and a bill measured in seconds rather than tokens — and burying that behind a keyword would make it look like a mode of the conversational session rather than a separate protocol.

Parameters:

Name	Type	Description	Default
model	str	A model serving the live session kind (gpt-live-1).	required
**kwargs	Any	Any :func:~doc_processing_toolkit.realtime.open_realtime_session argument except session_kind.	{}

Returns:

Name	Type	Description
RealtimeSession	RealtimeSession	A connected :class:GPTLiveAdapter.

Raises:

Type	Description
RealtimeValidationError	The model does not serve a live session, or an option GPT-Live cannot express was supplied.
RealtimeConnectionError	The handshake did not complete.

Example

session = await open_live_session(
    "gpt-live-1",
    voice="marin",
    instructions="Be concise.",
    provider_options={"backend_model": "gpt-5.6-luna"},
    tools=[look_up_order],
)