Agent Package API

Package

doc_processing_toolkit.agent

Agent abstraction layer for doc-processing-toolkit.

This package provides the high-level Agent class that wraps the raw toolkit functionality, providing a simplified interface for building AI agents with tool calling, session management, and cost tracking.

Main exports:

• Agent: The main agent class for running AI workflows
• AgentBuilder: Fluent builder for constructing an Agent
• AgentResult: Dataclass containing the result of an agent run
• AgentSession: Class for managing multi-turn conversations
• AgentStreamEvent: Dataclass for a single event yielded by Agent.stream()
• tool: Decorator for defining tool functions
• ToolLoader: Class for loading tools from various formats
• MediaTransformer: Class for transforming media inputs to provider formats
• DetailedCostReport: Class for detailed cost tracking
• CompactionConfig: Configuration for conversation history compaction

See Also:

- [Agent architecture and lifecycle](../concepts/agent.md)
- [Tool definition and registration](../concepts/tools.md)

Agent

High-level Agent class for AI interactions.

Attributes:

Name	Type	Description
model		The model name.
instructions		System prompt/instructions.

Notes

The Agent class provides:

• Provider detection from model name
• Unified parameter handling across providers
• Tool loading from multiple formats
• Media transformation
• Session management
• YAML configuration support

Example

from doc_processing_toolkit import Agent, tool

@tool
def calculate(a: int, b: int) -> int:
    '''Add two numbers.'''
    return a + b

agent = Agent(
    model="claude-sonnet-4-5",
    instructions="You are a helpful math assistant.",
    tools=[calculate],
)

result = await agent.run("What is 2 + 3?")
print(result.output)  # "The sum of 2 and 3 is 5."

__aenter__() async

Enter the async context manager.

__aexit__(exc_type, exc_val, exc_tb) async

Exit the async context manager, cleaning up resources.

__init__(model, instructions='', *, tools=None, api_key=None, max_iterations=10, max_tokens=None, temperature=None, thinking=None, built_in_tools=None, strict_tools=True, auto_inject_memory_instructions=True, auto_inject_skills=True, compaction=False, compaction_config=None, session_history_config=None, logger=None, log_level='DEBUG', log_file=None, log_to_file=True, tool_choice='auto', code_execution_output_dir=None, service_tier=None, hooks=None, emit_request_start=False, max_tool_result_chars=None, tool_result_preview_chars=DEFAULT_TOOL_RESULT_PREVIEW_CHARS, truncated_result_read_chars=None, prompt_caching=None, anthropic_block_binding=None, strict_prefix_binding=False, prefix_audit=False, prefix_strip_fallback=False, remove_citations=False, on_unsupported_part='drop', media_policy='auto', media_store=None, media_max_inline_bytes=DEFAULT_MEDIA_MAX_INLINE_BYTES, estimate_input_tokens='lazy', defer_tools=None, detailed_costs=False, cost_output_dir=None, **provider_kwargs)

Initialize an Agent.

Parameters:

Name	Type	Description	Default
model	str	Model name (e.g., "claude-sonnet-4-5", "gpt-4o", "gemini-2.5-flash", etc.)	required
instructions	str	System prompt/instructions.	''
tools	Optional[List[Any]]	List of tools in any supported format. The Agent accepts 9 tool formats (including directory paths): @tool decorated functions — Direct Python callables decorated with @tool from doc_processing_toolkit. File path strings — Paths to .py files containing @tool decorated functions. The agent dynamically imports them. Directory paths — String paths to directories. All .py files in the directory are scanned for @tool functions which are imported dynamically. FastMCP server objects — In-process MCP servers created with FastMCP("name"). Tools are discovered from the server. MCP script path dicts — {"type": "mcp_script", "path": "/path/to/server.py", "name": "server-name"}. Launches a local MCP server from a Python script. MCP command dicts — {"type": "mcp_command", "command": "npx", "args": ["-y", "@mcp/server"], "name": "server-name"}. Launches a local MCP server via a shell command (npx, uvx, etc.). Remote MCP URL dicts — {"type": "mcp", "server_url": "https://...", "authorization_token": "...", "name": "server-name"}. The provider connects to the remote MCP server (Anthropic, OpenAI and Fireworks only; tool calls never pass through this process). Pre-built session dicts — {"session": ClientSession, "name": "server-name"}. Uses an already-established MCP ClientSession. Client-side remote MCP dicts — {"type": "mcp_remote_client", "url": "https://.../mcp", "authorization_token": "...", "name": "server-name"}. This process connects over Streamable HTTP (or "transport": "sse") and executes the remote tools like local ones, so they work on every provider (including Google), take the function-calling path (compatible with output_schema), and are visible to hooks and tools_used. Optional keys: headers, auth (an httpx.Auth, e.g. an OAuth provider), timeout, sse_read_timeout, allowed_tools. The session is opened once per event loop and reused until :meth:close.	None
api_key	Optional[str]	Optional API key override for this Agent's provider. When supplied, this key is used instead of the provider's environment variable (ANTHROPIC_API_KEY, OPENAI_API_KEY, GOOGLE_GEMINI_API_KEY, FIREWORKS_API_KEY) and the Agent constructs its own client, bypassing the shared process-level client cache — per-key clients are never shared between Agents, and this Agent's close() closes them. Without it, clients come from the shared cache and are left open by close().	None
max_iterations	int	Maximum tool calling iterations (default: 10).	10
max_tokens	Optional[int]	Maximum output tokens per response.	None
temperature	Optional[float]	Sampling temperature (0.0 to 2.0).	None
thinking	Optional[Dict[str, Any]]	Thinking/reasoning configuration dict. Format varies by provider: Anthropic: {"budget_tokens": int, "interleaved": bool} or {"type": "adaptive"} for models supporting adaptive thinking. {"type": "disabled"} (or {"type": "between_tools"}, the same request) turns thinking off; what is sent comes from the model's registry row (thinking_off_type, thinking_off_max_effort), and a model that cannot turn thinking off runs adaptive with a warning. OpenAI: {"effort": "low"|"medium"|"high"|"xhigh", "summary": "auto"|"detailed"}. GPT-5.6 models additionally accept "mode": "standard"|"pro" — pro mode requires effort of at least "medium" (allowed values are registry-driven via supported_reasoning_modes / supported_reasoning_values_pro_mode in model_information.json; an unsupported effort in pro mode is dropped so the API default applies). Google: {"budget_tokens": int} or {"level": "minimal"|"low"|"medium"|"high"} Fireworks: {"effort": "low"|"medium"|"high"}	None
built_in_tools	Optional[Dict[str, Any]]	Built-in tool configuration dict. Keys: "web_search": BuiltInToolConfig(enabled=True, config={"max_uses": 10}) "code_execution": BuiltInToolConfig(enabled=True) "code_execution" with container reuse (Anthropic only): BuiltInToolConfig(enabled=True, config={"container_id": "<prior_id>"}) reuses an existing Anthropic code execution container so uploaded files and REPL state persist across runs (up to 30 days). Supported by: Anthropic, OpenAI, Google (not Fireworks).	None
strict_tools	bool	Enable strict schema validation (default: True). Anthropic only. Each request stays within Anthropic's per-request strict limits (20 strict tools, 24 optional properties, 16 union-typed parameters); a tool that does not fit is sent without strict mode, with one WARNING (DPT_ANTHROPIC_STRICT_BUDGET=off restores the per-tool check).	True
auto_inject_memory_instructions	bool	Auto-detect memory tools and inject memory instructions into the system prompt (default: True).	True
auto_inject_skills	bool	Detect a skills MCP server — one exposing list_skills and load_skill (the doc-processing-std-tools contract, 0.2+; the server's name does not matter) — call list_skills and inject the catalog (name, description, path) into the system prompt as <available_skills> (default: True). On an empty catalog the activation tools load_skill, load_skill_resource and execute_skill_script are hidden from the model. Skipped — no call, no injection, no hidden tools, one INFO line per run — while the agent's instructions already carry an <available_skills> block (an app that builds its own catalog). A failed or malformed list_skills reply leaves the activation tools advertised and logs one WARNING naming the server.	True
compaction	bool	Enable provider-agnostic context compaction (default: False). When True, the toolkit monitors conversation length and automatically compresses history when it approaches the model's context window limit.	False
compaction_config	Optional[Dict[str, Any]]	Configuration dict for compaction behaviour. Keys include: strategy (str): "summarize", "sliding_window", "tool_result_clearing", or "combined" (default "summarize"). trigger_threshold (float|int): Float 0.0–1.0 for percentage of context window, or int for absolute token count (default 0.8). keep_recent_messages (int): Messages to preserve verbatim in sliding_window/combined (default 5). keep_recent_tool_results (int): Tool results to keep in tool_result_clearing/combined (default 3). min_messages_before_compaction (int): Minimum messages before compaction can trigger (default 4). summary_model (str|None): Model for summarization (defaults to agent's model). summary_prompt (str|None): Inline custom summarization prompt. summary_prompt_file (str|None): File path to custom summarization prompt. archive_history (bool): Archive the original conversation before compacting (default True). archive_dir (str): Directory for archive files (default "./compaction_archives"). Ignored when archive_sink is set. archive_sink (Callable|None): Receives the archived conversation instead of a local file, for a host whose filesystem is ephemeral (default None). Called with the sanitised, JSON-serialisable message list; a string it returns is used wherever the file path would have been. An exception from it is logged at WARNING and never fails the run. Python only — a callable has no YAML spelling. include_archive_path_in_summary (bool): Include the archive's path (or the sink's locator) in summary output (default True). clear_tool_inputs (bool): Also clear tool call parameters (default False). combined_order (list[str]): Order of strategies for combined (default ["tool_result_clearing", "sliding_window", "summarize"]). max_compactions_per_session (int|None): Limit on compaction count (default None). summary_max_tokens (int): Maximum output tokens for the summarization LLM call (default 4096). Must be a positive integer and should not exceed the model's max_output_tokens_request. sliding_window_include_truncation_notice (bool): When True, the sliding_window strategy prepends a synthetic user message informing the model that earlier context was truncated. This helps the model understand that it may be missing prior conversation context. Defaults to True.	None
session_history_config	Optional[Dict[str, Any]]	Default session history configuration dict. Applied to all sessions created via agent.session() unless overridden per-session. Supported keys: mode (str): History filtering mode — "full", "messages_only", "no_tool_results", "no_reasoning", "compact_tool_results", or "text_summary" (default "full"). include_system_prompt (bool): Include system prompt in history (default True). compact_tool_results_max_chars (int): Max chars per tool result when mode="compact_tool_results" (default 500).	None
logger	Optional[Logger]	Optional pre-configured Logger instance. If provided, used as-is; otherwise one is created via _setup_logger.	None
log_level	str	Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL, NOTSET). Will be overwritten by the AGENT_LOG_LEVEL environment variable if set.	'DEBUG'
log_file	Optional[str]	Optional log file path. If not provided, defaults to ./logs/Agent-{model}_{timestamp}.log.	None
log_to_file	bool	Write the log file (default True). False — or the DPT_AGENT_LOG_TO_FILE=0 environment variable — makes the logger built when logger is None console-only: no file is opened, ./logs/ is not created, and log_file is ignored with a WARNING. Has no effect when logger is supplied.	True
tool_choice	str	Tool selection strategy: "auto", "required", "none", "any", "validated", or a specific tool name. Default: "auto" (Google default is "validated").	'auto'
code_execution_output_dir	Optional[str]	Output directory for the files a provider's code execution writes. None (the default) creates a fresh ./<provider>_code_execution_files_<timestamp>_<random>/ for each run that produces files (0.5.3; before, the timestamp alone, so two runs in one second shared a folder). An explicit directory is shared across runs by design, and a file an earlier run left there is overwritten as before. A name the run itself already wrote is written as name (2).ext and the record in downloaded_files carries the name written; AgentResult.files_created lists the paths written. Claude downloads on both of its paths since 0.5.3; until then the tool-calling path (local tools present) downloaded only when this directory or skills was set.	None
service_tier	Optional[str]	Requested service tier, e.g. "priority". On OpenAI this maps to Fast mode (2x token pricing, higher tokens/sec and priority handling); on Google to priority inference (1.8x pricing, non-sheddable processing). "flex" (half price, slower) and, on gpt-6-astra only, "ultrafast" (6x Standard, about three times Fast's output speed) are requested the same way. Ignored with a log line for providers/models without a matching service_tiers pricing row in model_information.json (the tier vocabulary is data-driven). Billing always follows the tier that actually SERVED each request: a request served at another tier is billed at the tier that ran and sets AgentResult.tier_downgraded.	None
hooks	Optional[Any]	Optional hooks object or callable for real-time steering (see :mod:doc_processing_toolkit.agent.hooks). Awaited at run/iteration boundaries and around every tool call; may veto tool calls (skip_tool), inject user-role messages at turn boundaries (inject_message), pause (resume via agent.run_handle.resume()), or abort (raises AgentAborted). Coexists with stream_callback: the callback is a passive tap, hooks are the control channel. Text/thinking deltas are not routed to hooks (use the callback for those) — awaiting a hook per token would stall streaming.	None
emit_request_start	bool	Emit a request_start event immediately before every ACTUAL provider send — the first request, each tool-loop iteration, each max_tokens / truncation / empty-turn re-request, each pause_turn continuation and each compaction summarisation call (B10). Default False, so a consumer that enumerates event types sees exactly the events earlier releases produced. When True the event goes to stream_callback on both public paths and, on the tool-calling path, to hooks as well — where it is OBSERVATION-ONLY: a directive other than continue is ignored with a WARNING, because a send point has no pending tool call to veto and no legal place to insert a message. metadata carries request_index (1-based, monotonic across the whole run), iteration, attempt (the tenacity attempt number), purpose, model and provider. run_start is unchanged and still fires once per run.	False
max_tool_result_chars	Optional[int]	Character cap on each tool result fed back to the model. None (the default) derives it from the registry: context_window_tokens * 4 * 0.25 — a quarter of the window at four characters per token, i.e. 200,000 characters on a 200k-token model and 1,000,000 on a 1M-token model (utils.TOOL_RESULT_CAP_CONTEXT_FRACTION / utils.TOOL_RESULT_CAP_CHARS_PER_TOKEN). A longer result is cut and ends with … [truncated: N of M characters shown; raise max_tool_result_chars to keep more], with one WARNING. Applies identically on all four providers to every result the toolkit executes (@tool functions and every client-side MCP format, mcp_remote_client included); results of provider-executed remote MCP servers never pass through the toolkit and are not capped. 0 turns the cap off (the kill switch — the previous behaviour), as does any value larger than a result can be. The raw result in hooks and stream_callback events is untouched.	None
tool_result_preview_chars	int	How much of each tool result the tool_call_result stream and hook events carry: the client's copy, which is independent of the model's copy that max_tool_result_chars cuts, so it never changes what the model is sent. 500 (the default) is the preview the events always carried. 0 carries the whole result, and adds result_data: the result as JSON data (a tool that returned a dict gives the dict; other MCP results give their content items, JSON text decoded; see utils.tool_result_data). Every event also carries result_truncated, so a client knows whether it holds everything. Keep the default where events are logged or stored: one result can be megabytes (DEBUG_AGENT_STREAMING mirrors every event into the log).	DEFAULT_TOOL_RESULT_PREVIEW_CHARS
truncated_result_read_chars	Optional[int]	How many more characters of a truncated tool result the model may read back, per result, beyond what it was first shown. None (the default) or 0 turns it off, and nothing changes. When set, the built-in read_truncated_result(tool_call_id, offset) tool is on the tool list from the first request of every run with client-side tools, and the marker on a result max_tool_result_chars cuts becomes … [truncated: N of M characters shown. Call read_truncated_result(tool_call_id="…", offset=N) to read up to K more.]. Each call returns the next slice, at most max_tool_result_chars long, ending with where to continue; every character returned counts against the allowance, and past it the tool answers with a short refusal. Truncated results are kept in full in memory for the run, or for an AgentSession's lifetime, and are never exported with a session. Has no effect with max_tool_result_chars=0 (nothing is cut) or a forcing tool_choice (the run ends after one tool round), and is not offered when a loaded tool is already named read_truncated_result (one WARNING for each).	None
prompt_caching	Optional[Any]	Opt-in Anthropic prompt caching. True or a dict {"ttl": "5m" | "1h"}. Places cache_control breakpoints on the system block (caching tools + system together) and on the most recent turn so each tool roundtrip reads the prior prefix from cache (reads bill at 0.1x input, writes at 1.25x for 5m / 2x for 1h — tracked via cached_input_tokens / cache_write_tokens). Prefixes below the model's minimum cacheable length (cache_min_prompt_tokens in the registry) are simply not cached by the API. Ignored for non-Anthropic providers (OpenAI/Google/Fireworks cache automatically server-side).	None
anthropic_block_binding	Optional[str]	How the API should treat a replayed thinking block whose conversation prefix no longer matches. "drop_block" has the server drop the stale blocks and report them in :attr:AgentResult.input_transformations; "error" fails the request; "off" sends nothing. None (the default) defers to DPT_ANTHROPIC_BLOCK_BINDING and then to the model's registry row. Four rows carry the registry flag (claude-fable-5-1, claude-opus-5-5, claude-sonnet-5-5, claude-haiku-5-5) and send "drop_block" plus the beta on every thinking request by default; on any other extended-thinking row "drop_block" is the opt-in that makes a model-binding drop (history produced by a model the serving one cannot read) visible in :attr:AgentResult.model_binding_mismatches. This setting has no effect on that drop itself: an unreadable block is always dropped. A caller-supplied thinking={"block_binding": ...} always wins. Anthropic-only.	None
strict_prefix_binding	bool	Sugar for anthropic_block_binding="error". Turn it on in CI so a history edit, an MCP server whose tool list moved, or a rebuilt instructions string fails loudly in test while production stays tolerant. Also makes a mid-session tool-surface change fatal rather than frozen. Anthropic-only (default: False).	False
prefix_audit	bool	Shadow mode. Fingerprints system, tools and every message before each Anthropic request and logs a WARNING naming what changed since the previous request — without altering a single request byte, on every model, with no beta header. The only mechanism that catches MCP tool-list drift at a client site. Also enabled process-wide by DPT_PREFIX_AUDIT (default: False).	False
prefix_strip_fallback	bool	For a client that cannot send the binding beta header. Strips all thinking and redacted_thinking blocks from the replayed history once per run, at the completed-turn boundary before the first request. The model loses that reasoning; the request never carries a stale binding. Anthropic-only (default: False).	False
remove_citations	bool	Remove citation markers from output (default: False).	False
on_unsupported_part	str	What to do with a prompt part the model cannot accept — an image sent to a text-only model, a video sent to a model without video input, a PDF sent to Fireworks (B12). "drop" (default): today's behaviour — log a warning and send the request without the part. "raise": refuse the run with UnsupportedPartError before any request is sent, so a confident answer about a document the model never received is impossible. Either way every dropped part is listed on AgentResult.dropped_parts.	'drop'
media_policy	str	How this agent's sessions write media when saved (to_dict / to_json; session-portability P2): "auto" (default) externalises when a media_store is configured and keeps media inline otherwise; "inline" writes every part exactly as stored; "externalize" hands every inline payload to the store and leaves a $media:<ref> marker; "strip" replaces every payload by a text placeholder. A per-call media= overrides it.	'auto'
media_store	Any	Where externalised bytes live — a directory path (a DirectoryMediaStore) or any object with put(sha256, mime_type, data) -> ref and get(ref) -> bytes. Inherited by every session of this agent; also the store a restore reads from unless from_dict / from_json are given one.	None
media_max_inline_bytes	Optional[int]	Per-part cap (8 MiB by default) on the bytes of a Google Files API upload that a session keeps beside the stored file_uri when no store is configured, so a restore after the 48-hour expiry can re-upload. Above the cap the bytes are not kept (a WARNING and a media_report entry); None lifts the cap. Inline parts on the other providers are never removed by the cap.	DEFAULT_MEDIA_MAX_INLINE_BYTES
estimate_input_tokens	Any	How the pre-flight input-token estimate is produced. The estimate seeds _estimated_input_tokens, which only the stream-failure cost record reads (B14). "lazy" (the default): a purely local approximation — no API call and, on Google, no Files API upload. Before this, every run uploaded each media part to count it and deleted it again, so a request with one image uploaded the same bytes twice (measured 4-5 s of that on the event loop per turn). True: the previous behaviour verbatim — the provider's own token-counting API, including that upload. False: no estimate at all (0), as the previous behaviour already did whenever the count raised.	'lazy'
defer_tools	Optional[Any]	Which client tools to hide behind the provider's native tool-search tool (C4), so a large tool surface stops consuming the cached prompt prefix. Either a set of tool names, or a predicate called as predicate(tool_name, server_name) where server_name is the MCP server the tool came from (None for a tool with no server). Requires built_in_tools={"tool_search": BuiltInToolConfig(enabled=True)} on a model whose registry row lists tool_search — Anthropic and OpenAI only. Nothing here can make a tool unreachable: if tool search is not enabled, or the provider has no native tool search (Google, Fireworks), every tool loads eagerly and one warning names why. The tools actually deferred are listed on AgentResult.deferred_tools. Example: agent = Agent( model="claude-haiku-4-5", tools=[big_mcp_server], built_in_tools={"tool_search": BuiltInToolConfig(enabled=True)}, defer_tools=lambda name, server: server == "catalog", )	None
detailed_costs	bool	Enable detailed cost tracking with per-iteration metrics.	False
cost_output_dir	Optional[str]	Directory for cost reports. Defaults to agent_costs. When detailed_costs=True, reports are saved as JSON files.	None
**provider_kwargs	Any	Provider-specific parameters passed directly to provider functions. A keyword this provider's entry points do not accept, or one the Agent supplies itself (system_prompt, client, logger and similar), is still passed through but logs one WARNING at construction naming it (with the Agent parameter to use instead where there is one, e.g. instructions= for system_prompt=). Supported parameters by provider: Anthropic: - effort_config (str): Computational effort — "low", "medium", "high", or "max" (Opus 4.6 only). - skills: Agent Skills for document tasks — "xlsx", "pptx", "docx", "pdf" or custom skill file paths. Supported formats: - Built-in skill name (str): "xlsx", "pptx", "docx", or "pdf" - Multiple built-in skills (List[str]): ["xlsx", "pptx"] - Custom skill directory path (str): "/path/to/custom_skill/" (must contain SKILL.md) - Mixed skills (List): ["/path/to/skill/", "xlsx", "/path/to/skill2/"] - Advanced dict format: {"type": "anthropic", "skill_id": "xlsx", "version": "20251013"} - max_pause_turn_iterations (int): Max iterations for pause_turn stop reason (default: 10). OpenAI: - verbosity_config (str): Output verbosity — "low", "medium", "high". - parallel_tool_calls (bool): Allow parallel tool calls (default: True). - image_detail (str): Image detail level applied uniformly to all images in the request (not per-image) — "low", "high", "auto", or "original" (gpt-5.4+ only). - ``openai_transport`` (str): ``"http"`` (the default) or ``"websocket"`` (0.5.2). ``"websocket"`` sends the requests of a run with local tools over one WebSocket to the Responses API, with the same results, billing and retries as HTTP. Unset, ``DPT_OPENAI_TRANSPORT`` is the process default. An unrecognised value is ignored with one warning, and a run with no local tools stays on HTTP. Google: - ``video_metadata`` (dict): Video preprocessing config — {"start_offset": str, "end_offset": str, "fps": float}. - ``media_resolution`` (str): Global media resolution setting. Supported values (model-dependent): - "MEDIA_RESOLUTION_UNSPECIFIED": Default setting (varies by model) - "MEDIA_RESOLUTION_LOW": Lower detail, faster processing, lower cost - "MEDIA_RESOLUTION_MEDIUM": Balance of detail, cost, and latency - "MEDIA_RESOLUTION_HIGH": Higher detail (recommended for most use cases) - ``file_active_timeout_s`` (float): Seconds to wait for a Files API upload to reach ``ACTIVE`` before the request proceeds. Applies to every media part. Defaults to 60 s for images and 120 s for video/audio/document, which a long MP4 exceeds — a ten-minute video needs roughly 600. Bounds 1-3600; out of range raises. Environment default: ``DPT_GOOGLE_FILE_ACTIVE_TIMEOUT_S``. - ``file_poll_interval_s`` (float): Seconds between state checks while waiting. Bounds 0.05-60, must not exceed the timeout. Environment default: ``DPT_GOOGLE_FILE_POLL_INTERVAL_S``. - ``file_poll_backoff`` (float): Multiplier applied to the poll interval after each check (bounds 1-10). ``1.0`` keeps the fixed cadence. Environment default: ``DPT_GOOGLE_FILE_POLL_BACKOFF``.	{}

Raises:

Type	Description
ValueError	If model is invalid or parameters are out of range.

builder() classmethod

Create an AgentBuilder for fluent configuration.

Returns:

Type	Description
AgentBuilder	AgentBuilder instance.

close() async

Close the agent and release all resources.

This method should be called when the agent is no longer needed to properly clean up MCP server connections, client sessions, and other resources.

Can be called explicitly or via the async context manager:

Example

# Explicit close
agent = Agent(model="claude-sonnet-4-5")
try:
    result = await agent.run("Hello")
finally:
    await agent.close()

# Context manager (recommended)
async with Agent(model="claude-sonnet-4-5") as agent:
    result = await agent.run("Hello")

Notes

MCP teardown is symmetric with the connect that run() performs: the sessions and the tools they registered are dropped, but the server configs are kept, so the Agent is reusable — the next run() reconnects them. Provider clients served from the shared process cache are deliberately left open; only an Agent built with an explicit api_key owns and closes its clients.

from_config(config_path, **overrides) classmethod

Create an Agent from a YAML configuration file.

Parameters:

Name	Type	Description	Default
config_path	str	Path to YAML configuration file.	required
**overrides	Any	Runtime parameter overrides. max_iterations, temperature and max_tokens replace the constraints values; any other key replaces the top-level YAML key of the same name. max_tool_result_chars, tool_result_preview_chars and truncated_result_read_chars are validated after the override is applied. logger is passed through. log_file and log_level apply only when the YAML has no logging block, and log_to_file only when that block has no to_file key.	{}

Returns:

Type	Description
Agent	Configured Agent instance.

Raises:

Type	Description
FileNotFoundError	If config file doesn't exist.
ConfigValidationError	If config is invalid.

Example

model: claude-sonnet-4-5
instructions: |
  You are a helpful assistant.

api_key: ${ANTHROPIC_API_KEY}  # Optional, uses env var by default
tool_choice: auto  # auto, required, none, any
max_tool_result_chars: 50000  # tool-result cap; 0 = off, omit = registry default
tool_result_preview_chars: 0  # events carry the whole result; omit = 500
truncated_result_read_chars: 200000  # read_truncated_result allowance; omit = off
code_execution_output_dir: ./output

constraints:
  max_iterations: 10
  max_tokens: 4096
  temperature: 0.7

thinking:
  budget_tokens: 10000
  interleaved: true

built_in_tools:
  web_search:
    enabled: true
    config:
      max_uses: 5

tools:
  - path/to/tools.py
  - type: mcp_remote
    url: https://api.example.com/mcp

logging:
  level: INFO
  file: agent.log
  to_file: true   # false: console only, no ./logs/ file

provider_overrides:
  effort_config: high

from_config_string(yaml_content, **overrides) classmethod

Create an Agent from a YAML string.

Parameters:

Name	Type	Description	Default
yaml_content	str	YAML configuration as a string.	required
**overrides	Any	Runtime parameter overrides. Applied exactly as in :meth:from_config: constraint keys replace constraints values, other keys replace top-level YAML keys, max_tool_result_chars, tool_result_preview_chars and truncated_result_read_chars are validated afterwards, and log_file / log_level / log_to_file yield to the YAML logging block.	{}

Returns:

Type	Description
Agent	Configured Agent instance.

Raises:

Type	Description
ConfigParseError	If YAML parsing fails.
ConfigValidationError	If config is invalid.

run(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None) async

Run the agent with a message.

Parameters:

Name	Type	Description	Default
message	str	The user message/task.	required
files	Optional[List[Any]]	Optional list of files to include with the message. Accepted formats: str — File path (e.g., "data.csv", "image.png"), URL, or YouTube URL. pathlib.Path — Path object pointing to a local file. bytes — Raw file bytes (media type auto-detected or set via kwargs). dict — Pre-structured content: {"type": "base64", "media_type": str, "data": str} or {"type": "document", "media_type": str, "data": bytes}. Supported media types vary by provider (images: all providers; PDF/text: Anthropic, OpenAI, Google; video/audio: Google only).	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool (where supported by Anthropic, OpenAI, and Google). Files should be in the same format as the files argument. File type validation is performed by the provider API. Files are uploaded automatically via the provider's Files API (Anthropic Files API, OpenAI Files API, Google Files API). Unsupported providers (e.g., Fireworks) are filtered automatically.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. The schema is used to VALIDATE the model's JSON output; the value placed on AgentResult.output is the parsed dict, not an instance of the schema class.	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). List of resolution strings matching the order of media parts in files. Use None for parts that should use the global/default resolution. Non-media parts will be ignored and don't need an entry in the list. Supported values (model-dependent): "MEDIA_RESOLUTION_UNSPECIFIED": Default setting (varies by model) "MEDIA_RESOLUTION_LOW": Lower detail, faster processing, lower cost "MEDIA_RESOLUTION_MEDIUM": Balance of detail, cost, and latency "MEDIA_RESOLUTION_HIGH": Higher detail (recommended for most use cases) "MEDIA_RESOLUTION_ULTRA_HIGH": Highest detail (Gemini 3 per-part only). Example: ["MEDIA_RESOLUTION_HIGH", "MEDIA_RESOLUTION_LOW", None] for a message with 3 media parts. Non-Google providers will gracefully ignore this parameter.	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse from a prior run. Files uploaded via tool_files and REPL state inside the container persist for up to 30 days and can be resumed by passing the container id returned in AgentResult.container_id. Auto-enables code_execution if the Agent did not declare it. Implicitly instructs the Anthropic provider to preserve (not delete) uploaded tool-file ids and custom-skill ids so they remain usable by the reused container on later runs. Ignored (with a debug log) for non-Anthropic providers.	None

Returns:

Type	Description
AgentResult	AgentResult with output and metadata.

run_sync(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None)

Synchronous wrapper for run().

This method handles the event loop internally, including nested event loop scenarios (e.g., Jupyter notebooks, async test frameworks).

Parameters:

Name	Type	Description	Default
message	str	The user message/task.	required
files	Optional[List[Any]]	Optional list of files to include with the message. Accepted formats: str — File path (e.g., "data.csv", "image.png"), URL, or YouTube URL. pathlib.Path — Path object pointing to a local file. bytes — Raw file bytes (media type auto-detected or set via kwargs). dict — Pre-structured content: {"type": "base64", "media_type": str, "data": str} or {"type": "document", "media_type": str, "data": bytes}. Supported media types vary by provider (images: all providers; PDF/text: Anthropic, OpenAI, Google; video/audio: Google only).	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool (where supported by Anthropic, OpenAI, and Google). Files should be in the same format as the files argument. File type validation is performed by the provider API. Files are uploaded automatically via the provider's Files API. Unsupported providers (e.g., Fireworks) are filtered automatically.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. As with run(), the schema validates the output and AgentResult.output is the parsed dict, not a model instance.	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). See run() for full documentation.	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse from a prior run. See run() for full documentation.	None

Returns:

Type	Description
AgentResult	AgentResult with output and metadata.

Notes

Local MCP servers (stdio scripts/commands, pre-built sessions) live for one event loop. asyncio.run creates and closes a loop per call, so each run_sync call connects them, runs, and releases them inside that loop. Use await run() on a long-lived loop to reuse one server process across runs until :meth:close.

session(session_id=None, session_history_config=None, reuse_anthropic_container=False, late_injection_policy='surface', max_follow_up_turns=1)

Create a new session for multi-turn conversations.

Parameters:

Name	Type	Description	Default
session_id	Optional[str]	Optional session ID. Auto-generated if not provided.	None
session_history_config	Optional[Dict[str, Any]]	Optional session history config dict. See AgentSession.__init__ for supported keys.	None
reuse_anthropic_container	bool	When True, capture the Anthropic code execution container_id from the first run and auto-inject it into every subsequent run in this session so uploaded files and REPL state persist. Ignored with a warning log for non-Anthropic providers. See AgentSession.__init__ for details.	False
late_injection_policy	str	What the session does with a hook message that arrived too late for any turn boundary to deliver it. "surface" (default) parks it for the host; "new_turn" sends it as the next turn automatically. See AgentSession.__init__.	'surface'
max_follow_up_turns	int	Automatic follow-up turns allowed per call under late_injection_policy="new_turn" (default 1).	1

Returns:

Type	Description
AgentSession	AgentSession instance.

Raises:

Type	Description
ValueError	If late_injection_policy or max_follow_up_turns is invalid.

stream(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None) async

Stream the agent's response as typed events.

Yields AgentStreamEvent objects in real time as the model generates output. Uses the same internal execution as run(), but emits deltas incrementally via the stream_callback mechanism.

Parameters:

Name	Type	Description	Default
message	str	The user message/task.	required
files	Optional[List[Any]]	Optional list of files to include with the message. Accepted formats: str — File path, URL, or YouTube URL. pathlib.Path — Path object pointing to a local file. bytes — Raw file bytes. dict — Pre-structured content dict.	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool (where supported by Anthropic, OpenAI, and Google). Files should be in the same format as the files argument. File type validation is performed by the provider API. Files are uploaded automatically via the provider's Files API. Unsupported providers (e.g., Fireworks) are filtered automatically.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. When supplied, text_delta events carry the streaming raw JSON payload; the final parsed dict is attached to the complete event's metadata["output"] (the schema is used for validation only — the value is a dict, not a model instance).	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). See run() for full documentation.	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse from a prior run. See run() for full documentation.	None

Yields:

Type	Description
AsyncIterator[AgentStreamEvent]	AgentStreamEvent objects with event_type and content/metadata.

Notes

Event types emitted by the stream. Seven are always available; two more (turn_start / turn_complete) appear only under AgentSession(late_injection_policy="new_turn"), and request_start only under Agent(emit_request_start=True).

• "text_delta": Partial text from the model. content is the text chunk. metadata is None for ordinary prose and {"structured_output": True} for the deltas of a schema-constrained (Path A) structured-output request, so a UI can suppress the terminal JSON without guessing. On Path B (tool-based structured output) the payload never reaches this event.
• "thinking_delta": Partial reasoning output when the provider exposes it.
• "tool_call_start": Tool call started; metadata may include tool_name, tool_call_id, arguments, and iteration.
• "tool_call_result": Tool call finished; metadata may include tool_name, tool_call_id, result, result_truncated, success, duration_ms, and iteration. result is the first tool_result_preview_chars characters (500 by default) of the result's str(); at 0 it is the whole result and result_data carries it as JSON data. A tool that reported spend with the _dpt_spend marker adds tool_reported_cost and tool_reported_cost_estimated (0.5.1).
• "iteration_complete": Iteration finished; metadata may include input_tokens, output_tokens, cost, tools_called, and stop_reason. The usage covers every request billed since the previous event, so the events sum to the run's requests up to the last event (0.5.1).
• "error": Unrecoverable error. content is the error message.
• "complete": Stream finished. metadata is the serialized AgentResult.
• "turn_start" / "turn_complete": Only under AgentSession(late_injection_policy="new_turn"). See :class:~doc_processing_toolkit.agent.cost.AgentStreamEvent.
• "request_start": Only when the Agent was built with emit_request_start=True. One event immediately before every ACTUAL provider send — including each retry, each max_tokens or truncation re-request, each pause_turn continuation and each compaction summarisation call — so the count matches len(result.request_cost_records) for a clean run. metadata carries request_index, iteration, attempt, purpose, model and provider. run_start is unchanged: it is a hook point, fires once per run, and is not a stream event.

Failures and aborts. Unlike run(), stream() never raises: a failed run yields an error event and then still terminates with a complete event, so teardown bound to complete runs exactly once per stream. The error event's metadata carries input_tokens, output_tokens, cost and iterations recovered from the failure (token_source is "stream_partial" when partial data was recovered, "unknown" otherwise), plus aborted (True when a hook aborted the run via HookDirective(action="abort") or RunHandle.abort) and reason (the hook's rationale, else None). The trailing complete event's metadata is an AgentResult dict with error set and the same partial usage — which is what lets AgentSession.stream() bank the spend of an aborted turn.

Stopping early. Closing the generator (aclose(), or contextlib.aclosing around a break) cancels the run within one chunk and waits at most :data:STREAM_CANCEL_WAIT_SECONDS for it. What the run had already spent is then reported as a complete-style record with "abandoned": True on Agent.last_abandoned_stream, to hooks (observation only) and, for AgentSession.stream(), into the session's totals. A bare break with the generator still referenced does not close it. A streamed run without local tools makes its provider call in a worker thread, so its deltas arrive as they are generated. DPT_STREAM_CANCEL_ON_CLOSE=off and DPT_STREAM_STRUCTURED_OFFLOOP=off restore the previous behaviour of each.

