Metadata-Version: 2.4
Name: anyformat
Version: 1.0.0rc1
Summary: Python SDK for the AnyFormat API
Author-email: Anyformat <info@anyformat.ai>
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27.0
Requires-Dist: pillow>=12.2.0
Requires-Dist: pydantic>=2.0
Requires-Dist: pypdfium2>=4.30.0
Requires-Dist: rich>=13.0
Requires-Dist: typer>=0.15
Description-Content-Type: text/markdown

# anyformat

Python SDK for the [AnyFormat](https://anyformat.ai) 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.

```bash
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.

```python
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:

```python
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.

```python
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:

```python
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:

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

### 2. Poll a run yourself / list runs

```python
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:

```python
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).

```python
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:

```python
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.

```python
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

```python
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

```python
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

```python
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=`.

```python
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`.

```python
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:

```python
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:

```python
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:

```bash
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 `@overload`s — 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`](../RELEASING.md). Break-glass: `./publish.sh [--build-only|--dry-run]`.
