Python SDK Guide

pip install anyformat

import os
from anyformat.sdk import Client

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

# Or from the environment
client = Client(api_key=os.environ["ANYFORMAT_API_KEY"])

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

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

# 1. Build & create the workflow (parse → extract)
workflow = (
    client.workflow("Invoice Processor")
    .parse()
    .extract([
        Schema.string("invoice_number", "The unique invoice identifier"),
        Schema.float("total_amount",    "Total invoice amount"),
        Schema.date("issue_date",       "Date when the invoice was issued"),
    ])
    .create()
)
print(f"Workflow id: {workflow.id}")

# 2. Submit a document — accepts Path, str (path), or bytes.
run = workflow.run("invoice.pdf")

# 3. Wait for results. Polls /results/ (412 → still processing, 200 → done).
result = run.wait()  # default: timeout=300s, poll_interval=3s

print(result.fields["invoice_number"].value)
print(result.fields["total_amount"].value)

import asyncio
import os

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

async def main():
    client = AsyncClient(api_key=os.environ["ANYFORMAT_API_KEY"])
    try:
        workflow = await (
            client.workflow("Invoice Processor")
            .parse()
            .extract([Schema.string("invoice_number", "...")])
            .create()
        )
        run = await workflow.run("invoice.pdf")
        result = await run.wait()
        print(result.fields["invoice_number"].value)
    finally:
        await client.aclose()

asyncio.run(main())

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)   # resolved from the catalog

# List your workflows, then run the first one
workflows = client.list_workflows(page_size=50)
for wf in workflows:
    print(wf.id, wf.name)

result = workflows[0].run("invoice.pdf").wait()

# Delete by id
client.delete_workflow(workflows[0].id)

result = run.wait()

# Scalar fields, for linear workflows (single untagged extraction)
inv = result.fields["invoice_number"]          # ExtractedField
print(inv.value, inv.confidence, inv.evidence)

# Parse output (markdown + per-block confidence)
parse_markdown = result.parse.markdown if result.parse else None

# Anything else — nested objects, split workflows, classifications:
# read result.raw, the validated wire envelope as a dict.
for extraction in result.raw["extractions"]:
    for name, field in extraction["fields"].items():
        print(name, field)

import time

from anyformat.sdk import (
    APIError,
    BadRequest,        # 400 — validation error, bad request body
    Unauthorized,      # 401 — invalid or missing API key
    Forbidden,         # 403 — disallowed by the server
    NotFound,          # 404 — workflow / file / webhook does not exist
    RateLimited,       # 429 — slow down; .retry_after has the suggested delay
    ServerError,       # 5xx — internal anyformat error
    SDKTimeout,        # local timeout — `.wait()` exceeded `timeout`
)

try:
    result = workflow.run("invoice.pdf").wait(timeout=60)
except RateLimited as e:
    time.sleep(e.retry_after or 5)
except NotFound:
    print("workflow or file is gone")
except APIError as e:
    print(f"HTTP {e.status_code}: {e.detail}")
except SDKTimeout:
    print("polling timed out — try webhooks for production")

import os

import httpx

webhook = httpx.post(
    "https://api.anyformat.ai/v2/webhooks/",
    headers={"Authorization": f"Bearer {os.environ['ANYFORMAT_API_KEY']}"},
    json={"url": "https://your-server.com/hook", "events": ["extraction.completed"]},
).json()
# webhook["secret"] is only returned once — store it.