Examples:

Basic usage:

async for event in agent.stream("Hello"):
    if event.event_type == "text_delta":
        print(event.content, end="", flush=True)
    elif event.event_type == "complete":
        result = event.metadata  # Full AgentResult as dict

Stopping early, with the spend reported:

async with contextlib.aclosing(agent.stream("Write an essay")) as stream:
    async for event in stream:
        if event.event_type == "text_delta" and should_stop():
            break
spent = agent.last_abandoned_stream  # None if it completed

to_config()

The data half of this agent's constructor, in the YAML shape from_config reads.

Session-portability P6. A host stores the recipe beside a session envelope and rebuilds the same agent later with Agent.from_config_string(yaml.safe_dump(agent.to_config())) or by writing the dict to a YAML file for Agent.from_config. The dict is plain data: model, instructions, constraints, the tool and media policies, thinking, built-in tools, compaction, the session history mode, the Anthropic prefix controls, logging and provider_overrides (the extra provider keywords the agent was built with). tools carries every entry that is data — a path, a YAML tool dict, an MCP server spec — exactly as it was given.

What cannot be serialised is listed, never silently dropped, under the top-level unserialisable key: Python callables and objects given as tools (by name; since 0.5.1 also a pre-built {"session": ...} entry and the {"type": "fastmcp", "server": ...} dict form, which hold a live object), a hooks object, a custom media_store (a DirectoryMediaStore is emitted as its directory path), a logger instance, and whether an api_key was set (the key itself is never emitted). Since 0.5.3 the env of an mcp_command / mcp_script entry is treated the same way: an API key may be among its values, so the entry is emitted without env and the variable names are listed under unserialisable["tool_env"] ({server name: [names]}); a path-like cwd is emitted as a string. from_config ignores that key, so the recipe loads as it is; the host supplies those objects and values again.

Returns:

Type	Description
Dict[str, Any]	A JSON-clean, YAML-safe dict.

Example

import yaml

recipe = agent.to_config()
(checkpoints / "agent.yaml").write_text(yaml.safe_dump(recipe))
again = Agent.from_config_string(yaml.safe_dump(recipe), tools=[add])   # re-supply callables

AgentAborted

Bases: Exception

Raised when a hook directs the run to abort.

Attributes:

Name	Type	Description
reason		The hook-supplied rationale (may be None).
partial		Best-effort partial state at the abort point — {"input_tokens", "output_tokens", "cost", "iterations"}.
undelivered_injections	List[str]	Hook-injected texts that were still queued when the run ended and were therefore never shown to the model. Agent._execute_run populates this as the run unwinds, so an abort at the final turn boundary hands the text back instead of destroying it. Empty unless a hook queued a message that no drain reached.

AgentBuilder

Builder pattern for Agent construction.

Provides a fluent interface for configuring agents.

Example

agent = (
    Agent.builder()
    .model("claude-sonnet-4-5")
    .instructions("You are helpful.")
    .add_tool(my_tool)
    .with_thinking(budget_tokens=10000)
    .build()
)

add_tool(tool)

Add a tool.

anthropic_block_binding(behaviour='drop_block')

Set the Anthropic thinking block-binding behaviour.

Parameters:

Name	Type	Description	Default
behaviour	Optional[str]	"drop_block", "error", "off", or None to fall back to DPT_ANTHROPIC_BLOCK_BINDING and the registry.	'drop_block'

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

auto_inject_skills(enabled=True)

Enable/disable auto skills injection.

build()

Build and return the Agent.

instructions(instructions)

Set the system instructions.

max_iterations(n)

Set max iterations.

max_tokens(n)

Set max tokens.

media_max_inline_bytes(limit)

Set the per-part cap on Google upload bytes kept in the envelope (None lifts it).

Parameters:

Name	Type	Description	Default
limit	Optional[int]	See Agent(media_max_inline_bytes=...).	required

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

media_policy(policy='auto')

Set how the agent's sessions write media when saved (auto / inline / externalize / strip).

Parameters:

Name	Type	Description	Default
policy	str	See Agent(media_policy=...).	'auto'

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

media_store(store)

Set where externalised media bytes live: a directory path or a MediaStore.

Parameters:

Name	Type	Description	Default
store	Any	See Agent(media_store=...).	required

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

model(model_name)

Set the model name.

prefix_audit(enabled=True)

Log would-be prefix mismatches without changing any request byte.

Parameters:

Name	Type	Description	Default
enabled	bool	True turns the shadow audit on for this agent.	True

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

prefix_strip_fallback(enabled=True)

Strip all replayed thinking instead of sending the binding beta.

Parameters:

Name	Type	Description	Default
enabled	bool	True removes every thinking block from the history a run replays, at the completed-turn boundary.	True

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

prompt_caching(enabled=True)

Opt in to Anthropic prompt caching. Pass True for the default 5m TTL or a dict such as {"ttl": "1h"}; non-Anthropic providers cache automatically and ignore this.

service_tier(tier)

Request a service tier ("priority", "flex" or "ultrafast"): OpenAI Fast, Flex and Ultrafast mode / Google priority and flex inference; validated data-driven against the model's service_tiers pricing rows.

strict_prefix_binding(enabled=True)

Fail loudly on a thinking prefix mismatch instead of dropping blocks.

Parameters:

Name	Type	Description	Default
enabled	bool	True maps to anthropic_block_binding="error" and makes a mid-session tool-surface change fatal.	True

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

temperature(t)

Set temperature.

with_built_in_tools(**kwargs)

Configure built-in tools.

with_compaction(enabled=True, **config)

Configure compaction.

with_thinking(**kwargs)

Configure thinking/reasoning.

AgentHooks

Bases: Protocol

Protocol for hook objects: a single awaited entry point.

AgentResult dataclass

Result of an Agent.run() call.

Contains the output, token usage, cost, and metadata about the agent execution. Includes all unique output fields across all provider structured output and function calling functions.

Attributes:

Name	Type	Description
output	Any	The agent's response. Can be a string, or a dict parsed from the model's JSON output when output_schema was provided.
raw_response	Any	The raw response from the provider API (for debugging).
messages	List[Dict[str, Any]]	The conversation messages including tool calls and responses.
injected_messages	List[str]	The flat, ordered list of hook-injected texts this run's turn-boundary drains actually delivered to the model. Each consumed injection appears exactly once, in the order the hook queued it, and is also present inline in :attr:messages as an ordinary user-role message. Empty on every run without hooks.
undelivered_injections	List[str]	Hook-injected texts that arrived after the run's last turn boundary and were therefore never shown to the model — queued at the final iteration_complete, returned by the run_end hook itself, or queued at a forced-tool_choice exit that sits above the loop's drain. They are not in :attr:messages and not in :attr:injected_messages. The toolkit deliberately does not extend the run to deliver them (see docs/concepts/hooks.md); deliver them as the next turn's user message. Empty on every run without hooks.
media_uploads	List[Dict[str, Any]]	One record per Google Files API upload this run made (or reused from the in-process cache), in prompt order: {"part_index", "uri", "name", "mime_type", "size", "sha256", "uploaded_at", "expiration_time"}, all JSON-clean. An AgentSession reads it to keep the bytes and the expiry beside the stored file_uri, so a restore after the 48-hour expiry can re-upload. Empty on every other provider and on a Google run with no uploads.
dropped_parts	List[Dict[str, Any]]	Every prompt part this run's model could not accept, in prompt order, recorded unconditionally — under the default on_unsupported_part="drop" as well as under "raise" (B12). Each entry is {"kind", "index", "reason", "provider", "model"}, where kind is the input-type key the validator built ("video", "base64-application/pdf", …) and index is the part's position in user_prompt_parts. Empty on every run whose parts were all supported. Before this field the only evidence was a log line, so an image sent to a text-only model produced a confident answer about a picture the model never saw.
iterations	int	Number of tool-calling iterations performed.
tools_used	List[str]	List of custom MCP tool names that were called during execution.
built_in_tools_used	List[str]	List of built-in tool names CONFIGURED for the run (e.g., web_search, code_execution) — an echo of the request configuration, not evidence of use. See built_in_tools_invoked.
built_in_tools_invoked	List[str]	List of built-in tool names with evidence they actually RAN during this run (observed usage: searches counted, containers observed, grounding metadata, MCP call items). A provider-executed remote MCP call adds the neutral "remote_mcp" beside the tool's own name, on every provider. Empty when a configured tool was never invoked.
deferred_tools	List[str]	The client tool names actually stamped defer_loading: true on this run's wire tools array, in wire order (C4). Empty whenever deferral did not apply — no defer_tools, no tool_search built-in tool, a provider without a native tool search, or a name that matched nothing. So an empty list beside a non-empty defer_tools argument is the signal that tools loaded eagerly, and the reason is in the run's warnings. Anthropic and OpenAI only; the other two providers never populate it.
input_tokens	int	Total input tokens consumed across all iterations.
output_tokens	int	Total output tokens generated across all iterations.
reasoning_tokens	int	Thinking/reasoning tokens across the run. A SUBSET of :attr:output_tokens, never additional to it, on all four providers — Anthropic reports them inside output_tokens, OpenAI and Fireworks inside their completion total, and Google reports them separately but the toolkit folds them into output_tokens before pricing. They are therefore already billed and are surfaced for observability only. 0 when thinking was off or the provider reported no breakdown.
cached_input_tokens	int	Prompt-cache READ tokens across the run. A SUBSET of :attr:input_tokens, never additional to it — costing subtracts this share out of the input total and re-prices it at the cache-read rate. Reported by every provider.
cache_write_tokens	int	Prompt-cache WRITE tokens. On Anthropic they are billed on top of input at the write rate (1.25x for a 5m TTL, 2x for 1h); on OpenAI GPT-5.6+ they are a SUBSET of input_tokens, re-priced at 1.25x the uncached input rate. Zero for providers and models that report no writes.
request_cost_records	List[Dict[str, Any]]	One record per API request — input/output/cached tokens, served_model, requested/served tier and cost. The run's token cost is the SUM of these records, never a single costing of the summed tokens (tiered models overbill that way). Structured-output runs make one request, so their single record is synthesised from the run-level fields and carries synthesised_from_run_totals. Google records also carry served_tier_source ("body", "header" or "none"), which says how the served tier was established. An OpenAI tool-loop record carries estimated: True (0.5.2) when a failure event ended its request on the first attempt after part of the reply had streamed: its output tokens are a count of the streamed text, priced at standard rates with no input tokens, and :attr:cost_estimated is True.
tier_downgraded	bool	True when any request in the run was not served at the tier it asked for. For a "priority" request that means it ran and billed at standard rates — the silent downgrade this flag was named for. An "ultrafast" request served at "priority" or at standard is billed at that cheaper row. For a "flex" request the mismatch goes the other way: flex is the cheaper row, so being served "standard" bills MORE than asked. The flag is the same "you did not get the tier you asked for" signal either way; read requested_tier / served_tier on :attr:request_cost_records, or :attr:service_tier_served, for the direction. False on every run that requested no tier.
cost	float	Estimated cost in USD for this run: the request records, the built-in tool charges and :attr:tool_reported_cost.
cost_estimated	bool	Whether the cost is an estimate (True) or actual (False). Also True when a tool reported its spend as an estimate.
tool_costs_estimated	Dict[str, bool]	Dict mapping tool names to whether their costs are estimated: built-in tools, and every tool that reported spend.
tool_reported_cost	float	What the run's own tools reported spending through the _dpt_spend marker (the multimedia MCP tools, the agent factory's create_agent and suggest_agent_config, or any tool returning utils.tool_spend_marker(...)), in USD. Already included in :attr:cost. 0.0 with DPT_AGENT_TOOL_REPORTED_COST=off.
tool_reported_costs	List[Dict[str, Any]]	One record per reporting tool call: {"tool_name", "tool_call_id", "usd", "estimated", "iteration"}.
token_source	Optional[str]	Where the token counts came from. One documented vocabulary: "api_response" (the provider reported them), "stream_partial" (accumulated from a stream that then failed), "rejected_before_stream" (the provider refused the request before streaming, so nothing was billed), "estimated" and the "*_metadata_missing" variants (counted locally), or None. The legacy spelling "api_metadata", which the single-call provider paths emit for the same fact as "api_response", is normalized to "api_response" on the way in and still accepted everywhere it is read (log-review F12).
token_totals_by_model	Dict[str, Dict[str, int]]	Billed tokens attributed to the model that ran them, e.g. {"claude-opus-4-8": {"input_tokens": 412, "output_tokens": 264}}. Contains only the requested model unless Anthropic server-side fallback ran (Claude Fable 5), where attempts may be billed at different models' rates; a billed attempt that declined is included (0.5.2). Empty for providers/results that do not report it.
reasoning_content	Optional[str]	The model's thinking/reasoning text, kept OUT of :attr:output — replaying a thought summary as the assistant's answer is a defect, not a feature. What arrives differs by provider: Anthropic returns full thinking blocks; Fireworks returns reasoning_content (or <think> tags); Google and OpenAI return a summary only, and only because the toolkit opts in (include_thoughts=True / reasoning_config={"summary": ...}) — neither exposes the raw chain of thought. "" or None when the run produced none.
error	Optional[str]	Error message if the run failed, None on success.
run_id	Optional[str]	Unique identifier for this run (for cost tracking correlation).
iteration_metrics	List[IterationMetrics]	Per-request cost/token breakdown (if detailed_costs enabled). One entry per API request on all four providers; a round's first request and its follow-up share iteration_number and differ in purpose.
files_created	List[str]	List of file paths created during execution (e.g., by code execution).
distinct_urls	List[str]	List of unique URLs returned by web search (if used).
detailed_cost_file	Optional[str]	Path to detailed cost report file if created.
input_transformations	List[Dict[str, Any]]	Every input_transformations entry Anthropic returned across the run, each annotated with the 1-based request_index of the API request that reported it (Anthropic-specific). The server sends this array when it changed something about the request's input — for example dropping a preserved thinking block whose binding no longer matches. Entries whose type or reason the toolkit does not recognise are kept verbatim, because Anthropic adds values over time. Populated only when a request carried the thinking-binding-controls-2026-08-01 beta: every request of a thinking run on claude-fable-5-1, claude-opus-5-5, claude-sonnet-5-5 or claude-haiku-5-5 (the rows with enforces_thinking_prefix_check), or on any other extended-thinking row that opted in with anthropic_block_binding="drop_block". Empty otherwise, even when the API dropped blocks the serving model could not read.
prefix_binding_mismatches	int	How many entries in :attr:input_transformations had type == "thinking_dropped" and reason == "prefix_binding_mismatch" — the alarm. A thinking_mismatch_allowed entry (the block failed the prefix check but reached the model and was billed) is not counted. Each one is a turn whose preserved reasoning the server dropped because the conversation prefix it was bound to had been rewritten (compaction, a changed system prompt, a moved tool list). Under sliding_window these are expected on every window slide; the cost is reasoning quality on the retained turns, not a failed request. 0 on every non-Anthropic provider and on every model that does not run the check.
model_binding_mismatches	int	How many entries had type == "thinking_dropped" and reason == "model_binding_mismatch" — blocks the serving model could not read because another model produced them, so the API dropped them. Expected after a model downgrade or a server-side fallback and not a bug in the integration, so it is counted separately from the alarm above. Recorded only when the request carried the thinking-binding-controls-2026-08-01 beta (every thinking request on claude-fable-5-1 / claude-opus-5-5 / claude-sonnet-5-5 / claude-haiku-5-5, or a run with anthropic_block_binding="drop_block"): a client-side switch to an ungated row reports 0 although the API dropped the blocks. 0 on every non-Anthropic provider.
prefix_binding_retry	bool	True when the toolkit caught a prefix-binding 400 and re-sent the request once with prefix_mismatch_behavior="drop_block" plus the binding beta header. The visible form of "don't hide the 400": the retry is also logged at WARNING, and every later request of the run carries the same choice. On a thinking-off run on a model that checks the prefix the retry removes the thinking blocks from the request's history instead, because an off value cannot carry the field. False on a clean run and on every other provider.
skills_used	List[str]	List of skill IDs used (Anthropic-specific).
pause_turn_iterations	int	Number of pause_turn iterations if any occurred (Anthropic-specific).
container_id	Optional[str]	Anthropic code execution container id (Anthropic-specific). Surfaced for every Anthropic run that created or reused a container (plain code_execution or Skills). Pass this back on a later run via Agent.run(..., anthropic_container_id=...), a BuiltInToolConfig(..., config={"container_id": ...}), or by opening a session with reuse_anthropic_container=True to reuse the same container — files uploaded via tool_files and REPL state persist up to 30 days.
compaction_count	int	Number of compaction events during this run (across both Level 1 and Level 2).
compaction_input_tokens	int	Total input tokens consumed by summarization LLM calls.
compaction_output_tokens	int	Total output tokens from summarization LLM calls.
compaction_cost	float	Total cost in USD of summarization calls.
compaction_archives	List[str]	List of file paths to archived conversation histories.
retry_count	int	Total number of failed-and-retried API attempts during this run. Zero on a clean run. Populated from the provider-level RetryAccumulator wired through :meth:Agent.run.
retry_cost	float	USD cost of the failed retry attempts only — that is, the portion of :attr:cost attributable to attempts that did not produce the final response. cost always includes this value; subtract retry_cost from cost to get the cost of the successful (or final-exhaustion) attempt alone. Zero on a clean run.
retry_input_tokens	int	Input tokens consumed by the failed-and-retried attempts only (already included in :attr:input_tokens). Zero on a clean run.
retry_output_tokens	int	Output tokens produced by the failed-and-retried attempts only (already included in :attr:output_tokens). Zero on a clean run.

Examples:

Basic usage:

result = await agent.run("What is 2 + 2?")
print(result.output)  # "The answer is 4"
print(result.cost)    # 0.0003
print(result.tools_used)  # ["calculator"]

priority_request_count property

How many of this run's requests were served at the "priority" tier (C7).

Returns:

Name	Type	Description
int	int	The number of :attr:request_cost_records whose served_tier
	int	is "priority" — that is, the requests actually billed at the
	int	model's priority row. Zero on a run that never asked for a tier, and
	int	zero on a run whose priority requests were all downgraded.

service_tier_served property

The tier this run's requests were actually SERVED at (C7).

Derived from :attr:request_cost_records, because tier resolution is per request: a tool-calling run can have one request served at the requested tier and the next silently downgraded, and its cost is the sum of records priced individually.

Returns:

Type	Description
Optional[str]	str or None: "priority", "flex", "ultrafast", "standard" or
Optional[str]	"unverifiable" when every tier-requesting record agrees;
Optional[str]	"mixed" when they do not; None when no request asked for a
Optional[str]	tier at all — deliberately not "standard", because "never asked"
Optional[str]	and "asked and was refused" are different facts.
Optional[str]	"unverifiable" is Google-specific and means the toolkit could not
Optional[str]	read a served-tier signal, billed standard, and may therefore be
Optional[str]	UNDER-stating the cost (see docs/concepts/service_tiers.md).

Example

result = await agent.run("Summarize this filing")
if result.service_tier_served == "mixed":
    print(f"{result.priority_request_count} requests ran at priority")

success property

Whether the run completed successfully.

total_tokens property

Total billed tokens for the run: input_tokens + output_tokens.

:attr:reasoning_tokens is deliberately not added. It is a subset of :attr:output_tokens on every provider (see that attribute), so adding it counted the thinking share twice — the same trap the toolkit avoids for :attr:cached_input_tokens, which is a subset of :attr:input_tokens and is likewise not added here.

Returns:

Name	Type	Description
int	int	The run's input plus output tokens.

Example

result = agent.run("...")
assert result.total_tokens == result.input_tokens + result.output_tokens

__post_init__()

Normalize the token_source vocabulary (log-review F12).

The single-call provider paths report "api_metadata" and the loop/session paths "api_response" for the identical fact — "the provider reported these counts" — so a caller asserting on either one passed or failed depending on which path a run happened to take. The run-level surface now always says "api_response"; the old spelling is still ACCEPTED here and by :func:~doc_processing_toolkit.agent.cost.normalize_token_source, so a persisted result or an older provider release loads unchanged.

from_dict(data) classmethod

Create from dictionary representation.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary with result fields.	required

Returns:

Type	Description
AgentResult	AgentResult instance.

to_dict()

Convert to dictionary representation.

Returns:

Type	Description
Dict[str, Any]	Dictionary with all result fields, plus the two derived run-level
Dict[str, Any]	tier keys service_tier_served and priority_request_count
Dict[str, Any]	(properties, so asdict would otherwise drop them). They are
Dict[str, Any]	recomputed from request_cost_records by :meth:from_dict and are
Dict[str, Any]	therefore ignored on the way back in.

AgentSession

Stateful conversation session with message history.

AgentSession maintains conversation context across multiple runs, automatically passing previous messages to the agent for continuity.

Examples:

Synchronous context manager:

```python
with agent.session() as session:
    result1 = session.run_sync("Hello, I'm Alice")
    result2 = session.run_sync("What's my name?")  # Remembers "Alice"
```

Async context manager:

```python
async with agent.session() as session:
    result = await session.run("Hello!")
```

Attributes:

Name	Type	Description
agent		The Agent instance this session is bound to.
session_id	str	Unique identifier for this session.
messages	List[Dict[str, Any]]	Conversation message history. Holds the messages circulated to the provider (user turns, tool calls and tool results) and each turn's final assistant reply, which the provider loops never send anywhere and therefore do not record themselves — it is appended here so later turns and compaction can see it.
metadata	Dict[str, Any]	Session metadata accumulated across runs. Always carries created_at, total_tokens, total_cost, total_input_tokens, total_output_tokens, run_count, compaction_count and compaction_archives; the first run additionally populates total_cached_input_tokens, total_cache_write_tokens, total_retry_cost, total_retry_count, total_retry_input_tokens, total_retry_output_tokens and total_cost_inclusive. The retry totals are summed from each run's RUN-LEVEL retry_* fields; the per-iteration retry_* fields on iteration_metrics are reserved and always zero. total_cost is already retry-inclusive (it sums AgentResult.cost), so total_cost_inclusive simply equals it — the key is retained for backward compatibility.
restored_from_schema_version	Optional[int]	The schema_version of the envelope this session was restored from, or None on a session that was never restored. An envelope written before versioning reports 1.
prefix_changed_on_resume	bool	True when the envelope's prefix fingerprint does not match the Agent doing the restore — the instructions or the tool set moved while the conversation was saved. Always False on a live session and on an envelope with no fingerprint to compare. The restore still succeeds; this is a report.
prefix_change_report	Optional[Dict[str, Any]]	Details behind :attr:prefix_changed_on_resume: {"changed": ["system"|"tools"], "saved": {...}, "current": {...}, "schema_version": int}, or None when nothing changed.
converted_from	Optional[str]	The dialect the history was converted from on the last restore or :meth:rebind ("anthropic", "openai", "google" or "fireworks"), or None when no conversion ran. The entries are on :attr:conversion_report.
dropped_parts	List[Dict[str, Any]]	Media parts of a restored history that the target model's input_types do not list, each {"kind", "index", "reason", "provider", "model"}; replaced by placeholders under on_unsupported_part="drop". Empty on a live session.
last_turn_results	List[AgentResult]	Every AgentResult the most recent :meth:run/:meth:stream call produced, in turn order. Length 1 on an ordinary call; longer only when late_injection_policy="new_turn" started follow-up turns. Reset at the start of each call. Nothing is ever discarded, so the chain's cost is sum(r.cost for r in session.last_turn_results).

conversion_report property

Every artefact a cross-provider conversion dropped and every id it re-keyed (a copy).

The list rides metadata["conversion_report"] and grows across conversions. Each entry carries kind ("dropped" or "rekeyed"), the stored message index it concerns, from and to (the dialects) and at (UTC); a drop names its artefact (thinking, thought, thought_signature, reasoning, reasoning_content, provider_item, provider_block, provider_part, media, tool_result_media, container, unknown_part) with a detail; a re-key carries name, from_id and to_id. Empty when no conversion ever ran.

Example

restored = AgentSession.from_json(path, agent=Agent(model="gpt-6-luna"))
if restored.converted_from:
    for entry in restored.conversion_report:
        print(entry["kind"], entry.get("artefact"), entry["index"])

interrupted_turn property

The most recent turn that did not complete, or None.

A copy of metadata["interrupted_turn"]: the user message as it was sent, how many files were attached, the error, when it failed and the run number it would have had. Present only between a failed turn and the next run() / stream() call, and it survives the envelope, so a host that checkpoints after the failure can decide to resend the message in another process. DPT_SESSION_INTERRUPTED_TURN=off disables the record.

Example

try:
    await session.run("Summarise clause 4.")
except Exception:
    session.to_json(path=checkpoint)   # interrupted_turn rides along
...
if session.interrupted_turn:
    await session.run(session.interrupted_turn["message"])

late_injection_policy property

The configured late-injection policy (read-only).

"surface" or "new_turn". See :meth:__init__.

max_follow_up_turns property

Automatic follow-up turns allowed per call (read-only).

Only consulted under late_injection_policy="new_turn".

media_report property

What a save or a restore could not keep, as a list of records (a copy).

Each record carries a reason: "over_inline_threshold" (a Google upload larger than media_max_inline_bytes with no store, so its bytes were not kept), "stripped" (a part removed by media="strip", or at storage time under session_history_config={"preserve_media": False}, which adds "source": "preserve_media"), "missing_blob" (a store reference the restore could not read), "expired_reuploaded" (a dead Files API URI re-uploaded from the kept bytes) or "expired_no_bytes" (a dead URI with nothing to re-upload; the part became a placeholder). The list rides the envelope and grows across saves and restores.

Example

for entry in restored.media_report:
    print(entry["reason"], entry.get("mime_type"), entry.get("size"))

pending_injections property

Texts a run could not deliver and the session has not sent.

A copy of metadata["pending_injections"]. Empty on every session that has never had a hook message arrive after the last turn boundary. Under "surface" these are the host's to deliver; under "new_turn" they are only what the budget, the identical-text guard or a failed turn left over.

persistence_report property

What :meth:to_dict would write, measured, so a host can choose a policy first.

Returns:

Type	Description
Dict[str, Any]	``{"size_bytes": ,
Dict[str, Any]	"message_count": int, "media_count": int, "media_bytes": <decoded
Dict[str, Any]	bytes carried inline>, "media": []}``.
Dict[str, Any]	Each media record is ``{"message_index", "role", "kind",
Dict[str, Any]	"mime_type", "size_bytes", "ref"}withkind`` one of
Dict[str, Any]	"base64" (bytes in the envelope), "google_file_uri",
Dict[str, Any]	"provider_file" or "url" (references, size_bytes 0).

Example

report = session.persistence_report
if report["media_bytes"] > 8 * 1024 * 1024:
    ...  # externalise before writing

run_boundaries property

Where each recorded turn's messages sit in :attr:messages.

A copy of metadata["run_boundaries"]: one {"run": int, "start": int, "end": int} record per completed :meth:run/:meth:stream turn, oldest first. Turn i owns session.messages[record["start"]:record["end"]], so a host can truncate "before turn N" — see :meth:truncate_to_turn — without parsing provider-native messages.

Empty on a session that has run nothing, and on one restored from an envelope written before boundaries were recorded. Not every turn the session ever ran is necessarily listed: compaction rebases the list (see metadata["boundaries_reset_at_run"]) and :meth:truncate_to drops the records it cut through, so len(run_boundaries) counts the turns still addressable in stored history, not the turns ever run — that is metadata["run_count"].

Returns:

Type	Description
List[Dict[str, int]]	A list of boundary records. Mutating it does not affect the
List[Dict[str, int]]	session.

Example

result = await session.run("First question")
result = await session.run("Second question")
first, second = session.run_boundaries
assert session.messages[second["start"] : second["end"]]

session_id property

Get the unique session identifier.

tool_surface_drift property

Every mid-session change to the Anthropic tools array (read-only).

The envelope's prefix_fingerprint records the tool surface an Agent declares, which deliberately excludes MCP connection state — so a server that changes which tools it serves between turns does not move it. This is the per-request half of that signal: the provider re-derives tools from the live servers on every run, and each time the result differs from the array this session started with, a record lands here and the frozen array is sent instead.

Returns:

Type	Description
List[Dict[str, Any]]	A copy of the drift records, each a dict with event,
List[Dict[str, Any]]	failed_servers, added, removed and redefined. Empty
List[Dict[str, Any]]	for every session whose tool set never moved — which is every
List[Dict[str, Any]]	session that does not lose an MCP server mid-flight.

Example

await session.run("first")
await session.run("second")
if session.tool_surface_drift:
    logging.warning("MCP tools moved: %s", session.tool_surface_drift)

turn_in_flight property

Whether a run() / stream() call is currently running on this session.

__aenter__() async

Enter async context manager.

__aexit__(*args) async

Exit async context manager.

__enter__()

Enter sync context manager.

__exit__(*args)

Exit sync context manager.

__init__(agent, session_id=None, session_history_config=None, reuse_anthropic_container=False, late_injection_policy='surface', max_follow_up_turns=1)

Initialize an AgentSession.

Parameters:

Name	Type	Description	Default
agent	Agent	The Agent instance this session belongs to.	required
session_id	Optional[str]	Optional unique identifier. Generated if not provided.	None
session_history_config	Optional[Dict[str, Any]]	Optional history configuration dict. Keys: mode (str): One of "full", "messages_only", "no_tool_results", "no_reasoning", "compact_tool_results", "text_summary" (default "full"). preserve_media (bool): Keep media in stored history (default True). False replaces every media part of a turn — inline bytes and Google upload references alike — by a text placeholder when the turn is stored, so no later turn replays it; each replacement is recorded on media_report with "source": "preserve_media". compact_tool_results_max_chars (int): Max chars per tool result when mode="compact_tool_results" (default 500).	None
reuse_anthropic_container	bool	When True, capture the Anthropic code execution container_id from the first run and automatically inject it into every subsequent run (and stream) in this session. Uploaded files and REPL state inside the container persist up to 30 days, so downstream turns can reference earlier uploads without re-uploading. Also implicitly instructs the Anthropic provider to preserve (not delete) uploaded tool-file ids and custom-skill ids across every run in this session — including the first run, before a container id has been captured — so they remain usable by the reused container. Ignored (with a warning log) for non-Anthropic providers — safe to set when provider swaps are possible. Default False.	False
late_injection_policy	str	What happens to a hook message that arrived too late for any turn boundary to deliver it (the texts a run returns on AgentResult.undelivered_injections). One of: "surface" (default): nothing automatic. The texts are parked in metadata["pending_injections"] — which survives to_dict()/from_dict() — and cleared at the start of the next call. The host decides whether and when to send them. This is behaviour-neutral: a session under this policy issues exactly the requests it issues today. "new_turn": once a turn succeeds with undelivered texts, the session immediately runs another turn whose user message is those texts joined by a blank line, bounded by max_follow_up_turns. No follow-up ever starts after an error, an abort or a raised exception, and a follow-up whose text repeats the message that produced it is refused (a stuck host inbox). Anything not sent stays parked and logged, never silently dropped. Default "surface".	'surface'
max_follow_up_turns	int	Maximum automatic follow-up turns per :meth:run/:meth:stream call under late_injection_policy="new_turn". The budget is counted per call, so a hook that injects at every final boundary cannot loop. 0 makes "new_turn" behave like "surface". Default 1.	1

Raises:

Type	Description
ValueError	If late_injection_policy is not "surface" or "new_turn", or max_follow_up_turns is negative.

__repr__()

String representation.

clear()

Clear conversation history while keeping session open.

Notes

Metadata is deliberately preserved — cumulative cost and token counters are the history of what this session spent, not of what it currently holds. The recorded run boundaries are not: they are message indices, and after the history is gone every one of them points at nothing, so they are reset and the run at which that happened is stamped on metadata["boundaries_reset_at_run"]. See :meth:truncate_to to remove only part of the history. The truncated tool results kept for read_truncated_result are forgotten with the history.

close()

Close the session, preventing further runs.

Also releases the truncated tool results kept for read_truncated_result.

from_dict(data, agent, *, media_store=None) classmethod

Restore session from serialized data.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary from to_dict().	required
agent	Agent	Agent instance to bind to.	required
media_store	Any	The store for the envelope's $media:<ref> markers and Google upload references (a directory path or a MediaStore); None uses the agent's store. A missing blob becomes a placeholder plus a media_report entry.	None

Returns:

Type	Description
AgentSession	Restored AgentSession. :attr:restored_from_schema_version,
AgentSession	attr:prefix_changed_on_resume and :attr:prefix_change_report
AgentSession	describe what the restore found.

Raises:

Type	Description
SessionCompatibilityError	If the stored history is written in one provider's dialect, agent is on another, and the history cannot be converted: it ends inside a tool round, a tool call has no name, or a tool result matches no call. The message names the session, the target provider and the turn at fault, and nothing partial is stored. With DPT_SESSION_CROSS_PROVIDER_CONVERSION=off every provider-native history is refused this way. A history of plain text turns is provider-neutral and is never refused, whichever provider saved it; a version-3 envelope whose provider differs from the agent's but whose history is neutral restores with one INFO line.
UnsupportedPartError	Under on_unsupported_part="raise", when a restored media part is not in the target model's input_types.

Notes

Loading never fails on a version or a fingerprint. An envelope from an older version is migrated; one from a newer version is loaded with a warning and its unknown keys ignored, because refusing would strand a conversation this reader can read. A moved prefix is reported, not rejected: changing the instructions or the tool set mid-conversation is a legitimate thing to do, and the host is the only party that knows whether it was deliberate. A restore under another model of the same provider is quiet by design: the history reaches the new model unchanged.

A restore under another provider converts a provider-native history through the neutral representation and stores the converted messages: tool rounds and attachments cross, and what cannot (signed thinking, thought parts and signatures, reasoning items, provider-run tool traffic, media the target row does not take) is dropped, WARNING-logged and recorded on :attr:conversion_report; :attr:converted_from names the source dialect, and run boundaries are not carried across. The dialect is read from the messages themselves, so a version-1 or version-2 envelope, which records no provider, converts the same way. The one refusal is the unconvertible history above, because the alternative is a provider 400 on the first request that names none of this.

from_json(source, *, agent, media_store=None) classmethod

Restore a session from JSON text or a JSON file.

