Python SDK

The anyformat package is the Python client for API v3. It is hand-written on httpx. You build a workflow with a fluent builder, run a file through it, and read typed results. The TypeScript SDK mirrors it method for method. Two import paths: anyformat.sdk holds the client, the handles and the errors; anyformat.workflow holds the schema factories and the node types.

Install and auth

Requires Python 3.10 or later. The 1.0 line is a release candidate, so pip needs --pre to pick it up; plain pip install anyformat still resolves to the 0.x (v2) line until 1.0.0 final ships.

pip install --pre anyformat

Pass the API key to the client. The client does not read the environment for you.

import os

from anyformat.sdk import Client

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

See Authentication for how to mint a key. The package also installs the afx command, which does read ANYFORMAT_API_KEY.

Client

Client is synchronous. AsyncClient has the same methods with await. Both take the same arguments.

Argument	Default	What it does
`api_key`	required	Your `af_...` key.
`base_url`	`https://api.anyformat.ai`	Override for a non-production deployment.
`timeout`	`60.0`	Per-request timeout in seconds.

Both clients are context managers that close the connection pool. Prefer that over a bare module-level client.

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

with Client(api_key="af_...") as client:
    workflow = (
        client.workflow("Invoices")
        .parse()
        .extract([Schema.string("vendor", "Vendor name on the invoice.")])
        .create()
    )
    result = workflow.run("invoice.pdf").wait()
    print(result.fields["vendor"].value)

import asyncio

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

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

asyncio.run(main())

In the async client, create(), update(), run(), upload(), wait() and every client method are awaitable. The builder verbs (parse, extract, and so on) are not.

Build a workflow

client.workflow(name, description=None) returns a builder. Chain one verb per node, then .create() to register the workflow and get a Workflow handle, or .update(workflow_id) to replace an existing workflow’s graph. .build() returns the typed WorkflowDefinition without a network call. The builder assigns node ids (parse_1, extract_1, …) and wires the edges for you. Pass id= to any verb to choose the id yourself. Each verb constructs the same typed node the API accepts, so the node schema is the reference for every argument. A builder mistake, such as extract() before parse(), raises WorkflowBuilderError before any request is sent. The builder has verbs for Parse, Classify, Split, Extract, Validate and Edit. For a graph with an If/Else, Slack alert or Knowledge node, build a WorkflowDefinition from the node classes in anyformat.workflow.nodes and pass it to client.create_workflow(definition).

parse

Adds the one Parse node. Call it first and once. mode selects the tier, and the type checker only offers the knobs that belong to that tier.

client.workflow("Read only").parse()                                  # mode="standard"
client.workflow("Dense tables").parse(mode="agentic", effort="accurate")
client.workflow("Born-digital").parse(mode="flash", scanned_pages="skip")

Argument	Applies to	Default
`mode`	all	`"standard"`. One of `"standard"`, `"agentic"`, `"lite"`, `"flash"`.
`prompt_hint`	standard, agentic, flash	`None`
`figure_enhancement`	standard, agentic, flash (only standard uses it)	`False`
`cache`	all	`True`
`effort`	agentic	`"mid"`. One of `"low"`, `"mid"`, `"accurate"`.
`scanned_pages`	flash	`"ocr"`. One of `"ocr"`, `"skip"`, `"fail"`.

lite takes no knob of its own. All arguments are keyword-only.

classify

Adds a Classify node after Parse. Takes one or more ClassifyCategory objects.

from anyformat.workflow import ClassifyCategory

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

builder = client.workflow("Invoice or receipt").parse().classify(invoice, receipt)

Every extract() that follows needs branch=.

split

Adds a Split node. Takes one or more SplitterRule objects. After classify(), pass route_from= to name the category that feeds the splitter.

from anyformat.workflow import SplitterRule

statement = SplitterRule(id="STATEMENT", name="Statement", description="A bank statement.", partition_key="account_number")
check = SplitterRule(id="CHECK", name="Check", description="A scanned check.")

builder = client.workflow("Statement batch").parse().split(statement, check)

extract

Adds an Extract node. Repeatable: one per branch. fields is a list built with Schema.

from anyformat.workflow import Schema

