anyformat 1.0.0rc1

Python SDK for the AnyFormat API

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.file_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 10 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

---

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.

Status	Exception	error_code	Retryable?
400	BadRequest	VALIDATION_ERROR / TOPOLOGY_INVALID	no
401	Unauthorized	AUTH_FAILED	no
402	PaymentRequired	PAYMENT_REQUIRED (out of extraction credit)	no
403	Forbidden	ACCESS_DENIED	no
404	NotFound	NOT_FOUND	no
422	APIError	VALIDATION_ERROR	no
429	RateLimited (.retry_after)	RATE_LIMITED	yes
5xx	ServerError	INTERNAL_ERROR / GATEWAY_TIMEOUT	yes

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); StillParsing (parse op not done). 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

Call	Returns	Notes
Client(api_key, base_url=...)	Client	ANYFORMAT_API_KEY if api_key omitted
client.workflow(name)	builder	chain verbs, then .create()
.parse(mode="standard"|"agentic"|"lite")	builder	exactly one per workflow
.classify(*categories)	builder	branching; needs parse() first
.split(*rules, route_from=...)	builder	route_from required after classify()
.extract(fields, branch=..., lookup_files=..., lookup_suggestion=...)	builder	branch required after classify/split
.validate(*rules, branch=...)	builder	follows extract()
.create() / .update(workflow_id)	Workflow	create POSTs, update PATCHes /v3/workflows/{id}/
client.get_workflow(id)	WorkflowDefinition	typed graph; fields carry persistent_id
client.update_workflow(id, definition)	Workflow	PATCH — echo persistent_id to keep field identity
client.list_workflows(limit, cursor)	Page[Workflow]	keyset; follow next_cursor
client.iter_workflows(limit)	Iterator[Workflow]	auto-pages every workflow, lazily
client.delete_workflow(id)	str	deletes packets + runs too
workflow.run(file, files=, text=, idempotency_key=)	Run	upload + run in one call
workflow.upload(...)	Upload	stage without running
upload.run(idempotency_key=)	Run	trigger extraction
run.wait(timeout=300, poll_interval=3)	Result	polls run status; retries transient 429/5xx; ExtractionFailed/ExtractionCancelled on terminal failure
client.get_run(run_id)	RunDetail	.status, .result inline when processed
client.list_runs(workflow_id, limit, cursor)	Page[RunSummary]	newest first
client.iter_runs(workflow_id, limit)	Iterator[RunSummary]	auto-pages every run, lazily
client.get_document_packet(id)	DocumentPacketDetail	.files, .latest_run_id
client.delete_document_packet(id)	str	irreversible
client.run_document_packet(id, idempotency_key=)	Run	re-run on latest workflow version
client.list_document_packets(workflow_id, limit, cursor)	Page[DocumentPacketSummary]	slim rows
client.iter_document_packets(workflow_id, limit)	Iterator[DocumentPacketSummary]	auto-pages every packet, lazily
client.parse(file) / client.get_markdown(job_id)	Run / str	atomic parse op (v2 surface)
result.fields	dict[str, ExtractedField]	.value, .confidence, .evidence
result.parse	ParseView | None	.blocks, .markdown, .text, .draw()
result.raw	dict	full results payload

Node configuration knobs

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

.parse(...) — mode ("standard"/"agentic"/"lite"), prompt_hint, figure_enhancement, cache; agentic-only effort ("low"/"mid"/"accurate"); lite-only ocr_effort ("medium"/"high"), skip_routing_review, figures, text_formatting, routing_effort. 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].