The inverse of :meth:to_json. source is read as JSON text when it is bytes or a str whose first non-blank character is { or [, and as a file path otherwise (a :class:pathlib.Path is always a path). Everything after parsing is :meth:from_dict, with the same versioning, fingerprint and compatibility rules.

Parameters:

Name	Type	Description	Default
source	Union[str, bytes, Path]	JSON text, UTF-8 bytes, or the path of a file written by :meth:to_json.	required
agent	Agent	The agent to bind the restored session to.	required
media_store	Any	The store the envelope's $media:<ref> markers and Google upload references are read from (a directory path or a MediaStore); None uses the agent's store. A blob the store cannot return becomes a placeholder and a media_report entry, never an error.	None

Returns:

Type	Description
AgentSession	The restored session.

Raises:

Type	Description
ValueError	If the JSON does not decode to an object.
SessionCompatibilityError	If a history written in another provider's dialect cannot be converted for agent, or DPT_SESSION_CROSS_PROVIDER_CONVERSION is off (see :meth:from_dict).
FileNotFoundError	If source names a file that does not exist.

Example

session = AgentSession.from_json("checkpoints/contract-review.json", agent=agent)
result = await session.run("Continue where we left off.")

get_history()

Get a copy of the conversation history.

Returns:

Type	Description
List[Dict[str, Any]]	Copy of the messages list, in provider-native shape: user prompts,
List[Dict[str, Any]]	assistant messages, tool calls, tool results, and reasoning items.

Notes

The run's final assistant reply is part of this history. The provider loops' own message lists hold only what was circulated to the API, and the last answer is never sent anywhere, so the session appends it after each run. Before this was fixed, every turn's actual answer was absent — later turns could not refer back to it and compaction summarized a conversation with no replies in it.

rebind(new_agent)

Continue this session's next turn on new_agent, in place.

The one-line answer to "continue this conversation on another model" between turns, without serialising: after a fallback decision, a cost cap, or the user picking another model. Same provider: the history is kept as it is. Another provider: the history is converted through the neutral representation exactly as from_dict converts it, with the same conversion_report, and refused with the same SessionCompatibilityError in the same impossible cases. The prefix fingerprint is recomputed against the new agent (prefix_changed_on_resume / prefix_change_report), the state that cannot cross an agent boundary is reset (the Anthropic container id, the frozen tool snapshot), the Google media table stays (it is data, not connection state), and the hop is recorded in metadata["served_models"].

Parameters:

Name	Type	Description	Default
new_agent	Agent	The agent the next turn runs on.	required

Returns:

Type	Description
AgentSession	This session, for chaining.

Raises:

Type	Description
RuntimeError	If a turn is running on this session.
SessionCompatibilityError	If the history cannot be converted for the new provider.

Example

session = Agent(model="gpt-6-luna", instructions="Be brief.").session()
await session.run("Summarise clause 4.", files=["contract.pdf"])
session.rebind(Agent(model="claude-haiku-4-5", instructions="Be brief."))
await session.run("And clause 5?")          # runs on Anthropic, history converted
print(session.converted_from, session.conversion_report)

run(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None) async

Run the agent within session context.

Automatically preserves conversation history and updates session state.

Parameters:

Name	Type	Description	Default
message	str	User message to send.	required
files	Optional[List[Any]]	Optional files to include with the message. Accepted formats: str — File path, URL, or YouTube URL. pathlib.Path — Path object pointing to a local file. bytes — Raw file bytes. dict — Pre-structured content dict.	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool (where supported by Anthropic, OpenAI, and Google). Files should be in the same format as the files argument. File type validation is performed by the provider API. Files are uploaded automatically via the provider's Files API. Unsupported providers (e.g., Fireworks) are filtered automatically.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. As with run(), the schema validates the output and AgentResult.output is the parsed dict, not a model instance.	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). See Agent.run() for full documentation.	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse for this call. Takes precedence over any id captured via reuse_anthropic_container auto-tracking. See Agent.run().	None

Returns:

Type	Description
AgentResult	AgentResult from the run. Under
AgentResult	late_injection_policy="new_turn" this is the last turn's
AgentResult	result — the answer to the thing the user most recently said.
AgentResult	result.cost is then that turn's cost only; the chain is on
AgentResult	attr:last_turn_results and its total is the
AgentResult	metadata["total_cost"] delta.

Raises:

Type	Description
RuntimeError	If session has been closed.

Notes

Under late_injection_policy="new_turn" one call can run several turns. Each turn is an ordinary turn: its own request, its own AgentResult, its own request_cost_records and its own metadata["run_count"] increment, so the session totals are the sum of the per-turn costs with nothing folded in invisibly. The caller's files, tool_files and media_resolution_list belong to the message they were passed with and are not resent on a follow-up turn; output_schema is forwarded, so every turn yields its own schema-valid output.

Example

Letting the session answer a late steer automatically:

session = agent.session(late_injection_policy="new_turn")
result = await session.run("Summarise the contract")
# result is the answer to the steer if one arrived late.
for turn in session.last_turn_results:
    print(turn.output, turn.cost)

run_sync(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None)

Synchronous wrapper for run().

Parameters:

Name	Type	Description	Default
message	str	User message to send.	required
files	Optional[List[Any]]	Optional files to include with the message. See run().	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool. See run().	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. See run().	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). See run().	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse for this call. See run().	None

Returns:

Type	Description
AgentResult	AgentResult from the run. Under
AgentResult	late_injection_policy="new_turn" this is the last turn's
AgentResult	result; see :meth:run and :attr:last_turn_results.

Notes

Each call runs in its own asyncio.run loop, so the agent's local MCP servers are connected and released per call (see Agent.run_sync). Use await run() to reuse them across turns.

stream(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None) async

Stream agent events within session context.

Delegates to Agent._stream_with_history with session history management. Loads conversation history before streaming and tracks costs from the complete event after streaming finishes.

Parameters:

Name	Type	Description	Default
message	str	User message to send.	required
files	Optional[List[Any]]	Optional files to include with the message.	None
tool_files	Optional[List[Any]]	Optional files for the provider code execution tool.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. See Agent.stream().	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only).	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse for this stream. Takes precedence over any id captured via reuse_anthropic_container auto-tracking.	None

Yields:

Type	Description
AsyncIterator[AgentStreamEvent]	AgentStreamEvent objects from Agent._stream_with_history.

Raises:

Type	Description
RuntimeError	If session has been closed.

Notes

A plain-text streamed turn stores a lightweight history shape: [user, injected…, assistant]. The opening user message, then each hook injection the run actually delivered (AgentResult.injected_messages, in order), then the final answer. Unlike :meth:run, that path does not store the provider-native transcript, so tool calls and tool results from a plain-text streamed turn are not visible to later turns. A "new_turn" follow-up therefore sees the previous turn's answer but not the tools it called, and may call one of them again.

A streamed turn with output_schema is stored exactly as :meth:run stores it: the provider-native transcript when the run returned one — including the retained signed final turn on the preserved-thinking rows, which DPT_ANTHROPIC_RETAIN_FINAL_TURN=off turns back into plain text — and otherwise the [user, answer] pair.

Under late_injection_policy="new_turn" one call can stream several turns, and exactly one complete event is still emitted per call — the documented teardown contract is unchanged. A non-final turn ends with turn_complete (the same metadata the complete event would have carried, plus turn_index and auto_follow_up) and the next one opens with turn_start (turn_index, auto_follow_up, trigger="late_injection" and the message being sent). Those two events appear only when a follow-up actually starts, so a stream under the default policy is event-for-event what it is today. The final complete then also carries turns, turn_costs and total_cost.

agent.run_handle is per-run and is None between turns: a host that steers must re-capture it on turn_start. An abort during a turn suppresses the follow-up entirely.

Closing the stream early (aclose() / contextlib.aclosing) closes the turn's inner generators at once, which cancels the request. A turn closed before its complete event is still banked: its spend (agent.last_abandoned_stream, marked "abandoned": True) is added to metadata and last_turn_results, and no answer is stored in history. A turn whose consumer stopped at complete is folded in normally. DPT_STREAM_CANCEL_ON_CLOSE=off restores the previous handling, in which neither was banked.

to_dict(*, media=None, media_store=UNSET, media_max_inline_bytes=UNSET)

Serialize session state to a dictionary.

Parameters:

Name	Type	Description	Default
media	Optional[str]	The media policy for this save — "auto" (the default: externalize when a store is in force, else inline), "inline" (every media part written exactly as stored), "externalize" (every inline payload handed to the store and replaced by a $media:<ref> marker) or "strip" (every inline payload replaced by a text placeholder). None uses the agent's media_policy.	None
media_store	Any	The store for this save (a directory path or a MediaStore); omitted means the agent's store, None means no store.	UNSET
media_max_inline_bytes	Any	Per-part cap on the Google upload bytes the envelope's media table keeps when no store is in force; omitted means the agent's cap, None lifts it. A per-call cap can only tighten what the session kept.	UNSET

Returns:

Type	Description
Dict[str, Any]	Dictionary containing session state, including schema_version
Dict[str, Any]	and prefix_fingerprint. Version 3 also records provider
Dict[str, Any]	and model (of the agent that saved it), toolkit_version,
Dict[str, Any]	saved_at (UTC, ISO 8601), sdk ({"name", "version"} of
Dict[str, Any]	the provider SDK), media_policy (the policy this save applied),
Dict[str, Any]	media (the media table: one entry per Google Files API upload —
Dict[str, Any]	URI, expiry, and its bytes or store reference — and per externalised
Dict[str, Any]	part) and media_report (what a save or restore could not keep).
Dict[str, Any]	The result is a detached copy: messages are normalised to
Dict[str, Any]	JSON-clean values and metadata is deep-copied, so a host that
Dict[str, Any]	mutates the envelope never mutates the live session through it.

Raises:

Type	Description
ValueError	If media is not a known policy, or is "externalize" with no store in force.

Notes

The envelope carries the conversation, never the agent's instructions. A session envelope is routinely handed to another process or another party, and a system prompt is the one part of a run that is often confidential and is always reconstructible from the :class:Agent that loads it (B16). A leading system message is therefore dropped on the way out; it can only be there at all in an envelope written before Fireworks took ownership of its own system prompt. prefix_fingerprint holds SHA-256 digests of the system prompt and the tool surface for the same reason — enough to notice that the request prefix moved, not enough to reconstruct it.

The per-turn run boundaries (B13) ride metadata, which already round-trips, so they need no envelope key and no schema_version bump: an envelope written before they existed simply has none, which reads as an empty list and lets a caller fall back. They are copied out of the live metadata and rebased onto the messages this envelope actually carries — dropping the leading system message moves every index down one — and they are re-validated against the restored messages on the way back in.