builder = (
    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
)

Argument	Default	What it does
`fields`	required	List of `Schema.*` fields. At least one.
`branch`	`None`	The category or rule this extract hangs off. Required after `classify()` or `split()`, forbidden otherwise. A `ClassifyCategory`, a `SplitterRule`, or its id.
`lookup_files`	`None`	Paths of Smart Lookup reference files. The SDK reads each one and sends it inline as `lookup_file_uploads`.
`lookup_file_uploads`	`None`	Already-encoded `LookupFileUpload(filename, content)` entries.
`lookup_suggestion`	`None`	Free-form hint for the lookup matcher.
`id`	auto	Node id.

Schema has one factory per field type: string, integer, float, boolean, date, datetime, enum(name, description, options=[Schema.option(...)]), multi_select, and object(name, description, fields=[...]). Every factory accepts source="smart_lookup" or source="lookup_if_missing", and persistent_id= to pin a field’s identity across updates.

client.workflow(...).extract() does not take mode, use_images or lookup_reasoning_effort yet. Set them through the API JSON, in Studio, or with client.get_workflow() + client.update_workflow() on the typed definition.

validate

Adds a Validate node after the most recent extract(), or after the extract on branch=. Takes one or more ValidationRule objects.

from anyformat.workflow import ValidationRule, ArithmeticCheck

builder = (
    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="vendor-legit", description="The vendor is a real, named company."),
        ValidationRule(
            id="totals-add-up",
            kind="deterministic",
            check=ArithmeticCheck(type="arithmetic", operands=["subtotal", "tax"], equals="total", tolerance=0.01),
        ),
    )
)

edit

Adds an Edit node after Parse. Call it once.

references = client.upload_edit_references(workflow_id, ["company-profile.csv"])

builder = (
    client.workflow("Fill onboarding form")
    .parse()
    .edit(
        instructions="Name: ACME SL; Date: 2026-01-01; I accept the terms: yes",
        reference_document_ids=[r.id for r in references],
    )
)

upload_edit_references(workflow_id, files) uploads the standing data an Edit node fills from (.csv, .txt, .md, .rst, .pdf). A PDF comes back pending while it is parsed once; poll list_edit_references(workflow_id) until ready. delete_edit_reference(workflow_id, document_id) removes one. The builder does not set font or output_mode; set those in the API JSON.

Run and read

Upload and run

workflow.run(...) uploads the document as one document packet and starts a run. It accepts a path string, a Path, raw bytes, a list of up to 10 files via files=, or text=. Exactly one of the three.

run = workflow.run("invoice.pdf")
run = workflow.run(files=["invoice.pdf", "annex.pdf"])   # one packet, one document
run = workflow.run(text="Plain text body")               # sent as a .txt file
run = workflow.run("invoice.pdf", idempotency_key="ingest-2026-08-25-0001")

Filenames are unique within a workflow. A colliding name is rejected with 409; pass on_conflict="rename" to auto-rename instead. To upload without spending credits, use workflow.upload(...). It returns an Upload with document_packet_id and files; call upload.run() later.

upload = workflow.upload("invoice.pdf")
run = upload.run(idempotency_key="run-2026-08-25-0001")

Retrying with the same idempotency_key replays the original upload or run instead of creating and billing a second one.

wait

run.wait(timeout=300, poll_interval=3) polls GET /v3/runs/{run_id}/ until the status is terminal and returns a Result. A transient 429 or 5xx while polling is retried with backoff and honours Retry-After. wait() raises ExtractionFailed on error, ExtractionCancelled on cancelled, and SDKTimeout when the budget runs out.

result = run.wait(timeout=600)

Result

Attribute	Type	What it holds
`result.fields`	`dict[str, ExtractedField \| list[dict]]`	The untagged extraction’s fields. A scalar field is an `ExtractedField` with `.value`, `.confidence` (0 to 100) and `.evidence`; an object field is a list of row dicts.
`result.field(name)`	`ExtractedField \| None`	One scalar field, or `None` when missing or list-typed. Pass a type as the second argument to type `.value` for the checker: `result.field("total", float)`.
`result.partitions`	`list[Partition]`	One entry per split partition, each with `split_name`, `partition` and its own `fields`. Empty without a Split node.
`result.parse`	`ParseView \| None`	The Parse node’s output: `.markdown`, `.text`, `.blocks`, `.parse_confidence`, `.layout_confidence`, and `.draw()`.
`result.edits`	`list[Edit]`	One filled form per file from an Edit node, with `fields`, `unmatched_instructions` and a `download_url` valid for 15 minutes.
`result.raw`	`dict`	The full results envelope, including `classifications` and `splits`.

