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