metadata["served_models"] (one record per run and served model, from the run's request_cost_records) rides metadata like the boundaries do, so which model actually answered each turn survives the round trip without an envelope key of its own.

Example

envelope = session.to_dict()
envelope["provider"], envelope["model"]   # who saved it
json.dumps(envelope)                      # always JSON-clean

to_json(*, path=None, indent=None, media=None, media_store=UNSET, media_max_inline_bytes=UNSET)

Serialize the session to a JSON string, optionally writing it to a file.

The string is json.dumps of :meth:to_dict (ensure_ascii=False, so non-ASCII text is written as it is). With path the same text is written to that file atomically — through a sibling temporary file and a rename — so a checkpoint interrupted mid-write never replaces the previous good envelope with a truncated one. This is the checkpoint primitive: call it after every completed turn and the file always holds a resumable session.

Parameters:

Name	Type	Description	Default
path	Optional[Union[str, Path]]	Optional destination file. Parent directories are created.	None
indent	Optional[int]	json.dumps indentation; None (the default) writes the compact form.	None
media	Optional[str]	The media policy for this save; see :meth:to_dict.	None
media_store	Any	The store for this save; see :meth:to_dict.	UNSET
media_max_inline_bytes	Any	The inline cap for this save; see :meth:to_dict.	UNSET

Returns:

Type	Description
str	The JSON text that was (or would have been) written.

Example

result = await session.run("Summarise the contract")
session.to_json(path="checkpoints/contract-review.json")
# or, with the bytes in a directory beside the file:
session.to_json(path="checkpoints/contract-review.json", media_store="checkpoints/media")

truncate_to(message_count, *, drop_orphans=True)

Cut stored history down to its first message_count messages.

The history editor's primitive: "delete everything from here on", for a host implementing edit-and-resend or delete-from-turn-k. Use :meth:truncate_to_turn to express the cut as a turn instead of a message count.

Truncation is a pure prefix cut. Nothing that survives is rewritten, so every retained thinking block, redacted_thinking block and Google thought_signature stays byte-identical to what the provider signed — the property that makes the next request's cached prefix and its block bindings valid. The one adjustment is to the cut point: a count that would leave a tool call without its result is moved backward, never repaired in place, so the invariant holds by construction rather than by editing.

Parameters:

Name	Type	Description	Default
message_count	int	How many messages to keep, counted from the start of :attr:messages. 0 keeps nothing; len(session.messages) keeps everything.	required
drop_orphans	bool	Keep the result provider-legal (the default). The cut point is moved back past any tool call whose result it would have severed, and any tool result already stored without its call is dropped with a warning. False cuts at exactly message_count and makes legality the caller's problem — for a host that knows the shape of its own history and wants an exact count.	True

Returns:

Type	Description
int	The number of messages actually kept. It is <= message_count,
int	and strictly less when the cut had to move back past a tool round.

Raises:

Type	Description
ValueError	If message_count is not an int, is negative, or is greater than the number of stored messages.
RuntimeError	If the session has been closed.

Notes

What is kept. Metadata survives exactly as it does through :meth:clear. The cumulative counters — total_cost, total_tokens, total_input_tokens, total_output_tokens, run_count, compaction_count — are the history of what this session spent, not of what it currently holds, and are deliberately not rewound: the requests were made and billed, and a host that rewound them would report a cost lower than its invoice. metadata["pending_injections"] survives for the same reason (a message the host still owes the model does not become delivered because history shrank), as do the Anthropic container id, :attr:last_turn_results (a record of the last call, not of history) and the request prefix fingerprint (a property of the :class:Agent, which truncation does not touch).

What is rewound. metadata["run_boundaries"] — the only metadata made of message indices. Records whose turn no longer fits entirely inside the kept prefix are dropped, so every remaining record still addresses the messages it names. Their messages are not deleted, they are simply no longer addressable as a turn, which is the same state a session restored from a pre-boundary envelope is in.

AgentResult.injected_messages is a property of a run's result and is not session state, so truncation has nothing to do to it.

Example

Drop the last turn and ask again:

keep = session.run_boundaries[-1]["start"]
session.truncate_to(keep)
result = await session.run("Let me rephrase that")

truncate_to_turn(turn_index, *, drop_orphans=True)

Cut stored history to just before recorded turn turn_index.

The public-index form of :meth:truncate_to: turn_index counts turns in :attr:run_boundaries, so a host can say "go back to before turn 3" without knowing anything about provider-native message shapes. The turn index is resolved to a message count and the same legality rules then apply, so a public turn index always maps to a private provider prefix the next request can legally send.

Parameters:

Name	Type	Description	Default
turn_index	int	The first turn to drop, 0-based over :attr:run_boundaries. 0 drops every recorded turn; len(session.run_boundaries) drops none.	required
drop_orphans	bool	As :meth:truncate_to.	True

Returns:

Type	Description
int	The number of messages kept.

Raises:

Type	Description
ValueError	If turn_index is not an int, is negative, or is greater than the number of recorded turns. The message names metadata["boundaries_reset_at_run"] when a compaction is why the turn is not addressable.
RuntimeError	If the session has been closed.

Notes

Turns that ran before a compaction, or before a restore from an envelope written without boundaries, are not addressable — their messages may still be stored, but nothing recorded where they begin. turn_index=0 therefore cuts to the start of the oldest recorded turn, which is not necessarily 0; call truncate_to(0) to empty the history outright.

Example

session.truncate_to_turn(2)  # keep turns 0 and 1, drop the rest

AgentStreamEvent dataclass

A single event emitted during Agent.stream().

Notes

Event types:

• "request_start": Opt-in (Agent(emit_request_start=True)); not emitted otherwise, so a consumer that enumerates event types is unaffected by default. Fires immediately before an ACTUAL provider send — the first request, each tool-loop iteration, each max_tokens / truncation / empty-turn re-request, each pause_turn continuation and each compaction summarisation call — so one event per API request, in order. content is None; metadata carries request_index (1-based and monotonic across the run), iteration, attempt (the tenacity attempt number), purpose ("generation", "tool_followup", "max_tokens_continuation", "truncation_retry", "empty_turn_retry", "pause_turn_continuation" or "compaction_summary"), model and provider. Reaching hooks as well on the tool-calling path, where it is observation-only.
• "text_delta": Partial text output. content is the text chunk. metadata is None for ordinary prose. On a schema-constrained request (Path A native structured output) it is {"structured_output": True}, so a UI can suppress the terminal JSON payload without a JSON-head heuristic (B11). On Path B (tool-based structured output) the payload is returned as tool input and never reaches text_delta at all, so nothing is tagged there.
• "thinking_delta": Partial reasoning/thinking output. content is the visible reasoning chunk when the provider exposes it.
• "tool_call_start": A tool call is starting. metadata includes tool_name, tool_call_id, arguments, and iteration when available.
• "tool_call_result": A tool call completed. metadata includes tool_name, tool_call_id, result, success, duration_ms, and iteration when available, plus tool_reported_cost and tool_reported_cost_estimated when the tool reported spend with the _dpt_spend marker (0.5.1). result is the first Agent(tool_result_preview_chars=) characters (500 by default) of the result's str(), and result_truncated says whether that is shorter than the whole; at 0 it is the whole result and result_data carries it as JSON data (0.5.1).
• "iteration_complete": One tool-calling iteration finished. metadata includes iteration, input_tokens, output_tokens, cost, tools_called, and stop_reason when available. The usage covers every request billed since the previous event, so the events sum to the run's requests up to the last event (0.5.1).
• "error": An unrecoverable error occurred. content is the error message. metadata carries input_tokens, output_tokens, cost, and token_source populated from the partial-stream data captured before the failure where available (token_source is then "stream_partial"); otherwise the token/cost fields are zero with token_source="unknown". It also carries undelivered_injections and, when a tool that ran before the failure reported spend, tool_reported_cost / tool_reported_costs (already inside cost). A failed run emits "error" first and still emits a trailing "complete" event, so consumers that short-circuit on "error" see the failure while teardown paths bound to "complete" continue to run.
• "complete": The stream finished. metadata is the full AgentResult serialized to a dict. Always emitted exactly once per stream() call, including after an "error" event when the run failed — metadata["error"] is populated in that case. It carries injected_messages (the hook injections this run delivered, in order) and undelivered_injections (texts that arrived after the last turn boundary and were never shown to the model — deliver them as the next turn's user message). When an AgentSession(late_injection_policy="new_turn") ran follow-up turns it also carries turn_index, turns, turn_costs and total_cost.
• "turn_complete": A non-final turn finished under AgentSession(late_injection_policy="new_turn"). metadata is that turn's AgentResult dict plus turn_index and auto_follow_up. Emitted only when a follow-up turn actually starts, in place of that turn's "complete", so exactly one "complete" still ends the stream.
• "turn_start": A follow-up turn is starting under the same policy. metadata carries turn_index, auto_follow_up, trigger ("late_injection") and message — the text being sent as the new user turn. Agent.run_handle is re-created per run, so this is the host's signal to re-capture it.

CompactionConfig dataclass

Configuration for context compaction behaviour.

Attributes:

Name	Type	Description
strategy	str	One of "summarize", "sliding_window", "tool_result_clearing", or "combined".
trigger_threshold	float | int	Float (0.0-1.0) interpreted as a percentage of the model's context window, or an int for an absolute token count.
summary_model	str | None	Model to use for the summarization LLM call. Defaults to the agent's own model when None.
summary_prompt_file	str | None	Path to a text file containing a custom summarization prompt.
summary_prompt	str | None	Inline custom summarization prompt. Takes precedence over summary_prompt_file.
keep_recent_messages	int	Number of recent messages to keep verbatim when using sliding_window or combined. Inside a tool loop the window never opens after the current turn's request, so it can keep more; DPT_COMPACTION_PROTECT_CURRENT_TURN=off restores the plain count.
keep_recent_tool_results	int	Number of recent tool results to preserve in tool_result_clearing / combined.
clear_tool_inputs	bool	Also clear tool call parameter payloads during tool_result_clearing.
archive_history	bool	Whether to archive the original conversation before compacting.
archive_dir	str	Directory for archive files. Ignored when archive_sink is set.
archive_sink	Callable[[list[Any]], str | None] | None	Optional callable receiving the archived conversation instead of writing a local file, for a host whose filesystem is ephemeral (a sandbox, a container) or who persists elsewhere. It is called with the same sanitised, JSON-serialisable payload that would have been written, and may return a locator string (an object key, a URL) which is then used everywhere the file path would have been: CompactionResult.archive_path, AgentResult.compaction_archives and the summary's reference section. Raising is contained — the compaction and the run continue, with a WARNING that names the exception type and nothing else. Ignored when archive_history is False.
include_archive_path_in_summary	bool	Include the archive file path in the compaction summary output.
combined_order	list[str]	Ordered list of strategies for the combined strategy.
min_messages_before_compaction	int	Minimum number of messages required before compaction can trigger.
max_compactions_per_session	int | None	Maximum compaction events per session. None means unlimited.
summary_max_tokens	int	Maximum output tokens for the summarization LLM call. Must be a positive integer no greater than the model's max_output_tokens_request (validated at config time when the model is known). Defaults to 4096.
sliding_window_include_truncation_notice	bool	When True, the sliding_window strategy prepends a synthetic user message informing the model that earlier context was truncated. This helps the model understand that it may be missing prior conversation context. Defaults to True.

ConfirmAfterNToolCalls

Built-in hook: pause the run after every N executed tool calls.

The "has been working for a while — continue?" pattern a host app shows its user. Counts tool_call_result events (one iteration can execute many tools), pauses at the following safe point, and leaves resuming to the host via the run handle (deliver the user's answer back to the process holding the handle, and apply a timeout policy so an unanswered confirmation doesn't hold resources forever). The count restarts at every run_start, so one instance can be shared across an Agent's runs.

Examples:

```python
confirm = ConfirmAfterNToolCalls(5)
agent = Agent(model="claude-haiku-4-5", tools=[...], hooks=confirm)
# elsewhere, when the user answers: agent.run_handle.resume()
```

DetailedCostReport dataclass

Detailed cost tracking report for agent runs.

Captures per-iteration costs, token breakdowns, and cumulative totals. Can be serialized to JSON for storage and analysis.

Attributes:

Name	Type	Description
run_id	str	Unique identifier for this run.
agent_name	Optional[str]	Optional name identifier for the agent.
timestamp	str	When the report was created.
provider	str	Provider name (anthropic, openai, google, fireworks).
model_name	str	Full model name used.
total_input_tokens	int	Total input tokens across all iterations.
total_output_tokens	int	Total output tokens across all iterations.
total_reasoning_tokens	int	Total reasoning/thinking tokens.
total_cached_input_tokens	int	Total cached input tokens across all iterations (the cache-read share of total_input_tokens, all providers).
total_cache_write_tokens	int	Total prompt-cache write tokens. Anthropic bills them on top at the cache-write rate; on OpenAI they are a subset of input re-priced at cache_write_input_multiplier. Zero for providers that do not report them.
total_cost	float	Total cost in USD for the run, INCLUDING the failed-and-retried attempts. Reports built by :class:~doc_processing_toolkit.agent.core.Agent take this straight from AgentResult.cost (the sum of the per-request token costs plus per-run tool costs), which is already retry-inclusive; hand-built reports accumulate it from :meth:add_iteration, whose per-iteration cost is likewise retry-inclusive. Subtract total_retry_cost to get the successful attempts alone.
cost_estimated	bool	Whether the cost is an estimate.
total_duration_ms	int	Total execution time in milliseconds.
thinking_enabled	bool	Whether the run carried thinking: any request's flag when the run has per-request entries, else what the configuration given resolves to (0.5.3).
thinking_config	Optional[Dict[str, Any]]	Thinking configuration if enabled.
max_iterations	int	Maximum iterations configured.
temperature	float	Temperature setting used.
iterations	List[IterationMetrics]	List of per-iteration cost breakdowns.
total_iterations	int	Actual number of iterations performed.
tools_available	List[str]	List of tool names available to agent.
tools_used	List[str]	List of tool names that were called.
tool_call_count	int	Total number of tool calls.
tool_success_rate	float	Success rate of tool calls (0.0 to 1.0).
tool_usage_summary	Dict[str, Any]	Summary of tool usage by tool name.
pricing_info	Dict[str, float]	Pricing rates used for calculation.
start_time	Optional[str]	When the run started.
end_time	Optional[str]	When the run completed.
total_retry_cost	float	USD cost of the failed-and-retried attempts only. Set from the run-level AgentResult.retry_cost when the report is built by :class:~doc_processing_toolkit.agent.core.Agent. The :meth:add_iteration accumulation below it is retained for hand-built reports but contributes nothing, because the per-iteration retry_* fields are reserved and always zero.
total_retry_input_tokens	int	Input tokens consumed by retried attempts. Run-level, on the same convention as total_retry_cost.
total_retry_output_tokens	int	Output tokens produced by retried attempts. Run-level, on the same convention as total_retry_cost.
total_retry_count	int	Total number of failed-and-retried API attempts. Run-level, on the same convention as total_retry_cost.
request_cost_records	List[Dict[str, Any]]	One record per API request (input/output/cached tokens, served_model, requested/served tier, cost). The run's token cost is the sum of these records, never a re-pricing of the aggregate totals. :attr:service_tier_served and :attr:priority_request_count are derived from them.
total_tool_reported_cost	float	What the run's tools reported spending (AgentResult.tool_reported_cost), already part of :attr:total_cost.
tool_reported_costs	List[Dict[str, Any]]	One record per reporting tool call (AgentResult.tool_reported_costs).
prefix_binding_mismatches	int	Preserved-thinking blocks the Anthropic API dropped in this run because the conversation prefix they were bound to had been rewritten (the alarm). Mirrors AgentResult.prefix_binding_mismatches so the count survives into the cost/telemetry sink clients already read. 0 everywhere else.
model_binding_mismatches	int	Preserved-thinking blocks dropped because the serving model could not read them. Expected after a model change or a server-side fallback — never an integration bug. Both counters count dropped blocks only (type == "thinking_dropped"). The model count is recorded only when the request carried the thinking-binding-controls-2026-08-01 beta (claude-fable-5-1, claude-opus-5-5, claude-sonnet-5-5, claude-haiku-5-5, or anthropic_block_binding="drop_block"), so a client-side switch to an ungated row reports 0 although the API dropped the blocks. 0 elsewhere.
prefix_binding_retry	bool	True when this run re-sent a request once with prefix_mismatch_behavior="drop_block" after a prefix-binding 400 (or, on a thinking-off run, with the thinking blocks removed). False elsewhere.
schema_version	int	Persisted-report schema marker, bumped when the report shape changes. Version 2 adds per-request records and cached-token fields (2026-08).

Notes

:meth:to_dict emits one key with no backing attribute: total_cost_inclusive. Because total_cost is already retry-inclusive, the key simply equals total_cost and is retained for backward compatibility (an earlier derivation added total_retry_cost on top, double-counting retries). It is not read back by :meth:from_dict.

Examples:

Basic usage:

agent = Agent(
    model="claude-sonnet-4-5",
    detailed_costs=True,
    cost_output_dir="./costs",
)
result = await agent.run("Complex task...")
# Cost report is automatically saved to ./costs/cost_report_<run_id>.json

priority_request_count property

How many requests in this run were served at the "priority" tier (C7).

Returns:

Name	Type	Description
int	int	The count of records billed at the model's priority row.

service_tier_served property

The tier this run's requests were SERVED at, or "mixed" (C7).

Returns:

Type	Description
Optional[str]	str or None: See :func:summarize_served_tiers. None when no
Optional[str]	request asked for a tier.

__post_init__()

Initialize default values.

add_iteration(input_tokens, output_tokens, reasoning_tokens, cost, duration_ms=0, thinking_enabled=False, stop_reason='', tool_calls=None, cached_input_tokens=0, cache_write_tokens=0, purpose=None)

Add a new iteration to the report.

Parameters:

Name	Type	Description	Default
input_tokens	int	Input tokens for this iteration.	required
output_tokens	int	Output tokens for this iteration.	required
reasoning_tokens	int	Reasoning tokens for this iteration.	required
cost	float	Cost for this iteration in USD.	required
duration_ms	int	Duration in milliseconds.	0
thinking_enabled	bool	Whether this request carried thinking (per request).	False
stop_reason	str	Reason the iteration stopped.	''
tool_calls	Optional[List[ToolCallMetrics]]	Detailed tool call metrics.	None
cached_input_tokens	int	Cached share of this iteration's input tokens.	0
cache_write_tokens	int	Prompt-cache write tokens for this iteration (Anthropic and OpenAI).	0
purpose	Optional[str]	Why the iteration's request was made (see :attr:IterationMetrics.purpose).	None

finalize()

Finalize the report, setting end_time and calculating totals.

from_dict(data) classmethod

Create a report from a dictionary.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary from to_dict().	required

Returns:

Type	Description
DetailedCostReport	DetailedCostReport instance.

load(file_path) classmethod

Load a report from a JSON file.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the JSON file.	required

Returns:

Type	Description
DetailedCostReport	DetailedCostReport instance.

save(directory)

Save the report to a JSON file.

Parameters:

Name	Type	Description	Default
directory	str	Directory to save the file in.	required

Returns:

Type	Description
str	Full path to the saved file.

to_dict()

Convert the report to a dictionary.

Returns:

Type	Description
Dict[str, Any]	Dictionary representation of the report. Every attribute is emitted,
Dict[str, Any]	plus three derived keys with no backing attribute:
Dict[str, Any]	total_cost_inclusive, which equals total_cost (already
Dict[str, Any]	retry-inclusive; kept for backward compatibility), and the run-level
Dict[str, Any]	tier pair service_tier_served / priority_request_count, both
Dict[str, Any]	recomputed from request_cost_records and therefore ignored by
Dict[str, Any]	meth:from_dict.

to_json(indent=2)

Serialize the report to JSON.

Parameters:

Name	Type	Description	Default
indent	int	JSON indentation level.	2

Returns:

Type	Description
str	JSON string representation.

HookDirective dataclass

Directive returned by a hook to steer the run.

Attributes:

Name	Type	Description
action	Literal['continue', 'skip_tool', 'inject_message', 'pause', 'abort']	One of "continue", "skip_tool", "inject_message", "pause", "abort".
message	Optional[str]	For inject_message: the user-role text to inject at the next turn boundary. For skip_tool: the text of the synthetic declined tool result (defaults to a standard notice).
reason	Optional[str]	Optional free-text rationale (logged; for abort it is carried on the raised :class:AgentAborted).

MediaTransformer

Transformer for converting media inputs to provider-specific formats.

The MediaTransformer normalizes file paths, URLs, and raw bytes to each provider's expected format.

Notes

Different providers have different requirements for media content

• Anthropic: Base64-encoded with media_type; filename preserved when supplied
• OpenAI: Base64 with media_type; filename preserved when supplied (required for documents, recommended for tool_files uploads of any category)
• Google: Various formats including bytes for video/audio; filename preserved when supplied and used as the Files API display_name for tool_files
• Fireworks: Base64 for images only

Example

transformer = MediaTransformer(provider="anthropic", logger=logger)

# Transform file paths
parts = transformer.transform(["image.png", "document.pdf"])

__init__(provider, logger=None)

Initialize the MediaTransformer.

Parameters:

Name	Type	Description	Default
provider	str	Target provider ("anthropic", "openai", "google", "fireworks").	required
logger	Optional[Logger]	Logger instance for warnings/errors.	None

Raises:

Type	Description
ValueError	If provider is not supported.

detect_media_type(file_path)

Detect the MIME type of a file.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path to the file.	required

Returns:

Type	Description
str	MIME type string (e.g., "image/png", "application/pdf").

encode_base64(file_path)

Read a file and encode it as base64.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path to the file.	required

Returns:

Type	Description
str	Base64-encoded string of the file contents.

get_supported_types(provider=None)

Get list of supported MIME types for a provider.

Parameters:

Name	Type	Description	Default
provider	Optional[str]	Provider name (uses instance provider if not specified).	None

Returns:

Type	Description
List[str]	List of supported MIME types.

is_supported(file_path, provider=None)

Check if a file type is supported by a provider.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path to the file.	required
provider	Optional[str]	Provider name (uses instance provider if not specified).	None

Returns:

Type	Description
bool	True if the file type is supported, False otherwise.

transform(files, **kwargs)

Transform a list of files to provider-specific format.

Parameters:

Name	Type	Description	Default
files	List[Union[str, Path, bytes, Dict[str, Any]]]	List of file inputs. Each can be: str: File path or URL Path: Path object bytes: Raw file bytes (requires media_type in kwargs or detection) dict: Pre-structured content	required
**kwargs	Any	Provider-specific options.	{}

Returns:

Type	Description
List[Dict[str, Any]]	List of content parts in provider format.

Raises:

Type	Description
FileNotFoundError	If a file doesn't exist.
ValueError	If file type not supported by provider.

transform_files(files, provider, **kwargs)

Alias for transform() for backwards compatibility.

Parameters:

Name	Type	Description	Default
files	List[Union[str, Path]]	List of file paths.	required
provider	str	Target provider name (ignored, uses instance provider).	required
**kwargs	Any	Provider-specific options.	{}

Returns:

Type	Description
List[Dict[str, Any]]	List of content parts in provider format.

RunHandle

Handle for controlling a running (possibly paused) agent run.

Exposed as Agent.run_handle for the duration of a run. pause is an in-process asyncio.Event: the loop blocks at its next safe point until :meth:resume is called (or :meth:abort fails the run cleanly).

:meth:resume, :meth:abort and :meth:inject may be called from any thread. A run started with run_sync blocks its caller inside asyncio.run, so the host necessarily releases it from another thread; the wake-up is then marshalled onto the run's event loop with call_soon_threadsafe — a bare asyncio.Event.set from a foreign thread would leave the loop parked in its selector, i.e. the run would stay paused indefinitely.

paused property

True while the run is paused.

abort(reason=None)

Request a clean abort at the next safe point (also unblocks a pause; any thread).

Like :meth:pause, this cannot pre-empt a tool that is already executing — that tool finishes and its result is delivered first, and the abort is raised at the following safe point. "Clean" means exactly that: no provider call is cut off mid-request and no tool call is left without its result.

Parameters:

Name	Type	Description	Default
reason	Optional[str]	Optional text carried on the resulting :class:AgentAborted.	None

inject(text)

Queue a user-role message for the run's next turn boundary.

The host-facing half of HookDirective(action="inject_message"): it lands in the same queue a hook's directive lands in, so the delivery rule is uniform and the accounting is unchanged — delivered at the next tool boundary if one comes, returned on AgentResult.undelivered_injections if not. It removes the need for a hook that polls the host's inbox on every event.

Safe to call from any thread: the queue is guarded by a threading.Lock, and this appends to it without touching the run's event loop. It does not pre-empt a tool that is already executing — drain latency is bounded by the host's longest-running tool.

Parameters:

Name	Type	Description	Default
text	str	The user-role message to queue. Must be a non-empty string.	required

Raises:

Type	Description
ValueError	If text is not a non-empty string.
RuntimeError	If this handle is not bound to a run, or the run has already ended. A message queued into a finished run would never be seen by anyone, so it is refused loudly rather than accepted and dropped.

Example

Steering a run from the host thread:

import threading

agent = Agent(model="gpt-5.6-luna", instructions="...", hooks=my_hooks)
thread = threading.Thread(target=lambda: agent.run_sync("summarise"))
thread.start()
# ... later, from the UI thread:
agent.run_handle.inject("actually, only the indemnity clauses")

pause()

Pause the run at its next safe point.

A request, not an interrupt: a tool that is already executing runs to completion and its result is delivered before the loop stops. Nothing cancels a tool mid-flight, so the latency between this call and the run actually stopping is bounded by the host's longest-running tool.

resume()

Resume a paused run (safe to call from any thread).

SessionCompatibilityError

Bases: ValueError

A stored session cannot be restored under the agent it was given.

Raised by AgentSession.from_dict / from_json / rebind when the history cannot be converted for the target provider: an open tool round (calls with no results), a tool call with no name, a tool result that matches no call — or, with DPT_SESSION_CROSS_PROVIDER_CONVERSION=off, for any history written in another provider's dialect. The message names the stored dialect, the target provider and, when one turn is at fault, the turn. A history of plain text turns is provider-neutral and is never refused.

Example

try:
    session = AgentSession.from_json(path, agent=Agent(model="gpt-6-luna"))
except SessionCompatibilityError as exc:
    print(exc)

ToolLoader

Loader for converting various tool formats into a unified representation.

Notes

The ToolLoader is responsible for

• Detecting the format of each tool input
• Loading/connecting to tools as needed
• Converting tools to provider-specific formats
• Managing tool lifecycle (cleanup on shutdown)

Supported formats:

1. @tool decorated function - Direct Python functions
2. File path string - .py file containing @tool functions
3. Directory path string - Directory of .py files with @tool functions
4. FastMCP server - In-process MCP server instance
5. MCP script path dict - {"type": "mcp_script", "path": "..."}
6. MCP command dict - {"type": "mcp_command", "command": "npx", "args": [...]}
7. Provider-native remote MCP URL dict - {"type": "mcp", "server_url": "..."}
8. Pre-built session dict - {"session": ClientSession, "name": "..."}
9. Client-side remote MCP dict - {"type": "mcp_remote_client", "url": "..."}, connected and executed by this loader rather than by the provider

Sessions for formats 4, 5, 6, 8 and 9 are opened once per event loop by :meth:connect_mcp_servers and reused until :meth:disconnect_mcp_servers, which keeps the configs so a later connect reconnects them.

Formats 5 and 6 start a subprocess, which sees only the MCP SDK's short allow-list of this process's environment (PATH, the home directory and a few others; the list differs by platform). Both take env (a dict merged over that allow-list — name every variable the server needs, an API key or a DPTTOOLS_* setting) and cwd (the directory it starts in) since 0.5.3; neither value is logged.

Every format the toolkit runs itself takes allowed_tools: the dicts of formats 5, 6, 8 and 9 carry the key, and format 4 takes it in the dict form {"type": "fastmcp", "server": server, "allowed_tools": [...]}. No key exposes every tool; a listed name the server lacks logs one WARNING.

Example

loader = ToolLoader()

# Load from mixed formats
tools = loader.load([
    my_tool_function,
    "/path/to/tools.py",
    fastmcp_server,
    {"type": "mcp", "server_url": "https://mcp.example.com"},
])

# Convert to provider format
anthropic_tools = loader.convert_to_provider_format(tools, "anthropic")

__init__()

Initialize the ToolLoader.

connect_mcp_servers() async

Connect to all MCP servers and create sessions for local tools.

This method creates MCP-compatible sessions for all local tool types:

• @tool decorated functions → LocalFunctionSession
• FastMCP server objects → FastMCPSession
• MCP scripts → subprocess-based session
• MCP commands (npx/uvx) → subprocess-based session
• Pre-built ClientSessions → Load tools asynchronously
• mcp_remote_client configs → client-side Streamable HTTP / SSE session

Provider-native remote MCP configs ({"type": "mcp", "server_url": ...}) are NOT connected here - they're handled natively by the provider functions.

Should be called before running the agent with local tools. Calling it again on the same event loop is a no-op: sessions are opened once and reused until :meth:disconnect_mcp_servers. If the sessions were opened on a different event loop (run_sync creates a fresh loop per call), the stale connections are discarded and the servers are reconnected on the current loop.

convert_to_provider_format(tools, provider)

Convert tools to provider-specific format.

Parameters:

Name	Type	Description	Default
tools	List[Union[LoadedTool, Any]]	List of LoadedTool objects or raw tool inputs.	required
provider	str	Target provider ("anthropic", "openai", "google", "fireworks").	required

Returns:

Type	Description
List[Dict[str, Any]]	List of tool definitions in the provider's expected format.

Raises:

Type	Description
ValueError	If provider is not supported.

detect_format(tool)

Detect the format of a tool input.

Parameters:

Name	Type	Description	Default
tool	Any	A tool in any supported format.	required

Returns:

Type	Description
str	Format string: "function", "file_path", "directory",
str	"fastmcp", "mcp_script", "mcp_command", "mcp_remote",
str	"client_session", or "mcp_remote_client".

Raises:

Type	Description
TypeError	If the format cannot be determined.

disconnect_mcp_servers() async

Disconnect from all connected MCP servers.

Should be called when the agent is done to clean up resources. Properly cleans up async context managers via the exit stack.

Teardown is symmetric with :meth:connect_mcp_servers: the tools that connect registered (mcp_script, mcp_command, client_session, mcp_remote_client, and deferred FastMCP servers) are dropped with their sessions, while the server configs are kept so the next connect_mcp_servers() call reconnects them. Sessions are only closed from the event loop that opened them; from any other loop (or with the opening loop already closed) the references are discarded without awaiting the dead context managers.

execute_tool(name, arguments) async

Execute a tool by name with the given arguments.

Parameters:

Name	Type	Description	Default
name	str	The tool name.	required
arguments	Dict[str, Any]	Tool arguments as a dictionary.	required

Returns:

Type	Description
Any	The tool's return value.

Raises:

Type	Description
ValueError	If tool not found.
RuntimeError	If tool execution fails.

get_remote_mcp_configs()

Get remote MCP configurations for provider-native handling.

get_tool_by_name(name)

Get a loaded tool by its name.

Parameters:

Name	Type	Description	Default
name	str	The tool name to look up.	required

Returns:

Type	Description
Optional[LoadedTool]	LoadedTool if found, None otherwise.

has_local_mcp_tools()

Check if any tools require the function calling path.

Returns True if there are any local tools that need to be passed to the LLM for function calling. This includes:

• @tool decorated functions
• FastMCP servers (including deferred fastmcp>=3 servers whose tools are listed at connect time)
• MCP scripts/commands
• Pre-built client sessions
• Client-side remote MCP servers (mcp_remote_client)
• Python files/directories with @tool functions

Returns False only when:

• There are no tools at all, or
• Only mcp_remote tools are present (handled by structured output with provider-native remote MCP support)

Returns:

Type	Description
bool	True if local MCP tools are present that need function calling.

load(tools)

Load tools from various formats into unified representation.

Parameters:

Name	Type	Description	Default
tools	Optional[List[Any]]	List of tools in any supported format.	required

Returns:

Type	Description
List[LoadedTool]	List of LoadedTool objects.

Raises:

Type	Description
TypeError	If a tool format is not recognized.
FileNotFoundError	If a file path doesn't exist.

Notes

Each call replaces the loader's tool set: pending MCP server, remote MCP and FastMCP registrations are rebuilt alongside the loaded tools, so loading twice never accumulates duplicate server registrations.

to_provider_format(tools, provider)

Alias for convert_to_provider_format.

tool(func=None, *, name=None, description=None)

Decorator to mark a function as a tool for use with Agent.

The decorator extracts metadata from the function signature and docstring to create an MCP-compatible tool definition. It generates a JSON Schema for the tool's parameters based on type hints.

Parameters:

Name	Type	Description	Default
func	Optional[F]	The function to decorate (auto-passed when used without parentheses).	None
name	Optional[str]	Optional override for the tool name (defaults to function name).	None
description	Optional[str]	Optional override for the description (defaults to docstring).	None

Returns:

Type	Description
Any	The decorated function with _tool_metadata attribute attached.

Raises:

Type	Description
TypeError	If the function has parameters without type hints.
ValueError	If the function is already decorated with @tool.

Notes

Can be used with or without parentheses

@tool
def my_func(x: int) -> str: ...

@tool(name="custom_name", description="Custom description")
def my_func(x: int) -> str: ...

Examples:

Basic usage:

@tool
def calculate(a: float, b: float, operation: str) -> float:
    '''Perform a calculation.

    Args:
        a: First number
        b: Second number
        operation: One of 'add', 'subtract', 'multiply', 'divide'
    '''
    ...

calculate._tool_metadata
{
    'name': 'calculate',
    'description': 'Perform a calculation.',
    'parameters': {
        'type': 'object',
        'properties': {
            'a': {'type': 'number', 'description': 'First number'},
            'b': {'type': 'number', 'description': 'Second number'},
            'operation': {'type': 'string', 'description': "One of 'add'..."}
        },
        'required': ['a', 'b', 'operation']
    },
    'function': calculate
}

Agent Core

doc_processing_toolkit.agent.core

Agent class - High-level interface for AI agents.

The Agent class provides a unified, provider-agnostic interface for building AI agents with tools, media support, thinking capabilities, and sessions.

Key Features: - Provider detection from model name - Intelligent routing between function calling and structured output - Tool support (decorated functions, FastMCP, MCP servers, remote URLs) - Media transformation for images, documents, video, audio - Session management for multi-turn conversations - YAML configuration support - Cost tracking and result aggregation

Agent

High-level Agent class for AI interactions.

Attributes:

Name	Type	Description
model		The model name.
instructions		System prompt/instructions.

Notes

The Agent class provides:

• Provider detection from model name
• Unified parameter handling across providers
• Tool loading from multiple formats
• Media transformation
• Session management
• YAML configuration support

Example

from doc_processing_toolkit import Agent, tool

@tool
def calculate(a: int, b: int) -> int:
    '''Add two numbers.'''
    return a + b

agent = Agent(
    model="claude-sonnet-4-5",
    instructions="You are a helpful math assistant.",
    tools=[calculate],
)

result = await agent.run("What is 2 + 3?")
print(result.output)  # "The sum of 2 and 3 is 5."

__aenter__() async

Enter the async context manager.

__aexit__(exc_type, exc_val, exc_tb) async

Exit the async context manager, cleaning up resources.

__init__(model, instructions='', *, tools=None, api_key=None, max_iterations=10, max_tokens=None, temperature=None, thinking=None, built_in_tools=None, strict_tools=True, auto_inject_memory_instructions=True, auto_inject_skills=True, compaction=False, compaction_config=None, session_history_config=None, logger=None, log_level='DEBUG', log_file=None, log_to_file=True, tool_choice='auto', code_execution_output_dir=None, service_tier=None, hooks=None, emit_request_start=False, max_tool_result_chars=None, tool_result_preview_chars=DEFAULT_TOOL_RESULT_PREVIEW_CHARS, truncated_result_read_chars=None, prompt_caching=None, anthropic_block_binding=None, strict_prefix_binding=False, prefix_audit=False, prefix_strip_fallback=False, remove_citations=False, on_unsupported_part='drop', media_policy='auto', media_store=None, media_max_inline_bytes=DEFAULT_MEDIA_MAX_INLINE_BYTES, estimate_input_tokens='lazy', defer_tools=None, detailed_costs=False, cost_output_dir=None, **provider_kwargs)

Initialize an Agent.

Parameters:

Name	Type	Description	Default
model	str	Model name (e.g., "claude-sonnet-4-5", "gpt-4o", "gemini-2.5-flash", etc.)	required
instructions	str	System prompt/instructions.	''
tools	Optional[List[Any]]	List of tools in any supported format. The Agent accepts 9 tool formats (including directory paths): @tool decorated functions — Direct Python callables decorated with @tool from doc_processing_toolkit. File path strings — Paths to .py files containing @tool decorated functions. The agent dynamically imports them. Directory paths — String paths to directories. All .py files in the directory are scanned for @tool functions which are imported dynamically. FastMCP server objects — In-process MCP servers created with FastMCP("name"). Tools are discovered from the server. MCP script path dicts — {"type": "mcp_script", "path": "/path/to/server.py", "name": "server-name"}. Launches a local MCP server from a Python script. MCP command dicts — {"type": "mcp_command", "command": "npx", "args": ["-y", "@mcp/server"], "name": "server-name"}. Launches a local MCP server via a shell command (npx, uvx, etc.). Remote MCP URL dicts — {"type": "mcp", "server_url": "https://...", "authorization_token": "...", "name": "server-name"}. The provider connects to the remote MCP server (Anthropic, OpenAI and Fireworks only; tool calls never pass through this process). Pre-built session dicts — {"session": ClientSession, "name": "server-name"}. Uses an already-established MCP ClientSession. Client-side remote MCP dicts — {"type": "mcp_remote_client", "url": "https://.../mcp", "authorization_token": "...", "name": "server-name"}. This process connects over Streamable HTTP (or "transport": "sse") and executes the remote tools like local ones, so they work on every provider (including Google), take the function-calling path (compatible with output_schema), and are visible to hooks and tools_used. Optional keys: headers, auth (an httpx.Auth, e.g. an OAuth provider), timeout, sse_read_timeout, allowed_tools. The session is opened once per event loop and reused until :meth:close.	None
api_key	Optional[str]	Optional API key override for this Agent's provider. When supplied, this key is used instead of the provider's environment variable (ANTHROPIC_API_KEY, OPENAI_API_KEY, GOOGLE_GEMINI_API_KEY, FIREWORKS_API_KEY) and the Agent constructs its own client, bypassing the shared process-level client cache — per-key clients are never shared between Agents, and this Agent's close() closes them. Without it, clients come from the shared cache and are left open by close().	None
max_iterations	int	Maximum tool calling iterations (default: 10).	10
max_tokens	Optional[int]	Maximum output tokens per response.	None
temperature	Optional[float]	Sampling temperature (0.0 to 2.0).	None
thinking	Optional[Dict[str, Any]]	Thinking/reasoning configuration dict. Format varies by provider: Anthropic: {"budget_tokens": int, "interleaved": bool} or {"type": "adaptive"} for models supporting adaptive thinking. {"type": "disabled"} (or {"type": "between_tools"}, the same request) turns thinking off; what is sent comes from the model's registry row (thinking_off_type, thinking_off_max_effort), and a model that cannot turn thinking off runs adaptive with a warning. OpenAI: {"effort": "low"|"medium"|"high"|"xhigh", "summary": "auto"|"detailed"}. GPT-5.6 models additionally accept "mode": "standard"|"pro" — pro mode requires effort of at least "medium" (allowed values are registry-driven via supported_reasoning_modes / supported_reasoning_values_pro_mode in model_information.json; an unsupported effort in pro mode is dropped so the API default applies). Google: {"budget_tokens": int} or {"level": "minimal"|"low"|"medium"|"high"} Fireworks: {"effort": "low"|"medium"|"high"}	None
built_in_tools	Optional[Dict[str, Any]]	Built-in tool configuration dict. Keys: "web_search": BuiltInToolConfig(enabled=True, config={"max_uses": 10}) "code_execution": BuiltInToolConfig(enabled=True) "code_execution" with container reuse (Anthropic only): BuiltInToolConfig(enabled=True, config={"container_id": "<prior_id>"}) reuses an existing Anthropic code execution container so uploaded files and REPL state persist across runs (up to 30 days). Supported by: Anthropic, OpenAI, Google (not Fireworks).	None
strict_tools	bool	Enable strict schema validation (default: True). Anthropic only. Each request stays within Anthropic's per-request strict limits (20 strict tools, 24 optional properties, 16 union-typed parameters); a tool that does not fit is sent without strict mode, with one WARNING (DPT_ANTHROPIC_STRICT_BUDGET=off restores the per-tool check).	True
auto_inject_memory_instructions	bool	Auto-detect memory tools and inject memory instructions into the system prompt (default: True).	True
auto_inject_skills	bool	Detect a skills MCP server — one exposing list_skills and load_skill (the doc-processing-std-tools contract, 0.2+; the server's name does not matter) — call list_skills and inject the catalog (name, description, path) into the system prompt as <available_skills> (default: True). On an empty catalog the activation tools load_skill, load_skill_resource and execute_skill_script are hidden from the model. Skipped — no call, no injection, no hidden tools, one INFO line per run — while the agent's instructions already carry an <available_skills> block (an app that builds its own catalog). A failed or malformed list_skills reply leaves the activation tools advertised and logs one WARNING naming the server.	True
compaction	bool	Enable provider-agnostic context compaction (default: False). When True, the toolkit monitors conversation length and automatically compresses history when it approaches the model's context window limit.	False
compaction_config	Optional[Dict[str, Any]]	Configuration dict for compaction behaviour. Keys include: strategy (str): "summarize", "sliding_window", "tool_result_clearing", or "combined" (default "summarize"). trigger_threshold (float|int): Float 0.0–1.0 for percentage of context window, or int for absolute token count (default 0.8). keep_recent_messages (int): Messages to preserve verbatim in sliding_window/combined (default 5). keep_recent_tool_results (int): Tool results to keep in tool_result_clearing/combined (default 3). min_messages_before_compaction (int): Minimum messages before compaction can trigger (default 4). summary_model (str|None): Model for summarization (defaults to agent's model). summary_prompt (str|None): Inline custom summarization prompt. summary_prompt_file (str|None): File path to custom summarization prompt. archive_history (bool): Archive the original conversation before compacting (default True). archive_dir (str): Directory for archive files (default "./compaction_archives"). Ignored when archive_sink is set. archive_sink (Callable|None): Receives the archived conversation instead of a local file, for a host whose filesystem is ephemeral (default None). Called with the sanitised, JSON-serialisable message list; a string it returns is used wherever the file path would have been. An exception from it is logged at WARNING and never fails the run. Python only — a callable has no YAML spelling. include_archive_path_in_summary (bool): Include the archive's path (or the sink's locator) in summary output (default True). clear_tool_inputs (bool): Also clear tool call parameters (default False). combined_order (list[str]): Order of strategies for combined (default ["tool_result_clearing", "sliding_window", "summarize"]). max_compactions_per_session (int|None): Limit on compaction count (default None). summary_max_tokens (int): Maximum output tokens for the summarization LLM call (default 4096). Must be a positive integer and should not exceed the model's max_output_tokens_request. sliding_window_include_truncation_notice (bool): When True, the sliding_window strategy prepends a synthetic user message informing the model that earlier context was truncated. This helps the model understand that it may be missing prior conversation context. Defaults to True.	None
session_history_config	Optional[Dict[str, Any]]	Default session history configuration dict. Applied to all sessions created via agent.session() unless overridden per-session. Supported keys: mode (str): History filtering mode — "full", "messages_only", "no_tool_results", "no_reasoning", "compact_tool_results", or "text_summary" (default "full"). include_system_prompt (bool): Include system prompt in history (default True). compact_tool_results_max_chars (int): Max chars per tool result when mode="compact_tool_results" (default 500).	None
logger	Optional[Logger]	Optional pre-configured Logger instance. If provided, used as-is; otherwise one is created via _setup_logger.	None
log_level	str	Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL, NOTSET). Will be overwritten by the AGENT_LOG_LEVEL environment variable if set.	'DEBUG'
log_file	Optional[str]	Optional log file path. If not provided, defaults to ./logs/Agent-{model}_{timestamp}.log.	None
log_to_file	bool	Write the log file (default True). False — or the DPT_AGENT_LOG_TO_FILE=0 environment variable — makes the logger built when logger is None console-only: no file is opened, ./logs/ is not created, and log_file is ignored with a WARNING. Has no effect when logger is supplied.	True
tool_choice	str	Tool selection strategy: "auto", "required", "none", "any", "validated", or a specific tool name. Default: "auto" (Google default is "validated").	'auto'
code_execution_output_dir	Optional[str]	Output directory for the files a provider's code execution writes. None (the default) creates a fresh ./<provider>_code_execution_files_<timestamp>_<random>/ for each run that produces files (0.5.3; before, the timestamp alone, so two runs in one second shared a folder). An explicit directory is shared across runs by design, and a file an earlier run left there is overwritten as before. A name the run itself already wrote is written as name (2).ext and the record in downloaded_files carries the name written; AgentResult.files_created lists the paths written. Claude downloads on both of its paths since 0.5.3; until then the tool-calling path (local tools present) downloaded only when this directory or skills was set.	None
service_tier	Optional[str]	Requested service tier, e.g. "priority". On OpenAI this maps to Fast mode (2x token pricing, higher tokens/sec and priority handling); on Google to priority inference (1.8x pricing, non-sheddable processing). "flex" (half price, slower) and, on gpt-6-astra only, "ultrafast" (6x Standard, about three times Fast's output speed) are requested the same way. Ignored with a log line for providers/models without a matching service_tiers pricing row in model_information.json (the tier vocabulary is data-driven). Billing always follows the tier that actually SERVED each request: a request served at another tier is billed at the tier that ran and sets AgentResult.tier_downgraded.	None
hooks	Optional[Any]	Optional hooks object or callable for real-time steering (see :mod:doc_processing_toolkit.agent.hooks). Awaited at run/iteration boundaries and around every tool call; may veto tool calls (skip_tool), inject user-role messages at turn boundaries (inject_message), pause (resume via agent.run_handle.resume()), or abort (raises AgentAborted). Coexists with stream_callback: the callback is a passive tap, hooks are the control channel. Text/thinking deltas are not routed to hooks (use the callback for those) — awaiting a hook per token would stall streaming.	None
emit_request_start	bool	Emit a request_start event immediately before every ACTUAL provider send — the first request, each tool-loop iteration, each max_tokens / truncation / empty-turn re-request, each pause_turn continuation and each compaction summarisation call (B10). Default False, so a consumer that enumerates event types sees exactly the events earlier releases produced. When True the event goes to stream_callback on both public paths and, on the tool-calling path, to hooks as well — where it is OBSERVATION-ONLY: a directive other than continue is ignored with a WARNING, because a send point has no pending tool call to veto and no legal place to insert a message. metadata carries request_index (1-based, monotonic across the whole run), iteration, attempt (the tenacity attempt number), purpose, model and provider. run_start is unchanged and still fires once per run.	False
max_tool_result_chars	Optional[int]	Character cap on each tool result fed back to the model. None (the default) derives it from the registry: context_window_tokens * 4 * 0.25 — a quarter of the window at four characters per token, i.e. 200,000 characters on a 200k-token model and 1,000,000 on a 1M-token model (utils.TOOL_RESULT_CAP_CONTEXT_FRACTION / utils.TOOL_RESULT_CAP_CHARS_PER_TOKEN). A longer result is cut and ends with … [truncated: N of M characters shown; raise max_tool_result_chars to keep more], with one WARNING. Applies identically on all four providers to every result the toolkit executes (@tool functions and every client-side MCP format, mcp_remote_client included); results of provider-executed remote MCP servers never pass through the toolkit and are not capped. 0 turns the cap off (the kill switch — the previous behaviour), as does any value larger than a result can be. The raw result in hooks and stream_callback events is untouched.	None
tool_result_preview_chars	int	How much of each tool result the tool_call_result stream and hook events carry: the client's copy, which is independent of the model's copy that max_tool_result_chars cuts, so it never changes what the model is sent. 500 (the default) is the preview the events always carried. 0 carries the whole result, and adds result_data: the result as JSON data (a tool that returned a dict gives the dict; other MCP results give their content items, JSON text decoded; see utils.tool_result_data). Every event also carries result_truncated, so a client knows whether it holds everything. Keep the default where events are logged or stored: one result can be megabytes (DEBUG_AGENT_STREAMING mirrors every event into the log).	DEFAULT_TOOL_RESULT_PREVIEW_CHARS
truncated_result_read_chars	Optional[int]	How many more characters of a truncated tool result the model may read back, per result, beyond what it was first shown. None (the default) or 0 turns it off, and nothing changes. When set, the built-in read_truncated_result(tool_call_id, offset) tool is on the tool list from the first request of every run with client-side tools, and the marker on a result max_tool_result_chars cuts becomes … [truncated: N of M characters shown. Call read_truncated_result(tool_call_id="…", offset=N) to read up to K more.]. Each call returns the next slice, at most max_tool_result_chars long, ending with where to continue; every character returned counts against the allowance, and past it the tool answers with a short refusal. Truncated results are kept in full in memory for the run, or for an AgentSession's lifetime, and are never exported with a session. Has no effect with max_tool_result_chars=0 (nothing is cut) or a forcing tool_choice (the run ends after one tool round), and is not offered when a loaded tool is already named read_truncated_result (one WARNING for each).	None
prompt_caching	Optional[Any]	Opt-in Anthropic prompt caching. True or a dict {"ttl": "5m" | "1h"}. Places cache_control breakpoints on the system block (caching tools + system together) and on the most recent turn so each tool roundtrip reads the prior prefix from cache (reads bill at 0.1x input, writes at 1.25x for 5m / 2x for 1h — tracked via cached_input_tokens / cache_write_tokens). Prefixes below the model's minimum cacheable length (cache_min_prompt_tokens in the registry) are simply not cached by the API. Ignored for non-Anthropic providers (OpenAI/Google/Fireworks cache automatically server-side).	None
anthropic_block_binding	Optional[str]	How the API should treat a replayed thinking block whose conversation prefix no longer matches. "drop_block" has the server drop the stale blocks and report them in :attr:AgentResult.input_transformations; "error" fails the request; "off" sends nothing. None (the default) defers to DPT_ANTHROPIC_BLOCK_BINDING and then to the model's registry row. Four rows carry the registry flag (claude-fable-5-1, claude-opus-5-5, claude-sonnet-5-5, claude-haiku-5-5) and send "drop_block" plus the beta on every thinking request by default; on any other extended-thinking row "drop_block" is the opt-in that makes a model-binding drop (history produced by a model the serving one cannot read) visible in :attr:AgentResult.model_binding_mismatches. This setting has no effect on that drop itself: an unreadable block is always dropped. A caller-supplied thinking={"block_binding": ...} always wins. Anthropic-only.	None
strict_prefix_binding	bool	Sugar for anthropic_block_binding="error". Turn it on in CI so a history edit, an MCP server whose tool list moved, or a rebuilt instructions string fails loudly in test while production stays tolerant. Also makes a mid-session tool-surface change fatal rather than frozen. Anthropic-only (default: False).	False
prefix_audit	bool	Shadow mode. Fingerprints system, tools and every message before each Anthropic request and logs a WARNING naming what changed since the previous request — without altering a single request byte, on every model, with no beta header. The only mechanism that catches MCP tool-list drift at a client site. Also enabled process-wide by DPT_PREFIX_AUDIT (default: False).	False
prefix_strip_fallback	bool	For a client that cannot send the binding beta header. Strips all thinking and redacted_thinking blocks from the replayed history once per run, at the completed-turn boundary before the first request. The model loses that reasoning; the request never carries a stale binding. Anthropic-only (default: False).	False
remove_citations	bool	Remove citation markers from output (default: False).	False
on_unsupported_part	str	What to do with a prompt part the model cannot accept — an image sent to a text-only model, a video sent to a model without video input, a PDF sent to Fireworks (B12). "drop" (default): today's behaviour — log a warning and send the request without the part. "raise": refuse the run with UnsupportedPartError before any request is sent, so a confident answer about a document the model never received is impossible. Either way every dropped part is listed on AgentResult.dropped_parts.	'drop'
media_policy	str	How this agent's sessions write media when saved (to_dict / to_json; session-portability P2): "auto" (default) externalises when a media_store is configured and keeps media inline otherwise; "inline" writes every part exactly as stored; "externalize" hands every inline payload to the store and leaves a $media:<ref> marker; "strip" replaces every payload by a text placeholder. A per-call media= overrides it.	'auto'
media_store	Any	Where externalised bytes live — a directory path (a DirectoryMediaStore) or any object with put(sha256, mime_type, data) -> ref and get(ref) -> bytes. Inherited by every session of this agent; also the store a restore reads from unless from_dict / from_json are given one.	None
media_max_inline_bytes	Optional[int]	Per-part cap (8 MiB by default) on the bytes of a Google Files API upload that a session keeps beside the stored file_uri when no store is configured, so a restore after the 48-hour expiry can re-upload. Above the cap the bytes are not kept (a WARNING and a media_report entry); None lifts the cap. Inline parts on the other providers are never removed by the cap.	DEFAULT_MEDIA_MAX_INLINE_BYTES
estimate_input_tokens	Any	How the pre-flight input-token estimate is produced. The estimate seeds _estimated_input_tokens, which only the stream-failure cost record reads (B14). "lazy" (the default): a purely local approximation — no API call and, on Google, no Files API upload. Before this, every run uploaded each media part to count it and deleted it again, so a request with one image uploaded the same bytes twice (measured 4-5 s of that on the event loop per turn). True: the previous behaviour verbatim — the provider's own token-counting API, including that upload. False: no estimate at all (0), as the previous behaviour already did whenever the count raised.	'lazy'
defer_tools	Optional[Any]	Which client tools to hide behind the provider's native tool-search tool (C4), so a large tool surface stops consuming the cached prompt prefix. Either a set of tool names, or a predicate called as predicate(tool_name, server_name) where server_name is the MCP server the tool came from (None for a tool with no server). Requires built_in_tools={"tool_search": BuiltInToolConfig(enabled=True)} on a model whose registry row lists tool_search — Anthropic and OpenAI only. Nothing here can make a tool unreachable: if tool search is not enabled, or the provider has no native tool search (Google, Fireworks), every tool loads eagerly and one warning names why. The tools actually deferred are listed on AgentResult.deferred_tools. Example: agent = Agent( model="claude-haiku-4-5", tools=[big_mcp_server], built_in_tools={"tool_search": BuiltInToolConfig(enabled=True)}, defer_tools=lambda name, server: server == "catalog", )	None
detailed_costs	bool	Enable detailed cost tracking with per-iteration metrics.	False
cost_output_dir	Optional[str]	Directory for cost reports. Defaults to agent_costs. When detailed_costs=True, reports are saved as JSON files.	None
**provider_kwargs	Any	Provider-specific parameters passed directly to provider functions. A keyword this provider's entry points do not accept, or one the Agent supplies itself (system_prompt, client, logger and similar), is still passed through but logs one WARNING at construction naming it (with the Agent parameter to use instead where there is one, e.g. instructions= for system_prompt=). Supported parameters by provider: Anthropic: - effort_config (str): Computational effort — "low", "medium", "high", or "max" (Opus 4.6 only). - skills: Agent Skills for document tasks — "xlsx", "pptx", "docx", "pdf" or custom skill file paths. Supported formats: - Built-in skill name (str): "xlsx", "pptx", "docx", or "pdf" - Multiple built-in skills (List[str]): ["xlsx", "pptx"] - Custom skill directory path (str): "/path/to/custom_skill/" (must contain SKILL.md) - Mixed skills (List): ["/path/to/skill/", "xlsx", "/path/to/skill2/"] - Advanced dict format: {"type": "anthropic", "skill_id": "xlsx", "version": "20251013"} - max_pause_turn_iterations (int): Max iterations for pause_turn stop reason (default: 10). OpenAI: - verbosity_config (str): Output verbosity — "low", "medium", "high". - parallel_tool_calls (bool): Allow parallel tool calls (default: True). - image_detail (str): Image detail level applied uniformly to all images in the request (not per-image) — "low", "high", "auto", or "original" (gpt-5.4+ only). - ``openai_transport`` (str): ``"http"`` (the default) or ``"websocket"`` (0.5.2). ``"websocket"`` sends the requests of a run with local tools over one WebSocket to the Responses API, with the same results, billing and retries as HTTP. Unset, ``DPT_OPENAI_TRANSPORT`` is the process default. An unrecognised value is ignored with one warning, and a run with no local tools stays on HTTP. Google: - ``video_metadata`` (dict): Video preprocessing config — {"start_offset": str, "end_offset": str, "fps": float}. - ``media_resolution`` (str): Global media resolution setting. Supported values (model-dependent): - "MEDIA_RESOLUTION_UNSPECIFIED": Default setting (varies by model) - "MEDIA_RESOLUTION_LOW": Lower detail, faster processing, lower cost - "MEDIA_RESOLUTION_MEDIUM": Balance of detail, cost, and latency - "MEDIA_RESOLUTION_HIGH": Higher detail (recommended for most use cases) - ``file_active_timeout_s`` (float): Seconds to wait for a Files API upload to reach ``ACTIVE`` before the request proceeds. Applies to every media part. Defaults to 60 s for images and 120 s for video/audio/document, which a long MP4 exceeds — a ten-minute video needs roughly 600. Bounds 1-3600; out of range raises. Environment default: ``DPT_GOOGLE_FILE_ACTIVE_TIMEOUT_S``. - ``file_poll_interval_s`` (float): Seconds between state checks while waiting. Bounds 0.05-60, must not exceed the timeout. Environment default: ``DPT_GOOGLE_FILE_POLL_INTERVAL_S``. - ``file_poll_backoff`` (float): Multiplier applied to the poll interval after each check (bounds 1-10). ``1.0`` keeps the fixed cadence. Environment default: ``DPT_GOOGLE_FILE_POLL_BACKOFF``.	{}

Raises:

Type	Description
ValueError	If model is invalid or parameters are out of range.

builder() classmethod

Create an AgentBuilder for fluent configuration.

Returns:

Type	Description
AgentBuilder	AgentBuilder instance.

close() async

Close the agent and release all resources.

This method should be called when the agent is no longer needed to properly clean up MCP server connections, client sessions, and other resources.

Can be called explicitly or via the async context manager:

Example

# Explicit close
agent = Agent(model="claude-sonnet-4-5")
try:
    result = await agent.run("Hello")
finally:
    await agent.close()

# Context manager (recommended)
async with Agent(model="claude-sonnet-4-5") as agent:
    result = await agent.run("Hello")

Notes

MCP teardown is symmetric with the connect that run() performs: the sessions and the tools they registered are dropped, but the server configs are kept, so the Agent is reusable — the next run() reconnects them. Provider clients served from the shared process cache are deliberately left open; only an Agent built with an explicit api_key owns and closes its clients.

from_config(config_path, **overrides) classmethod

Create an Agent from a YAML configuration file.

Parameters:

Name	Type	Description	Default
config_path	str	Path to YAML configuration file.	required
**overrides	Any	Runtime parameter overrides. max_iterations, temperature and max_tokens replace the constraints values; any other key replaces the top-level YAML key of the same name. max_tool_result_chars, tool_result_preview_chars and truncated_result_read_chars are validated after the override is applied. logger is passed through. log_file and log_level apply only when the YAML has no logging block, and log_to_file only when that block has no to_file key.	{}

Returns:

Type	Description
Agent	Configured Agent instance.

Raises:

Type	Description
FileNotFoundError	If config file doesn't exist.
ConfigValidationError	If config is invalid.

Example

model: claude-sonnet-4-5
instructions: |
  You are a helpful assistant.

api_key: ${ANTHROPIC_API_KEY}  # Optional, uses env var by default
tool_choice: auto  # auto, required, none, any
max_tool_result_chars: 50000  # tool-result cap; 0 = off, omit = registry default
tool_result_preview_chars: 0  # events carry the whole result; omit = 500
truncated_result_read_chars: 200000  # read_truncated_result allowance; omit = off
code_execution_output_dir: ./output

constraints:
  max_iterations: 10
  max_tokens: 4096
  temperature: 0.7

thinking:
  budget_tokens: 10000
  interleaved: true

built_in_tools:
  web_search:
    enabled: true
    config:
      max_uses: 5

tools:
  - path/to/tools.py
  - type: mcp_remote
    url: https://api.example.com/mcp

logging:
  level: INFO
  file: agent.log
  to_file: true   # false: console only, no ./logs/ file

provider_overrides:
  effort_config: high

from_config_string(yaml_content, **overrides) classmethod

Create an Agent from a YAML string.

Parameters:

Name	Type	Description	Default
yaml_content	str	YAML configuration as a string.	required
**overrides	Any	Runtime parameter overrides. Applied exactly as in :meth:from_config: constraint keys replace constraints values, other keys replace top-level YAML keys, max_tool_result_chars, tool_result_preview_chars and truncated_result_read_chars are validated afterwards, and log_file / log_level / log_to_file yield to the YAML logging block.	{}

Returns:

Type	Description
Agent	Configured Agent instance.

Raises:

Type	Description
ConfigParseError	If YAML parsing fails.
ConfigValidationError	If config is invalid.

run(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None) async

Run the agent with a message.

Parameters:

Name	Type	Description	Default
message	str	The user message/task.	required
files	Optional[List[Any]]	Optional list of files to include with the message. Accepted formats: str — File path (e.g., "data.csv", "image.png"), URL, or YouTube URL. pathlib.Path — Path object pointing to a local file. bytes — Raw file bytes (media type auto-detected or set via kwargs). dict — Pre-structured content: {"type": "base64", "media_type": str, "data": str} or {"type": "document", "media_type": str, "data": bytes}. Supported media types vary by provider (images: all providers; PDF/text: Anthropic, OpenAI, Google; video/audio: Google only).	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool (where supported by Anthropic, OpenAI, and Google). Files should be in the same format as the files argument. File type validation is performed by the provider API. Files are uploaded automatically via the provider's Files API (Anthropic Files API, OpenAI Files API, Google Files API). Unsupported providers (e.g., Fireworks) are filtered automatically.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. The schema is used to VALIDATE the model's JSON output; the value placed on AgentResult.output is the parsed dict, not an instance of the schema class.	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). List of resolution strings matching the order of media parts in files. Use None for parts that should use the global/default resolution. Non-media parts will be ignored and don't need an entry in the list. Supported values (model-dependent): "MEDIA_RESOLUTION_UNSPECIFIED": Default setting (varies by model) "MEDIA_RESOLUTION_LOW": Lower detail, faster processing, lower cost "MEDIA_RESOLUTION_MEDIUM": Balance of detail, cost, and latency "MEDIA_RESOLUTION_HIGH": Higher detail (recommended for most use cases) "MEDIA_RESOLUTION_ULTRA_HIGH": Highest detail (Gemini 3 per-part only). Example: ["MEDIA_RESOLUTION_HIGH", "MEDIA_RESOLUTION_LOW", None] for a message with 3 media parts. Non-Google providers will gracefully ignore this parameter.	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse from a prior run. Files uploaded via tool_files and REPL state inside the container persist for up to 30 days and can be resumed by passing the container id returned in AgentResult.container_id. Auto-enables code_execution if the Agent did not declare it. Implicitly instructs the Anthropic provider to preserve (not delete) uploaded tool-file ids and custom-skill ids so they remain usable by the reused container on later runs. Ignored (with a debug log) for non-Anthropic providers.	None

Returns:

Type	Description
AgentResult	AgentResult with output and metadata.

run_sync(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None)

Synchronous wrapper for run().

This method handles the event loop internally, including nested event loop scenarios (e.g., Jupyter notebooks, async test frameworks).

Parameters:

Name	Type	Description	Default
message	str	The user message/task.	required
files	Optional[List[Any]]	Optional list of files to include with the message. Accepted formats: str — File path (e.g., "data.csv", "image.png"), URL, or YouTube URL. pathlib.Path — Path object pointing to a local file. bytes — Raw file bytes (media type auto-detected or set via kwargs). dict — Pre-structured content: {"type": "base64", "media_type": str, "data": str} or {"type": "document", "media_type": str, "data": bytes}. Supported media types vary by provider (images: all providers; PDF/text: Anthropic, OpenAI, Google; video/audio: Google only).	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool (where supported by Anthropic, OpenAI, and Google). Files should be in the same format as the files argument. File type validation is performed by the provider API. Files are uploaded automatically via the provider's Files API. Unsupported providers (e.g., Fireworks) are filtered automatically.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. As with run(), the schema validates the output and AgentResult.output is the parsed dict, not a model instance.	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). See run() for full documentation.	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse from a prior run. See run() for full documentation.	None

Returns:

Type	Description
AgentResult	AgentResult with output and metadata.

Notes

Local MCP servers (stdio scripts/commands, pre-built sessions) live for one event loop. asyncio.run creates and closes a loop per call, so each run_sync call connects them, runs, and releases them inside that loop. Use await run() on a long-lived loop to reuse one server process across runs until :meth:close.

session(session_id=None, session_history_config=None, reuse_anthropic_container=False, late_injection_policy='surface', max_follow_up_turns=1)

Create a new session for multi-turn conversations.

Parameters:

Name	Type	Description	Default
session_id	Optional[str]	Optional session ID. Auto-generated if not provided.	None
session_history_config	Optional[Dict[str, Any]]	Optional session history config dict. See AgentSession.__init__ for supported keys.	None
reuse_anthropic_container	bool	When True, capture the Anthropic code execution container_id from the first run and auto-inject it into every subsequent run in this session so uploaded files and REPL state persist. Ignored with a warning log for non-Anthropic providers. See AgentSession.__init__ for details.	False
late_injection_policy	str	What the session does with a hook message that arrived too late for any turn boundary to deliver it. "surface" (default) parks it for the host; "new_turn" sends it as the next turn automatically. See AgentSession.__init__.	'surface'
max_follow_up_turns	int	Automatic follow-up turns allowed per call under late_injection_policy="new_turn" (default 1).	1

Returns:

Type	Description
AgentSession	AgentSession instance.

Raises:

Type	Description
ValueError	If late_injection_policy or max_follow_up_turns is invalid.

stream(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None) async

Stream the agent's response as typed events.

Yields AgentStreamEvent objects in real time as the model generates output. Uses the same internal execution as run(), but emits deltas incrementally via the stream_callback mechanism.

Parameters:

Name	Type	Description	Default
message	str	The user message/task.	required
files	Optional[List[Any]]	Optional list of files to include with the message. Accepted formats: str — File path, URL, or YouTube URL. pathlib.Path — Path object pointing to a local file. bytes — Raw file bytes. dict — Pre-structured content dict.	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool (where supported by Anthropic, OpenAI, and Google). Files should be in the same format as the files argument. File type validation is performed by the provider API. Files are uploaded automatically via the provider's Files API. Unsupported providers (e.g., Fireworks) are filtered automatically.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. When supplied, text_delta events carry the streaming raw JSON payload; the final parsed dict is attached to the complete event's metadata["output"] (the schema is used for validation only — the value is a dict, not a model instance).	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). See run() for full documentation.	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse from a prior run. See run() for full documentation.	None

Yields:

Type	Description
AsyncIterator[AgentStreamEvent]	AgentStreamEvent objects with event_type and content/metadata.

Notes

Event types emitted by the stream. Seven are always available; two more (turn_start / turn_complete) appear only under AgentSession(late_injection_policy="new_turn"), and request_start only under Agent(emit_request_start=True).

• "text_delta": Partial text from the model. content is the text chunk. metadata is None for ordinary prose and {"structured_output": True} for the deltas of a schema-constrained (Path A) structured-output request, so a UI can suppress the terminal JSON without guessing. On Path B (tool-based structured output) the payload never reaches this event.
• "thinking_delta": Partial reasoning output when the provider exposes it.
• "tool_call_start": Tool call started; metadata may include tool_name, tool_call_id, arguments, and iteration.
• "tool_call_result": Tool call finished; metadata may include tool_name, tool_call_id, result, result_truncated, success, duration_ms, and iteration. result is the first tool_result_preview_chars characters (500 by default) of the result's str(); at 0 it is the whole result and result_data carries it as JSON data. A tool that reported spend with the _dpt_spend marker adds tool_reported_cost and tool_reported_cost_estimated (0.5.1).
• "iteration_complete": Iteration finished; metadata may include input_tokens, output_tokens, cost, tools_called, and stop_reason. The usage covers every request billed since the previous event, so the events sum to the run's requests up to the last event (0.5.1).
• "error": Unrecoverable error. content is the error message.
• "complete": Stream finished. metadata is the serialized AgentResult.
• "turn_start" / "turn_complete": Only under AgentSession(late_injection_policy="new_turn"). See :class:~doc_processing_toolkit.agent.cost.AgentStreamEvent.
• "request_start": Only when the Agent was built with emit_request_start=True. One event immediately before every ACTUAL provider send — including each retry, each max_tokens or truncation re-request, each pause_turn continuation and each compaction summarisation call — so the count matches len(result.request_cost_records) for a clean run. metadata carries request_index, iteration, attempt, purpose, model and provider. run_start is unchanged: it is a hook point, fires once per run, and is not a stream event.

Failures and aborts. Unlike run(), stream() never raises: a failed run yields an error event and then still terminates with a complete event, so teardown bound to complete runs exactly once per stream. The error event's metadata carries input_tokens, output_tokens, cost and iterations recovered from the failure (token_source is "stream_partial" when partial data was recovered, "unknown" otherwise), plus aborted (True when a hook aborted the run via HookDirective(action="abort") or RunHandle.abort) and reason (the hook's rationale, else None). The trailing complete event's metadata is an AgentResult dict with error set and the same partial usage — which is what lets AgentSession.stream() bank the spend of an aborted turn.

Stopping early. Closing the generator (aclose(), or contextlib.aclosing around a break) cancels the run within one chunk and waits at most :data:STREAM_CANCEL_WAIT_SECONDS for it. What the run had already spent is then reported as a complete-style record with "abandoned": True on Agent.last_abandoned_stream, to hooks (observation only) and, for AgentSession.stream(), into the session's totals. A bare break with the generator still referenced does not close it. A streamed run without local tools makes its provider call in a worker thread, so its deltas arrive as they are generated. DPT_STREAM_CANCEL_ON_CLOSE=off and DPT_STREAM_STRUCTURED_OFFLOOP=off restore the previous behaviour of each.

Examples:

Basic usage:

async for event in agent.stream("Hello"):
    if event.event_type == "text_delta":
        print(event.content, end="", flush=True)
    elif event.event_type == "complete":
        result = event.metadata  # Full AgentResult as dict

Stopping early, with the spend reported:

async with contextlib.aclosing(agent.stream("Write an essay")) as stream:
    async for event in stream:
        if event.event_type == "text_delta" and should_stop():
            break
spent = agent.last_abandoned_stream  # None if it completed

to_config()

The data half of this agent's constructor, in the YAML shape from_config reads.

Session-portability P6. A host stores the recipe beside a session envelope and rebuilds the same agent later with Agent.from_config_string(yaml.safe_dump(agent.to_config())) or by writing the dict to a YAML file for Agent.from_config. The dict is plain data: model, instructions, constraints, the tool and media policies, thinking, built-in tools, compaction, the session history mode, the Anthropic prefix controls, logging and provider_overrides (the extra provider keywords the agent was built with). tools carries every entry that is data — a path, a YAML tool dict, an MCP server spec — exactly as it was given.

What cannot be serialised is listed, never silently dropped, under the top-level unserialisable key: Python callables and objects given as tools (by name; since 0.5.1 also a pre-built {"session": ...} entry and the {"type": "fastmcp", "server": ...} dict form, which hold a live object), a hooks object, a custom media_store (a DirectoryMediaStore is emitted as its directory path), a logger instance, and whether an api_key was set (the key itself is never emitted). Since 0.5.3 the env of an mcp_command / mcp_script entry is treated the same way: an API key may be among its values, so the entry is emitted without env and the variable names are listed under unserialisable["tool_env"] ({server name: [names]}); a path-like cwd is emitted as a string. from_config ignores that key, so the recipe loads as it is; the host supplies those objects and values again.

Returns:

Type	Description
Dict[str, Any]	A JSON-clean, YAML-safe dict.

Example

import yaml

recipe = agent.to_config()
(checkpoints / "agent.yaml").write_text(yaml.safe_dump(recipe))
again = Agent.from_config_string(yaml.safe_dump(recipe), tools=[add])   # re-supply callables

AgentBuilder

Builder pattern for Agent construction.

Provides a fluent interface for configuring agents.

Example

agent = (
    Agent.builder()
    .model("claude-sonnet-4-5")
    .instructions("You are helpful.")
    .add_tool(my_tool)
    .with_thinking(budget_tokens=10000)
    .build()
)

add_tool(tool)

Add a tool.

anthropic_block_binding(behaviour='drop_block')

Set the Anthropic thinking block-binding behaviour.

Parameters:

Name	Type	Description	Default
behaviour	Optional[str]	"drop_block", "error", "off", or None to fall back to DPT_ANTHROPIC_BLOCK_BINDING and the registry.	'drop_block'

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

auto_inject_skills(enabled=True)

Enable/disable auto skills injection.

build()

Build and return the Agent.

instructions(instructions)

Set the system instructions.

max_iterations(n)

Set max iterations.

max_tokens(n)

Set max tokens.

media_max_inline_bytes(limit)

Set the per-part cap on Google upload bytes kept in the envelope (None lifts it).

Parameters:

Name	Type	Description	Default
limit	Optional[int]	See Agent(media_max_inline_bytes=...).	required

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

media_policy(policy='auto')

Set how the agent's sessions write media when saved (auto / inline / externalize / strip).

Parameters:

Name	Type	Description	Default
policy	str	See Agent(media_policy=...).	'auto'

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

media_store(store)

Set where externalised media bytes live: a directory path or a MediaStore.

Parameters:

Name	Type	Description	Default
store	Any	See Agent(media_store=...).	required

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

model(model_name)

Set the model name.

prefix_audit(enabled=True)

Log would-be prefix mismatches without changing any request byte.

Parameters:

Name	Type	Description	Default
enabled	bool	True turns the shadow audit on for this agent.	True

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

prefix_strip_fallback(enabled=True)

Strip all replayed thinking instead of sending the binding beta.

Parameters:

Name	Type	Description	Default
enabled	bool	True removes every thinking block from the history a run replays, at the completed-turn boundary.	True

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

prompt_caching(enabled=True)

Opt in to Anthropic prompt caching. Pass True for the default 5m TTL or a dict such as {"ttl": "1h"}; non-Anthropic providers cache automatically and ignore this.

service_tier(tier)

Request a service tier ("priority", "flex" or "ultrafast"): OpenAI Fast, Flex and Ultrafast mode / Google priority and flex inference; validated data-driven against the model's service_tiers pricing rows.

strict_prefix_binding(enabled=True)

Fail loudly on a thinking prefix mismatch instead of dropping blocks.

Parameters:

Name	Type	Description	Default
enabled	bool	True maps to anthropic_block_binding="error" and makes a mid-session tool-surface change fatal.	True

Returns:

Type	Description
AgentBuilder	This builder, for chaining.

temperature(t)

Set temperature.

with_built_in_tools(**kwargs)

Configure built-in tools.

with_compaction(enabled=True, **config)

Configure compaction.

with_thinking(**kwargs)

Configure thinking/reasoning.

Agent Result

doc_processing_toolkit.agent.result

AgentResult dataclass for representing the output of an Agent run.

AgentResult dataclass

Result of an Agent.run() call.

Contains the output, token usage, cost, and metadata about the agent execution. Includes all unique output fields across all provider structured output and function calling functions.

Attributes:

Name	Type	Description
output	Any	The agent's response. Can be a string, or a dict parsed from the model's JSON output when output_schema was provided.
raw_response	Any	The raw response from the provider API (for debugging).
messages	List[Dict[str, Any]]	The conversation messages including tool calls and responses.
injected_messages	List[str]	The flat, ordered list of hook-injected texts this run's turn-boundary drains actually delivered to the model. Each consumed injection appears exactly once, in the order the hook queued it, and is also present inline in :attr:messages as an ordinary user-role message. Empty on every run without hooks.
undelivered_injections	List[str]	Hook-injected texts that arrived after the run's last turn boundary and were therefore never shown to the model — queued at the final iteration_complete, returned by the run_end hook itself, or queued at a forced-tool_choice exit that sits above the loop's drain. They are not in :attr:messages and not in :attr:injected_messages. The toolkit deliberately does not extend the run to deliver them (see docs/concepts/hooks.md); deliver them as the next turn's user message. Empty on every run without hooks.
media_uploads	List[Dict[str, Any]]	One record per Google Files API upload this run made (or reused from the in-process cache), in prompt order: {"part_index", "uri", "name", "mime_type", "size", "sha256", "uploaded_at", "expiration_time"}, all JSON-clean. An AgentSession reads it to keep the bytes and the expiry beside the stored file_uri, so a restore after the 48-hour expiry can re-upload. Empty on every other provider and on a Google run with no uploads.
dropped_parts	List[Dict[str, Any]]	Every prompt part this run's model could not accept, in prompt order, recorded unconditionally — under the default on_unsupported_part="drop" as well as under "raise" (B12). Each entry is {"kind", "index", "reason", "provider", "model"}, where kind is the input-type key the validator built ("video", "base64-application/pdf", …) and index is the part's position in user_prompt_parts. Empty on every run whose parts were all supported. Before this field the only evidence was a log line, so an image sent to a text-only model produced a confident answer about a picture the model never saw.
iterations	int	Number of tool-calling iterations performed.
tools_used	List[str]	List of custom MCP tool names that were called during execution.
built_in_tools_used	List[str]	List of built-in tool names CONFIGURED for the run (e.g., web_search, code_execution) — an echo of the request configuration, not evidence of use. See built_in_tools_invoked.
built_in_tools_invoked	List[str]	List of built-in tool names with evidence they actually RAN during this run (observed usage: searches counted, containers observed, grounding metadata, MCP call items). A provider-executed remote MCP call adds the neutral "remote_mcp" beside the tool's own name, on every provider. Empty when a configured tool was never invoked.
deferred_tools	List[str]	The client tool names actually stamped defer_loading: true on this run's wire tools array, in wire order (C4). Empty whenever deferral did not apply — no defer_tools, no tool_search built-in tool, a provider without a native tool search, or a name that matched nothing. So an empty list beside a non-empty defer_tools argument is the signal that tools loaded eagerly, and the reason is in the run's warnings. Anthropic and OpenAI only; the other two providers never populate it.
input_tokens	int	Total input tokens consumed across all iterations.
output_tokens	int	Total output tokens generated across all iterations.
reasoning_tokens	int	Thinking/reasoning tokens across the run. A SUBSET of :attr:output_tokens, never additional to it, on all four providers — Anthropic reports them inside output_tokens, OpenAI and Fireworks inside their completion total, and Google reports them separately but the toolkit folds them into output_tokens before pricing. They are therefore already billed and are surfaced for observability only. 0 when thinking was off or the provider reported no breakdown.
cached_input_tokens	int	Prompt-cache READ tokens across the run. A SUBSET of :attr:input_tokens, never additional to it — costing subtracts this share out of the input total and re-prices it at the cache-read rate. Reported by every provider.
cache_write_tokens	int	Prompt-cache WRITE tokens. On Anthropic they are billed on top of input at the write rate (1.25x for a 5m TTL, 2x for 1h); on OpenAI GPT-5.6+ they are a SUBSET of input_tokens, re-priced at 1.25x the uncached input rate. Zero for providers and models that report no writes.
request_cost_records	List[Dict[str, Any]]	One record per API request — input/output/cached tokens, served_model, requested/served tier and cost. The run's token cost is the SUM of these records, never a single costing of the summed tokens (tiered models overbill that way). Structured-output runs make one request, so their single record is synthesised from the run-level fields and carries synthesised_from_run_totals. Google records also carry served_tier_source ("body", "header" or "none"), which says how the served tier was established. An OpenAI tool-loop record carries estimated: True (0.5.2) when a failure event ended its request on the first attempt after part of the reply had streamed: its output tokens are a count of the streamed text, priced at standard rates with no input tokens, and :attr:cost_estimated is True.
tier_downgraded	bool	True when any request in the run was not served at the tier it asked for. For a "priority" request that means it ran and billed at standard rates — the silent downgrade this flag was named for. An "ultrafast" request served at "priority" or at standard is billed at that cheaper row. For a "flex" request the mismatch goes the other way: flex is the cheaper row, so being served "standard" bills MORE than asked. The flag is the same "you did not get the tier you asked for" signal either way; read requested_tier / served_tier on :attr:request_cost_records, or :attr:service_tier_served, for the direction. False on every run that requested no tier.
cost	float	Estimated cost in USD for this run: the request records, the built-in tool charges and :attr:tool_reported_cost.
cost_estimated	bool	Whether the cost is an estimate (True) or actual (False). Also True when a tool reported its spend as an estimate.
tool_costs_estimated	Dict[str, bool]	Dict mapping tool names to whether their costs are estimated: built-in tools, and every tool that reported spend.
tool_reported_cost	float	What the run's own tools reported spending through the _dpt_spend marker (the multimedia MCP tools, the agent factory's create_agent and suggest_agent_config, or any tool returning utils.tool_spend_marker(...)), in USD. Already included in :attr:cost. 0.0 with DPT_AGENT_TOOL_REPORTED_COST=off.
tool_reported_costs	List[Dict[str, Any]]	One record per reporting tool call: {"tool_name", "tool_call_id", "usd", "estimated", "iteration"}.
token_source	Optional[str]	Where the token counts came from. One documented vocabulary: "api_response" (the provider reported them), "stream_partial" (accumulated from a stream that then failed), "rejected_before_stream" (the provider refused the request before streaming, so nothing was billed), "estimated" and the "*_metadata_missing" variants (counted locally), or None. The legacy spelling "api_metadata", which the single-call provider paths emit for the same fact as "api_response", is normalized to "api_response" on the way in and still accepted everywhere it is read (log-review F12).
token_totals_by_model	Dict[str, Dict[str, int]]	Billed tokens attributed to the model that ran them, e.g. {"claude-opus-4-8": {"input_tokens": 412, "output_tokens": 264}}. Contains only the requested model unless Anthropic server-side fallback ran (Claude Fable 5), where attempts may be billed at different models' rates; a billed attempt that declined is included (0.5.2). Empty for providers/results that do not report it.
reasoning_content	Optional[str]	The model's thinking/reasoning text, kept OUT of :attr:output — replaying a thought summary as the assistant's answer is a defect, not a feature. What arrives differs by provider: Anthropic returns full thinking blocks; Fireworks returns reasoning_content (or <think> tags); Google and OpenAI return a summary only, and only because the toolkit opts in (include_thoughts=True / reasoning_config={"summary": ...}) — neither exposes the raw chain of thought. "" or None when the run produced none.
error	Optional[str]	Error message if the run failed, None on success.
run_id	Optional[str]	Unique identifier for this run (for cost tracking correlation).
iteration_metrics	List[IterationMetrics]	Per-request cost/token breakdown (if detailed_costs enabled). One entry per API request on all four providers; a round's first request and its follow-up share iteration_number and differ in purpose.
files_created	List[str]	List of file paths created during execution (e.g., by code execution).
distinct_urls	List[str]	List of unique URLs returned by web search (if used).
detailed_cost_file	Optional[str]	Path to detailed cost report file if created.
input_transformations	List[Dict[str, Any]]	Every input_transformations entry Anthropic returned across the run, each annotated with the 1-based request_index of the API request that reported it (Anthropic-specific). The server sends this array when it changed something about the request's input — for example dropping a preserved thinking block whose binding no longer matches. Entries whose type or reason the toolkit does not recognise are kept verbatim, because Anthropic adds values over time. Populated only when a request carried the thinking-binding-controls-2026-08-01 beta: every request of a thinking run on claude-fable-5-1, claude-opus-5-5, claude-sonnet-5-5 or claude-haiku-5-5 (the rows with enforces_thinking_prefix_check), or on any other extended-thinking row that opted in with anthropic_block_binding="drop_block". Empty otherwise, even when the API dropped blocks the serving model could not read.
prefix_binding_mismatches	int	How many entries in :attr:input_transformations had type == "thinking_dropped" and reason == "prefix_binding_mismatch" — the alarm. A thinking_mismatch_allowed entry (the block failed the prefix check but reached the model and was billed) is not counted. Each one is a turn whose preserved reasoning the server dropped because the conversation prefix it was bound to had been rewritten (compaction, a changed system prompt, a moved tool list). Under sliding_window these are expected on every window slide; the cost is reasoning quality on the retained turns, not a failed request. 0 on every non-Anthropic provider and on every model that does not run the check.
model_binding_mismatches	int	How many entries had type == "thinking_dropped" and reason == "model_binding_mismatch" — blocks the serving model could not read because another model produced them, so the API dropped them. Expected after a model downgrade or a server-side fallback and not a bug in the integration, so it is counted separately from the alarm above. Recorded only when the request carried the thinking-binding-controls-2026-08-01 beta (every thinking request on claude-fable-5-1 / claude-opus-5-5 / claude-sonnet-5-5 / claude-haiku-5-5, or a run with anthropic_block_binding="drop_block"): a client-side switch to an ungated row reports 0 although the API dropped the blocks. 0 on every non-Anthropic provider.
prefix_binding_retry	bool	True when the toolkit caught a prefix-binding 400 and re-sent the request once with prefix_mismatch_behavior="drop_block" plus the binding beta header. The visible form of "don't hide the 400": the retry is also logged at WARNING, and every later request of the run carries the same choice. On a thinking-off run on a model that checks the prefix the retry removes the thinking blocks from the request's history instead, because an off value cannot carry the field. False on a clean run and on every other provider.
skills_used	List[str]	List of skill IDs used (Anthropic-specific).
pause_turn_iterations	int	Number of pause_turn iterations if any occurred (Anthropic-specific).
container_id	Optional[str]	Anthropic code execution container id (Anthropic-specific). Surfaced for every Anthropic run that created or reused a container (plain code_execution or Skills). Pass this back on a later run via Agent.run(..., anthropic_container_id=...), a BuiltInToolConfig(..., config={"container_id": ...}), or by opening a session with reuse_anthropic_container=True to reuse the same container — files uploaded via tool_files and REPL state persist up to 30 days.
compaction_count	int	Number of compaction events during this run (across both Level 1 and Level 2).
compaction_input_tokens	int	Total input tokens consumed by summarization LLM calls.
compaction_output_tokens	int	Total output tokens from summarization LLM calls.
compaction_cost	float	Total cost in USD of summarization calls.
compaction_archives	List[str]	List of file paths to archived conversation histories.
retry_count	int	Total number of failed-and-retried API attempts during this run. Zero on a clean run. Populated from the provider-level RetryAccumulator wired through :meth:Agent.run.
retry_cost	float	USD cost of the failed retry attempts only — that is, the portion of :attr:cost attributable to attempts that did not produce the final response. cost always includes this value; subtract retry_cost from cost to get the cost of the successful (or final-exhaustion) attempt alone. Zero on a clean run.
retry_input_tokens	int	Input tokens consumed by the failed-and-retried attempts only (already included in :attr:input_tokens). Zero on a clean run.
retry_output_tokens	int	Output tokens produced by the failed-and-retried attempts only (already included in :attr:output_tokens). Zero on a clean run.

Examples:

Basic usage:

result = await agent.run("What is 2 + 2?")
print(result.output)  # "The answer is 4"
print(result.cost)    # 0.0003
print(result.tools_used)  # ["calculator"]

priority_request_count property

How many of this run's requests were served at the "priority" tier (C7).

Returns:

Name	Type	Description
int	int	The number of :attr:request_cost_records whose served_tier
	int	is "priority" — that is, the requests actually billed at the
	int	model's priority row. Zero on a run that never asked for a tier, and
	int	zero on a run whose priority requests were all downgraded.

service_tier_served property

The tier this run's requests were actually SERVED at (C7).

Derived from :attr:request_cost_records, because tier resolution is per request: a tool-calling run can have one request served at the requested tier and the next silently downgraded, and its cost is the sum of records priced individually.

Returns:

Type	Description
Optional[str]	str or None: "priority", "flex", "ultrafast", "standard" or
Optional[str]	"unverifiable" when every tier-requesting record agrees;
Optional[str]	"mixed" when they do not; None when no request asked for a
Optional[str]	tier at all — deliberately not "standard", because "never asked"
Optional[str]	and "asked and was refused" are different facts.
Optional[str]	"unverifiable" is Google-specific and means the toolkit could not
Optional[str]	read a served-tier signal, billed standard, and may therefore be
Optional[str]	UNDER-stating the cost (see docs/concepts/service_tiers.md).

Example

result = await agent.run("Summarize this filing")
if result.service_tier_served == "mixed":
    print(f"{result.priority_request_count} requests ran at priority")

success property

Whether the run completed successfully.

total_tokens property

Total billed tokens for the run: input_tokens + output_tokens.

:attr:reasoning_tokens is deliberately not added. It is a subset of :attr:output_tokens on every provider (see that attribute), so adding it counted the thinking share twice — the same trap the toolkit avoids for :attr:cached_input_tokens, which is a subset of :attr:input_tokens and is likewise not added here.

Returns:

Name	Type	Description
int	int	The run's input plus output tokens.

Example

result = agent.run("...")
assert result.total_tokens == result.input_tokens + result.output_tokens

__post_init__()

Normalize the token_source vocabulary (log-review F12).

The single-call provider paths report "api_metadata" and the loop/session paths "api_response" for the identical fact — "the provider reported these counts" — so a caller asserting on either one passed or failed depending on which path a run happened to take. The run-level surface now always says "api_response"; the old spelling is still ACCEPTED here and by :func:~doc_processing_toolkit.agent.cost.normalize_token_source, so a persisted result or an older provider release loads unchanged.

from_dict(data) classmethod

Create from dictionary representation.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary with result fields.	required

Returns:

Type	Description
AgentResult	AgentResult instance.

to_dict()

Convert to dictionary representation.

Returns:

Type	Description
Dict[str, Any]	Dictionary with all result fields, plus the two derived run-level
Dict[str, Any]	tier keys service_tier_served and priority_request_count
Dict[str, Any]	(properties, so asdict would otherwise drop them). They are
Dict[str, Any]	recomputed from request_cost_records by :meth:from_dict and are
Dict[str, Any]	therefore ignored on the way back in.

Session Management

doc_processing_toolkit.agent.session

AgentSession for managing stateful conversation sessions.

AgentSession

Stateful conversation session with message history.

AgentSession maintains conversation context across multiple runs, automatically passing previous messages to the agent for continuity.

Examples:

Synchronous context manager:

```python
with agent.session() as session:
    result1 = session.run_sync("Hello, I'm Alice")
    result2 = session.run_sync("What's my name?")  # Remembers "Alice"
```

Async context manager:

```python
async with agent.session() as session:
    result = await session.run("Hello!")
```

Attributes:

Name	Type	Description
agent		The Agent instance this session is bound to.
session_id	str	Unique identifier for this session.
messages	List[Dict[str, Any]]	Conversation message history. Holds the messages circulated to the provider (user turns, tool calls and tool results) and each turn's final assistant reply, which the provider loops never send anywhere and therefore do not record themselves — it is appended here so later turns and compaction can see it.
metadata	Dict[str, Any]	Session metadata accumulated across runs. Always carries created_at, total_tokens, total_cost, total_input_tokens, total_output_tokens, run_count, compaction_count and compaction_archives; the first run additionally populates total_cached_input_tokens, total_cache_write_tokens, total_retry_cost, total_retry_count, total_retry_input_tokens, total_retry_output_tokens and total_cost_inclusive. The retry totals are summed from each run's RUN-LEVEL retry_* fields; the per-iteration retry_* fields on iteration_metrics are reserved and always zero. total_cost is already retry-inclusive (it sums AgentResult.cost), so total_cost_inclusive simply equals it — the key is retained for backward compatibility.
restored_from_schema_version	Optional[int]	The schema_version of the envelope this session was restored from, or None on a session that was never restored. An envelope written before versioning reports 1.
prefix_changed_on_resume	bool	True when the envelope's prefix fingerprint does not match the Agent doing the restore — the instructions or the tool set moved while the conversation was saved. Always False on a live session and on an envelope with no fingerprint to compare. The restore still succeeds; this is a report.
prefix_change_report	Optional[Dict[str, Any]]	Details behind :attr:prefix_changed_on_resume: {"changed": ["system"|"tools"], "saved": {...}, "current": {...}, "schema_version": int}, or None when nothing changed.
converted_from	Optional[str]	The dialect the history was converted from on the last restore or :meth:rebind ("anthropic", "openai", "google" or "fireworks"), or None when no conversion ran. The entries are on :attr:conversion_report.
dropped_parts	List[Dict[str, Any]]	Media parts of a restored history that the target model's input_types do not list, each {"kind", "index", "reason", "provider", "model"}; replaced by placeholders under on_unsupported_part="drop". Empty on a live session.
last_turn_results	List[AgentResult]	Every AgentResult the most recent :meth:run/:meth:stream call produced, in turn order. Length 1 on an ordinary call; longer only when late_injection_policy="new_turn" started follow-up turns. Reset at the start of each call. Nothing is ever discarded, so the chain's cost is sum(r.cost for r in session.last_turn_results).

conversion_report property

Every artefact a cross-provider conversion dropped and every id it re-keyed (a copy).

The list rides metadata["conversion_report"] and grows across conversions. Each entry carries kind ("dropped" or "rekeyed"), the stored message index it concerns, from and to (the dialects) and at (UTC); a drop names its artefact (thinking, thought, thought_signature, reasoning, reasoning_content, provider_item, provider_block, provider_part, media, tool_result_media, container, unknown_part) with a detail; a re-key carries name, from_id and to_id. Empty when no conversion ever ran.

Example

restored = AgentSession.from_json(path, agent=Agent(model="gpt-6-luna"))
if restored.converted_from:
    for entry in restored.conversion_report:
        print(entry["kind"], entry.get("artefact"), entry["index"])

interrupted_turn property

The most recent turn that did not complete, or None.

A copy of metadata["interrupted_turn"]: the user message as it was sent, how many files were attached, the error, when it failed and the run number it would have had. Present only between a failed turn and the next run() / stream() call, and it survives the envelope, so a host that checkpoints after the failure can decide to resend the message in another process. DPT_SESSION_INTERRUPTED_TURN=off disables the record.

Example

try:
    await session.run("Summarise clause 4.")
except Exception:
    session.to_json(path=checkpoint)   # interrupted_turn rides along
...
if session.interrupted_turn:
    await session.run(session.interrupted_turn["message"])

late_injection_policy property

The configured late-injection policy (read-only).

"surface" or "new_turn". See :meth:__init__.

max_follow_up_turns property

Automatic follow-up turns allowed per call (read-only).

Only consulted under late_injection_policy="new_turn".

media_report property

What a save or a restore could not keep, as a list of records (a copy).

Each record carries a reason: "over_inline_threshold" (a Google upload larger than media_max_inline_bytes with no store, so its bytes were not kept), "stripped" (a part removed by media="strip", or at storage time under session_history_config={"preserve_media": False}, which adds "source": "preserve_media"), "missing_blob" (a store reference the restore could not read), "expired_reuploaded" (a dead Files API URI re-uploaded from the kept bytes) or "expired_no_bytes" (a dead URI with nothing to re-upload; the part became a placeholder). The list rides the envelope and grows across saves and restores.

Example

for entry in restored.media_report:
    print(entry["reason"], entry.get("mime_type"), entry.get("size"))

pending_injections property

Texts a run could not deliver and the session has not sent.

A copy of metadata["pending_injections"]. Empty on every session that has never had a hook message arrive after the last turn boundary. Under "surface" these are the host's to deliver; under "new_turn" they are only what the budget, the identical-text guard or a failed turn left over.

persistence_report property

What :meth:to_dict would write, measured, so a host can choose a policy first.

Returns:

Type	Description
Dict[str, Any]	``{"size_bytes": ,
Dict[str, Any]	"message_count": int, "media_count": int, "media_bytes": <decoded
Dict[str, Any]	bytes carried inline>, "media": []}``.
Dict[str, Any]	Each media record is ``{"message_index", "role", "kind",
Dict[str, Any]	"mime_type", "size_bytes", "ref"}withkind`` one of
Dict[str, Any]	"base64" (bytes in the envelope), "google_file_uri",
Dict[str, Any]	"provider_file" or "url" (references, size_bytes 0).

Example

report = session.persistence_report
if report["media_bytes"] > 8 * 1024 * 1024:
    ...  # externalise before writing

run_boundaries property

Where each recorded turn's messages sit in :attr:messages.

A copy of metadata["run_boundaries"]: one {"run": int, "start": int, "end": int} record per completed :meth:run/:meth:stream turn, oldest first. Turn i owns session.messages[record["start"]:record["end"]], so a host can truncate "before turn N" — see :meth:truncate_to_turn — without parsing provider-native messages.

Empty on a session that has run nothing, and on one restored from an envelope written before boundaries were recorded. Not every turn the session ever ran is necessarily listed: compaction rebases the list (see metadata["boundaries_reset_at_run"]) and :meth:truncate_to drops the records it cut through, so len(run_boundaries) counts the turns still addressable in stored history, not the turns ever run — that is metadata["run_count"].

Returns:

Type	Description
List[Dict[str, int]]	A list of boundary records. Mutating it does not affect the
List[Dict[str, int]]	session.

Example

result = await session.run("First question")
result = await session.run("Second question")
first, second = session.run_boundaries
assert session.messages[second["start"] : second["end"]]

session_id property

Get the unique session identifier.

tool_surface_drift property

Every mid-session change to the Anthropic tools array (read-only).

The envelope's prefix_fingerprint records the tool surface an Agent declares, which deliberately excludes MCP connection state — so a server that changes which tools it serves between turns does not move it. This is the per-request half of that signal: the provider re-derives tools from the live servers on every run, and each time the result differs from the array this session started with, a record lands here and the frozen array is sent instead.

Returns:

Type	Description
List[Dict[str, Any]]	A copy of the drift records, each a dict with event,
List[Dict[str, Any]]	failed_servers, added, removed and redefined. Empty
List[Dict[str, Any]]	for every session whose tool set never moved — which is every
List[Dict[str, Any]]	session that does not lose an MCP server mid-flight.

Example

await session.run("first")
await session.run("second")
if session.tool_surface_drift:
    logging.warning("MCP tools moved: %s", session.tool_surface_drift)

turn_in_flight property

Whether a run() / stream() call is currently running on this session.

__aenter__() async

Enter async context manager.

__aexit__(*args) async

Exit async context manager.

__enter__()

Enter sync context manager.

__exit__(*args)

Exit sync context manager.

__init__(agent, session_id=None, session_history_config=None, reuse_anthropic_container=False, late_injection_policy='surface', max_follow_up_turns=1)

Initialize an AgentSession.

Parameters:

Name	Type	Description	Default
agent	Agent	The Agent instance this session belongs to.	required
session_id	Optional[str]	Optional unique identifier. Generated if not provided.	None
session_history_config	Optional[Dict[str, Any]]	Optional history configuration dict. Keys: mode (str): One of "full", "messages_only", "no_tool_results", "no_reasoning", "compact_tool_results", "text_summary" (default "full"). preserve_media (bool): Keep media in stored history (default True). False replaces every media part of a turn — inline bytes and Google upload references alike — by a text placeholder when the turn is stored, so no later turn replays it; each replacement is recorded on media_report with "source": "preserve_media". compact_tool_results_max_chars (int): Max chars per tool result when mode="compact_tool_results" (default 500).	None
reuse_anthropic_container	bool	When True, capture the Anthropic code execution container_id from the first run and automatically inject it into every subsequent run (and stream) in this session. Uploaded files and REPL state inside the container persist up to 30 days, so downstream turns can reference earlier uploads without re-uploading. Also implicitly instructs the Anthropic provider to preserve (not delete) uploaded tool-file ids and custom-skill ids across every run in this session — including the first run, before a container id has been captured — so they remain usable by the reused container. Ignored (with a warning log) for non-Anthropic providers — safe to set when provider swaps are possible. Default False.	False
late_injection_policy	str	What happens to a hook message that arrived too late for any turn boundary to deliver it (the texts a run returns on AgentResult.undelivered_injections). One of: "surface" (default): nothing automatic. The texts are parked in metadata["pending_injections"] — which survives to_dict()/from_dict() — and cleared at the start of the next call. The host decides whether and when to send them. This is behaviour-neutral: a session under this policy issues exactly the requests it issues today. "new_turn": once a turn succeeds with undelivered texts, the session immediately runs another turn whose user message is those texts joined by a blank line, bounded by max_follow_up_turns. No follow-up ever starts after an error, an abort or a raised exception, and a follow-up whose text repeats the message that produced it is refused (a stuck host inbox). Anything not sent stays parked and logged, never silently dropped. Default "surface".	'surface'
max_follow_up_turns	int	Maximum automatic follow-up turns per :meth:run/:meth:stream call under late_injection_policy="new_turn". The budget is counted per call, so a hook that injects at every final boundary cannot loop. 0 makes "new_turn" behave like "surface". Default 1.	1

Raises:

Type	Description
ValueError	If late_injection_policy is not "surface" or "new_turn", or max_follow_up_turns is negative.

__repr__()

String representation.

clear()

Clear conversation history while keeping session open.

Notes

Metadata is deliberately preserved — cumulative cost and token counters are the history of what this session spent, not of what it currently holds. The recorded run boundaries are not: they are message indices, and after the history is gone every one of them points at nothing, so they are reset and the run at which that happened is stamped on metadata["boundaries_reset_at_run"]. See :meth:truncate_to to remove only part of the history. The truncated tool results kept for read_truncated_result are forgotten with the history.

close()

Close the session, preventing further runs.

Also releases the truncated tool results kept for read_truncated_result.

from_dict(data, agent, *, media_store=None) classmethod

Restore session from serialized data.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary from to_dict().	required
agent	Agent	Agent instance to bind to.	required
media_store	Any	The store for the envelope's $media:<ref> markers and Google upload references (a directory path or a MediaStore); None uses the agent's store. A missing blob becomes a placeholder plus a media_report entry.	None

Returns:

Type	Description
AgentSession	Restored AgentSession. :attr:restored_from_schema_version,
AgentSession	attr:prefix_changed_on_resume and :attr:prefix_change_report
AgentSession	describe what the restore found.

Raises:

Type	Description
SessionCompatibilityError	If the stored history is written in one provider's dialect, agent is on another, and the history cannot be converted: it ends inside a tool round, a tool call has no name, or a tool result matches no call. The message names the session, the target provider and the turn at fault, and nothing partial is stored. With DPT_SESSION_CROSS_PROVIDER_CONVERSION=off every provider-native history is refused this way. A history of plain text turns is provider-neutral and is never refused, whichever provider saved it; a version-3 envelope whose provider differs from the agent's but whose history is neutral restores with one INFO line.
UnsupportedPartError	Under on_unsupported_part="raise", when a restored media part is not in the target model's input_types.

Notes

Loading never fails on a version or a fingerprint. An envelope from an older version is migrated; one from a newer version is loaded with a warning and its unknown keys ignored, because refusing would strand a conversation this reader can read. A moved prefix is reported, not rejected: changing the instructions or the tool set mid-conversation is a legitimate thing to do, and the host is the only party that knows whether it was deliberate. A restore under another model of the same provider is quiet by design: the history reaches the new model unchanged.

A restore under another provider converts a provider-native history through the neutral representation and stores the converted messages: tool rounds and attachments cross, and what cannot (signed thinking, thought parts and signatures, reasoning items, provider-run tool traffic, media the target row does not take) is dropped, WARNING-logged and recorded on :attr:conversion_report; :attr:converted_from names the source dialect, and run boundaries are not carried across. The dialect is read from the messages themselves, so a version-1 or version-2 envelope, which records no provider, converts the same way. The one refusal is the unconvertible history above, because the alternative is a provider 400 on the first request that names none of this.

from_json(source, *, agent, media_store=None) classmethod

Restore a session from JSON text or a JSON file.

The inverse of :meth:to_json. source is read as JSON text when it is bytes or a str whose first non-blank character is { or [, and as a file path otherwise (a :class:pathlib.Path is always a path). Everything after parsing is :meth:from_dict, with the same versioning, fingerprint and compatibility rules.

Parameters:

Name	Type	Description	Default
source	Union[str, bytes, Path]	JSON text, UTF-8 bytes, or the path of a file written by :meth:to_json.	required
agent	Agent	The agent to bind the restored session to.	required
media_store	Any	The store the envelope's $media:<ref> markers and Google upload references are read from (a directory path or a MediaStore); None uses the agent's store. A blob the store cannot return becomes a placeholder and a media_report entry, never an error.	None

Returns:

Type	Description
AgentSession	The restored session.

Raises:

Type	Description
ValueError	If the JSON does not decode to an object.
SessionCompatibilityError	If a history written in another provider's dialect cannot be converted for agent, or DPT_SESSION_CROSS_PROVIDER_CONVERSION is off (see :meth:from_dict).
FileNotFoundError	If source names a file that does not exist.

Example

session = AgentSession.from_json("checkpoints/contract-review.json", agent=agent)
result = await session.run("Continue where we left off.")

get_history()

Get a copy of the conversation history.

Returns:

Type	Description
List[Dict[str, Any]]	Copy of the messages list, in provider-native shape: user prompts,
List[Dict[str, Any]]	assistant messages, tool calls, tool results, and reasoning items.

Notes

The run's final assistant reply is part of this history. The provider loops' own message lists hold only what was circulated to the API, and the last answer is never sent anywhere, so the session appends it after each run. Before this was fixed, every turn's actual answer was absent — later turns could not refer back to it and compaction summarized a conversation with no replies in it.

rebind(new_agent)

Continue this session's next turn on new_agent, in place.

The one-line answer to "continue this conversation on another model" between turns, without serialising: after a fallback decision, a cost cap, or the user picking another model. Same provider: the history is kept as it is. Another provider: the history is converted through the neutral representation exactly as from_dict converts it, with the same conversion_report, and refused with the same SessionCompatibilityError in the same impossible cases. The prefix fingerprint is recomputed against the new agent (prefix_changed_on_resume / prefix_change_report), the state that cannot cross an agent boundary is reset (the Anthropic container id, the frozen tool snapshot), the Google media table stays (it is data, not connection state), and the hop is recorded in metadata["served_models"].

Parameters:

Name	Type	Description	Default
new_agent	Agent	The agent the next turn runs on.	required

Returns:

Type	Description
AgentSession	This session, for chaining.

Raises:

Type	Description
RuntimeError	If a turn is running on this session.
SessionCompatibilityError	If the history cannot be converted for the new provider.

Example

session = Agent(model="gpt-6-luna", instructions="Be brief.").session()
await session.run("Summarise clause 4.", files=["contract.pdf"])
session.rebind(Agent(model="claude-haiku-4-5", instructions="Be brief."))
await session.run("And clause 5?")          # runs on Anthropic, history converted
print(session.converted_from, session.conversion_report)

run(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None) async

Run the agent within session context.

Automatically preserves conversation history and updates session state.

Parameters:

Name	Type	Description	Default
message	str	User message to send.	required
files	Optional[List[Any]]	Optional files to include with the message. Accepted formats: str — File path, URL, or YouTube URL. pathlib.Path — Path object pointing to a local file. bytes — Raw file bytes. dict — Pre-structured content dict.	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool (where supported by Anthropic, OpenAI, and Google). Files should be in the same format as the files argument. File type validation is performed by the provider API. Files are uploaded automatically via the provider's Files API. Unsupported providers (e.g., Fireworks) are filtered automatically.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. As with run(), the schema validates the output and AgentResult.output is the parsed dict, not a model instance.	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). See Agent.run() for full documentation.	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse for this call. Takes precedence over any id captured via reuse_anthropic_container auto-tracking. See Agent.run().	None

Returns:

Type	Description
AgentResult	AgentResult from the run. Under
AgentResult	late_injection_policy="new_turn" this is the last turn's
AgentResult	result — the answer to the thing the user most recently said.
AgentResult	result.cost is then that turn's cost only; the chain is on
AgentResult	attr:last_turn_results and its total is the
AgentResult	metadata["total_cost"] delta.

Raises:

Type	Description
RuntimeError	If session has been closed.

Notes

Under late_injection_policy="new_turn" one call can run several turns. Each turn is an ordinary turn: its own request, its own AgentResult, its own request_cost_records and its own metadata["run_count"] increment, so the session totals are the sum of the per-turn costs with nothing folded in invisibly. The caller's files, tool_files and media_resolution_list belong to the message they were passed with and are not resent on a follow-up turn; output_schema is forwarded, so every turn yields its own schema-valid output.

Example

Letting the session answer a late steer automatically:

session = agent.session(late_injection_policy="new_turn")
result = await session.run("Summarise the contract")
# result is the answer to the steer if one arrived late.
for turn in session.last_turn_results:
    print(turn.output, turn.cost)

run_sync(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None)

Synchronous wrapper for run().

Parameters:

Name	Type	Description	Default
message	str	User message to send.	required
files	Optional[List[Any]]	Optional files to include with the message. See run().	None
tool_files	Optional[List[Any]]	Optional list of files for the provider code execution tool. See run().	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. See run().	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only). See run().	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse for this call. See run().	None

Returns:

Type	Description
AgentResult	AgentResult from the run. Under
AgentResult	late_injection_policy="new_turn" this is the last turn's
AgentResult	result; see :meth:run and :attr:last_turn_results.

Notes

Each call runs in its own asyncio.run loop, so the agent's local MCP servers are connected and released per call (see Agent.run_sync). Use await run() to reuse them across turns.

stream(message, *, files=None, tool_files=None, output_schema=None, media_resolution_list=None, anthropic_container_id=None) async

Stream agent events within session context.

Delegates to Agent._stream_with_history with session history management. Loads conversation history before streaming and tracks costs from the complete event after streaming finishes.

Parameters:

Name	Type	Description	Default
message	str	User message to send.	required
files	Optional[List[Any]]	Optional files to include with the message.	None
tool_files	Optional[List[Any]]	Optional files for the provider code execution tool.	None
output_schema	Optional[Type[BaseModel]]	Optional Pydantic model for structured output. See Agent.stream().	None
media_resolution_list	Optional[List[Optional[str]]]	Optional per-part media resolution overrides (Google only).	None
anthropic_container_id	Optional[str]	Optional Anthropic code execution container id to reuse for this stream. Takes precedence over any id captured via reuse_anthropic_container auto-tracking.	None

Yields:

Type	Description
AsyncIterator[AgentStreamEvent]	AgentStreamEvent objects from Agent._stream_with_history.

Raises:

Type	Description
RuntimeError	If session has been closed.

Notes

A plain-text streamed turn stores a lightweight history shape: [user, injected…, assistant]. The opening user message, then each hook injection the run actually delivered (AgentResult.injected_messages, in order), then the final answer. Unlike :meth:run, that path does not store the provider-native transcript, so tool calls and tool results from a plain-text streamed turn are not visible to later turns. A "new_turn" follow-up therefore sees the previous turn's answer but not the tools it called, and may call one of them again.

A streamed turn with output_schema is stored exactly as :meth:run stores it: the provider-native transcript when the run returned one — including the retained signed final turn on the preserved-thinking rows, which DPT_ANTHROPIC_RETAIN_FINAL_TURN=off turns back into plain text — and otherwise the [user, answer] pair.

Under late_injection_policy="new_turn" one call can stream several turns, and exactly one complete event is still emitted per call — the documented teardown contract is unchanged. A non-final turn ends with turn_complete (the same metadata the complete event would have carried, plus turn_index and auto_follow_up) and the next one opens with turn_start (turn_index, auto_follow_up, trigger="late_injection" and the message being sent). Those two events appear only when a follow-up actually starts, so a stream under the default policy is event-for-event what it is today. The final complete then also carries turns, turn_costs and total_cost.

agent.run_handle is per-run and is None between turns: a host that steers must re-capture it on turn_start. An abort during a turn suppresses the follow-up entirely.

Closing the stream early (aclose() / contextlib.aclosing) closes the turn's inner generators at once, which cancels the request. A turn closed before its complete event is still banked: its spend (agent.last_abandoned_stream, marked "abandoned": True) is added to metadata and last_turn_results, and no answer is stored in history. A turn whose consumer stopped at complete is folded in normally. DPT_STREAM_CANCEL_ON_CLOSE=off restores the previous handling, in which neither was banked.

to_dict(*, media=None, media_store=UNSET, media_max_inline_bytes=UNSET)

Serialize session state to a dictionary.

Parameters:

Name	Type	Description	Default
media	Optional[str]	The media policy for this save — "auto" (the default: externalize when a store is in force, else inline), "inline" (every media part written exactly as stored), "externalize" (every inline payload handed to the store and replaced by a $media:<ref> marker) or "strip" (every inline payload replaced by a text placeholder). None uses the agent's media_policy.	None
media_store	Any	The store for this save (a directory path or a MediaStore); omitted means the agent's store, None means no store.	UNSET
media_max_inline_bytes	Any	Per-part cap on the Google upload bytes the envelope's media table keeps when no store is in force; omitted means the agent's cap, None lifts it. A per-call cap can only tighten what the session kept.	UNSET

Returns:

Type	Description
Dict[str, Any]	Dictionary containing session state, including schema_version
Dict[str, Any]	and prefix_fingerprint. Version 3 also records provider
Dict[str, Any]	and model (of the agent that saved it), toolkit_version,
Dict[str, Any]	saved_at (UTC, ISO 8601), sdk ({"name", "version"} of
Dict[str, Any]	the provider SDK), media_policy (the policy this save applied),
Dict[str, Any]	media (the media table: one entry per Google Files API upload —
Dict[str, Any]	URI, expiry, and its bytes or store reference — and per externalised
Dict[str, Any]	part) and media_report (what a save or restore could not keep).
Dict[str, Any]	The result is a detached copy: messages are normalised to
Dict[str, Any]	JSON-clean values and metadata is deep-copied, so a host that
Dict[str, Any]	mutates the envelope never mutates the live session through it.

Raises:

Type	Description
ValueError	If media is not a known policy, or is "externalize" with no store in force.

Notes

The envelope carries the conversation, never the agent's instructions. A session envelope is routinely handed to another process or another party, and a system prompt is the one part of a run that is often confidential and is always reconstructible from the :class:Agent that loads it (B16). A leading system message is therefore dropped on the way out; it can only be there at all in an envelope written before Fireworks took ownership of its own system prompt. prefix_fingerprint holds SHA-256 digests of the system prompt and the tool surface for the same reason — enough to notice that the request prefix moved, not enough to reconstruct it.

The per-turn run boundaries (B13) ride metadata, which already round-trips, so they need no envelope key and no schema_version bump: an envelope written before they existed simply has none, which reads as an empty list and lets a caller fall back. They are copied out of the live metadata and rebased onto the messages this envelope actually carries — dropping the leading system message moves every index down one — and they are re-validated against the restored messages on the way back in.

metadata["served_models"] (one record per run and served model, from the run's request_cost_records) rides metadata like the boundaries do, so which model actually answered each turn survives the round trip without an envelope key of its own.

Example

envelope = session.to_dict()
envelope["provider"], envelope["model"]   # who saved it
json.dumps(envelope)                      # always JSON-clean

to_json(*, path=None, indent=None, media=None, media_store=UNSET, media_max_inline_bytes=UNSET)

Serialize the session to a JSON string, optionally writing it to a file.

The string is json.dumps of :meth:to_dict (ensure_ascii=False, so non-ASCII text is written as it is). With path the same text is written to that file atomically — through a sibling temporary file and a rename — so a checkpoint interrupted mid-write never replaces the previous good envelope with a truncated one. This is the checkpoint primitive: call it after every completed turn and the file always holds a resumable session.

Parameters:

Name	Type	Description	Default
path	Optional[Union[str, Path]]	Optional destination file. Parent directories are created.	None
indent	Optional[int]	json.dumps indentation; None (the default) writes the compact form.	None
media	Optional[str]	The media policy for this save; see :meth:to_dict.	None
media_store	Any	The store for this save; see :meth:to_dict.	UNSET
media_max_inline_bytes	Any	The inline cap for this save; see :meth:to_dict.	UNSET

Returns:

Type	Description
str	The JSON text that was (or would have been) written.

Example

result = await session.run("Summarise the contract")
session.to_json(path="checkpoints/contract-review.json")
# or, with the bytes in a directory beside the file:
session.to_json(path="checkpoints/contract-review.json", media_store="checkpoints/media")

truncate_to(message_count, *, drop_orphans=True)

Cut stored history down to its first message_count messages.

The history editor's primitive: "delete everything from here on", for a host implementing edit-and-resend or delete-from-turn-k. Use :meth:truncate_to_turn to express the cut as a turn instead of a message count.

Truncation is a pure prefix cut. Nothing that survives is rewritten, so every retained thinking block, redacted_thinking block and Google thought_signature stays byte-identical to what the provider signed — the property that makes the next request's cached prefix and its block bindings valid. The one adjustment is to the cut point: a count that would leave a tool call without its result is moved backward, never repaired in place, so the invariant holds by construction rather than by editing.

Parameters:

Name	Type	Description	Default
message_count	int	How many messages to keep, counted from the start of :attr:messages. 0 keeps nothing; len(session.messages) keeps everything.	required
drop_orphans	bool	Keep the result provider-legal (the default). The cut point is moved back past any tool call whose result it would have severed, and any tool result already stored without its call is dropped with a warning. False cuts at exactly message_count and makes legality the caller's problem — for a host that knows the shape of its own history and wants an exact count.	True

Returns:

Type	Description
int	The number of messages actually kept. It is <= message_count,
int	and strictly less when the cut had to move back past a tool round.

Raises:

Type	Description
ValueError	If message_count is not an int, is negative, or is greater than the number of stored messages.
RuntimeError	If the session has been closed.

Notes

What is kept. Metadata survives exactly as it does through :meth:clear. The cumulative counters — total_cost, total_tokens, total_input_tokens, total_output_tokens, run_count, compaction_count — are the history of what this session spent, not of what it currently holds, and are deliberately not rewound: the requests were made and billed, and a host that rewound them would report a cost lower than its invoice. metadata["pending_injections"] survives for the same reason (a message the host still owes the model does not become delivered because history shrank), as do the Anthropic container id, :attr:last_turn_results (a record of the last call, not of history) and the request prefix fingerprint (a property of the :class:Agent, which truncation does not touch).

What is rewound. metadata["run_boundaries"] — the only metadata made of message indices. Records whose turn no longer fits entirely inside the kept prefix are dropped, so every remaining record still addresses the messages it names. Their messages are not deleted, they are simply no longer addressable as a turn, which is the same state a session restored from a pre-boundary envelope is in.

AgentResult.injected_messages is a property of a run's result and is not session state, so truncation has nothing to do to it.

Example

Drop the last turn and ask again:

keep = session.run_boundaries[-1]["start"]
session.truncate_to(keep)
result = await session.run("Let me rephrase that")

truncate_to_turn(turn_index, *, drop_orphans=True)

Cut stored history to just before recorded turn turn_index.

The public-index form of :meth:truncate_to: turn_index counts turns in :attr:run_boundaries, so a host can say "go back to before turn 3" without knowing anything about provider-native message shapes. The turn index is resolved to a message count and the same legality rules then apply, so a public turn index always maps to a private provider prefix the next request can legally send.

Parameters:

Name	Type	Description	Default
turn_index	int	The first turn to drop, 0-based over :attr:run_boundaries. 0 drops every recorded turn; len(session.run_boundaries) drops none.	required
drop_orphans	bool	As :meth:truncate_to.	True

Returns:

Type	Description
int	The number of messages kept.

Raises:

Type	Description
ValueError	If turn_index is not an int, is negative, or is greater than the number of recorded turns. The message names metadata["boundaries_reset_at_run"] when a compaction is why the turn is not addressable.
RuntimeError	If the session has been closed.

Notes

Turns that ran before a compaction, or before a restore from an envelope written without boundaries, are not addressable — their messages may still be stored, but nothing recorded where they begin. turn_index=0 therefore cuts to the start of the oldest recorded turn, which is not necessarily 0; call truncate_to(0) to empty the history outright.

Example

session.truncate_to_turn(2)  # keep turns 0 and 1, drop the rest

compute_prefix_fingerprint(agent)

Fingerprint the request prefix an Agent will send.

Anthropic binds a preserved thinking block to the conversation prefix it was produced under, so a resumed session whose instructions or tool set have moved is restoring into a different context. The envelope stores digests, never the instructions themselves: a session envelope travels between processes and parties, and a system prompt is often confidential (B16).

Parameters:

Name	Type	Description	Default
agent	Agent	The Agent the session is bound to.	required

Returns:

Type	Description
Dict[str, str]	{"system": <sha256>, "tools": <sha256>}. Two digests rather than
Dict[str, str]	one so a mismatch can name which half moved.

Example

fingerprint = compute_prefix_fingerprint(agent)
sorted(fingerprint)  # ["system", "tools"]

resolve_session_history_config(per_session=None, agent_level=None)

Resolve session history config using precedence chain.

Precedence (highest to lowest):

1. Per-session config
2. Agent-level config
3. Environment variable DOC_TOOLKIT_SESSION_HISTORY_MODE
4. Built-in default ({"mode": "full", "preserve_media": True})

Parameters:

Name	Type	Description	Default
per_session	Optional[Dict[str, Any]]	Config passed to agent.session(session_history_config=...).	None
agent_level	Optional[Dict[str, Any]]	Config stored on the Agent instance.	None

Returns:

Type	Description
Dict[str, Any]	Resolved config dict.

Cost Tracking

doc_processing_toolkit.agent.cost

DetailedCostReport class for tracking and reporting agent execution costs.

AgentStreamEvent dataclass

A single event emitted during Agent.stream().

Notes

Event types:

• "request_start": Opt-in (Agent(emit_request_start=True)); not emitted otherwise, so a consumer that enumerates event types is unaffected by default. Fires immediately before an ACTUAL provider send — the first request, each tool-loop iteration, each max_tokens / truncation / empty-turn re-request, each pause_turn continuation and each compaction summarisation call — so one event per API request, in order. content is None; metadata carries request_index (1-based and monotonic across the run), iteration, attempt (the tenacity attempt number), purpose ("generation", "tool_followup", "max_tokens_continuation", "truncation_retry", "empty_turn_retry", "pause_turn_continuation" or "compaction_summary"), model and provider. Reaching hooks as well on the tool-calling path, where it is observation-only.
• "text_delta": Partial text output. content is the text chunk. metadata is None for ordinary prose. On a schema-constrained request (Path A native structured output) it is {"structured_output": True}, so a UI can suppress the terminal JSON payload without a JSON-head heuristic (B11). On Path B (tool-based structured output) the payload is returned as tool input and never reaches text_delta at all, so nothing is tagged there.
• "thinking_delta": Partial reasoning/thinking output. content is the visible reasoning chunk when the provider exposes it.
• "tool_call_start": A tool call is starting. metadata includes tool_name, tool_call_id, arguments, and iteration when available.
• "tool_call_result": A tool call completed. metadata includes tool_name, tool_call_id, result, success, duration_ms, and iteration when available, plus tool_reported_cost and tool_reported_cost_estimated when the tool reported spend with the _dpt_spend marker (0.5.1). result is the first Agent(tool_result_preview_chars=) characters (500 by default) of the result's str(), and result_truncated says whether that is shorter than the whole; at 0 it is the whole result and result_data carries it as JSON data (0.5.1).
• "iteration_complete": One tool-calling iteration finished. metadata includes iteration, input_tokens, output_tokens, cost, tools_called, and stop_reason when available. The usage covers every request billed since the previous event, so the events sum to the run's requests up to the last event (0.5.1).
• "error": An unrecoverable error occurred. content is the error message. metadata carries input_tokens, output_tokens, cost, and token_source populated from the partial-stream data captured before the failure where available (token_source is then "stream_partial"); otherwise the token/cost fields are zero with token_source="unknown". It also carries undelivered_injections and, when a tool that ran before the failure reported spend, tool_reported_cost / tool_reported_costs (already inside cost). A failed run emits "error" first and still emits a trailing "complete" event, so consumers that short-circuit on "error" see the failure while teardown paths bound to "complete" continue to run.
• "complete": The stream finished. metadata is the full AgentResult serialized to a dict. Always emitted exactly once per stream() call, including after an "error" event when the run failed — metadata["error"] is populated in that case. It carries injected_messages (the hook injections this run delivered, in order) and undelivered_injections (texts that arrived after the last turn boundary and were never shown to the model — deliver them as the next turn's user message). When an AgentSession(late_injection_policy="new_turn") ran follow-up turns it also carries turn_index, turns, turn_costs and total_cost.
• "turn_complete": A non-final turn finished under AgentSession(late_injection_policy="new_turn"). metadata is that turn's AgentResult dict plus turn_index and auto_follow_up. Emitted only when a follow-up turn actually starts, in place of that turn's "complete", so exactly one "complete" still ends the stream.
• "turn_start": A follow-up turn is starting under the same policy. metadata carries turn_index, auto_follow_up, trigger ("late_injection") and message — the text being sent as the new user turn. Agent.run_handle is re-created per run, so this is the host's signal to re-capture it.

DetailedCostReport dataclass

Detailed cost tracking report for agent runs.

Captures per-iteration costs, token breakdowns, and cumulative totals. Can be serialized to JSON for storage and analysis.

Attributes:

Name	Type	Description
run_id	str	Unique identifier for this run.
agent_name	Optional[str]	Optional name identifier for the agent.
timestamp	str	When the report was created.
provider	str	Provider name (anthropic, openai, google, fireworks).
model_name	str	Full model name used.
total_input_tokens	int	Total input tokens across all iterations.
total_output_tokens	int	Total output tokens across all iterations.
total_reasoning_tokens	int	Total reasoning/thinking tokens.
total_cached_input_tokens	int	Total cached input tokens across all iterations (the cache-read share of total_input_tokens, all providers).
total_cache_write_tokens	int	Total prompt-cache write tokens. Anthropic bills them on top at the cache-write rate; on OpenAI they are a subset of input re-priced at cache_write_input_multiplier. Zero for providers that do not report them.
total_cost	float	Total cost in USD for the run, INCLUDING the failed-and-retried attempts. Reports built by :class:~doc_processing_toolkit.agent.core.Agent take this straight from AgentResult.cost (the sum of the per-request token costs plus per-run tool costs), which is already retry-inclusive; hand-built reports accumulate it from :meth:add_iteration, whose per-iteration cost is likewise retry-inclusive. Subtract total_retry_cost to get the successful attempts alone.
cost_estimated	bool	Whether the cost is an estimate.
total_duration_ms	int	Total execution time in milliseconds.
thinking_enabled	bool	Whether the run carried thinking: any request's flag when the run has per-request entries, else what the configuration given resolves to (0.5.3).
thinking_config	Optional[Dict[str, Any]]	Thinking configuration if enabled.
max_iterations	int	Maximum iterations configured.
temperature	float	Temperature setting used.
iterations	List[IterationMetrics]	List of per-iteration cost breakdowns.
total_iterations	int	Actual number of iterations performed.
tools_available	List[str]	List of tool names available to agent.
tools_used	List[str]	List of tool names that were called.
tool_call_count	int	Total number of tool calls.
tool_success_rate	float	Success rate of tool calls (0.0 to 1.0).
tool_usage_summary	Dict[str, Any]	Summary of tool usage by tool name.
pricing_info	Dict[str, float]	Pricing rates used for calculation.
start_time	Optional[str]	When the run started.
end_time	Optional[str]	When the run completed.
total_retry_cost	float	USD cost of the failed-and-retried attempts only. Set from the run-level AgentResult.retry_cost when the report is built by :class:~doc_processing_toolkit.agent.core.Agent. The :meth:add_iteration accumulation below it is retained for hand-built reports but contributes nothing, because the per-iteration retry_* fields are reserved and always zero.
total_retry_input_tokens	int	Input tokens consumed by retried attempts. Run-level, on the same convention as total_retry_cost.
total_retry_output_tokens	int	Output tokens produced by retried attempts. Run-level, on the same convention as total_retry_cost.
total_retry_count	int	Total number of failed-and-retried API attempts. Run-level, on the same convention as total_retry_cost.
request_cost_records	List[Dict[str, Any]]	One record per API request (input/output/cached tokens, served_model, requested/served tier, cost). The run's token cost is the sum of these records, never a re-pricing of the aggregate totals. :attr:service_tier_served and :attr:priority_request_count are derived from them.
total_tool_reported_cost	float	What the run's tools reported spending (AgentResult.tool_reported_cost), already part of :attr:total_cost.
tool_reported_costs	List[Dict[str, Any]]	One record per reporting tool call (AgentResult.tool_reported_costs).
prefix_binding_mismatches	int	Preserved-thinking blocks the Anthropic API dropped in this run because the conversation prefix they were bound to had been rewritten (the alarm). Mirrors AgentResult.prefix_binding_mismatches so the count survives into the cost/telemetry sink clients already read. 0 everywhere else.
model_binding_mismatches	int	Preserved-thinking blocks dropped because the serving model could not read them. Expected after a model change or a server-side fallback — never an integration bug. Both counters count dropped blocks only (type == "thinking_dropped"). The model count is recorded only when the request carried the thinking-binding-controls-2026-08-01 beta (claude-fable-5-1, claude-opus-5-5, claude-sonnet-5-5, claude-haiku-5-5, or anthropic_block_binding="drop_block"), so a client-side switch to an ungated row reports 0 although the API dropped the blocks. 0 elsewhere.
prefix_binding_retry	bool	True when this run re-sent a request once with prefix_mismatch_behavior="drop_block" after a prefix-binding 400 (or, on a thinking-off run, with the thinking blocks removed). False elsewhere.
schema_version	int	Persisted-report schema marker, bumped when the report shape changes. Version 2 adds per-request records and cached-token fields (2026-08).

Notes

:meth:to_dict emits one key with no backing attribute: total_cost_inclusive. Because total_cost is already retry-inclusive, the key simply equals total_cost and is retained for backward compatibility (an earlier derivation added total_retry_cost on top, double-counting retries). It is not read back by :meth:from_dict.

Examples:

Basic usage:

agent = Agent(
    model="claude-sonnet-4-5",
    detailed_costs=True,
    cost_output_dir="./costs",
)
result = await agent.run("Complex task...")
# Cost report is automatically saved to ./costs/cost_report_<run_id>.json

priority_request_count property

How many requests in this run were served at the "priority" tier (C7).

Returns:

Name	Type	Description
int	int	The count of records billed at the model's priority row.

service_tier_served property

The tier this run's requests were SERVED at, or "mixed" (C7).

Returns:

Type	Description
Optional[str]	str or None: See :func:summarize_served_tiers. None when no
Optional[str]	request asked for a tier.

__post_init__()

Initialize default values.

add_iteration(input_tokens, output_tokens, reasoning_tokens, cost, duration_ms=0, thinking_enabled=False, stop_reason='', tool_calls=None, cached_input_tokens=0, cache_write_tokens=0, purpose=None)

Add a new iteration to the report.

Parameters:

Name	Type	Description	Default
input_tokens	int	Input tokens for this iteration.	required
output_tokens	int	Output tokens for this iteration.	required
reasoning_tokens	int	Reasoning tokens for this iteration.	required
cost	float	Cost for this iteration in USD.	required
duration_ms	int	Duration in milliseconds.	0
thinking_enabled	bool	Whether this request carried thinking (per request).	False
stop_reason	str	Reason the iteration stopped.	''
tool_calls	Optional[List[ToolCallMetrics]]	Detailed tool call metrics.	None
cached_input_tokens	int	Cached share of this iteration's input tokens.	0
cache_write_tokens	int	Prompt-cache write tokens for this iteration (Anthropic and OpenAI).	0
purpose	Optional[str]	Why the iteration's request was made (see :attr:IterationMetrics.purpose).	None

finalize()

Finalize the report, setting end_time and calculating totals.

from_dict(data) classmethod

Create a report from a dictionary.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary from to_dict().	required

Returns:

Type	Description
DetailedCostReport	DetailedCostReport instance.

load(file_path) classmethod

Load a report from a JSON file.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the JSON file.	required

Returns:

Type	Description
DetailedCostReport	DetailedCostReport instance.

save(directory)

Save the report to a JSON file.

Parameters:

Name	Type	Description	Default
directory	str	Directory to save the file in.	required

Returns:

Type	Description
str	Full path to the saved file.

to_dict()

Convert the report to a dictionary.

Returns:

Type	Description
Dict[str, Any]	Dictionary representation of the report. Every attribute is emitted,
Dict[str, Any]	plus three derived keys with no backing attribute:
Dict[str, Any]	total_cost_inclusive, which equals total_cost (already
Dict[str, Any]	retry-inclusive; kept for backward compatibility), and the run-level
Dict[str, Any]	tier pair service_tier_served / priority_request_count, both
Dict[str, Any]	recomputed from request_cost_records and therefore ignored by
Dict[str, Any]	meth:from_dict.

to_json(indent=2)

Serialize the report to JSON.

Parameters:

Name	Type	Description	Default
indent	int	JSON indentation level.	2

Returns:

Type	Description
str	JSON string representation.

IterationMetrics dataclass

Metrics for a single API request of an agent run.

There is one entry per request on all four providers, not one per iteration: a round's first request and its follow-up share an iteration_number and are told apart by :attr:purpose.

Attributes:

Name	Type	Description
iteration_number	int	1-based iteration number.
input_tokens	int	Input tokens for this iteration. Includes any failed- and-retried attempts rolled up from the RetryAccumulator.
output_tokens	int	Output tokens for this iteration. Includes retry attempts.
reasoning_tokens	int	Reasoning/thinking tokens for this iteration.
cached_input_tokens	int	Prompt-cache read tokens for this iteration — a SUBSET of :attr:input_tokens, priced at the model's cached rate rather than added on top.
cache_write_tokens	int	Prompt-cache WRITE tokens for this iteration. On Anthropic they are billed on top of input at the cache-write rate; on OpenAI (rows with cache_write_input_multiplier) they are a SUBSET of :attr:input_tokens, re-priced at that multiple.
cost	float	Cost in USD for this iteration. Includes retry attempts — subtract retry_cost to get the successful attempt cost alone.
duration_ms	int	Duration in milliseconds.
tool_calls	List[ToolCallMetrics]	List of tool call metrics.
thinking_enabled	bool	Whether this request carried thinking (per request; populated since 0.5.3, False on every entry before).
stop_reason	str	Reason the iteration stopped.
timestamp	str	When this iteration occurred.
compaction_triggered	bool	Whether compaction was triggered during this iteration.
compaction_strategy	Optional[str]	The compaction strategy used, if compaction was triggered.
compaction_tokens_cleared	int	Number of tokens cleared by compaction.
retry_count	int	Number of failed-and-retried API attempts for this iteration. Zero on a clean iteration.
retry_cost	float	USD cost of the failed retry attempts only — that is, the portion of :attr:cost attributable to attempts that did not produce the final response for this iteration.
retry_input_tokens	int	Input tokens consumed by failed retry attempts only.
retry_output_tokens	int	Output tokens produced by failed retry attempts only.
retry_details	List[Dict[str, Any]]	Per-attempt breakdown of the retry attempts as a list of RetryAttemptCost-shaped dicts (each with attempt_number, input_tokens, output_tokens, reasoning_tokens, cached_input_tokens, cost, token_source, error, duration_ms, cost_estimated).
model	Optional[str]	The model that served this iteration's final API call. Differs from the requested model only when Anthropic server-side fallback ran (Claude Fable 5). None for providers/results that do not report it.
purpose	Optional[str]	Why this entry's request was made — the request_start vocabulary (agent.hooks.REQUEST_PURPOSES): "generation" (the first request, or a fresh turn), "tool_followup" (the answer to a tool round), "max_tokens_continuation" / "truncation_retry" (a re-request after an output-limit cut), "empty_turn_retry" (Google's one re-request of an empty final turn) or "pause_turn_continuation" (an Anthropic pause_turn continuation). Every request of a tool loop has one entry, so a round's first request and its follow-up share an iteration_number and are told apart here. None on entries that predate the field.

RetryAccumulator dataclass

Accumulates cost data across retry attempts for a single logical request.

Used with tenacity's before_sleep callback to capture cost data from each failed attempt before the retry sleep.

Properties provide aggregate totals across all attempts. The tenacity hook records failed-and-retried attempts only, so every entry is a failed attempt; the run-level retry figures (retry_count, retry_cost, retry_input_tokens, retry_output_tokens) live on AgentResult, rolled up by utils.attach_retry_summary from :attr:attempts.

total_cached_input_tokens property

Prompt-cache read tokens across all attempts.

A SUBSET of :attr:total_input_tokens, never additional to it.

Returns:

Name	Type	Description
int	int	The sum of every attempt's cached_input_tokens.

total_cost property

Total cost across all attempts (including successful).

total_input_tokens property

Total input tokens across all attempts.

total_output_tokens property

Total output tokens across all attempts.

add_attempt(attempt)

Add a retry attempt's cost data.

to_list()

Serialize all attempts to a list of dicts.

RetryAttemptCost dataclass

Cost data from a single API call attempt (successful or failed).

Attributes:

Name	Type	Description
attempt_number	int	1-based attempt number
input_tokens	int	Known or estimated input token count
output_tokens	int	Known (from usage) or estimated (from partial stream)
reasoning_tokens	int	If applicable (e.g., thinking models)
cached_input_tokens	int	Prompt-cache READ tokens for this attempt — a SUBSET of input_tokens, on the same convention as everywhere else in the toolkit (subtracted out and re-priced, never added on top). A failed attempt on a long cached prefix reads the cache just like a successful one, so without this field its input was billed at the full rate; the field is populated whenever the provider attaches a cached count to the failed attempt's partial-stream data. Zero when the provider reports none.
cost	float	Calculated cost in USD
token_source	str	How tokens were determined. "api_usage": From API usage metadata. "stream_partial": From accumulated partial stream output. "billable_error_input_only": Input estimated, no output (error classified as billable). "no_charge": Non-billable error with no stream data. "rejected_before_stream": The provider rejected the request before streaming (an HTTP error response, or no connection), so nothing was billed and no pre-flight estimate is charged. "estimated": From count_tokens() estimation.
error	Optional[str]	Error message if this attempt failed (None for successful)
duration_ms	int	Wall time for this attempt
cost_estimated	bool	True when cost rests on an estimated token count — the attempt's input came from the caller's pre-flight _estimated_input_tokens rather than provider usage — so the run's cost_estimated is raised with it.

ToolCallMetrics dataclass

Metrics for a single tool call.

normalize_token_source(token_source)

Map a token_source value onto the canonical vocabulary (F12).

Parameters:

Name	Type	Description	Default
token_source	str or None	A value from a provider result, a persisted report, or an older release.	required

Returns:

Type	Description
Optional[str]	str or None: The canonical spelling. "api_metadata" becomes
Optional[str]	"api_response"; every other value, including None,
Optional[str]	"stream_partial", "estimated" and any provider-specific
Optional[str]	"*_metadata_missing" variant, is returned unchanged.

Example

normalize_token_source("api_metadata")   # "api_response"
normalize_token_source("stream_partial") # "stream_partial"

summarize_served_tiers(request_cost_records)

Aggregate per-request served tiers into one run-level answer (C7).

Tier resolution is per request, not per run: a long tool-calling run can have some requests served at the requested tier and others silently downgraded, and the run's cost is the sum of records priced individually. This collapses that to the single value a caller wants on :class:AgentResult.

Only records that actually asked for a tier take part. A run that requested nothing reports None rather than "standard", because "never asked" and "asked and was refused" are different facts.

Parameters:

Name	Type	Description	Default
request_cost_records	list	The run's per-request cost records, each a dict with requested_tier and served_tier keys.	required

Returns:

Name	Type	Description
tuple	tuple[Optional[str], int]	(service_tier_served, priority_request_count). service_tier_served (str or None): the one tier every tier-requesting record was served at — typically "priority", "flex", "ultrafast", "standard" or "unverifiable" — or :data:SERVICE_TIER_MIXED when they disagree, or None when no request asked for a tier. priority_request_count (int): how many requests in the run were actually served at the "priority" tier, and therefore billed at its premium row.

Example

served, priority_count = summarize_served_tiers(result.request_cost_records)
if served == "mixed":
    print(f"{priority_count} of {len(result.request_cost_records)} ran at priority")

Tool Decorator

doc_processing_toolkit.agent.tools

@tool decorator for defining tool functions.

The @tool decorator transforms regular Python functions into MCP-compatible tools that can be used with the Agent class. It extracts metadata from function signatures and docstrings to generate JSON Schema definitions.

get_tool_metadata(func)

Get the tool metadata from a @tool decorated function.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	A @tool decorated function.	required

Returns:

Type	Description
Optional[Dict[str, Any]]	Metadata dict if func is a tool, None otherwise.

is_tool(obj)

Check if an object is a @tool decorated function.

Parameters:

Name	Type	Description	Default
obj	Any	Object to check.	required

Returns:

Type	Description
bool	True if obj has been decorated with @tool.

tool(func=None, *, name=None, description=None)

Decorator to mark a function as a tool for use with Agent.

The decorator extracts metadata from the function signature and docstring to create an MCP-compatible tool definition. It generates a JSON Schema for the tool's parameters based on type hints.

Parameters:

Name	Type	Description	Default
func	Optional[F]	The function to decorate (auto-passed when used without parentheses).	None
name	Optional[str]	Optional override for the tool name (defaults to function name).	None
description	Optional[str]	Optional override for the description (defaults to docstring).	None

Returns:

Type	Description
Any	The decorated function with _tool_metadata attribute attached.

Raises:

Type	Description
TypeError	If the function has parameters without type hints.
ValueError	If the function is already decorated with @tool.

Notes

Can be used with or without parentheses

@tool
def my_func(x: int) -> str: ...

@tool(name="custom_name", description="Custom description")
def my_func(x: int) -> str: ...

Examples:

Basic usage:

@tool
def calculate(a: float, b: float, operation: str) -> float:
    '''Perform a calculation.

    Args:
        a: First number
        b: Second number
        operation: One of 'add', 'subtract', 'multiply', 'divide'
    '''
    ...

calculate._tool_metadata
{
    'name': 'calculate',
    'description': 'Perform a calculation.',
    'parameters': {
        'type': 'object',
        'properties': {
            'a': {'type': 'number', 'description': 'First number'},
            'b': {'type': 'number', 'description': 'Second number'},
            'operation': {'type': 'string', 'description': "One of 'add'..."}
        },
        'required': ['a', 'b', 'operation']
    },
    'function': calculate
}

Configuration Loading

doc_processing_toolkit.agent.config

Configuration loading and validation for Agent.

Handles loading agent configuration from YAML files, validating parameter combinations, and environment variable interpolation.

AgentConfig

Agent configuration container.

Provides attribute-style access to configuration values with type hints.

__init__(config_dict)

Initialize from a configuration dictionary.

Parameters:

Name	Type	Description	Default
config_dict	Dict[str, Any]	Validated configuration dictionary.	required

from_string(yaml_content) classmethod

Load configuration from a YAML string.

Parameters:

Name	Type	Description	Default
yaml_content	str	YAML content as a string.	required

Returns:

Type	Description
AgentConfig	AgentConfig instance.

from_yaml(file_path) classmethod

Load configuration from a YAML file.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the YAML file.	required

Returns:

Type	Description
AgentConfig	AgentConfig instance.

merge(overrides)

Create a new config with overrides applied.

Parameters:

Name	Type	Description	Default
overrides	Dict[str, Any]	Values to override.	required

Returns:

Type	Description
AgentConfig	New AgentConfig with overrides.

to_dict()

Convert to dictionary.

ConfigParseError

Bases: Exception

Raised when YAML parsing fails.

ConfigValidationError

Bases: Exception

Raised when configuration validation fails.

load_config(file_path)

Load agent configuration from a YAML file.

Notes

Performs:

1. YAML parsing
2. Environment variable interpolation
3. Validation

Parameters:

Name	Type	Description	Default
file_path	str	Path to the YAML configuration file.	required

Returns:

Type	Description
Dict[str, Any]	Validated configuration dictionary.

Raises:

Type	Description
FileNotFoundError	If the file doesn't exist.
ConfigParseError	If YAML parsing fails.
ConfigValidationError	If validation fails.

load_config_from_string(yaml_content)

Load agent configuration from a YAML string.

Parameters:

Name	Type	Description	Default
yaml_content	str	YAML content as a string.	required

Returns:

Type	Description
Dict[str, Any]	Validated configuration dictionary.

Raises:

Type	Description
ConfigParseError	If YAML parsing fails.
ConfigValidationError	If validation fails.

merge_configs(base, overrides)

Merge base configuration with runtime overrides.

Overrides take precedence. Nested dictionaries are merged recursively.

Parameters:

Name	Type	Description	Default
base	Dict[str, Any]	Base configuration dictionary.	required
overrides	Dict[str, Any]	Dictionary of override values.	required

Returns:

Type	Description
Dict[str, Any]	New merged configuration dictionary.

validate_config(config)

Validate an agent configuration dictionary.

Notes

Checks for:

• Required fields (model)
• Valid model name (exists in model_information.json)
• Valid parameter values (ranges, types)
• Compatible parameter combinations

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Configuration dictionary to validate.	required

Returns:

Type	Description
Dict[str, Any]	Validated configuration with defaults applied.

Raises:

Type	Description
ConfigValidationError	If configuration is invalid.

validate_max_tool_result_chars(config)

Validate the optional top-level max_tool_result_chars key.

Maps to Agent(max_tool_result_chars=...): omitted or null keeps the registry-derived default, 0 turns the tool-result cap off, and a positive integer is the cap in characters. Checked both for a YAML key and for a from_config / from_config_string override, so a bad value fails at load time rather than at the first tool call.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	The configuration (YAML plus any overrides).	required

Returns:

Name	Type	Description
None	None	Returns when the key is absent, None or a non-negative int.

Raises:

Type	Description
ConfigValidationError	If the value is not a non-negative integer (booleans are rejected).

Example

validate_max_tool_result_chars({"max_tool_result_chars": 50_000})  # ok
validate_max_tool_result_chars({"max_tool_result_chars": -1})  # raises

validate_tool_result_preview_chars(config)

Validate the optional top-level tool_result_preview_chars key.

Maps to Agent(tool_result_preview_chars=...): omitted keeps the 500-character preview, 0 makes the tool_call_result events carry the whole result (plus result_data), and a positive integer is the preview length. Checked both for a YAML key and for a from_config / from_config_string override.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	The configuration (YAML plus any overrides).	required

Returns:

Name	Type	Description
None	None	Returns when the key is absent or a non-negative int.

Raises:

Type	Description
ConfigValidationError	If the value is not a non-negative integer (booleans and null are rejected).

Example

validate_tool_result_preview_chars({"tool_result_preview_chars": 0})  # ok
validate_tool_result_preview_chars({"tool_result_preview_chars": -1})  # raises

validate_truncated_result_read_chars(config)

Validate the optional top-level truncated_result_read_chars key.

Maps to Agent(truncated_result_read_chars=...): omitted or null keeps the feature off, 0 is off too, and a positive integer is how many more characters of each truncated tool result the model may read back with the built-in read_truncated_result tool. Checked both for a YAML key and for a from_config / from_config_string override.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	The configuration (YAML plus any overrides).	required

Returns:

Name	Type	Description
None	None	Returns when the key is absent, null or a non-negative int.

Raises:

Type	Description
ConfigValidationError	If the value is not a non-negative integer or null (booleans are rejected).

Example

validate_truncated_result_read_chars({"truncated_result_read_chars": 200000})  # ok
validate_truncated_result_read_chars({"truncated_result_read_chars": -1})  # raises

Tool Loader

doc_processing_toolkit.agent.loader

ToolLoader class for loading tools from various formats.

The ToolLoader handles all 9 supported tool input formats:

1. @tool decorated functions
2. File paths to .py files with @tool functions
3. Directory paths containing .py files with @tool functions
4. FastMCP server objects
5. MCP server script paths
6. MCP commands (npx/uvx)
7. Remote MCP URLs
8. Pre-built ClientSession dicts
9. Client-side remote MCP connections ({"type": "mcp_remote_client", "url": ...})

FastMCPSession

MCP-compatible session wrapper for FastMCP server objects.

This wraps a FastMCP server and provides the MCP session interface by directly accessing the server's internal tool registry.

__init__(server, name='fastmcp-server')

Initialize the FastMCP session wrapper.

Parameters:

Name	Type	Description	Default
server	Any	A FastMCP server object.	required
name	str	Name for this session.	'fastmcp-server'

call_tool(name, arguments) async

Call a tool on the FastMCP server.

A registry tool (SDK-bundled FastMCP, fastmcp<3) is called through its own run since 0.5.3 unless DPT_FASTMCP_VALIDATED_CALL=off, so arguments are validated and coerced as the server does, and a synchronous tool runs on a worker thread unless DPT_FASTMCP_SYNC_TOOLS_OFFLOOP=off (:func:_call_fastmcp_tool). A fastmcp>=3 server is called through its public call_tool. Errors become isError results.

list_tools() async

List all tools from the FastMCP server.

Reads the private registry the SDK-bundled mcp.server.fastmcp.FastMCP and standalone fastmcp<3 expose; falls back to the public async list_tools() of standalone fastmcp>=3.

LoadedTool dataclass

Unified representation of a loaded tool.

Attributes:

Name	Type	Description
name	str	Tool name.
description	str	Tool description.
input_schema	Dict[str, Any]	JSON Schema for tool parameters.
function	Optional[Callable[..., Any]]	The callable function (for @tool decorated functions).
is_async	bool	Whether the function is async.
source_type	str	Original format type.
server_name	Optional[str]	Name of the MCP server (for MCP tools).
session	Optional[Any]	MCP ClientSession (for MCP tools).

LocalFunctionSession

MCP-compatible session wrapper for local @tool decorated functions.

This class provides the same interface as an MCP ClientSession, allowing the provider functions to work with local Python functions as if they were MCP tools.

This enables @tool decorated functions to work seamlessly with the function calling path without requiring actual MCP infrastructure.

Notes

The session implements:

• list_tools(): Returns tool definitions in MCP format
• call_tool(name, arguments): Executes the tool and returns results

__init__(name='local-functions')

Initialize the local function session.

Parameters:

Name	Type	Description	Default
name	str	Name for this session (used for identification in logs).	'local-functions'

add_tool(loaded_tool)

Add a LoadedTool to this session.

Parameters:

Name	Type	Description	Default
loaded_tool	LoadedTool	A LoadedTool with function and schema information.	required

call_tool(name, arguments) async

Call a tool by name with given arguments (MCP-compatible interface).

Parameters:

Name	Type	Description	Default
name	str	Tool name to call.	required
arguments	Dict[str, Any]	Dictionary of arguments.	required

Returns:

Type	Description
MCPCallToolResult	MCPCallToolResult with the tool's output.

list_tools() async

List all tools in this session (MCP-compatible interface).

Returns:

Type	Description
MCPToolsResponse	MCPToolsResponse containing tool schemas.

MCPCallToolResult

MCP-compatible tool call result.

MCPToolSchema

MCP-compatible tool schema representation.

MCPToolsResponse

MCP-compatible tools list response.

ToolLoader

Loader for converting various tool formats into a unified representation.

Notes

The ToolLoader is responsible for

• Detecting the format of each tool input
• Loading/connecting to tools as needed
• Converting tools to provider-specific formats
• Managing tool lifecycle (cleanup on shutdown)

Supported formats:

1. @tool decorated function - Direct Python functions
2. File path string - .py file containing @tool functions
3. Directory path string - Directory of .py files with @tool functions
4. FastMCP server - In-process MCP server instance
5. MCP script path dict - {"type": "mcp_script", "path": "..."}
6. MCP command dict - {"type": "mcp_command", "command": "npx", "args": [...]}
7. Provider-native remote MCP URL dict - {"type": "mcp", "server_url": "..."}
8. Pre-built session dict - {"session": ClientSession, "name": "..."}
9. Client-side remote MCP dict - {"type": "mcp_remote_client", "url": "..."}, connected and executed by this loader rather than by the provider

Sessions for formats 4, 5, 6, 8 and 9 are opened once per event loop by :meth:connect_mcp_servers and reused until :meth:disconnect_mcp_servers, which keeps the configs so a later connect reconnects them.

Formats 5 and 6 start a subprocess, which sees only the MCP SDK's short allow-list of this process's environment (PATH, the home directory and a few others; the list differs by platform). Both take env (a dict merged over that allow-list — name every variable the server needs, an API key or a DPTTOOLS_* setting) and cwd (the directory it starts in) since 0.5.3; neither value is logged.

Every format the toolkit runs itself takes allowed_tools: the dicts of formats 5, 6, 8 and 9 carry the key, and format 4 takes it in the dict form {"type": "fastmcp", "server": server, "allowed_tools": [...]}. No key exposes every tool; a listed name the server lacks logs one WARNING.

Example

loader = ToolLoader()

# Load from mixed formats
tools = loader.load([
    my_tool_function,
    "/path/to/tools.py",
    fastmcp_server,
    {"type": "mcp", "server_url": "https://mcp.example.com"},
])

# Convert to provider format
anthropic_tools = loader.convert_to_provider_format(tools, "anthropic")

__init__()

Initialize the ToolLoader.

connect_mcp_servers() async

Connect to all MCP servers and create sessions for local tools.

This method creates MCP-compatible sessions for all local tool types:

• @tool decorated functions → LocalFunctionSession
• FastMCP server objects → FastMCPSession
• MCP scripts → subprocess-based session
• MCP commands (npx/uvx) → subprocess-based session
• Pre-built ClientSessions → Load tools asynchronously
• mcp_remote_client configs → client-side Streamable HTTP / SSE session

Provider-native remote MCP configs ({"type": "mcp", "server_url": ...}) are NOT connected here - they're handled natively by the provider functions.

Should be called before running the agent with local tools. Calling it again on the same event loop is a no-op: sessions are opened once and reused until :meth:disconnect_mcp_servers. If the sessions were opened on a different event loop (run_sync creates a fresh loop per call), the stale connections are discarded and the servers are reconnected on the current loop.

convert_to_provider_format(tools, provider)

Convert tools to provider-specific format.

Parameters:

Name	Type	Description	Default
tools	List[Union[LoadedTool, Any]]	List of LoadedTool objects or raw tool inputs.	required
provider	str	Target provider ("anthropic", "openai", "google", "fireworks").	required

Returns:

Type	Description
List[Dict[str, Any]]	List of tool definitions in the provider's expected format.

Raises:

Type	Description
ValueError	If provider is not supported.

detect_format(tool)

Detect the format of a tool input.

Parameters:

Name	Type	Description	Default
tool	Any	A tool in any supported format.	required

Returns:

Type	Description
str	Format string: "function", "file_path", "directory",
str	"fastmcp", "mcp_script", "mcp_command", "mcp_remote",
str	"client_session", or "mcp_remote_client".

Raises:

Type	Description
TypeError	If the format cannot be determined.

disconnect_mcp_servers() async

Disconnect from all connected MCP servers.

Should be called when the agent is done to clean up resources. Properly cleans up async context managers via the exit stack.

Teardown is symmetric with :meth:connect_mcp_servers: the tools that connect registered (mcp_script, mcp_command, client_session, mcp_remote_client, and deferred FastMCP servers) are dropped with their sessions, while the server configs are kept so the next connect_mcp_servers() call reconnects them. Sessions are only closed from the event loop that opened them; from any other loop (or with the opening loop already closed) the references are discarded without awaiting the dead context managers.

execute_tool(name, arguments) async

Execute a tool by name with the given arguments.

Parameters:

Name	Type	Description	Default
name	str	The tool name.	required
arguments	Dict[str, Any]	Tool arguments as a dictionary.	required

Returns:

Type	Description
Any	The tool's return value.

Raises:

Type	Description
ValueError	If tool not found.
RuntimeError	If tool execution fails.

get_remote_mcp_configs()

Get remote MCP configurations for provider-native handling.

get_tool_by_name(name)

Get a loaded tool by its name.

Parameters:

Name	Type	Description	Default
name	str	The tool name to look up.	required

Returns:

Type	Description
Optional[LoadedTool]	LoadedTool if found, None otherwise.

has_local_mcp_tools()

Check if any tools require the function calling path.

Returns True if there are any local tools that need to be passed to the LLM for function calling. This includes:

• @tool decorated functions
• FastMCP servers (including deferred fastmcp>=3 servers whose tools are listed at connect time)
• MCP scripts/commands
• Pre-built client sessions
• Client-side remote MCP servers (mcp_remote_client)
• Python files/directories with @tool functions

Returns False only when:

• There are no tools at all, or
• Only mcp_remote tools are present (handled by structured output with provider-native remote MCP support)

Returns:

Type	Description
bool	True if local MCP tools are present that need function calling.

load(tools)

Load tools from various formats into unified representation.

Parameters:

Name	Type	Description	Default
tools	Optional[List[Any]]	List of tools in any supported format.	required

Returns:

Type	Description
List[LoadedTool]	List of LoadedTool objects.

Raises:

Type	Description
TypeError	If a tool format is not recognized.
FileNotFoundError	If a file path doesn't exist.

Notes

Each call replaces the loader's tool set: pending MCP server, remote MCP and FastMCP registrations are rebuilt alongside the loaded tools, so loading twice never accumulates duplicate server registrations.

to_provider_format(tools, provider)

Alias for convert_to_provider_format.

TruncatedResultSession

MCP-compatible session serving the built-in read_truncated_result tool.

Agent adds it to a run's servers when truncated_result_read_chars is set, from the first request, so the tool list never changes mid-run. It reads from the :class:~doc_processing_toolkit.utils.TruncatedResultStore that the tool loop's size cap fills.

__init__(store)

Create the session over a store.

Parameters:

Name	Type	Description	Default
store	Any	The run's or the session's TruncatedResultStore.	required

call_tool(name, arguments) async

Return the next slice of a truncated result (MCP-compatible interface).

Parameters:

Name	Type	Description	Default
name	str	Tool name; only read_truncated_result is served.	required
arguments	Dict[str, Any]	tool_call_id and offset.	required

Returns:

Type	Description
MCPCallToolResult	MCPCallToolResult with the slice, or an error the model can read.

list_tools() async

List the one tool (MCP-compatible interface).

Returns:

Type	Description
MCPToolsResponse	MCPToolsResponse holding the read_truncated_result schema.

fastmcp_sync_tools_offloop_enabled()

Whether a synchronous in-process FastMCP tool runs on a worker thread.

Controlled by DPT_FASTMCP_SYNC_TOOLS_OFFLOOP (default on). off restores the behaviour before 0.5.3 exactly: the function runs on the event-loop thread, and a slow call stalls streaming, hooks and every other task on the loop (a concurrent run, a session's other agent) for its duration. Async tools and fastmcp>=3 servers (called through their own call_tool) are not affected either way. See :func:_fastmcp_switch_enabled for the values.

Returns:

Name	Type	Description
bool	bool	True unless the variable is off.

Example

import os

os.environ["DPT_FASTMCP_SYNC_TOOLS_OFFLOOP"] = "off"
fastmcp_sync_tools_offloop_enabled()  # False

fastmcp_validated_call_enabled()

Whether an in-process registry tool is called through its own run.

Controlled by DPT_FASTMCP_VALIDATED_CALL (default on). On, the SDK-bundled FastMCP validates and coerces the arguments as its server does over stdio, which a tool written against the old direct call can notice: a pydantic-model parameter arrives as the model instance rather than a dict, date / Path / Enum parameters arrive converted, and a value the schema rejects (2.5 for an int, null for a parameter that is not Optional) is an error instead of being passed through. off restores the behaviour before 0.5.3 exactly: fn(**arguments) with the arguments as the model sent them (an unknown argument then raises, and a tool asking for a Context cannot be called). See :func:_fastmcp_switch_enabled for the values.

Returns:

Name	Type	Description
bool	bool	True unless the variable is off.

Example

import os

os.environ["DPT_FASTMCP_VALIDATED_CALL"] = "off"
fastmcp_validated_call_enabled()  # False

Media Transformer

doc_processing_toolkit.agent.media

MediaTransformer class for transforming media inputs to provider formats.

The MediaTransformer handles conversion of file paths, URLs, bytes, and pre-encoded content to the format required by each provider's API.

When a filename is supplied (via dict input, _from_path, or kwargs on raw bytes), it is preserved in the formatted output for Anthropic, OpenAI, and Google so that tool_files upload loops can pass the original filename through to each provider's Files API and the user-supplied name survives into the code-execution container. Fireworks does not use file uploads.

Provider Media Support:

• Anthropic: Images (base64), PDFs (base64), Office docs (base64); filename preserved when provided
• OpenAI: Images (base64), PDFs (base64), Office docs (base64); filename preserved when provided
• Google: Images (base64), PDFs (bytes), Video (bytes), Audio (bytes), YouTube URLs, Office docs (bytes); filename preserved when provided
• Fireworks: Images only (base64)

MediaTransformer

Transformer for converting media inputs to provider-specific formats.

The MediaTransformer normalizes file paths, URLs, and raw bytes to each provider's expected format.

Notes

Different providers have different requirements for media content

• Anthropic: Base64-encoded with media_type; filename preserved when supplied
• OpenAI: Base64 with media_type; filename preserved when supplied (required for documents, recommended for tool_files uploads of any category)
• Google: Various formats including bytes for video/audio; filename preserved when supplied and used as the Files API display_name for tool_files
• Fireworks: Base64 for images only

Example

transformer = MediaTransformer(provider="anthropic", logger=logger)

# Transform file paths
parts = transformer.transform(["image.png", "document.pdf"])

__init__(provider, logger=None)

Initialize the MediaTransformer.

Parameters:

Name	Type	Description	Default
provider	str	Target provider ("anthropic", "openai", "google", "fireworks").	required
logger	Optional[Logger]	Logger instance for warnings/errors.	None

Raises:

Type	Description
ValueError	If provider is not supported.

detect_media_type(file_path)

Detect the MIME type of a file.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path to the file.	required

Returns:

Type	Description
str	MIME type string (e.g., "image/png", "application/pdf").

encode_base64(file_path)

Read a file and encode it as base64.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path to the file.	required

Returns:

Type	Description
str	Base64-encoded string of the file contents.

get_supported_types(provider=None)

Get list of supported MIME types for a provider.

Parameters:

Name	Type	Description	Default
provider	Optional[str]	Provider name (uses instance provider if not specified).	None

Returns:

Type	Description
List[str]	List of supported MIME types.

is_supported(file_path, provider=None)

Check if a file type is supported by a provider.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path to the file.	required
provider	Optional[str]	Provider name (uses instance provider if not specified).	None

Returns:

Type	Description
bool	True if the file type is supported, False otherwise.

transform(files, **kwargs)

Transform a list of files to provider-specific format.

Parameters:

Name	Type	Description	Default
files	List[Union[str, Path, bytes, Dict[str, Any]]]	List of file inputs. Each can be: str: File path or URL Path: Path object bytes: Raw file bytes (requires media_type in kwargs or detection) dict: Pre-structured content	required
**kwargs	Any	Provider-specific options.	{}

Returns:

Type	Description
List[Dict[str, Any]]	List of content parts in provider format.

Raises:

Type	Description
FileNotFoundError	If a file doesn't exist.
ValueError	If file type not supported by provider.

transform_files(files, provider, **kwargs)

Alias for transform() for backwards compatibility.

Parameters:

Name	Type	Description	Default
files	List[Union[str, Path]]	List of file paths.	required
provider	str	Target provider name (ignored, uses instance provider).	required
**kwargs	Any	Provider-specific options.	{}

Returns:

Type	Description
List[Dict[str, Any]]	List of content parts in provider format.

Media Policy

doc_processing_toolkit.agent.media_policy

Media policy for persisted sessions: inventory, externalise, strip, re-inline.

An AgentSession stores media the way each provider's API took it: base64 inside an Anthropic source, a data URL inside an OpenAI or Fireworks part, inline_data inside a Gemini part, and a Files API file_data URI for the media Gemini took by upload. This module is the one place that knows those shapes, so the session can:

• inventory them (:func:media_parts, behind persistence_report);
• externalise the bytes to a :class:MediaStore and leave a $media:<ref> marker in the slot (:func:externalize_messages);
• strip them to a text placeholder (:func:strip_messages);
• re-inline the markers from the store on restore (:func:reinline_messages), reporting a missing blob instead of failing;
• classify each part against a model's registry input_types (:func:registry_key_for_part).

Nothing here touches a provider or the network. The markers and the top-level media table are the only envelope additions, and both are ignored by an older reader (which therefore cannot replay an externalised or stripped envelope — the documented limit).

See Also

• docs/concepts/history_checkpointing.md

DirectoryMediaStore

A :class:MediaStore that keeps each blob as <sha256>.<ext> in one directory.

References are the file names, not paths, so an envelope and its directory can be moved together. A blob already present is not rewritten (put is idempotent by digest). Paths past Windows MAX_PATH work.

Parameters:

Name	Type	Description	Default
directory	Union[str, PathLike]	The directory; created on the first put.	required

Example

store = DirectoryMediaStore("checkpoints/media")
session.to_json(path="checkpoints/s.json", media_store=store)
restored = AgentSession.from_json("checkpoints/s.json", agent=agent, media_store=store)

get(ref)

Read the blob stored under ref.

ref comes from a checkpoint envelope, so it is checked before it is used: only a <sha256>.<ext> name as :meth:put returns is accepted, and the file it names must be directly inside the store's directory. A reference with a directory separator, a .. segment or an absolute path is refused, which the session reports as a missing blob.

Parameters:

Name	Type	Description	Default
ref	str	A file name returned by :meth:put.	required

Returns:

Type	Description
bytes	The bytes.

Raises:

Type	Description
ValueError	If ref is not a plain <sha256>.<ext> file name inside the directory.
FileNotFoundError	If no such blob exists.

put(sha256, mime_type, data)

Write data as <sha256>.<ext> and return that file name.

Parameters:

Name	Type	Description	Default
sha256	str	Hex digest of data.	required
mime_type	Optional[str]	Used to pick the extension (bin when unknown).	required
data	bytes	The bytes.	required

Returns:

Type	Description
str	The file name inside the directory.

Raises:

Type	Description
ValueError	If sha256 is not a hex digest (the file name is built from it, so anything else could name a path outside the directory).

MediaSlot

One media part found in a stored history, with the way to rewrite it.

Attributes:

Name	Type	Description
kind		"base64" for bytes carried in the envelope, "google_file_uri", "provider_file" or "url" for references.
dialect		"anthropic", "openai", "fireworks", "google" or "openai_generated" — decides the placeholder shape.
slot		"base64" (a bare base64 string) or "data_url" (a data:<mime>;base64,<payload> string); None for references.
mime_type		The declared MIME type, if any.
payload		The base64 payload for kind == "base64", else None.
ref		The reference for a reference kind (URI, file id, URL), else None.
message_index		Index of the message the part belongs to.
role		The message's role.
part		The part dict itself.

size_bytes property

Decoded size of an inline payload (0 for a reference).

record()

The inventory record persistence_report lists.

MediaStore

Bases: Protocol

Where externalised media bytes live between a save and a restore.

Any object with these two methods qualifies; :class:DirectoryMediaStore is the bundled one, and an object-storage host writes eight lines against the same shape. put is called at save time with the SHA-256 of the bytes, so a store can de-duplicate by returning the reference it already holds for that digest; get is called at restore time and may raise (KeyError, FileNotFoundError, anything), which the session reports as a missing blob rather than failing the restore. The ref handed to get is read from the envelope, so a store must treat it as data, never as a path: the bundled directory store accepts only the <sha256>.<ext> names its own put returns, and an object store should key on the digest the same way rather than interpolate the reference into a path.

Example

class S3MediaStore:
    def __init__(self, bucket, prefix=""):
        self.bucket, self.prefix = bucket, prefix

    def put(self, sha256, mime_type, data):
        key = f"{self.prefix}{sha256}"
        s3.put_object(Bucket=self.bucket, Key=key, Body=data, ContentType=mime_type or "application/octet-stream")
        return key

    def get(self, ref):
        return s3.get_object(Bucket=self.bucket, Key=ref)["Body"].read()

get(ref)

Return the bytes stored under ref.

put(sha256, mime_type, data)

Store data and return the reference to write into the envelope.

as_media_store(value)

Normalise a media_store argument.

Parameters:

Name	Type	Description	Default
value	Any	None, an object with put and get, or a directory path (str / os.PathLike) for a :class:DirectoryMediaStore.	required

Returns:

Type	Description
Optional[MediaStore]	The store, or None.

Raises:

Type	Description
TypeError	If value is none of those.

base64_payload_size(data)

Decoded size in bytes of a base64 string (0 for anything else).

classify_part(part, message_index=0, role=None)

Recognise part as a media part of any dialect.

Parameters:

Name	Type	Description	Default
part	Any	A content block / item / Gemini part dict.	required
message_index	int	For the record.	0
role	Any	For the record.	None

Returns:

Name	Type	Description
A	Optional[MediaSlot]	class:MediaSlot, or None when part carries no media.

data_url_payload(value)

(mime_type, base64_payload) for a data: URL, else None.

externalize_messages(messages, store, table)

Move every inline payload to store and leave a marker in its slot.

Parameters:

Name	Type	Description	Default
messages	List[Any]	The stored history (not mutated).	required
store	MediaStore	Receives the bytes.	required
table	Dict[str, Dict[str, Any]]	The envelope's media table, extended with one entry per reference: {"kind": "external", "ref", "sha256", "mime_type", "size", "slot"}.	required

Returns:

Type	Description
List[Any]	The rewritten copy.

google_files_api_uri(value)

Whether value is a Gemini Files API URI (as opposed to a YouTube URL).

is_marker(value)

Whether value is a $media:<ref> store reference.

iter_slots(messages)

Yield a :class:MediaSlot for every media part in messages.

marker_ref(value)

The reference inside a marker string.

media_parts(messages)

Inventory every media part a stored history carries, in message order.

Parameters:

Name	Type	Description	Default
messages	List[Any]	The stored session messages.	required

Yields:

Type	Description
Dict[str, Any]	One record per part: ``{"message_index", "role", "kind", "mime_type",
Dict[str, Any]	"size_bytes", "ref"}``.

registry_key_for_part(slot, provider)

The input_types key a stored media part corresponds to.

Mirrors the current-turn check in utils.validate_user_prompt_parts: base64-<mime> for images and PDFs on every provider, and on Google the category names video, audio, document and youtube.

Parameters:

Name	Type	Description	Default
slot	MediaSlot	The part.	required
provider	Optional[str]	The provider the history will be sent to.	required

Returns:

Type	Description
Optional[str]	The key, or None when the part cannot be judged (a reference whose
Optional[str]	type is unknown).

reinline_messages(messages, table, store, report)

Put the bytes back into every $media:<ref> slot from store.

A blob the store cannot return (no store configured, get raised, reference absent from the table) becomes a text placeholder plus a {"reason": "missing_blob", ...} report entry; the restore never fails on media.

Parameters:

Name	Type	Description	Default
messages	List[Any]	The restored history (not mutated).	required
table	Dict[str, Dict[str, Any]]	The envelope's media table.	required
store	Optional[MediaStore]	The store to read from, or None.	required
report	List[Dict[str, Any]]	Extended with one entry per missing blob.	required

Returns:

Type	Description
List[Any]	The rewritten copy.

replace_slot(messages, target, replacement)

Replace target's part in place inside messages (identity match).

resolve_media_policy(policy, store)

Turn auto into the concrete policy and validate the name.

Parameters:

Name	Type	Description	Default
policy	Optional[str]	One of :data:MEDIA_POLICIES, or None for auto.	required
store	Optional[MediaStore]	The store in force, if any.	required

Returns:

Type	Description
str	"inline", "externalize" or "strip".

Raises:

Type	Description
ValueError	If policy is not a known name, or is externalize without a store.

rewrite_messages(messages, transform)

Return a deep copy of messages with every media part passed through transform.

Parameters:

Name	Type	Description	Default
messages	List[Any]	The stored history.	required
transform	Callable[[MediaSlot], Any]	Called once per media part with its :class:MediaSlot; returns the replacement part (or the same part to keep it).	required

Returns:

Type	Description
List[Any]	The rewritten copy; the input is never mutated.

strip_messages(messages, report, *, references=False, label='removed on save', source=None)

Replace every inline payload by a text placeholder.

Parameters:

Name	Type	Description	Default
messages	List[Any]	The stored history (not mutated).	required
report	List[Dict[str, Any]]	Extended with one {"reason": "stripped", ...} entry per part.	required
references	bool	Also replace reference parts (a Google file_uri, a URL, a provider file id), which the save-time policy leaves alone. session_history_config={"preserve_media": False} uses this, because a replayed reference still costs the media's tokens.	False
label	str	The closing words of the placeholder text.	'removed on save'
source	Optional[str]	Recorded on every report entry when given ("preserve_media"), so a host can tell a storage-time strip from a save-time one.	None

Returns:

Type	Description
List[Any]	The rewritten copy.

validate_max_inline_bytes(value)

Validate media_max_inline_bytes: a non-negative int, or None (no cap).

Parameters:

Name	Type	Description	Default
value	Any	The caller's value.	required

Returns:

Type	Description
Optional[int]	The validated value.

Raises:

Type	Description
ValueError	If value is negative, a bool, or not an int.

History Conversion

doc_processing_toolkit.agent.history_conversion

Cross-provider conversion of stored session history (session-portability P3).

An AgentSession stores its conversation in the dialect of the provider that produced it: Anthropic content blocks, OpenAI Responses items, Gemini parts, Fireworks (Chat Completions) tool_calls. To continue such a conversation on another provider the history is read into a neutral intermediate representation and written back out in the target's dialect:

• a user turn is a list of parts — text, or media by bytes (base64) or URL;
• an assistant turn is optional text plus the tool calls it issued (name, arguments, an id);
• a tool turn is the results of those calls (id, name, text output).

What cannot cross a provider boundary is dropped and reported, never silently: signed thinking (Anthropic), thought parts and signatures (Google), reasoning items (OpenAI), reasoning_content (Fireworks), provider-run tool blocks and items (web search, code execution, MCP), media a target row's input_types do not list, and media whose bytes are not available. Tool-call ids are re-keyed to the target's convention and every re-key is reported. A history that cannot be converted at all — an open tool round, a tool call without a name, a result with no call — raises :class:SessionCompatibilityError naming the turn, and nothing partial is stored.

Nothing here touches a provider or the network.

See Also

• docs/concepts/sessions.md ("Continuing on another provider")
• docs/concepts/history_checkpointing.md

ConversionReport

The record of one conversion: dropped artefacts and re-keyed ids.

Each entry is a JSON-clean dict: kind ("dropped" or "rekeyed"), index (the stored message the entry concerns), from / to (the dialects), and for a drop artefact (thinking, thought, thought_signature, reasoning, reasoning_content, provider_item, provider_block, provider_part, media, tool_result_media, container, unknown_part) with detail; for a re-key name, from_id and to_id.

dropped(index, artefact, detail, **extra)

Record an artefact the conversion left out.

rekeyed(index, name, from_id, to_id)

Record a tool-call id rewritten to the target's convention.

SessionCompatibilityError

Bases: ValueError

A stored session cannot be restored under the agent it was given.

Raised by AgentSession.from_dict / from_json / rebind when the history cannot be converted for the target provider: an open tool round (calls with no results), a tool call with no name, a tool result that matches no call — or, with DPT_SESSION_CROSS_PROVIDER_CONVERSION=off, for any history written in another provider's dialect. The message names the stored dialect, the target provider and, when one turn is at fault, the turn. A history of plain text turns is provider-neutral and is never refused.

Example

try:
    session = AgentSession.from_json(path, agent=Agent(model="gpt-6-luna"))
except SessionCompatibilityError as exc:
    print(exc)

convert_history(messages, *, source, target, model=None, input_types=None, google_media=None)

Convert a stored history from one provider's dialect to another's.

Parameters:

Name	Type	Description	Default
messages	List[Any]	The stored session messages.	required
source	str	Their dialect.	required
target	str	The provider to write for.	required
model	Optional[str]	The target model (for placeholders and the report).	None
input_types	Optional[List[str]]	The target row's input_types.	None
google_media	Optional[GoogleMediaLookup]	Resolver for Files API URIs in a Google history.	None

Returns:

Type	Description
Tuple[List[Dict[str, Any]], ConversionReport]	(converted_messages, report).

Raises:

Type	Description
SessionCompatibilityError	When the history cannot be converted (see :func:read_history).

Example

converted, report = convert_history(
    session.messages, source="anthropic", target="openai", model="gpt-6-luna"
)
for entry in report.entries:
    print(entry["kind"], entry.get("artefact"), entry["index"])

decode_media_bytes(part)

The raw bytes of a neutral media part (a convenience for hosts and tests).

google_part_kind(part)

Classify one stored Gemini part.

Parameters:

Name	Type	Description	Default
part	Any	A Part.model_dump() dict.	required

Returns:

Type	Description
str	"thought" (a thought-summary part), "text", "function_call",
str	"function_response", "inline_data", "file_data",
str	"provider" (executable code and its result) or "unknown".

iter_google_parts(msg)

Yield (kind, part) for every part of a stored Gemini message.

read_history(messages, source, report, *, google_media=None)

Read a stored history into neutral turns.

Parameters:

Name	Type	Description	Default
messages	List[Any]	The stored session messages.	required
source	str	Their dialect ("anthropic", "openai", "google", "fireworks").	required
report	ConversionReport	Receives every dropped artefact.	required
google_media	Optional[GoogleMediaLookup]	Resolves a Files API URI to (mime_type, base64); None means no bytes are available for any URI.	None

Returns:

Type	Description
List[Dict[str, Any]]	The neutral turns, validated: every tool round is closed and every
List[Dict[str, Any]]	result matches a call.

Raises:

Type	Description
SessionCompatibilityError	For an open tool round, a call without a name, or a result that matches no call.
ValueError	If source is not a known dialect.

summarize_report(entries)

One human-readable line per artefact kind, for the WARNING log.

Parameters:

Name	Type	Description	Default
entries	List[Dict[str, Any]]	ConversionReport.entries.	required

Returns:

Type	Description
List[str]	Lines such as "dropped 3 reasoning (messages 1, 4, 7)" and
List[str]	"re-keyed 2 tool-call ids".

write_history(turns, target, report, *, model=None, input_types=None)

Write neutral turns in target's dialect.

Parameters:

Name	Type	Description	Default
turns	List[Dict[str, Any]]	From :func:read_history.	required
target	str	The provider the history will be sent to.	required
report	ConversionReport	Receives every re-keyed id and every media part the target cannot take (replaced by a text placeholder).	required
model	Optional[str]	The target model, named in placeholders and report entries.	None
input_types	Optional[List[str]]	The target row's input_types; None writes every media part.	None

Returns:

Type	Description
List[Dict[str, Any]]	The stored-message shape the target provider's history builder takes.

Raises:

Type	Description
ValueError	If target is not a known dialect.

Compaction

doc_processing_toolkit.agent.compaction

Provider-agnostic context compaction for doc-processing-toolkit.

This module implements a two-level context compaction architecture:

• Level 1 (Pre-Call): Compacts session history before any API call in _execute_run(). Uses count_tokens() from core.py for token estimation. All four strategies are supported since session history now contains provider-native messages.

• Level 2 (Intra-Run): Compacts provider-native messages between tool roundtrips inside _run_tool_calling_loop_*() functions. Uses API-reported input_tokens for accurate triggering. All four strategies are supported.

Both levels archive the original conversation to a JSON file (when enabled) before applying compaction.

See Also:

- [Context compaction](../concepts/compaction.md)

CompactionConfig dataclass

Configuration for context compaction behaviour.

Attributes:

Name	Type	Description
strategy	str	One of "summarize", "sliding_window", "tool_result_clearing", or "combined".
trigger_threshold	float | int	Float (0.0-1.0) interpreted as a percentage of the model's context window, or an int for an absolute token count.
summary_model	str | None	Model to use for the summarization LLM call. Defaults to the agent's own model when None.
summary_prompt_file	str | None	Path to a text file containing a custom summarization prompt.
summary_prompt	str | None	Inline custom summarization prompt. Takes precedence over summary_prompt_file.
keep_recent_messages	int	Number of recent messages to keep verbatim when using sliding_window or combined. Inside a tool loop the window never opens after the current turn's request, so it can keep more; DPT_COMPACTION_PROTECT_CURRENT_TURN=off restores the plain count.
keep_recent_tool_results	int	Number of recent tool results to preserve in tool_result_clearing / combined.
clear_tool_inputs	bool	Also clear tool call parameter payloads during tool_result_clearing.
archive_history	bool	Whether to archive the original conversation before compacting.
archive_dir	str	Directory for archive files. Ignored when archive_sink is set.
archive_sink	Callable[[list[Any]], str | None] | None	Optional callable receiving the archived conversation instead of writing a local file, for a host whose filesystem is ephemeral (a sandbox, a container) or who persists elsewhere. It is called with the same sanitised, JSON-serialisable payload that would have been written, and may return a locator string (an object key, a URL) which is then used everywhere the file path would have been: CompactionResult.archive_path, AgentResult.compaction_archives and the summary's reference section. Raising is contained — the compaction and the run continue, with a WARNING that names the exception type and nothing else. Ignored when archive_history is False.
include_archive_path_in_summary	bool	Include the archive file path in the compaction summary output.
combined_order	list[str]	Ordered list of strategies for the combined strategy.
min_messages_before_compaction	int	Minimum number of messages required before compaction can trigger.
max_compactions_per_session	int | None	Maximum compaction events per session. None means unlimited.
summary_max_tokens	int	Maximum output tokens for the summarization LLM call. Must be a positive integer no greater than the model's max_output_tokens_request (validated at config time when the model is known). Defaults to 4096.
sliding_window_include_truncation_notice	bool	When True, the sliding_window strategy prepends a synthetic user message informing the model that earlier context was truncated. This helps the model understand that it may be missing prior conversation context. Defaults to True.

CompactionResult dataclass

Metadata returned from a compaction operation.

Attributes:

Name	Type	Description
triggered	bool	True if compaction was actually performed.
strategy	str | None	The strategy that was applied (None if not triggered).
tokens_before	int	Approximate token count before compaction.
tokens_after	int	Approximate token count after compaction. Computed by scaling tokens_before by a local pre/post size ratio (~4 chars/token estimate) — the compaction entry points have no API client, so no exact recount is possible.
tokens_cleared	int	Approximate tokens removed (tokens_before - tokens_after; same approximation).
archive_path	str | None	Path to the archive file (None if archiving disabled or not triggered).
compaction_number	int	Sequence number of this compaction event.
summary_input_tokens	int	Input tokens consumed by the summarization call.
summary_output_tokens	int	Output tokens produced by the summarization call.
summary_cost	float	USD cost of the summarization call.

check_and_perform_compaction(messages, input_tokens, config, model_name, provider, context_window_tokens, compaction_count, session_id, logger, request_start_emitter=None) async

Level 2 (intra-run) compaction check and execution.

Called from _run_tool_calling_loop_*() between complete tool roundtrips. Uses API-reported input_tokens for triggering.

Parameters:

Name	Type	Description	Default
messages	list[Any]	Provider-native message list (modified in-place on compaction).	required
input_tokens	int	Token count reported by the latest API response.	required
config	CompactionConfig	Resolved compaction configuration.	required
model_name	str	Model name for summarization.	required
provider	str	Provider name.	required
context_window_tokens	int	Context-window limit for the model.	required
compaction_count	int	How many compaction events have already occurred.	required
session_id	str	Session identifier for archiving.	required
logger	Logger	Logger instance.	required

Returns:

Type	Description
CompactionResult	A CompactionResult describing what happened.

Notes

The current turn is protected. summarize, the summarise step of combined and sliding_window compact only the messages before the current turn's user message — the last plain-text user message (_is_plain_user_turn) — and keep that message, the in-flight assistant draft and the open tool round verbatim. The next request therefore continues the turn itself, not a summary's account of it, and no continuation turn is needed because the list still ends on the pending tool result. sliding_window returns its ordinary window whenever that window already contains the turn's user message, and otherwise opens the window on it. When nothing before the turn is left to compact — the whole list is the current turn, no plain-text user message exists, or everything before it is already an earlier compaction's output — the call falls back to tool_result_clearing on the current turn (CompactionResult.strategy then reads "tool_result_clearing") and logs why at INFO. Setting DPT_COMPACTION_PROTECT_CURRENT_TURN=off restores the folding described below byte for byte; tool_result_clearing alone and Level-1 compaction are unaffected either way.

The pending user turn is preserved. summarize and combined replace the whole list with a summary, stranding the question the current call is about to answer — Google rejects a request ending on a model turn outright ("Requests ending with a model turn are not supported") and the other providers answer the summary instead. So a trailing plain-text user message is re-appended after compaction.

The re-append is skipped only when the compacted tail already is that pending message — not when the tail merely carries the user role. Google's recap is itself a single user-role Content (a request may not end on a model turn), so a role test would drop the question on that provider; the question follows the recap as a second consecutive user turn, which every provider here accepts.

Only plain-text user messages are re-appended. A trailing tool-result message (Anthropic tool_result blocks and Google function_response parts also ride user-role messages) is folded into the summary instead: re-appending one would orphan it from its tool_use and would make an aggressive mid-loop compaction re-pose the question every pass, looping the model until max_iterations. sliding_window and tool_result_clearing keep the tail already, so this is a no-op for them.

When no plain-text tail exists to restore — the mid-tool-loop case — the compacted list would end on the assistant summary. A request whose last message is an assistant turn is an assistant prefill, which Anthropic prohibits whenever thinking is enabled and Google rejects outright, so a synthetic user continuation (COMPACTION_CONTINUATION_TEXT) is appended instead.

A second, always-on rule applies at this seam: the latest assistant message may not be modified while the request still returns tool_result blocks. _protect_latest_tool_round restores that round verbatim if a strategy rewrote it.

check_and_perform_pre_call_compaction(history, estimated_tokens, config, model_name, provider, context_window_tokens, compaction_count, session_id, logger, request_start_emitter=None) async

Level 1 (pre-call) compaction check and execution.

Called from Agent._execute_run() before any API call when session history is present. Operates on session-format messages which now contain provider-native structures (tool calls, tool results, etc.).

Parameters:

Name	Type	Description	Default
history	list[Any]	Session-format message list ([{role, content}] dicts), ending on the pending user turn. Modified in-place on compaction. Since 0.5.1 summarize (and combined with a summarize step) compacts the messages before a plain user turn at the tail and keeps that turn verbatim after the recap.	required
estimated_tokens	int	Pre-computed token estimate for the full context (system prompt + history + user prompt).	required
config	CompactionConfig	Resolved compaction configuration.	required
model_name	str	Model name for summarization.	required
provider	str	Provider name.	required
context_window_tokens	int	Context-window limit for the model.	required
compaction_count	int	How many compaction events have already occurred.	required
session_id	str	Session identifier for archiving.	required
logger	Logger	Logger instance.	required

Returns:

Type	Description
CompactionResult	A CompactionResult describing what happened.

resolve_compaction_config(raw=None)

Merge a raw configuration dict with CompactionConfig defaults.

Parameters:

Name	Type	Description	Default
raw	dict[str, Any] | None	Optional dict of overrides. Keys that do not correspond to CompactionConfig fields are silently ignored.	None

Returns:

Type	Description
CompactionConfig	A fully resolved CompactionConfig instance.

Raises:

Type	Description
ValueError	If strategy is not one of the valid options or trigger_threshold is not positive.

resolve_context_window(model_name, provider, logger)

Resolve a model's context window from the registry, warning loudly on a miss.

Shared by all four provider compaction blocks and the Level-1 pre-call check (they previously carried divergent inline fallbacks of 200000 and 128000 that silently decayed as the model lineup changed: 1M-context models compacted at ~15-20% utilization while sub-128k models could overflow before compaction fired).

On a registry miss this warns — naming the exact JSON path to fix — and floors at 128000. Warn-and-floor rather than raise: compaction is an optional resilience feature, and aborting a long run over a missing registry row would trade a recoverable inefficiency (premature compaction) for a total failure. The floor errs toward compacting too early, never toward context overflow.

Parameters:

Name	Type	Description	Default
model_name	str	Registry model name.	required
provider	str	Provider key under limits (anthropic/google/openai/fireworks).	required
logger	Logger	Logger for the loud-miss warning.	required

Returns:

Type	Description
int	The model's context_window_tokens, or 128000 on a miss.

resolve_threshold(threshold, context_window_tokens)

Convert a threshold value to an absolute token count.

A float in (0.0, 1.0] is interpreted as a fraction of the context window. An int (or float > 1.0) is used as-is.

Parameters:

Name	Type	Description	Default
threshold	float | int	The configured threshold.	required
context_window_tokens	int	The model's context-window size in tokens.	required

Returns:

Type	Description
int	Absolute token threshold.

Hooks and Steering

doc_processing_toolkit.agent.hooks

Hooks and real-time steering for Agent runs.

A hook is an async (or plain sync) callable handed to Agent(hooks=...) that the toolkit awaits at defined points in the tool-calling loop. The hook receives a typed :class:~doc_processing_toolkit.agent.cost.AgentStreamEvent and may return a :class:HookDirective that changes what happens next.

Two channels coexist by design:

• stream_callback — a synchronous, fire-and-forget observation tap whose return value never alters the loop (existing contract, unchanged).
• hooks — an awaited control channel: the loop genuinely stops at each hook point until the coroutine returns, and the returned directive can veto a tool call, inject a message, pause, or abort the run.

Event vocabulary: hooks are awaited at run_start, iteration_start, tool_call_start, before_tool_call (the veto point), tool_call_result, iteration_complete, and run_end, plus the opt-in request_start (B10) when Agent(emit_request_start=True). The streaming deltas (text_delta / thinking_delta) are not hook points — they reach stream_callback only.

Directive semantics (honored only at safe points). "Safe point" is a real constraint, not a hedge: a hook can stop a tool call from starting, and it can change what happens after one finishes, but nothing here pre-empts a tool that is already executing. skip_tool is the veto and it is offered before execution; pause and abort take effect at the next boundary, after the in-flight tool returns its result. A host that needs a hard stop mid-tool must build the timeout into the tool itself:

• continue (or returning None): pure side-effect hook — the loop proceeds exactly as if no hook existed. This is the normal case.
• skip_tool (valid at before_tool_call only): the pending tool call is NOT executed; the toolkit automatically answers the model's tool_use slot with a synthetic "declined" tool result (the message field customizes its text) — required because every provider demands a matching tool result for each tool call the model emitted; an unanswered slot is an API error (Anthropic 400s outright).
• inject_message: appends a user-role message at the next turn boundary (after tool results are appended, before the next API call — the same proven seam compaction uses). Returned from before_tool_call it is queued rather than applied immediately (a message cannot legally sit between a tool call and its result).
• pause: the loop awaits an asyncio.Event at the next safe point; resume via :meth:RunHandle.resume. In-process only (v1): a paused run does not survive process restarts and holds its resources while paused.
• abort: raises :class:AgentAborted carrying partial token/cost state.

request_start is observation-only: it fires immediately before a real provider send, where no tool call is pending and no message can legally be inserted, so any directive other than continue is ignored with a WARNING.

See Also

• Hooks and steering

AgentAborted

Bases: Exception

Raised when a hook directs the run to abort.

Attributes:

Name	Type	Description
reason		The hook-supplied rationale (may be None).
partial		Best-effort partial state at the abort point — {"input_tokens", "output_tokens", "cost", "iterations"}.
undelivered_injections	List[str]	Hook-injected texts that were still queued when the run ended and were therefore never shown to the model. Agent._execute_run populates this as the run unwinds, so an abort at the final turn boundary hands the text back instead of destroying it. Empty unless a hook queued a message that no drain reached.

AgentHooks

Bases: Protocol

Protocol for hook objects: a single awaited entry point.

ConfirmAfterNToolCalls

Built-in hook: pause the run after every N executed tool calls.

The "has been working for a while — continue?" pattern a host app shows its user. Counts tool_call_result events (one iteration can execute many tools), pauses at the following safe point, and leaves resuming to the host via the run handle (deliver the user's answer back to the process holding the handle, and apply a timeout policy so an unanswered confirmation doesn't hold resources forever). The count restarts at every run_start, so one instance can be shared across an Agent's runs.

Examples:

```python
confirm = ConfirmAfterNToolCalls(5)
agent = Agent(model="claude-haiku-4-5", tools=[...], hooks=confirm)
# elsewhere, when the user answers: agent.run_handle.resume()
```

HookDirective dataclass

Directive returned by a hook to steer the run.

Attributes:

Name	Type	Description
action	Literal['continue', 'skip_tool', 'inject_message', 'pause', 'abort']	One of "continue", "skip_tool", "inject_message", "pause", "abort".
message	Optional[str]	For inject_message: the user-role text to inject at the next turn boundary. For skip_tool: the text of the synthetic declined tool result (defaults to a standard notice).
reason	Optional[str]	Optional free-text rationale (logged; for abort it is carried on the raised :class:AgentAborted).

HookRunner

Bridges the provider loops to the user's hooks.

One instance per run. Provider loops call :meth:emit at each hook point; the runner awaits the hook, honors pause/abort inline, queues inject_message directives, and returns skip_tool directives to the before_tool_call site (the only place a veto is meaningful).

close_injections()

Close the queue and return anything stranded in it.

The backstop for the one path that never reaches :meth:take_undelivered: a CancelledError/GeneratorExit, where the consumer went away mid-run. Agent._execute_run calls this from its finally so every run, however it ended, leaves a queue that rejects further injections. Idempotent, and ordinarily returns an empty list because the end-of-run capture ran first.

Returns:

Type	Description
List[str]	Texts that were queued after the last capture and can no longer be
List[str]	handed to anyone.

drain_injections()

Return and clear the queued user-role injection texts.

Called at a provider loop's turn boundary, where each returned text is appended to the live message list. Every drained text is therefore also recorded, in order, on :attr:consumed_injections — the record that becomes AgentResult.injected_messages.

Returns:

Type	Description
List[str]	The queued injection texts, in the order the hook queued them.
List[str]	Empty when nothing is pending.

emit(event_type, content=None, metadata=None) async

Emit one event to the hook and enforce its directive.

Returns the directive ONLY when the caller must act on it locally (skip_tool at a before_tool_call site). pause and abort are handled here; inject_message is queued for :meth:drain_injections at the next turn boundary.

Events listed in :data:OBSERVATION_ONLY_EVENTS — request_start today — are delivered to the hook but never steer: a directive other than continue is logged at WARNING and dropped. The run handle's own pause/abort channel still applies, because that is the host speaking, not the hook's return value.

Raises:

Type	Description
AgentAborted	When the hook (or the run handle) requested an abort.

has_injections()

True when injection texts are queued for the next turn boundary.

queue_injection(text)

Append a user-role text to the pending queue. Thread-safe.

The single writer behind both channels: a hook's inject_message directive and :meth:RunHandle.inject from a host thread. Keeping them on one queue is what makes the delivery rule uniform — the next turn boundary drains whatever is there, and whatever is left at the end of the run is returned rather than destroyed.

Parameters:

Name	Type	Description	Default
text	str	The user-role message to queue.	required

Returns:

Type	Description
bool	True when the text was queued; False when the run's end-of-run
bool	capture has already taken the queue, which means nothing will ever
bool	read this text and the caller must be told.

take_undelivered()

Return and clear queued texts that were never shown to the model.

Called once at the end of a run (Agent._execute_run), after the run_end hook has had its own chance to queue a message. A text is still queued at that point only because it arrived after the run's last turn boundary — at the final iteration_complete, from the run_end hook itself, or at a forced-tool_choice exit that sits above the drain.

Deliberately does not record into :attr:consumed_injections: these texts were never appended to any message list, so recording them would double-count them against the run's history once the host re-sends them as the next turn's user message.

Taking the queue also closes it, atomically under the same lock: from here on :meth:queue_injection refuses, so a host thread calling RunHandle.inject a moment too late is told the run has ended instead of queueing a message nobody will ever read.

Returns:

Type	Description
List[str]	The undelivered injection texts, in queue order. Empty when the run
List[str]	delivered everything, which is the ordinary case.

track_usage(input_tokens=0, output_tokens=0, cost=0.0)

Accumulate best-effort partial state (used by AgentAborted).

RequestStartEmitter

Emits the opt-in request_start event at each real provider send (B10).

One instance per :meth:Agent.run / :meth:Agent.stream call, created only when Agent(emit_request_start=True); when the switch is off no emitter is built and the event stream is byte-identical to earlier releases. Providers take it as request_start_emitter and call :meth:emit (async tool loops) or :meth:emit_sync (the sync structured/text paths) immediately before every actual provider send — the first request, each tool-loop iteration, each max_tokens / truncation / empty-turn re-request, each pause_turn continuation, and each compaction summarisation call.

The counter is shared with every view returned by :meth:for_purpose, so request_index is monotonic across the whole run — including the summarisation requests compaction makes on the side.

Attributes:

Name	Type	Description
model		The model id the run was started with.
provider		"anthropic", "openai", "google" or "fireworks".
purpose		The default purpose stamped on payloads from this view.

Example

seen = []
agent = Agent(model="claude-haiku-4-5", emit_request_start=True)
async for event in agent.stream("hi"):
    if event.event_type == "request_start":
        seen.append(event.metadata["request_index"])

request_count property

How many provider sends this run has announced so far.

Returns:

Name	Type	Description
int	int	The current 1-based request_index high-water mark, shared
	int	across every view produced by :meth:for_purpose.

__init__(*, model, provider, stream_callback=None, hooks_runner=None, retry_accumulator=None, purpose='generation', _counter=None)

Build an emitter for one run.

Parameters:

Name	Type	Description	Default
model	str	The model id, echoed on every payload.	required
provider	str	The provider name, echoed on every payload.	required
stream_callback	Callable or None	The run's streaming tap.	None
hooks_runner	HookRunner or None	The run's hook channel. Only the tool-calling loops have one; the structured/text paths do not, so request_start reaches hooks on the tool path only.	None
retry_accumulator	Any or None	The run's RetryAccumulator. Its attempt list is how the emitter learns the current tenacity attempt number; without one every payload reports attempt: 1.	None
purpose	str	Default purpose for payloads from this view.	'generation'
_counter	list or None	Internal — the shared one-element request counter. Never passed by callers; :meth:for_purpose uses it.	None

bind_retry_accumulator(retry_accumulator)

Attach the run's RetryAccumulator so attempt can be derived.

The accumulator is built per provider call, after the emitter; this is how the two are joined. Views made by :meth:for_purpose before this call keep their own reference, so bind before handing views out.

Parameters:

Name	Type	Description	Default
retry_accumulator	Any	The run's RetryAccumulator.	required

emit(*, iteration=1, attempt=None, purpose=None) async

Announce a send from an asynchronous provider loop.

Reaches stream_callback first (the repo's event ordering) and then the hook channel, where the event is observation-only.

Parameters:

Name	Type	Description	Default
iteration	int	The 1-based tool-loop iteration.	1
attempt	int or None	Override the derived tenacity attempt number.	None
purpose	str or None	Override this view's default purpose.	None

emit_sync(*, iteration=1, attempt=None, purpose=None)

Announce a send from a synchronous provider path.

Reaches stream_callback only — the structured/text paths carry no hook channel.

Parameters:

Name	Type	Description	Default
iteration	int	The 1-based tool-loop iteration, or 1 on the single-request structured/text path.	1
attempt	int or None	Override the derived tenacity attempt number.	None
purpose	str or None	Override this view's default purpose.	None

for_purpose(purpose)

Return a view of this emitter with a different default purpose.

The counter is shared, so a request announced through the view advances the same request_index sequence. Used to hand compaction a "compaction_summary" view without threading a purpose argument through every summarisation call.

Parameters:

Name	Type	Description	Default
purpose	str	The default purpose for the returned view.	required

Returns:

Name	Type	Description
RequestStartEmitter	'RequestStartEmitter'	A new emitter sharing this one's counter,
	'RequestStartEmitter'	channels and identity fields.

RunHandle

Handle for controlling a running (possibly paused) agent run.

Exposed as Agent.run_handle for the duration of a run. pause is an in-process asyncio.Event: the loop blocks at its next safe point until :meth:resume is called (or :meth:abort fails the run cleanly).

:meth:resume, :meth:abort and :meth:inject may be called from any thread. A run started with run_sync blocks its caller inside asyncio.run, so the host necessarily releases it from another thread; the wake-up is then marshalled onto the run's event loop with call_soon_threadsafe — a bare asyncio.Event.set from a foreign thread would leave the loop parked in its selector, i.e. the run would stay paused indefinitely.

paused property

True while the run is paused.

abort(reason=None)

Request a clean abort at the next safe point (also unblocks a pause; any thread).

Like :meth:pause, this cannot pre-empt a tool that is already executing — that tool finishes and its result is delivered first, and the abort is raised at the following safe point. "Clean" means exactly that: no provider call is cut off mid-request and no tool call is left without its result.

Parameters:

Name	Type	Description	Default
reason	Optional[str]	Optional text carried on the resulting :class:AgentAborted.	None

inject(text)

Queue a user-role message for the run's next turn boundary.

The host-facing half of HookDirective(action="inject_message"): it lands in the same queue a hook's directive lands in, so the delivery rule is uniform and the accounting is unchanged — delivered at the next tool boundary if one comes, returned on AgentResult.undelivered_injections if not. It removes the need for a hook that polls the host's inbox on every event.

Safe to call from any thread: the queue is guarded by a threading.Lock, and this appends to it without touching the run's event loop. It does not pre-empt a tool that is already executing — drain latency is bounded by the host's longest-running tool.

Parameters:

Name	Type	Description	Default
text	str	The user-role message to queue. Must be a non-empty string.	required

Raises:

Type	Description
ValueError	If text is not a non-empty string.
RuntimeError	If this handle is not bound to a run, or the run has already ended. A message queued into a finished run would never be seen by anyone, so it is refused loudly rather than accepted and dropped.

Example

Steering a run from the host thread:

import threading

agent = Agent(model="gpt-5.6-luna", instructions="...", hooks=my_hooks)
thread = threading.Thread(target=lambda: agent.run_sync("summarise"))
thread.start()
# ... later, from the UI thread:
agent.run_handle.inject("actually, only the indemnity clauses")

pause()

Pause the run at its next safe point.

A request, not an interrupt: a tool that is already executing runs to completion and its result is delivered before the loop stops. Nothing cancels a tool mid-flight, so the latency between this call and the run actually stopping is bounded by the host's longest-running tool.

resume()

Resume a paused run (safe to call from any thread).

build_injected_user_message(provider, text)

Build a provider-native user-role message carrying text.

Parameters:

Name	Type	Description	Default
provider	str	"anthropic", "openai", "google", or "fireworks".	required
text	str	The message text to inject.	required

Returns:

Type	Description
Any	A provider-native message object/dict appendable to the loop's live
Any	message list.

stream_callback_accepts_metadata(stream_callback)

Whether stream_callback can be called with a third metadata argument.

stream_callback is a public provider parameter, and the historical text_delta call site passed only two arguments. B11 wants to attach {"structured_output": True} to the deltas that carry a schema-constrained payload, which must not break a host that wrote def cb(event_type, content). Callers inspect the signature once per provider call and keep the two-argument form when this returns False.

A callable whose signature cannot be read (a C builtin, an exotic wrapper) is treated as two-argument only, because breaking an existing tap is worse than omitting additive metadata.

Parameters:

Name	Type	Description	Default
stream_callback	Callable or None	The caller's streaming tap, or None.	required

Returns:

Name	Type	Description
bool	bool	True when a third positional/keyword metadata argument is
	bool	accepted, False otherwise (including for None).

Example

tag = {"structured_output": True} if stream_callback_accepts_metadata(cb) else None
cb("text_delta", chunk, tag) if tag else cb("text_delta", chunk)

Parameter Mapping

doc_processing_toolkit.agent.parameter_mapping

Parameter mapping for translating Agent parameters to provider-specific formats.

This module handles the mapping of universal Agent parameters to the specific parameter formats required by each provider's API.

ParameterMapper

Maps universal Agent parameters to provider-specific formats.

This class is the main entry point for parameter mapping, providing methods to map individual parameter types as well as full validation.

__init__()

Initialize the ParameterMapper.

get_provider_defaults(provider)

Get default parameter values for a provider.

Parameters:

Name	Type	Description	Default
provider	str	Provider name.	required

Returns:

Type	Description
Dict[str, Any]	Dictionary of default parameter values.

map_built_in_tools(provider, built_in_tools)

Map universal built-in tools config to provider format.

Parameters:

Name	Type	Description	Default
provider	str	Target provider name.	required
built_in_tools	Dict[str, Any]	Universal built-in tools configuration.	required

Returns:

Type	Description
Dict[str, Any]	Provider-specific built-in tools parameters.

map_thinking(provider, thinking_config)

Map universal thinking config to provider format.

Parameters:

Name	Type	Description	Default
provider	str	Target provider name.	required
thinking_config	Dict[str, Any]	Universal thinking configuration.	required

Returns:

Type	Description
Dict[str, Any]	Provider-specific thinking parameters.

map_tool_choice(provider, tool_choice)

Map universal tool_choice to provider format.

Parameters:

Name	Type	Description	Default
provider	str	Target provider name.	required
tool_choice	str	Universal tool choice ("auto", "required", "none", or tool name).	required

Returns:

Type	Description
Dict[str, Any]	Single-entry dict {"tool_choice": <value>} ready to merge into the
Dict[str, Any]	provider call kwargs. The value maps directly for all providers.

validate_provider_params(provider, params)

Validate that parameters are valid for a provider.

Parameters:

Name	Type	Description	Default
provider	str	Provider name.	required
params	Dict[str, Any]	Parameters to validate.	required

Returns:

Type	Description
Tuple[Dict[str, Any], List[str]]	Tuple of (validated_params, warnings)

Raises:

Type	Description
ValueError	If parameters are invalid for the provider.

ParameterValidator

Validates and maps parameters from unified Agent API to provider-specific formats.

Notes

Handles:

• Mapping universal parameter names to provider-specific names
• Transforming parameter values (e.g., thinking config)
• Validating provider-specific parameters aren't used with wrong providers
• Generating warnings for unsupported parameters

validate(provider, params)

Validate and filter parameters for a provider.

Parameters:

Name	Type	Description	Default
provider	str	Target provider name	required
params	Dict[str, Any]	Parameters to validate	required

Returns:

Type	Description
Dict[str, Any]	Tuple of (valid_params, warnings)
List[str]	valid_params: Parameters transformed for the provider
Tuple[Dict[str, Any], List[str]]	warnings: List of warning messages for ignored params

Raises:

Type	Description
ValueError	For provider-specific params used with wrong provider, or for conflicting parameters
TypeError	For a remote MCP server's allowed_tools in a form the provider does not take (raised by :func:transform_mcp_remote_servers; on Fireworks any allowed_tools is a ValueError).

apply_anthropic_allowed_tools(server, toolset_config, label)

Apply allowed_tools to one mcp_toolset entry of a direct Anthropic call (0.5.1).

Agent translates the key before the provider sees the config (see :func:_anthropic_allowed_tools_config), so a config that reaches the provider through Agent no longer carries it and this does nothing. A direct call_anthropic_* caller's key used to be ignored, so the model saw every tool; it is translated here the same way.

Parameters:

Name	Type	Description	Default
server	Dict[str, Any]	The caller's server config.	required
toolset_config	Dict[str, Any]	The {"type": "mcp_toolset", ...} entry being built, with any default_config / configs already copied; updated in place.	required
label	str	The server's name, for the error message.	required

Raises:

Type	Description
TypeError	If allowed_tools is not a list of tool-name strings.

check_provider_run_allowed_tools(server, provider)

Fail on a provider-run MCP server's allowed_tools before any request (0.5.1).

Agent calls this at construction for every provider-run server, so a value the request build would refuse fails there instead of at the first run. Fireworks refuses the key (:func:refuse_provider_ignored_allowed_tools), Anthropic takes a list of tool names, and OpenAI a list or its filter object (:func:openai_allowed_tools). Google has no provider-run MCP.

Parameters:

Name	Type	Description	Default
server	Dict[str, Any]	The provider-run MCP server config.	required
provider	str	The provider name.	required

Raises:

Type	Description
ValueError	If provider is "fireworks" and the config sets allowed_tools.
TypeError	If the value is not a form the provider takes.

detect_provider(model_name)

Detect provider from model name using model_information.json.

Parameters:

Name	Type	Description	Default
model_name	str	Model name (e.g., "claude-sonnet-4-5", "gpt-4o")	required

Note

_-prefixed keys are registry metadata at every level — section, provider and model row — and are never routable model IDs, so a _comments block sitting beside the model entries can never be resolved as a model.

Returns:

Type	Description
str	Provider name ("anthropic", "openai", "google", "fireworks")

Raises:

Type	Description
ValueError	If model not found in configuration
TypeError	If model_name is None

get_default_value(param_name, provider)

Get the default value for a parameter for a specific provider.

Parameters:

Name	Type	Description	Default
param_name	str	Parameter name	required
provider	str	Provider name	required

Returns:

Type	Description
Any	Default value, or None if no default

openai_allowed_tools(server, label)

Check the allowed_tools of an OpenAI provider-run MCP server or connector (0.5.1).

OpenAI takes a list of tool names or its own filter object {"tool_names": [...], "read_only": bool}. Both are sent as given, except that a tuple or set of names becomes a list. An empty list is sent too, so the model sees no tool from that server, as on every other format; before 0.5.1 the request dropped it and the model saw every tool. OpenAI reads an empty tool_names in the filter object as no filter.

Parameters:

Name	Type	Description	Default
server	Dict[str, Any]	The MCP server or connector config.	required
label	str	The server's label, for the error message.	required

Returns:

Name	Type	Description
Any	Any	None when the key is absent or null (send nothing: every
	Any	tool), else the list or the filter object to send.

Raises:

Type	Description
TypeError	If the value is neither a list of tool names nor a filter object whose tool_names is a list of names and whose read_only is a bool.

Example

openai_allowed_tools({"allowed_tools": ("search",)}, "docs")  # ["search"]
openai_allowed_tools({"allowed_tools": []}, "docs")  # [] (no tool)
openai_allowed_tools({}, "docs")  # None (every tool)

refuse_provider_ignored_allowed_tools(server, provider)

Refuse allowed_tools on a provider-run MCP server whose provider ignores it.

Fireworks accepts the field on a provider-run ({"type": "mcp"}) server and still exposes every tool: measured 2026-09-27 on kimi-k3 against the DeepWiki server, where the model listed all three tools with and without the filter, at the same input size. Passing the key on would expose tools the caller meant to hide, so the toolkit fails instead. mcp_remote_client runs the server from this process and filters it on every provider.

Parameters:

Name	Type	Description	Default
server	Dict[str, Any]	The provider-run MCP server config.	required
provider	str	The provider name.	required

Raises:

Type	Description
ValueError	If provider is "fireworks" and the config sets allowed_tools.

transform_mcp_remote_servers(servers, provider)

Transform remote MCP server configurations to provider-specific format.

Parameters:

Name	Type	Description	Default
servers	List[Dict[str, Any]]	List of remote MCP server configurations (unified format)	required
provider	str	Target provider name	required

Returns:

Type	Description
List[Dict[str, Any]]	List of provider-specific MCP server configurations

Raises:

Type	Description
ValueError	If provider doesn't support remote MCP servers

Notes

Each provider has a different format for remote MCP server configurations.

Examples:

# Anthropic format (per MCP connector docs):
{"type": "url", "url": "https://...", "name": "my-server", "authorization_token": "..."}

# OpenAI format:
{"type": "mcp", "server_label": "...", "server_url": "...", "require_approval": "never", "headers": {...}}

# Fireworks format (structured_output only): Similar to Anthropic format

# Google: Does not support remote MCP servers directly.

transform_thinking_param(thinking, provider)

Transform unified thinking config to provider-specific format.

Parameters:

Name	Type	Description	Default
thinking	Dict[str, Any]	Unified thinking configuration dict. For Anthropic, supports: Adaptive mode: {"type": "adaptive"} — recommended for models that support it (e.g., claude-opus-4-6). Claude dynamically decides when and how much to think. Use effort parameter for depth control. Legacy mode: {"budget_tokens": int, "interleaved": bool} — explicit budget-based thinking for older models. interleaved_thinking is accepted as an alias for interleaved. Off: {"type": "disabled"} (or {"type": "between_tools"}, the same request under the name claude-sonnet-5-5 uses). The type is forwarded as given and the registry row decides what the request carries, so either spelling works on every Claude row. budget_tokens and the interleaved flag are dropped; any other key is forwarded, and reaches the API only if the model cannot turn thinking off and the run thinks after all. Any other Anthropic key is preserved verbatim and reaches the API untouched, so a caller can use a thinking.* field the toolkit does not know about yet. For OpenAI, supports {"effort": str, "summary": str, "mode": str} — "mode" ("standard"/"pro", GPT-5.6+ models only) is forwarded only when explicitly set, since older models reject unknown reasoning fields. {"type": "disabled"} is not an off value on OpenAI or Fireworks: the type is not read, and the request carries the effort given, medium when there is none. One WARNING says so when no effort was given (0.5.2). Use {"effort": ...} with an effort the row lists (supported_reasoning_values in the registry) to reason less.	required
provider	str	Target provider name	required

Returns:

Type	Description
Any	Provider-specific thinking parameter value