> ## Documentation Index
> Fetch the complete documentation index at: https://docs.anyformat.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Python SDK
> The anyformat Python package: a fluent builder over the typed workflow graph, run and result handles, and the afx command line.
The `anyformat` package is the Python client for [API v3](/api-reference-v3/introduction). 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](/api-reference-v3/sdks/typescript) 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.
```bash theme={null}
pip install --pre anyformat
```
Pass the API key to the client. The client does not read the environment for you.
```python theme={null}
import os
from anyformat.sdk import Client
client = Client(api_key=os.environ["ANYFORMAT_API_KEY"])
```
See [Authentication](/api-reference-v3/introduction#authentication) for how to mint a key. The package also installs the [`afx` command](/api-reference-v3/cli), 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.
```python Sync theme={null}
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)
```
```python Async theme={null}
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](/api-reference-v3/node-schemas) 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](/guides/nodes/if-else), [Slack alert](/guides/nodes/slack-alert) or [Knowledge](/guides/nodes/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](/guides/nodes/parse) node. Call it first and once. `mode` selects the tier, and the type checker only offers the knobs that belong to that tier.
```python theme={null}
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](/guides/nodes/classify) node after Parse. Takes one or more `ClassifyCategory` objects.
```python theme={null}
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](/guides/nodes/split) node. Takes one or more `SplitterRule` objects. After `classify()`, pass `route_from=` to name the category that feeds the splitter.
```python theme={null}
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](/guides/nodes/extract) node. Repeatable: one per branch. `fields` is a list built with `Schema`.
```python theme={null}
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](/guides/nodes/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](/concepts/field-types): `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](/guides/nodes/validate) node after the most recent `extract()`, or after the extract on `branch=`. Takes one or more `ValidationRule` objects.
```python theme={null}
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](/guides/nodes/edit) node after Parse. Call it once.
```python theme={null}
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](/concepts/document-packets) 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.
```python theme={null}
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.
```python theme={null}
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.
```python theme={null}
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`. |
```python theme={null}
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.
```python theme={null}
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`.
```python theme={null}
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. |
```python theme={null}
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](/guides/nodes/knowledge) node answers questions about what its documents say. `workflow.ask(question)` and `client.ask(workflow_id, question)` call [Ask](/api-reference-v3/knowledge/ask) and return an `Answer`.
```python theme={null}
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](/concepts/evals) holds documents with known answers. Upload one document, with optional ground truth, as one packet:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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](/api-reference-v3/evals/launch).
## 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.
```python theme={null}
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](/api-reference-v3/introduction#errors): `.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.
```python theme={null}
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](/api-reference-v3/migrating-from-v2). |
Upgrading from 0.x changes uploads, runs, pagination and workflow updates. Follow the [0.x to 1.0 migration guide](/api-reference-v3/sdk-migration).
## Links
* [PyPI](https://pypi.org/project/anyformat/)
* [TypeScript SDK](/api-reference-v3/sdks/typescript): the same shape in TS
* [afx command line](/api-reference-v3/cli): the CLI this package installs
* [Node schemas](/api-reference-v3/node-schemas): every argument the builder maps to