Skip to main content
The official Python SDK for anyformat, built on httpx. Auto-generated from our OpenAPI spec using Stainless.

Installation

Requires Python 3.9+.
pip install anyformat

Authentication

The SDK reads ANYFORMAT_API_KEY from your environment automatically:
export ANYFORMAT_API_KEY="your_api_key_here"
from anyformat import Anyformat

# Auto-reads ANYFORMAT_API_KEY from environment
client = Anyformat()

# Or pass explicitly
client = Anyformat(api_key="your_api_key_here")
See Authentication for how to get your API key.

Basic usage

Complete flow: create a workflow, run it on a document, and get results.
from anyformat import Anyformat

client = Anyformat()

# 1. Create a workflow
# The SDK uses extra_body for request body fields because the OpenAPI spec
# does not yet expose them as named parameters.
workflow = client.workflows.create(
    extra_body={
        "name": "Invoice Processing",
        "description": "Extract key data from invoices",
        "fields": [
            {
                "name": "invoice_number",
                "description": "The unique invoice identifier",
                "data_type": "string"
            },
            {
                "name": "total_amount",
                "description": "Total invoice amount",
                "data_type": "float"
            }
        ]
    }
)
print(f"Created workflow: {workflow.id}")

# 2. Run the workflow on a document
with open("invoice.pdf", "rb") as f:
    result = client.workflows.run(workflow.id, file=f)
print(f"File submitted: {result.id}")

# 3. Poll for extraction results (returns 412 while processing)
import time
import anyformat

for attempt in range(60):
    try:
        extraction = client.files.get_extraction_results(result.id)
        print(extraction)
        break
    except anyformat.APIStatusError as e:
        if e.status_code == 412:
            time.sleep(min(5 * (1.5 ** min(attempt, 5)), 30))
        else:
            raise

Async usage

The SDK provides an async client for use with asyncio:
import asyncio
from anyformat import AsyncAnyformat

async def main():
    client = AsyncAnyformat()

    workflows = await client.workflows.list()
    print(workflows.results)

    await client.close()

asyncio.run(main())
Or use it as a context manager:
async with AsyncAnyformat() as client:
    workflows = await client.workflows.list()

Error handling

The SDK raises typed exceptions you can catch:
import anyformat
from anyformat import Anyformat

client = Anyformat()

try:
    workflow = client.workflows.retrieve("non-existent-id")
except anyformat.AuthenticationError:
    # 401 - Invalid or missing API key
    print("Check your API key")
except anyformat.NotFoundError:
    # 404 - Resource not found
    print("Workflow not found")
except anyformat.RateLimitError:
    # 429 - Too many requests
    print("Rate limited, back off")
except anyformat.BadRequestError:
    # 400 - Invalid request
    print("Bad request")
except anyformat.APIStatusError as e:
    # Other 4xx/5xx errors
    print(f"API error {e.status_code}: {e.message}")
except anyformat.APIConnectionError:
    # Network errors
    print("Could not reach the server")
These map to the error codes described in Error Handling.

Retries and timeouts

The SDK retries failed requests automatically:
  • Default retries: 2 (with exponential backoff)
  • Default timeout: 60 seconds
Configure them per-client or per-request:
# Per-client
client = Anyformat(
    max_retries=5,
    timeout=120.0,
)

# Per-request
client.workflows.list(timeout=30.0)

Available resources

SDK methodAPI endpoint
client.health.check()GET /health/
client.workflows.create()POST /v2/workflows/
client.workflows.retrieve(id)GET /v2/workflows/{id}/
client.workflows.list()GET /v2/workflows/
client.workflows.delete(id)DELETE /v2/workflows/{id}/
client.workflows.run(id)POST /v2/workflows/{id}/run/
client.workflows.upload(id)POST /v2/workflows/{id}/upload/
client.workflows.results(id)GET /v2/workflows/{id}/results/
client.workflows.list_runs(id)GET /v2/workflows/{id}/runs/
client.files.create()POST /v2/files/
client.files.list()GET /v2/files/
client.files.delete(id)DELETE /v2/files/{id}/
client.files.get_extraction_results(id)GET /v2/files/{id}/extraction/
client.webhooks.create()POST /v2/webhooks/
client.webhooks.list()GET /v2/webhooks/
client.webhooks.delete(id)DELETE /v2/webhooks/{id}/

Raw responses

Access the raw httpx.Response object when you need headers or status codes:
response = client.workflows.with_raw_response.list()

print(response.status_code)
print(response.headers)
workflows = response.parse()  # parse the response body