result = run.wait()

vendor = result.fields["vendor"]
print(vendor.value, vendor.confidence, [e.text for e in vendor.evidence])

for row in result.fields["line_items"]:
    print(row["sku"].value, row["amount"].value)

for partition in result.partitions:
    print(partition.split_name, partition.partition, partition.fields.keys())

for category in result.raw["classifications"]:
    print(category["category"], category["confidence"])

result.parse.draw("out/") renders every block’s bounding box onto its page, coloured by confidence, and writes one PNG per page. It uses the file you ran as the background; pass source= to use another.

Quick parse

For a one-off parse with no workflow, client.parse(file) runs the atomic parse operation and returns a handle. Poll client.get_markdown(run.file_id) until it stops raising StillParsing. This is the one call still served by the v2 operations surface.

import time

from anyformat.sdk import StillParsing

job = client.parse("contract.pdf")
while True:
    try:
        markdown = client.get_markdown(job.file_id)
        break
    except StillParsing:
        time.sleep(3)

Document packets and runs

Packets and runs stay addressable after wait() returns.

Method	Returns	Notes
`client.get_document_packet(packet_id)`	`DocumentPacketDetail`	`.status`, `.files`, `.latest_run_id`.
`client.run_document_packet(packet_id, idempotency_key=None)`	`Run`	Re-run on the workflow’s latest version. Each call is a new run.
`client.delete_document_packet(packet_id)`	`str`	Irreversible.
`client.list_document_packets(workflow_id, limit=20, cursor=None)`	`Page[DocumentPacketSummary]`	Newest first.
`client.get_run(run_id)`	`RunDetail`	`.status`, and `.result` inline once `processed`.
`client.list_runs(workflow_id, limit=20, cursor=None)`	`Page[RunSummary]`	Newest first.
`client.iter_runs(workflow_id, limit=20)`	`Iterator[RunSummary]`	Follows the cursor for you.

Run statuses: queued, in_progress, then processed, error or cancelled.

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

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

Manage workflows

Method	Returns	Notes
`client.get_workflow(workflow_id)`	`WorkflowDefinition`	The typed graph. Fields carry their `persistent_id`.
`client.update_workflow(workflow_id, definition)`	`Workflow`	Replaces the graph. Echo each `persistent_id` to keep field identity across renames.
`client.create_workflow(definition)`	`Workflow`	Create from a `WorkflowDefinition` you built by hand.
`client.list_workflows(limit=20, cursor=None)`	`Page[Workflow]`	Each item is a `Workflow` you can `.run()`.
`client.iter_workflows(limit=20)`	`Iterator[Workflow]`	Follows the cursor.
`client.delete_workflow(workflow_id)`	`str`	Deletes its packets and runs too.

definition = client.get_workflow(workflow.id)
for node in definition.nodes:
    if node.type == "extract":
        for field in node.extraction_schema.fields:
            if field.name == "vendor":
                field.name = "vendor_name"   # persistent_id travels with it
client.update_workflow(workflow.id, definition)

Knowledge ask

A workflow with a Knowledge node answers questions about what its documents say. workflow.ask(question) and client.ask(workflow_id, question) call Ask and return an Answer.

answer = workflow.ask("Which contracts renew before March and what notice do they need?")
print(answer.answer)
for citation in answer.citations:
    print(citation.path, citation.page, citation.quote)

Answer has .answer, .citations (each with path, quote, file_id, block_id, page, bbox), .steps_used and .thread_id. Pass thread_id="kb-..." to ask a follow-up in the context of earlier questions on that thread. A workflow without a Knowledge node raises APIError with error_code KNOWLEDGE_NOT_ENABLED.

Evals and datasets

