API reference

This is the canonical reference for the public Python API. CI compares every function heading below with the installed callable signature; update this page in the same change as any public signature.

How-to: Guide. Integration contract for generated code: For agents.

Extraction

Extractor(schema, model=None, instructions=None, *, style='direct', agent=None, model_settings=None, timeout=None, instrument=False, retry_policy=None, max_input_bytes=None, url_timeout=None)

Reusable synchronous extraction session. Enter it with with; then call extract(input_file, *, media_type=None) or extract_with_usage(input_file, *, media_type=None). The agent, model provider client, and URL-fetch client are constructed once and closed on context exit.

model accepts either a known model string or a configured pydantic_ai.models.Model. style selects how the model inspects the input (direct, search, or code; see Common arguments) and applies to every call in the session; search and code sessions build their harness agent and temporary workspace once on enter and remove both on exit. As an advanced alternative, agent accepts a fully configured Pydantic AI Agent; it is mutually exclusive with model, and its output is revalidated against schema. Non-direct styles cannot be combined with an injected agent.

AsyncExtractor(schema, model=None, instructions=None, *, style='direct', agent=None, model_settings=None, timeout=None, instrument=False, retry_policy=None, max_input_bytes=None, url_timeout=None)

Async session counterpart. Enter it with async with; then await extract or extract_with_usage. It shares one async HTTP client and one agent across calls, including concurrent calls made on the entering event loop. Close it manually with aclose() only when a context manager is impractical.

RetryPolicy(max_retries=0, backoff=1.0, max_backoff=60.0)

Frozen session retry configuration. Only transient ModelError failures are retried. The backoff and bounded provider Retry-After behavior match the one-shot function arguments.

ExtractionStyle

direct (default) sends resolved media to the model in one shot. search and code are text-only agentic styles powered by Pydantic AI Harness. Pass the enum (ExtractionStyle.SEARCH) or the string ("search"). Non-text inputs and a missing harness extra fail before the model call.

