Skip to main content
API v3 is available. This SDK targets the v2 API, which keeps working through its deprecation window. A new SDK major targeting v3 is in progress — see the v3 SDKs page and the v3 API reference.
The hand-written Python SDK for anyformat, built on httpx. It mirrors the TypeScript SDK and exposes a fluent builder over the same typed-graph workflow definition.

​
Installation

Requires Python 3.13 (the SDK pins to >=3.13,<3.14 today; this is expected to loosen before the official launch). 
pip install anyformat
The PyPI distribution is named anyformat; the import path uses the dotted namespace anyformat.sdk (transport) and anyformat.workflow (schema factories).

​
Authentication

Pass the API key to Client, or set ANYFORMAT_API_KEY in the environment and read it from there. 
import os
from anyformat.sdk import Client

# Explicit
client = Client(api_key="af_...")

# Or from the environment
client = Client(api_key=os.environ["ANYFORMAT_API_KEY"])
See Authentication for how to mint an API key.

​
Basic usage

Full flow: build a workflow with a fluent builder, run a document, await the typed result. 
import os
from anyformat.sdk import Client
from anyformat.workflow import Schema

client = Client(api_key=os.environ["ANYFORMAT_API_KEY"])

# 1. Build & create the workflow (parse → extract)
workflow = (
    client.workflow("Invoice Processor")
    .parse()
    .extract([
        Schema.string("invoice_number", "The unique invoice identifier"),
        Schema.float("total_amount",    "Total invoice amount"),
        Schema.date("issue_date",       "Date when the invoice was issued"),
    ])
    .create()
)
print(f"Workflow id: {workflow.id}")

# 2. Submit a document — accepts Path, str (path), or bytes.
run = workflow.run("invoice.pdf")

# 3. Wait for results. Polls /results/ (412 → still processing, 200 → done).
result = run.wait()  # default: timeout=300s, poll_interval=3s

print(result.fields["invoice_number"].value)
print(result.fields["total_amount"].value)
The Result exposes typed scalar accessors via result.fields[name] for linear workflows (one untagged extraction) and the full envelope at result.raw for everything else (parse markdown, multiple extractions per split, classifications, splits).

​
Async usage

AsyncClient is the async sibling — every builder, handle, and result method has an async counterpart. 
import asyncio
import os

from anyformat.sdk import AsyncClient
from anyformat.workflow import Schema

async def main():
    client = AsyncClient(api_key=os.environ["ANYFORMAT_API_KEY"])
    try:
        workflow = await (
            client.workflow("Invoice Processor")
            .parse()
            .extract([Schema.string("invoice_number", "...")])
            .create()
        )
        run = await workflow.run("invoice.pdf")
        result = await run.wait()
        print(result.fields["invoice_number"].value)
    finally:
        await client.aclose()

asyncio.run(main())

​
Builder methods

All node types in the typed graph are exposed as fluent methods. parse is required; the others are optional and can be chained in any topology the API allows.
MethodWhat it addsNotes
.parse(*, mode='standard', prompt_hint=None, figure_enhancement=False, cache=True)The required parse nodeAll args keyword-only. Knobs are mode-discriminated (per-mode @overloads): mode='agentic' adds effort='low'|'mid'|'accurate'; mode='lite' adds ocr_effort='medium'|'high', region='eu'|'global', skip_routing_review, figures, text_formatting, routing_effort.
.classify(*categories)A classify nodeCategories from ClassifyCategory(id=..., name=..., description=...)
.split(*rules, route_from=None)A splitter nodeUse route_from= after .classify() to wire which branch fans out
.extract(fields, *, branch=None, mode='standard', lookup_suggestion=None, lookup_reasoning_effort=None, lookup_file_uploads=None, use_images=False)An extract nodebranch= required after .classify() / .split(); Schema.* build fields. mode selects the tier ('max' staff-gated). lookup_reasoning_effort ('minimal'|'low'|'medium'|'high') + inline lookup_file_uploads + a field flagged lookup=True drive smart lookup. use_images feeds page images.
.validate(*rules, branch=None)A validate nodeAttaches to the most recent extract (or the one named by branch)
.create()Persists the workflowReturns a Workflow handle you can .run(...) on
.build()Same shape without the network callReturns a WorkflowDefinition — useful for tests and inspection
Workflow.run(file=None, *, text=None) accepts either file: bytes | pathlib.Path | str (a file path) or text= (raw text — useful for emails / plain-text bodies). Exactly one must be set. Returns a Run handle; Run.wait(timeout=300, poll_interval=3) returns the Result.