A dataset holds documents with known answers. Upload one document, with optional ground truth, as one packet:

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)

ground_truth keys are the schema’s field identifiers. A scalar maps to str | None; an object field maps to a list of row dicts. The call is atomic: a rejected file or a ground-truth failure stores nothing. Bulk upload loops one atomic call per document:

from anyformat.sdk import DatasetDocument

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

Then launch an eval over the whole dataset and poll it:

launched = client.launch_eval(workflow.id, idempotency_key="eval-2026-08-25")
print(launched.eval_id, launched.enqueued_count, launched.failed_count)

detail = client.get_eval(workflow.id, launched.eval_id)
while detail.status == "in_progress":
    time.sleep(10)
    detail = client.get_eval(workflow.id, launched.eval_id)
print(detail.accuracy, detail.matched, detail.mismatched, detail.ungraded)

launch_eval(workflow_id, version_id=None, idempotency_key=None) evaluates the current version unless you pass one. list_evals(workflow_id, limit, cursor) and iter_evals(workflow_id) list past evals. See Launch eval.

Pagination

Every list returns a Page(items, next_cursor). Pass next_cursor back as cursor= until it is None. There are no totals or page numbers; limit is capped at 100.

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

The iter_* methods (iter_workflows, iter_document_packets, iter_runs, iter_evals) do this loop for you and fetch each page lazily.

Errors and retries

Every HTTP failure raises a subclass of APIError carrying the v3 error envelope: .status_code, .error_code, .detail, .retryable and .request_id.

Status	Exception	Retry?
400	`BadRequest`	no. `VALIDATION_ERROR` or `TOPOLOGY_INVALID`.
401	`Unauthorized`	no
402	`PaymentRequired`	no. Top up credit first.
403	`Forbidden`	no
404	`NotFound`	no
422	`APIError`	no
429	`RateLimited`	yes, after `.retry_after` seconds
5xx	`ServerError`	yes

Outside HTTP: RunFailed when wait() sees a terminal failure, subclassed as ExtractionFailed (error) and ExtractionCancelled (cancelled), each with .run_id and .status; SDKTimeout when wait() runs out of budget; MissingResults when a processed run carries no results; StillParsing while a quick parse is not done; WorkflowBuilderError for builder misuse.

import time

from anyformat.sdk import APIError, RateLimited, RunFailed, SDKTimeout

try:
    result = workflow.run("invoice.pdf").wait(timeout=120)
except RateLimited as e:
    time.sleep(e.retry_after or 5)
except RunFailed as e:
    print(f"run {e.run_id} ended {e.status}")
except SDKTimeout:
    print("still running; poll client.get_run() later")
except APIError as e:
    print(e.status_code, e.error_code, e.detail, e.request_id)

wait() retries a transient 429 or 5xx itself, so those never reach your except while polling.

Version pins

Package	Version	API
`anyformat` (PyPI)	`1.0.0rc1`	v3. Release candidate; needs `pip install --pre`.
`anyformat` 0.7.x	stable	v2. Keeps working through the v2 deprecation window.

Upgrading from 0.x changes uploads, runs, pagination and workflow updates. Follow the 0.x to 1.0 migration guide.

Overview

Workflows

Upload

Document packets

Runs

Parse

Knowledge

Evals

SDKs and tools

API v2 (deprecated)

Install and auth

Client

Build a workflow

parse

classify

split

extract

validate

edit

Run and read

Upload and run

wait

Result

Quick parse

Document packets and runs

Manage workflows

Knowledge ask

Evals and datasets

Errors and retries

Version pins

Links

​Install and auth

​Client

​Build a workflow

​parse

​classify

​split

​extract

​validate

​edit

​Run and read

​Upload and run

​wait

​Result

​Quick parse

​Document packets and runs

​Manage workflows

​Knowledge ask

​Evals and datasets

​Pagination

​Errors and retries

​Version pins

​Links

Install and auth

Client

Build a workflow

parse

classify

split

extract

validate

edit

Run and read

Upload and run

wait

Result

Quick parse

Document packets and runs

Manage workflows

Knowledge ask

Evals and datasets

Pagination

Errors and retries

Version pins

Links