anyformat

Python SDK for the AnyFormat API (v3). Build a document workflow with a fluent builder, run a file through it, and read typed results — including drawing the parser's layout boxes back onto the page.

pip install --pre anyformat

Release candidate. The v3 SDK ships as 1.0.0rc1; pip skips pre-releases unless --pre (or an explicit pin) is given, so plain pip install anyformat keeps resolving to the stable 0.7.x (v2) line until 1.0.0 final ships after the RC bake period.

from anyformat.sdk import Client
from anyformat.workflow import Schema

client = Client(api_key="af_...")  # or set ANYFORMAT_API_KEY

workflow = (
    client.workflow("Invoices")
    .parse()
    .extract([Schema.string("vendor", "Vendor name on the invoice.")])
    .create()
)

run = workflow.run("invoice.pdf")            # uploads a document packet + starts a run
result = run.wait()                          # polls the run until it is processed
print(result.fields["vendor"].value)         # -> "Acme Corp"
result.parse.draw("out/")                    # -> out/page_1.png with boxes drawn

The client reads ANYFORMAT_API_KEY from the environment if you omit api_key. base_url defaults to https://api.anyformat.ai.

Concepts in 30 seconds

• client.workflow(name) returns a fluent builder. Chain verbs (.parse(), .classify(), .split(), .extract(), .validate()) then .create() to register it and get a Workflow.
• workflow.run(file) uploads the file(s) as one document packet and triggers a Run — one execution attempt; re-running the same packet creates a new run. run.wait() polls GET /v3/runs/{run_id}/ until the status is terminal and returns a Result.
• Run statuses: queued → in_progress → processed (success) | error | cancelled. wait() raises ExtractionFailed on error and ExtractionCancelled on cancelled (both subclass RunFailed, so an except RunFailed still catches either), and transparently retries a transient 429/5xx while polling. There is no 412 polling in v3.
• workflow.upload(file) stages a packet without running it (no credits spent); call .run() on the returned Upload later.
• Lists are keyset-paginated: they return Page(items, next_cursor); pass next_cursor back as cursor= until it is None. No totals, newest first.
• result.fields is the extracted data; result.parse the parser's output (markdown + per-block layout); result.raw the full envelope.
• All ids are hyphenated UUIDs.

Cookbook

Every recipe below is a complete, runnable script. They share this header:

from anyformat.sdk import Client
from anyformat.workflow import Schema, ClassifyCategory, SplitterRule, ValidationRule

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

Quick parse: one call → markdown

For a fast, no-setup parse, client.parse() runs the platform's atomic /parse operation (a fast lite parse) and returns a handle. Poll get_markdown() for the parsed markdown — no workflow to author or manage.

run = client.parse("contract.pdf")
while True:
    try:
        print(client.get_markdown(run.id))   # the parsed markdown
        break
    except StillParsing:
        time.sleep(3)

1. Upload → run → results (the full v3 flow, step by step)

workflow.run(file) fuses these steps; here they are separately:

workflow = (
    client.workflow("Invoice extraction")
    .parse()
    .extract([Schema.string("vendor", "Vendor name."), Schema.float("total", "Grand total.")])
    .create()
)

upload = workflow.upload("invoice.pdf", idempotency_key="ingest-2026-07-04-0001")
print(upload.document_packet_id, [f.name for f in upload.files])

run = upload.run(idempotency_key="run-2026-07-04-0001")
result = run.wait(timeout=300, poll_interval=3)
print(result.fields["vendor"].value)

packet = client.get_document_packet(upload.document_packet_id)
print(packet.status, packet.latest_run_id)      # latest_run_id == run.id

Retrying a request with the same Idempotency-Key replays the original upload/run — no duplicate packet, no second (billed) extraction.

A packet can hold up to 100 files treated as one document:

run = workflow.run(files=["invoice.pdf", "annex.pdf"])

2. Poll a run yourself / list runs

detail = client.get_run(run.id)
print(detail.status)                  # queued | in_progress | processed | error | cancelled
if detail.result:                     # inline once status == "processed"
    print(detail.result.fields)

cursor = None
while True:
    page = client.list_runs(workflow.id, limit=50, cursor=cursor)
    for r in page.items:
        print(r.id, r.status, r.document_packet_id)
    if page.next_cursor is None:
        break
    cursor = page.next_cursor

list_workflows() and list_document_packets(workflow_id) paginate the same way.

To walk every row without cursor bookkeeping, use the auto-paging iterators — they follow next_cursor for you and fetch each page lazily:

for run in client.iter_runs(workflow.id):
    print(run.id, run.status)

all_workflows = list(client.iter_workflows())
for packet in client.iter_document_packets(workflow.id):
    ...

3. Edit a workflow without losing field identity (persistent_id)

GET returns the typed graph whose fields carry a server-assigned persistent_id. Echo it back on update — including across renames — and analytics, ground truth and quality metrics stay attached to the field. Omit it for new fields (an omitted id falls back to name-matching).

definition = client.get_workflow(workflow.id)      # typed WorkflowDefinition

for node in definition.nodes:
    if node.type == "extract":
        for field in node.extraction_schema.fields:
            if field.name == "vendor":
                field.name = "vendor_name"         # rename; persistent_id kept

client.update_workflow(workflow.id, definition)    # PATCH /v3/workflows/{id}/

When authoring from scratch, Schema factories accept the knob directly:

Schema.string("vendor_name", "Vendor name.", persistent_id="0686bb97-8c30-70f0-8000-97669e00aaaa")

4. Extract fields (linear parse → extract)

Schema is the field factory. Mix any field types in one extract.

workflow = (
    client.workflow("Invoice extraction")
    .parse()
    .extract([
        Schema.string("vendor", "Vendor name on the invoice."),
        Schema.float("total", "Grand total amount."),
        Schema.date("issued_on", "Invoice issue date."),
        Schema.boolean("is_paid", "Whether the invoice is marked paid."),
        Schema.integer("line_count", "Number of line items."),
        Schema.enum("currency", "Currency of the totals.", options=[
            Schema.option("USD", "US Dollar."),
            Schema.option("EUR", "Euro."),
        ]),
        Schema.object("line_items", "One row per line item.", fields=[
            Schema.string("sku", "Stock keeping unit."),
            Schema.string("description", "Line description."),
            Schema.float("amount", "Line amount."),
        ]),
    ])
    .create()
)

result = workflow.run("invoice.pdf").wait()
field = result.fields["vendor"]
print(field.value, field.confidence, [e.text for e in field.evidence])

Field types: string, integer, float, boolean, date, datetime, enum, multi_select, object (nested).

5. Classify, then extract per category

invoice = ClassifyCategory(id="INVOICE", name="Invoice", description="A vendor invoice.")
receipt = ClassifyCategory(id="RECEIPT", name="Receipt", description="A point-of-sale receipt.")

workflow = (
    client.workflow("Invoice or Receipt")
    .parse()
    .classify(invoice, receipt)
    .extract([Schema.string("vendor", "Vendor name.")], branch=invoice)        # object form
    .extract([Schema.string("merchant", "Merchant name.")], branch="RECEIPT")  # id form
    .create()
)

result = workflow.run("doc.pdf").wait()
for category in result.raw["classifications"]:
    print(category["category"], category["confidence"])

6. Split a multi-document file, then extract

statements = SplitterRule(id="STMT", name="Statement", description="A bank statement.")
checks = SplitterRule(id="CHECK", name="Check", description="A scanned check.")

workflow = (
    client.workflow("Statement bundle")
    .parse()
    .split(statements, checks)
    .extract([Schema.string("account", "Account number.")], branch="STMT")
    .extract([Schema.float("amount", "Check amount.")], branch="CHECK")
    .create()
)

result = workflow.run("bundle.pdf").wait()
for split in result.raw["splits"]:
    print(split["name"], split["files"])

7. Validate extracted fields

workflow = (
    client.workflow("Invoice with checks")
    .parse()
    .extract([
        Schema.float("subtotal", "Subtotal before tax."),
        Schema.float("tax", "Tax amount."),
        Schema.float("total", "Grand total."),
    ])
    .validate(
        ValidationRule(id="totals", description="total must equal subtotal + tax.", severity="error"),
        ValidationRule(id="positive", description="All amounts must be positive.", severity="warning"),
    )
    .create()
)

8. Different file inputs

run() and upload() accept a path string, a Path, raw bytes, or a list via files=.

from pathlib import Path

workflow.run("invoice.pdf")                       # path string
workflow.run(Path("scans") / "page.png")          # Path (PDF or image)
workflow.run(open("invoice.pdf", "rb").read())    # raw bytes
workflow.run(files=["a.pdf", "b.pdf"])            # one packet, many files
workflow.run(text="Plain text body")              # text, synthesised as a .txt file

9. Async client

AsyncClient mirrors the sync API with await on create, run, and wait.

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