​
Smart lookup

A smart-lookup field is resolved by matching the document against a reference file (a CSV/catalog) instead of being read off the page. Flag the field with lookup=True, and pass the reference file to .extract(..., lookup_files=[...]) — the SDK reads each path and uploads it with the workflow. Here the model reads vendor_name, then resolves the canonical vendor_id from a vendor catalog: 
workflow = (
    client.workflow("Invoice + vendor lookup")
    .parse()
    .extract(
        [
            Schema.string("vendor_name", "Vendor as printed on the invoice."),
            Schema.float("total", "Grand total amount."),
            Schema.string("vendor_id", "Canonical vendor code from the catalog, joined on vendor_name.", lookup=True),
        ],
        lookup_files=["vendor_catalog.csv"],
        lookup_suggestion="Match the extracted vendor_name against the vendor_name column; return vendor_id.",
    )
    .create()
)

result = workflow.run("invoice.pdf").wait()
print(result.fields["vendor_id"].value)   # resolved from the catalog
lookup=True is accepted on every Schema.* factory. The looked-up field comes back in result.fields alongside the others — there’s no separate section for it. An extract with a lookup field but no lookup_files is rejected with 400. By default the lookup overwrites the field; add lookup_only_if_missing=True to extract it from the document too and let the lookup fill it only where extraction found no value.

​
Managing workflows

The client lists and deletes the workflows on your account:
MethodWhat it doesReturns
client.list_workflows(page=1, page_size=20, status=None)One page of your workflowslist[Workflow] — each is .run(...)-able (page_size max 100)
client.delete_workflow(workflow_id)Soft-deletes a workflowThe deleted workflow_id (a 404 raises NotFound)
# List your workflows, then run the first one
workflows = client.list_workflows(page_size=50)
for wf in workflows:
    print(wf.id, wf.name)

result = workflows[0].run("invoice.pdf").wait()

# Delete by id
client.delete_workflow(workflows[0].id)
Both have async equivalents on AsyncClient: await client.list_workflows(...) returns list[AsyncWorkflow] and await client.delete_workflow(id) returns the id.

​
Reading results

result = run.wait()

# Scalar fields, for linear workflows (single untagged extraction)
inv = result.fields["invoice_number"]          # ExtractedField
print(inv.value, inv.confidence, inv.evidence)

# Parse output (markdown + per-block confidence)
parse_markdown = result.parse.markdown if result.parse else None

# Anything else — nested objects, split workflows, classifications:
# read result.raw, the validated wire envelope as a dict.
for extraction in result.raw["extractions"]:
    for name, field in extraction["fields"].items():
        print(name, field)
See Response formats for the full shape of every section.

​
Error handling

The SDK raises typed exceptions you can catch: 
import time

from anyformat.sdk import (
    APIError,
    BadRequest,        # 400 — validation error, bad request body
    Unauthorized,      # 401 — invalid or missing API key
    Forbidden,         # 403 — disallowed by the server
    NotFound,          # 404 — workflow / file / webhook does not exist
    RateLimited,       # 429 — slow down; .retry_after has the suggested delay
    ServerError,       # 5xx — internal anyformat error
    SDKTimeout,        # local timeout — `.wait()` exceeded `timeout`
)

try:
    result = workflow.run("invoice.pdf").wait(timeout=60)
except RateLimited as e:
    time.sleep(e.retry_after or 5)
except NotFound:
    print("workflow or file is gone")
except APIError as e:
    print(f"HTTP {e.status_code}: {e.detail}")
except SDKTimeout:
    print("polling timed out — try webhooks for production")
APIError exposes .status_code, .error_code, and .detail. The wire-format error codes (e.g. EXTRACTION_FAILED, RATE_LIMITED) are documented at Errors.

​
Webhooks (not on the SDK surface yet)

Client doesn’t expose webhook endpoints today. Until it does, register and delete webhooks with httpx directly: 
import os

import httpx

webhook = httpx.post(
    "https://api.anyformat.ai/v2/webhooks/",
    headers={"Authorization": f"Bearer {os.environ['ANYFORMAT_API_KEY']}"},
    json={"url": "https://your-server.com/hook", "events": ["extraction.completed"]},
).json()
# webhook["secret"] is only returned once — store it.
See Webhooks for the payload shape and signature verification.