extract(schema, model, input_file=None, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

model also accepts an agent, and an agent that declares an output_schema can be passed as schema instead — extract(invoices, "doc.pdf") is the same call as extract(Invoice, invoices, "doc.pdf"). An agent that resolves to one local model runs as a normal one-shot call using the agent’s model, instructions, and style; an agent with subagents or a remote endpoint runs as a swarm and its outputs are reduced. Omitting input_file raises ValueError; the parameter is optional only so the agent form can shift it left. The same applies to the three siblings below.

Extract one input synchronously and return an instance of schema.

extract_async(schema, model, input_file=None, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Async counterpart to extract. It uses Agent.run and returns an instance of schema.

extract_with_usage(schema, model, input_file=None, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Extract one input synchronously and return (output, Usage). It has the same retry behavior as extract; Usage describes the successful model call.

extract_with_usage_async(schema, model, input_file=None, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Async counterpart to extract_with_usage; returns (output, Usage).

extract_many(schema, model, input_files, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_concurrency=5, return_exceptions=False, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Run concurrent extractions from synchronous code. Results preserve input order. When return_exceptions=True, per-item exceptions appear in the result list. Do not call this function from a running event loop; use extract_many_async instead.

extract_many_async(schema, model, input_files, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_concurrency=5, return_exceptions=False, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Async counterpart to extract_many; it has the same arguments, result ordering, and per-item retry behavior.

iter_extract_many_async(schema, model, input_files, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_concurrency=5, return_exceptions=False, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Return an async iterator of (input_index, result) pairs in completion order. Inputs are consumed lazily, at most max_concurrency items are scheduled, and results are available before the complete batch finishes. Simultaneous completions are yielded in input-index order.

This is the streaming counterpart to extract_many_async, which waits for every item and returns a list in input order. Keep the original input_files sequence (or a name on ExtractionInput) if you need to map input_index back to a path.

With return_exceptions=False, the first failure cancels and awaits outstanding work before being raised. With return_exceptions=True, item exceptions are yielded in the result position and streaming continues.

async for index, result in iter_extract_many_async(
    schema=PdfInfo,
    model="openai:gpt-5",
    input_files=paths,
    return_exceptions=True,
    max_concurrency=5,
):
    if isinstance(result, Exception):
        print(f"{index} failed: {result}")
    else:
        print(index, result.summary)

extract_many_with_results(schema, model, input_files, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_concurrency=5, return_exceptions=False, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Run a batch and return per-item ExtractionResult objects instead of bare schema instances. It has the same arguments, input ordering, concurrency, and retry semantics as extract_many, and each result carries token usage, attempt count, duration, model/media metadata, and a sanitized source label. With return_exceptions=True, failed items appear as Exception values in place. Use total_usage to aggregate token usage across the returned results.

extract_many_with_results_async(schema, model, input_files, instructions=None, *, style='direct', media_type=None, max_input_bytes=None, max_concurrency=5, return_exceptions=False, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Async counterpart to extract_many_with_results; it has the same arguments, result ordering, and per-item retry behavior.

total_usage(results)

Sum token usage across batch extraction results, for example the list returned by extract_many_with_results or extract_many_with_results_async. Returns a single Usage whose fields are the totals of the successful items.

Swarm

A swarm runs several agents over one input and reduces their outputs. The input is fetched and decoded once, so a swarm costs one load and N model calls. Each agent is told its position in the swarm so it works independently instead of assuming a peer covered a section.

Use a swarm when one pass under-recalls: a long document, a schema with many optional fields, or a job worth cross-checking with a second model. One document that one model handles well does not need one.

SwarmMember

SwarmMember(model, instructions=None, style=None) is one agent, where model is a model identifier, a configured pydantic-ai Model, or a RemoteAgent. Its instructions and style override the swarm-wide values, so a search reader and a direct reader can share the same swarm. A bare model identifier or a configured pydantic-ai Model is accepted anywhere a SwarmMember is.

SwarmResult

Returned by extract_swarm_with_results*. output is the reduced instance, agents holds each agent’s ExtractionResult or the exception it raised in agent order, usage sums the successful agents, and reduce is the strategy that produced output.

resolve_swarm_members(agents, size=None)

Expand the agents argument into one SwarmMember per agent. A single agent plus size fans it out size times (1..16); a list is used as-is and size may not contradict its length. Defined agents are flattened first, so a parent with subagents contributes one member per leaf.

extract_swarm(schema, agents, input_file, instructions=None, *, size=None, style='direct', reduce='merge', media_type=None, max_input_bytes=None, max_concurrency=None, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Run the agents concurrently over one input and return the reduced schema instance. max_concurrency defaults to min(5, agents). Agent failures are tolerated as long as one agent succeeds; if every agent fails, the first failure is raised. Raises RuntimeError from a running event loop.

extract_swarm_async(schema, agents, input_file, instructions=None, *, size=None, style='direct', reduce='merge', media_type=None, max_input_bytes=None, max_concurrency=None, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0)

Async counterpart to extract_swarm.

extract_swarm_with_results(schema, agents, input_file, instructions=None, *, size=None, style='direct', reduce='merge', media_type=None, max_input_bytes=None, max_concurrency=None, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0, on_agent_start=None, on_agent=None)

Same run, returning a SwarmResult. on_agent_start(index, total) and on_agent(index, total, result) report progress as agents start and finish.

extract_swarm_with_results_async(schema, agents, input_file, instructions=None, *, size=None, style='direct', reduce='merge', media_type=None, max_input_bytes=None, max_concurrency=None, max_retries=0, retry_backoff=1.0, retry_max_backoff=60.0, on_agent_start=None, on_agent=None)

Async counterpart to extract_swarm_with_results.

Agents

An agent packages what a swarm member would otherwise repeat at every call site — model, style, instructions, and output schema — so a caller imports a specialist instead of re-specifying it. Agents compose: a parent with subagents flattens into one swarm member per leaf, so an agent is accepted anywhere agents is.

define_agent(description, *, model=None, style=None, instructions=None, output_schema=None, subagents=())

Define a local agent. description is required. model is optional only when subagents is non-empty (a pure group). output_schema must be a pydantic.BaseModel subclass. Returns a frozen DefinedAgent.

define_remote_agent(url, description, *, auth=None, headers=None, path='/extract', output_schema=None)

Define an agent served over HTTP. Returns a frozen RemoteAgent. url may be a callable (sync or async) so a rotating deployment URL is resolved per request; headers and auth likewise. See Remote agent protocol.

flatten_agent(agent)

Expand one agent into the SwarmMember list it contributes: its own model first (when it has one), then each subagent’s members, depth first.

resolve_output_schema(agent)

Return the agent’s output_schema, or raise ValueError naming the agent when it declared none.

load_agent(spec)

Load one agent from a directory, a Python file, or a module:attribute path. A DefinedAgent / RemoteAgent passes through unchanged. A file or module must expose the agent as a module-level name (agent for files, the named attribute for module:attribute).

load_agents(spec)

Load several agents from a comma-separated string or a sequence of specs.

load_agent_directory(directory)

Load an agent from a directory laid out as:

invoices/
  agent.py          # defines `agent = define_agent(...)`
  instructions.md   # used when the agent declared no instructions
  subagents/
    line_items.py   # each defines `agent = ...`
    totals/         # nested directories load recursively

Entries beginning with . or _ are skipped and subagents load in sorted order. A directory with subagents but no agent.py becomes a group agent named after the directory; a directory with exactly one subagent resolves to it.

Remote agent protocol

RemoteAgent posts to url + path with {"schema": <JSON Schema>, "input": {"data": <base64>, "mediaType": <MIME>}, "instructions": ..., "style": ...} and reads {"output": ..., "usage": {...}} — a bare object is treated as the output, and {"error": "..."} (at any status) raises RemoteAgentError. usage is accepted in either inputTokens or input_tokens spelling. HTTP 408/409/425/429/5xx and transport failures are retryable and honour max_retries.

The agent URL goes through the same SSRF host allowlist as document URLs, so a local agent server needs OPENEXTRACT_ALLOW_PRIVATE_URLS=1.

openextract.auth

Outbound auth header providers for define_remote_agent(auth=...). Each is resolved per request, so a refreshed credential is picked up without redefining the agent.

bearer and basic accept the value itself or a callable (sync or async) returning it.

Swarm reduce

A swarm runs several agents over one input and folds their outputs into a single result. The fold strategy is SwarmReduce; the reducers are public so a caller can combine outputs it gathered itself.

SwarmReduce

merge (default) unions list fields and fills each scalar field from the first agent that produced a value. vote keeps the most frequent non-empty value per field, breaking ties toward the earlier agent; list fields have no majority, so they fall back to merge. first returns the first successful agent’s output untouched. Pass the enum (SwarmReduce.VOTE) or the string ("vote").

normalize_reduce(reduce='merge')

Return a valid SwarmReduce for an enum member or string, or raise ValueError naming the allowed strategies.

reduce_outputs(values, reduce='merge')

Fold a sequence of same-schema Pydantic model instances into one instance. merge and vote reduce the dumped payloads and re-validate the combined value, so the return value always satisfies the schema; a combination that no longer validates raises SchemaValidationError. An empty values raises ValueError.

Choosing a batch API

extract_many and extract_many_with_results raise RuntimeError from a running event loop; use the _async siblings or the iterator instead. See examples/batch/stream_batch_extract.py for a side-by-side run.

Input and result contracts

ExtractionInput

A frozen dataclass wrapping a single media source with optional per-item media metadata. Passing one to any extraction API (or mixing them into a batch) is equivalent to passing the raw source directly.

Batch item media types resolve per item: an ExtractionInput.media_type wins over the batch-wide media_type argument, which remains the fallback for raw items.

ExtractionResult

A frozen, generic dataclass returned by extract_many_with_results*. It never retains raw media, credentials, query strings, fragments, or provider internals; source is sanitized.

Common arguments

Batch functions also accept:

Usage

Usage is a frozen dataclass returned by the two usage helpers.

Configuration

The Python library reads provider configuration from the existing process environment and does not load .env files. The openextract CLI and bundled examples load .env explicitly as an application-level convenience.

Session model_settings are passed directly to Pydantic AI using its typed ModelSettings contract. timeout is an explicit shortcut for the model request timeout and overrides a timeout entry in model_settings. url_timeout separately controls URL input fetching. instrument=True enables Pydantic AI instrumentation; an InstrumentationSettings instance provides fine-grained control.

Extractor is bound to the thread that enters it and is not thread-safe. AsyncExtractor is bound to one event loop; concurrent calls within that loop are supported. Neither session can be reopened after it is closed.