async def main():
    client = AsyncClient(api_key="af_...")
    workflow = await (
        client.workflow("Async invoices")
        .parse()
        .extract([Schema.string("vendor", "Vendor name.")])
        .create()
    )
    run = await workflow.run("invoice.pdf")
    result = await run.wait()
    print(result.fields["vendor"].value)
    await client.aclose()

asyncio.run(main())

Both clients are context managers, which close the underlying connection pool for you — prefer this over a bare module-level client that leaks connections:

with Client(api_key="af_...") as client:
    result = client.workflow("Invoices").parse().extract([...]).create().run("invoice.pdf").wait()

async with AsyncClient(api_key="af_...") as client:
    ...

10. Enrich values with smart-lookup

Mark the field with source="smart_lookup" and pass the reference file to extract(..., lookup_files=[...]). A smart-lookup field is overwritten by the lookup; use source="lookup_if_missing" to extract it from the document too and let the lookup fill only the slots extraction left empty:

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.", source="smart_lookup"),
        ],
        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)   # -> "V-0001", resolved from the catalog

11. Upload documents into a workflow's dataset (ground truth)

Register a document into a workflow's dataset as one document packet — 1..100 files treated as one document, plus optional ground_truth (the expected extraction). Atomic: any rejected file or a ground-truth failure fails the whole call and stores nothing. Filenames are unique within a workflow, so by default a colliding name fails with 409; pass on_conflict="rename" to auto-rename instead (a rename is never silent):

entry = workflow.upload_to_dataset(
    "invoice.pdf",
    ground_truth={"vendor": "ACME", "total": "42.00", "line_items": [{"sku": "A1", "qty": "2"}]},
)
print(entry.document_packet_id, entry.ground_truth_saved)   # "069d…", True

ground_truth keys are the workflow schema's field identifiers (persistent_id); a scalar maps to str | None, a table/object field to a list of row dicts. Bulk ingestion loops one atomic call per document — there is no batch endpoint, so each call is bounded and independently retryable:

from anyformat.sdk import DatasetDocument

workflow.upload_documents_to_dataset([
    DatasetDocument(file="a.pdf", ground_truth={"vendor": "ACME"}),
    DatasetDocument(files=["b1.pdf", "b2.pdf"]),   # one 2-file document
])

Both are also available off the client: client.upload_to_dataset(workflow_id, ...) and client.upload_documents_to_dataset(workflow_id, documents).

Errors

Every v3 error carries {error, detail, error_code, retryable, request_id}; the raised exception exposes status_code, error_code, detail, retryable, and request_id. Dynamic values (e.g. the missing scope on a scope denial) are named in the error message, not a separate structured field.

The class is chosen by error_code where a condition is worth handling on its own, and otherwise by HTTP status. An unmapped code falls back to its status row (or bare APIError for an unmapped status).

StillParsing surfaces from read_markdown when a parse operation is polled before its body reports done.

Non-HTTP: RunFailed (run ended in a terminal failure — .run_id, .status), subclassed as ExtractionFailed (error) and ExtractionCancelled (cancelled) so you can branch on the exact outcome while except RunFailed still catches both; SDKTimeout (wait() budget exhausted); MissingResults (processed run with no inline results — .run_id). A transient 429/5xx during wait() polling is retried with backoff (honouring Retry-After), not surfaced.

CLI

The package installs an afx command for one-off parses without writing code:

export ANYFORMAT_API_KEY=af_...
afx parse invoice.pdf

API quick reference

Node configuration knobs

Every knob is a first-class .parse() / .extract() keyword argument.

.parse(...) — mode ("standard"/"agentic"/"lite"/"flash"), prompt_hint, figure_enhancement, cache; agentic-only effort ("low"/"mid"/"accurate"); lite: no knobs of its own — one OCR-only recipe that never reads prompt_hint; flash-only scanned_pages ("skip"/"fail") — no model call, no OCR. Mode-discriminated via per-mode @overloads — a wrong-mode knob is a type error.

.extract(...) — mode ("standard"/"agentic"/"lite"), use_images, lookup_suggestion, lookup_reasoning_effort ("minimal"/"low"/"medium"/"high"), lookup_file_uploads (inline base64).

Schema.<type>(...) — source="smart_lookup" (lookup overwrites) or source="lookup_if_missing" (extract too; lookup only fills empty slots), persistent_id= (pin field identity across updates).

Releasing

Cut on demand — a maintainer bumps the version in a PR, then pushes a sdk-python-vX.Y.Z tag on the release commit. Full flow + one-time setup: ../RELEASING.md. Break-glass: ./publish.sh [--build-only|--dry-run].