Client reference

Python SDK

from aiotic import AioticClient, AsyncAioticClient

client = AioticClient(base_url="https://acme.aiotic.ai", api_key="…", sync_api_key="…")   # or AioticClient() from AIOTIC_* env

Both clients expose the same resources; the async one is await-able and iterators are async for.

Behaviour

Feature	Detail
Retries	on connection errors and 408 425 429 502 503 504; backoff 0.5 s·2ⁿ + jitter, capped 30 s; Retry-After honoured; uploads are not retried after the first attempt
Rate limit	token bucket, rate_limit req/s (default 10; 0 = off); safe across threads
Keys	master-data paths use sync_api_key when set, everything else the integration key
Errors	typed exceptions with .status, .detail (string or object), .request_id; see Errors
Unknown fields	preserved on every model

client.orders

Method	Endpoint
upload(files, *, request_id=None, metadata=None) → OrderUploadResponse	POST /order/upload — files: paths, (name, bytes) or (name, bytes, content_type)
upload_raw_email(eml, *, request_id=None) → OrderUploadResponse	POST /order/raw/upload — raises AioticValidationError with structured detail for non-orders
classify_raw_email(eml) → EmailClassification	POST /order/raw/classify
get(request_id) → OrderStatus	GET /order_status/{id}
list(*, page=1, size=100) → OrderListResponse · iter_all(size=200, max_pages=None)	GET /order_status/list
group(email_group_id) → OrderGroup	GET /order/group/{id}
download_file(request_id, filename, *, preview=False) → bytes	GET /order/{id}/{filename}[/preview]
retry(request_id) → OrderUploadResponse	POST /order/retry/{id}
wait(request_id, *, until=LANDED_STATUSES, timeout=600, initial_interval=2, max_interval=15) → OrderStatus	polling helper

OrderUploadResponse.request_ids returns the ids to poll (children when split).

client.erp

Method	Endpoint
send(request_id) → ErpSendResponse	POST /erp/send/{id} — raises AioticErpRejectedError(.erp_error) on success: false, AioticConflictError when not sendable

ErpSendResponse.erp_order_number — your ERP's reference from the response body.

client.rejected

list(*, page, size, status="pending"), get(request_id), reprocess(request_id).

client.customers (sync key ok)

list(*, page, size), iter_all(size=500), search(query, *, top_k=10), get(number), upsert(number, CustomerUpsert | dict), delete(number).

client.products (sync key ok)

list(*, page, size, language_code=None), iter_all(...), get(item_number, language_code), upsert(item_number, language_code, ProductUpsert | dict), delete(item_number, language_code).

client.customer_products (sync key ok)

list(*, page, size, customer_number=None, customer_item_number=None, item_number=None, language_code=None), iter_all(...), get(customer_number, customer_item_number), upsert(customer_number, customer_item_number, CustomerProductUpsert | dict), delete(...).

Misc

client.health() → HealthCheck, client.system_status() → SystemStatus, client.mailbox.fetch_all().

Exceptions

AioticError
├── AioticAuthError            401
├── AioticNotFoundError        404
├── AioticConflictError        409
├── AioticValidationError      400 / 415 / 422   (.detail may be an object)
├── AioticErpRejectedError     422 on /erp/send  (.erp_error)
├── AioticUnavailableError     503
├── AioticServerError          5xx
└── AioticTransportError       network, after retries

Async example

import asyncio
from aiotic import AsyncAioticClient

async def main():
    async with AsyncAioticClient() as client:
        up = await client.orders.upload(["PO.pdf"])
        status = await client.orders.wait(up.request_id)
        async for c in client.customers.iter_all():
            ...

asyncio.run(main())

Testing your code

AioticClient(transport=httpx.MockTransport(handler)) injects a fake transport; or run the mock tenant in-process as the SDK's own tests do.