Python SDK

nace-sdk is the official Python client for https://console.nace.ai. It covers Drex's decision model, system_one and models.list, and Perception's five document tools, uploads, jobs, job files and saved schemas. It needs Python 3.11 or later and depends only on httpx, pydantic and typing-extensions.

The client is built on the TypeSafe Python SDK, so system_one, Noul, Choice, Score, SystemOneResponse with .nouls, .choices and .scores, RetryPolicy and the error classes match typesafe-sdk. Code written for TypeSafeClient runs on NaceClient once you rename the imports. The differences are Drex's: the 402, 409 and 529 error classes, a 60-second default timeout, instructions and criteria as text only, and legend as a tuple. See Migrate from TypeSafe.

pip install nace-sdk
export NACE_API_KEY="nace_sk_..."

Create a key as shown in Authentication.

Configure the clientCopy Anchor Link

NaceClient(**options)

An argument you pass wins over its environment variable, and an empty or whitespace-only variable counts as unset:

Each variable's name from before the packages were renamed from drex-* still works when the new one is unset or blank: DREX_API_KEY, DREX_BASE_URL, DREX_DEFAULT_MODEL and DREX_LOG_LEVEL. The new name wins when both are set.

NaceClient() never reads the CLI's ~/.nace/config.toml; use nace_sdk.credentials for that. The SDK sets Authorization, User-Agent, X-Nace-SDK, X-Nace-Runtime and X-Nace-Retry-Count itself, plus Idempotency-Key and Content-Type when a request carries them, so headers and extra_headers can't override those.

import httpx
from nace_sdk import NaceClient, RetryPolicy

client = NaceClient(
    base_url="https://console.nace.ai",
    timeout=httpx.Timeout(90, connect=5),
    retry=RetryPolicy(max_retries=4, timeout=500),  # room for 5 attempts of 90 s and the waits between them
    headers={"X-Team": "billing-bot"},
)
try:
    print(client.base_url)
finally:
    client.close()

Most methods of client.documents, client.jobs and client.extraction_schemas also take retry, timeout and extra_headers for one call, and the five tools also take extra_body. The methods that don't are documents.upload, documents.upload_part, documents.complete_upload_session, documents.get_upload_session, documents.abort_upload_session, jobs.iter, jobs.wait, jobs.events and jobs.download; they use the client's retry policy and timeout, except jobs.events, which is never retried. jobs.wait's own timeout is how long to keep polling.

AsyncCopy Anchor Link

AsyncNaceClient takes the same arguments, except that transport is an httpx.AsyncBaseTransport and http_client an httpx.AsyncClient. Its models, documents, jobs and extraction_schemas (AsyncModels, AsyncDocuments, AsyncJobs, AsyncExtractionSchemas) have the same methods and parameters as the sync ones. Await each call, iterate jobs.iter(), jobs.events(), extraction_schemas.iter() and extraction_schemas.iter_versions() with async for, and close the client with async with or await client.aclose().

import asyncio

from nace_sdk import AsyncNaceClient, Noul


async def main() -> None:
    async with AsyncNaceClient() as client:
        result = await client.system_one(
            state="I was charged twice for my March invoice.",
            questions={"wants_refund": Noul(instructions="Is the customer asking for a refund?")},
        )
        print(result.nouls["wants_refund"].noul)


asyncio.run(main())

The document methods are awaited the same way:

import asyncio

from nace_sdk import AsyncNaceClient


async def main() -> None:
    async with AsyncNaceClient() as client:
        job = await client.documents.parse("https://example.com/report.pdf")
        async for event in client.jobs.events(job.job_id):
            print(event.status, event.progress)
        job = await client.jobs.wait(job.job_id)
        print(job.result["document"]["markdown"])


asyncio.run(main())

Ask questionsCopy Anchor Link

client.system_one(state, questions, **options)

It returns a SystemOneResponse. A decision is billed per input token and only when it returns 200, so the SDK's retries never bill twice.

from nace_sdk import Choice, NaceClient, Noul, Score

with NaceClient() as client:
    result = client.system_one(
        state="I was charged twice for my March invoice.",
        questions={
            "wants_refund": Noul(instructions="Is the customer asking for a refund?"),
            "topic": Choice(instructions="Which topic is it?", criteria={"billing": None, "shipping": None, "other": None}),
            "urgency": Score(instructions="How urgent is it?", criteria=["low", "medium", "high"]),
        },
    )

print(result.nouls["wants_refund"].noul)
print(result.choices["topic"].choice, result.choices["topic"].confidence)
print(result.scores["urgency"].score, result.usage.input_tokens, result.request_id)

Question typesCopy Anchor Link

questions = {
    "wants_refund": {"type": "noul", "instructions": "Is the customer asking for a refund?"},
    "is_spam": Noul(criteria={"true": "Unsolicited advertising", "false": "A real customer message"}),
}

The SDK checks the questions before it sends anything and raises NaceError for no questions or more than 512, a Noul with neither instructions nor a described outcome, a Choice whose criteria isn't a nonempty mapping, a Score with no criteria, and a dictionary without a nonempty type (or, for choice and score, without criteria). An unknown field on a question object raises pydantic.ValidationError. Unset instructions and criteria are left out of the request.

Read the responseCopy Anchor Link

SystemOneResponse:

request_id and raw_http_response aren't fields, so model_dump() leaves them out. Every response type has them, and a from_http_response(response) class method that builds one from an httpx.Response you fetched yourself, raising the same errors as a call.

Answer is the union of the three, told apart by type. Score probabilities are keyed by int, so read them as result.scores["urgency"].probabilities[2]. An answer of a type this SDK version doesn't know is dropped with a warning on the nace_sdk logger; it is still in raw_http_response. See Reading answers.

Parse into your own modelCopy Anchor Link

Subclass SystemOneResponse and declare an answer per question name. The SDK lifts each declared answer out of answers into its field, and you keep request_id and raw_http_response:

from nace_sdk import Choice, ChoiceAnswer, NaceClient, Noul, NoulAnswer, SystemOneResponse


class Ticket(SystemOneResponse):
    wants_refund: NoulAnswer
    topic: ChoiceAnswer


with NaceClient() as client:
    ticket = client.system_one(
        state="I was charged twice for my March invoice.",
        questions={
            "wants_refund": Noul(instructions="Is the customer asking for a refund?"),
            "topic": Choice(criteria={"billing": None, "shipping": None}),
        },
        response_model=Ticket,
    )

print(ticket.wants_refund.noul, ticket.topic.choice, ticket.request_id)

Any other pydantic.BaseModel is validated against the raw JSON body and has no request_id or raw_http_response. A body that doesn't match raises NaceAPIResponseValidationError.

List modelsCopy Anchor Link

client.models.list(**options) takes retry, timeout and extra_headers, and returns a ListModelsResponse. It works at any balance and is free, so it doubles as a key check.

with NaceClient() as client:
    print([model.name for model in client.models.list().models])

See Models.

SourcesCopy Anchor Link

Every tool takes a source: the document to work on.

Each also has a type ("url", "workspace_file", "parse_result") that is set for you, and an unknown field raises pydantic.ValidationError. Source is the union of the three. You can also pass:

• A string, which must start with https:// and becomes a UrlSource. Anything else raises NaceError.
• A dictionary in the wire form, such as {"type": "workspace_file", "workspace_id": "...", "file_id": "..."}.

The document service picks the parser from a URL source's file_name and refuses an extension the tool doesn't read with unsupported_file_type (see Format support). When you leave file_name out, the SDK takes it from the URL: the last path segment, without its query or fragment, percent-decoded. That segment must have an extension, a . that is neither its first nor its last character; otherwise the SDK raises NaceError before sending anything. A file_name you pass is sent as it is.

client.documents.parse("https://example.com/files/report%20v2.pdf")  # file_name="report v2.pdf"
client.documents.parse(UrlSource(url="https://example.com/download?id=42", file_name="report.pdf"))
client.documents.parse(UrlSource(url="https://arxiv.org/pdf/1706.03762", file_name="attention.pdf"))

Name the file whenever the last segment isn't the file's real name: https://arxiv.org/pdf/1706.03762 would otherwise be sent as 1706.03762, which the document service refuses.

Create a jobCopy Anchor Link

The five tools create a job and return it as a DocumentJob: status="queued", or the finished job when wait_seconds was long enough. Besides source and the tool's own options, each takes:

Every other keyword argument, name and project_id included, becomes a body field of that tool as you give it, except that None values are left out. The document service refuses a field it doesn't know, a misspelled one included, with 422. A wait_seconds outside 0 to 60, or an empty or too-long idempotency_key, raises NaceError before anything is sent.

Parse, Split, Classify and Extract take page_ranges as a list of {"start": int, "end": int}: one-based and inclusive, at most 200 ranges and 500 pages in all.

wait_seconds is best effort: a job that takes longer comes back queued or running, with result still None. Check status before you read result:

job = client.documents.parse("https://example.com/report.pdf", wait_seconds=60)
if job.status != "succeeded":
    job = client.jobs.wait(job.job_id)  # raises NaceJobFailedError if it failed
print(job.result["document"]["markdown"])

A create that holds the request open is retried like any other call, but the default retry budget is 30 seconds in all, so an attempt that fails after waiting longer than that isn't retried. Raise RetryPolicy.timeout to retry those; see Retries.

Idempotency. Every create carries an Idempotency-Key, so the SDK's own retries return the same job and never bill twice. The minted key is new on each call, though. To retry a create yourself, after a 429, a 503 job_pending or a crash, pass your own idempotency_key and reuse it. The same key with a different body raises NaceConflictError (idempotency_conflict). See Retry safely.

Charge id. A create's response carries Drex's id for the charge in the x-drex-document-job header. Read it from the job's HTTP response, or from the error's headers when a create fails:

job = client.documents.parse("https://example.com/report.pdf")
print(job.request_id, job.raw_http_response.headers.get("x-drex-document-job"))

Quote both when you contact support about a charge.

ParseCopy Anchor Link

client.documents.parse(source, **options)

A parse_result source with any of the options marked "not with a parse_result source" set to something other than its default (for spreadsheet: sheets, an include_hidden_* of False, or include_formulas=True), or with a parse_mode other than low, is refused with 422 (NaceUnprocessableEntityError) before a job is created. OCR runs when the file needs it: the document service refuses any ocr setting other than the default with 422 invalid_request.

from nace_sdk import NaceClient

with NaceClient() as client:
    job = client.documents.parse(
        "https://example.com/report.pdf",
        page_ranges=[{"start": 1, "end": 3}],
        output={"formats": ["markdown", "blocks"]},
        chunking={"strategy": "page"},
        wait_seconds=60,
    )
    if job.status != "succeeded":
        job = client.jobs.wait(job.job_id)

print(job.result["document"]["markdown"])

For a very large file, document.markdown is only a preview and document.content_url links the full text: save it with jobs.download. See Parse for the result shape.

SplitCopy Anchor Link

client.documents.split(source, *, classes, **options)

from nace_sdk import NaceClient

with NaceClient() as client:
    job = client.documents.split(
        "https://example.com/packet.pdf",
        classes=[
            {"id": "invoice", "label": "Invoice", "description": "A supplier invoice"},
            {"id": "receipt", "label": "Receipt", "description": "A payment receipt"},
        ],
        wait_seconds=60,
    )
    if job.status != "succeeded":
        job = client.jobs.wait(job.job_id)

for segment in job.result["segments"]:
    print(segment["start_page"], segment["end_page"], segment["class"]["id"] if segment["class"] else None)

Class ids must be unique and can't contain /, and blank_page, unreadable_content and __ndi_unclassified are reserved. See Split.

ClassifyCopy Anchor Link

client.documents.classify(source, *, classes, **options)

from nace_sdk import NaceClient

with NaceClient() as client:
    job = client.documents.classify(
        "https://example.com/invoice.pdf",
        classes=[
            {"id": "invoice", "label": "Invoice", "description": "A supplier invoice"},
            {"id": "contract", "label": "Contract", "description": "A signed agreement between two parties"},
        ],
        granularity="page",
        wait_seconds=60,
    )
    if job.status != "succeeded":
        job = client.jobs.wait(job.job_id)

for unit in job.result["units"]:
    print(unit["page_range"], unit["labels"][0]["class_id"])

Class ids must be unique and can't contain /, and other, blank_page, unreadable_content and __ndi_unclassified are reserved. See Classify.

ExtractCopy Anchor Link

client.documents.extract(source, *, schema=None, schema_id=None, schema_version=None, instructions=None, **options)

Pass exactly one of schema and schema_id. Both, neither, or a schema_version without schema_id raises NaceError before anything is sent.

from nace_sdk import NaceClient, ParseResultSource

with NaceClient() as client:
    parsed = client.jobs.wait(client.documents.parse("https://example.com/invoice.pdf").job_id)
    job = client.documents.extract(
        ParseResultSource(job_id=parsed.job_id),
        schema={
            "type": "object",
            "properties": {
                "invoice_number": {"type": "string"},
                "total": {"type": "number"},
            },
            "required": ["invoice_number", "total"],
        },
        wait_seconds=60,
    )
    if job.status != "succeeded":
        job = client.jobs.wait(job.job_id)

print(job.result["data"])  # {'invoice_number': 'INV-1042', 'total': 1200.5}

See Extract for the result: data, shaped by your schema, and fields, one entry per schema field with its path, value, status (found, not_found or ambiguous), confidence and citations.

GroundCopy Anchor Link

client.documents.ground(source, *, targets, **options)

semantic belongs on each target; there is no top-level semantic. Ground takes neither page_ranges nor parse_mode.

from nace_sdk import NaceClient

with NaceClient() as client:
    job = client.documents.ground(
        "https://example.com/report.pdf",
        targets=[
            {"id": "total", "text": "1,200.50"},
            {"id": "signer", "text": "Jane Smith", "hint": "Signed on behalf of the supplier"},
            {"id": "late_fee", "text": "Late payment fee"},
        ],
        wait_seconds=60,
    )
    if job.status != "succeeded":
        job = client.jobs.wait(job.job_id)

for target in job.result["targets"]:
    print(target["id"], target["status"], target["matches"][:1])

job.result is a plain dict. Each target has id, status and matches; each match has rank, matched_text, match_method, confidence, cropped_image_url and location, whose kind decides its other keys: page_region (page, polygons), text_span (page, char_start, char_end), sheet_range (sheet, a1_range), row_range (start_row, end_row, column), json_pointer (pointer), jsonl_record (row_offset, column) or audio_range (start_ms, end_ms). See Matches.

UploadsCopy Anchor Link

Uploads are free and work at any balance. The bytes go from your machine straight to the document service with a one-use token, never through Drex and never with your API key. See Uploads.

Upload a fileCopy Anchor Link

client.documents.upload(content, *, path=None, file_name=None, on_conflict=None)

It returns an UploadedFile. Pass file.as_source() to any tool.

from nace_sdk import NaceClient

with NaceClient() as client:
    file = client.documents.upload("invoice.pdf", path="invoices/invoice.pdf", on_conflict="new_version")
    job = client.documents.extract(file.as_source(), schema={"type": "object", "properties": {"total": {"type": "number"}}}, wait_seconds=60)

• Keep the extension on path. The document service types an upload from its first bytes, and from the extension when those don't settle it; it refuses a file it can't type with unsupported_file_type. See Format support.
• Files of 32 MiB or more (UPLOAD_SESSION_THRESHOLD_BYTES) go up in parts through an upload session, automatically.
• A path conflict raises NaceConflictError with code path_conflict for a file sent whole. For a file sent in parts it is found when the parts are assembled, and raises NaceError naming the failed upload job and path_conflict.
• Uploading the same bytes to the same path again returns the file already there.
• A file still assembling after about ten minutes raises NaceError. The document service keeps assembling it; upload the same file to the same path again later to get it.

Upload grantsCopy Anchor Link

client.documents.create_upload_grant(**options) mints a one-use token for one upload into your workspace, so another client can upload without your API key. It returns an UploadGrant.

Mint the grant on your server and hand upload_url and token to the browser. The browser POSTs the file there as multipart form data, with a metadata part such as {"path": "...", "on_conflict": "new_version"}, only the X-Upload-Token header, and credentials: "omit". The upload's answer carries the file under result.file; send its workspace_id and file_id back to your server and use them as a WorkspaceFileSource. /v1 routes send no Access-Control-Allow-Origin header, so browser code can't call Drex itself; keep every Drex call on your server.

grant = client.documents.create_upload_grant(path="invoices/invoice.pdf", total_size_bytes=48213, ttl_seconds=600)
print(grant.upload_url, grant.expires_at)  # hand grant.upload_url and grant.token to the uploader

Upload sessionsCopy Anchor Link

A session uploads a large file in parts, and can resume after a dropped connection. upload runs one for you; drive one yourself to upload parts in parallel or across process restarts.

The part, status, complete and abort calls go straight to the session's upload_session_url with its session_token. complete_upload_session raises NaceError when the upload fails, for example with path_conflict, or when the file is still assembling after its wait; that message names the key to call it again with.

A session refuses calls that don't fit its state. None of these is retried:

• session_busy (NaceConflictError): sending a part or aborting while the session is completing, or completing it with a different key while a completion runs. Call complete_upload_session again with the same key to resume that completion.
• upload_expired (NaceConflictError): sending a part to a session that completed or was aborted, or completing an aborted one. Completing a completed session again returns its file.
• After expires_at the session token stops working, and every call on the session raises NaceAuthenticationError (401). Open a new session.

from pathlib import Path

from nace_sdk import NaceClient

source = Path("scan.tiff")
with NaceClient() as client:
    session = client.documents.create_upload_session(path="scans/scan.tiff", total_size_bytes=source.stat().st_size)
    with source.open("rb") as handle:
        # After a drop, send only the parts that haven't landed.
        landed = {part["part_number"] for part in client.documents.get_upload_session(session).parts}
        for number in range(session.total_parts):
            if number not in landed:
                handle.seek(number * session.chunk_size)
                client.documents.upload_part(session, number, handle.read(session.chunk_size))
    file = client.documents.complete_upload_session(session)

To resume from another process, save session.model_dump() and rebuild it with UploadSession.model_validate(saved).

JobsCopy Anchor Link

client.jobs reads, lists, waits for and cancels jobs. job_id is a str or a uuid.UUID; anything that isn't a job id raises NaceError before anything is sent.

list with status filters on the status Drex last recorded and leaves off a job that has moved on since, so a page can hold fewer jobs than limit, even none, while next_cursor says more follow. Keep paging until next_cursor is None, or use iter. A limit outside 1 to 50 raises NaceError.

client.jobs.list(limit=20, operation="parse", status="succeeded")
for job in client.jobs.iter(operation="extract"):
    print(job.job_id, job.status, job.credits)
client.jobs.delete(job_id)

See Jobs and results.

Wait for a jobCopy Anchor Link

wait polls get until status is succeeded, failed or cancelled: after 1 second, then doubling to every 10 seconds, for up to timeout seconds in all, 600 by default. Each poll is a get, with the client's retries.

• Failed or cancelled: raises NaceJobFailedError, with the job as error.job. With raise_on_failure=False, it returns the job instead.
• Out of time: raises NaceJobTimeoutError, with error.job and error.timeout. The job keeps running; wait again or read it later.

from nace_sdk import NaceJobFailedError, NaceJobTimeoutError

try:
    job = client.jobs.wait(job_id, timeout=1200)
except NaceJobFailedError as error:
    print(error.job.error_code, error.job.error_message)
except NaceJobTimeoutError as error:
    print("still", error.job.status)

Progress eventsCopy Anchor Link

events yields a JobEvent whenever the job's status or progress changes. The first event is the job's current state, and the stream ends when the job is terminal or after five minutes. It is a latency convenience: read the outcome with get or wait. The stream is never retried and counts against your per-minute limit. After a drop, pass the last event's sequence to continue its numbering:

last = None
for event in client.jobs.events(job_id):
    last = event
    print(event.sequence, event.status, event.progress)
if last is not None and last.status not in ("succeeded", "failed", "cancelled"):
    for event in client.jobs.events(job_id, last_event_id=str(last.sequence)):
        print(event.sequence, event.status)

Job filesCopy Anchor Link

A job file's path is a link from the result, such as document.content_url, a sheet's rows_url or a figure block's image_url, or the part after the job id, such as parse-images/<ref>. Its query and fragment are ignored. A path that is empty or holds . or .. raises NaceError, and so does a full link from another job's result, before any request: pass that job's id instead.

Downloads of job files don't count against your per-minute limit; the request echo and the event stream do. Results and their files expire on the document service's retention schedule. After that, job files and the request echo raise NaceNotFoundError (404, or 410 with result_expired), while get and wait still return the job, with result set to None and job.result_state saying why. See Jobs and results.

Download a fileCopy Anchor Link

download(job_id, path, destination, *, wait_timeout=600) saves any job file to destination, overwriting it, and returns its Path. A stored file comes from its signed link, without your API key; everything else streams from Drex.

Markdown the document service is still preparing answers 202: download waits as long as its retry-after header or retry_after_seconds says (else 3 seconds) and asks again, for up to wait_timeout seconds, then raises NaceError; the content keeps being prepared, so download it again later. Preparing content that fails raises NaceConflictError. Opening each request is retried under the client's policy; a transfer cut off midway is not.

from nace_sdk import NaceClient

with NaceClient() as client:
    job = client.jobs.wait(client.documents.parse("https://example.com/ledger.parquet").job_id)
    document = job.result["document"]
    if document.get("content_url"):  # markdown is only a preview
        client.jobs.download(job.job_id, document["content_url"], "ledger.md")

Signed linksCopy Anchor Link

file_link(job_id, path) returns a FileLink: url, which works for five minutes without your API key, to hand to a browser or another service, and expires_at, in Unix seconds. Only stored files have one: split-artifacts, ground-crops, ground-transcript, parse-images, parse-transcript, converted-pdf and spreadsheet-cell-maps. Any other path raises NaceError pointing at download before a request goes out; so does a stored file where the document service can't sign links, after the request. A stored-file path this job doesn't have raises NaceNotFoundError (404). download works in every case.

client.jobs.download(job_id, "parse-images/<ref>", "figure-1.png")
link = client.jobs.file_link(job_id, "parse-images/<ref>")
print(link.url, link.expires_at)

Job requestCopy Anchor Link

get_request(job_id) returns a JobRequest: request_type (the operation), and what the job was asked for, such as its source, options, schema, classes or targets, in model_extra. A PDF password is never returned. It expires with the job's result.

print(client.jobs.get_request(job_id).model_extra)

Spreadsheet rowsCopy Anchor Link

rows(job_id, path, *, start_row=None, limit=None) reads one page of a parsed sheet's rows, in file order: path is the sheet's rows_url, which only Parquet sources carry, start_row is zero-based (default 0), and limit is 1 to 100 (MAX_ROWS_PAGE_SIZE, default 50). A negative start_row or a limit out of range raises NaceError. It returns a RowsPage.

for sheet in (job.result["document"].get("spreadsheet") or {}).get("sheets", []):
    if not sheet.get("rows_url"):
        continue
    start = 0
    while True:
        page = client.jobs.rows(job.job_id, sheet["rows_url"], start_row=start, limit=100)
        print([column.name for column in page.columns], page.rows[:2])
        start += len(page.rows)
        if not page.rows or start >= page.total_rows:
            break

Saved schemasCopy Anchor Link

Save a JSON Schema once, then name it by schema_id on extract instead of sending it each time. Schemas can't be edited or deleted; add a version instead. Schema routes are free and work at any balance.

• Order: list and iter return your schemas and any Nace provides (owner="platform"), which can't take versions, sorted by schema_id the way the document service's database compares text: case-insensitively, with - and _ ignored at first, so not in byte order; sorted(ids) won't match it. Versions come oldest first, version 1 first; for the newest, call get.
• No automatic retries on creates: create and create_version aren't idempotent, since a retried create can save a second schema or version. They aren't retried unless you pass retry.
• The JSON Schema is schema_, because schema is a Pydantic method name. model_dump(by_alias=True)["schema"] gives it under its wire name.
• A limit outside 1 to 200, or a schema_id that is empty or holds /, ? or #, raises NaceError. Another account's schema_id raises NaceNotFoundError.

from nace_sdk import NaceClient

invoice = {
    "type": "object",
    "properties": {"invoice_number": {"type": "string"}, "total": {"type": "number"}},
}

with NaceClient() as client:
    saved = client.extraction_schemas.create("invoice", invoice, description="Invoice header fields")
    client.extraction_schemas.create_version(
        saved.schema_id,
        {**invoice, "properties": {**invoice["properties"], "due_date": {"type": "string"}}},
    )
    for version in client.extraction_schemas.iter_versions(saved.schema_id):
        print(version.version, version.schema_)
    job = client.documents.extract("https://example.com/invoice.pdf", schema_id=saved.schema_id, schema_version=1)

See Extract's saved schemas.

Types and constantsCopy Anchor Link

Every object a call returns directly also has request_id and raw_http_response, as in Read the response. Items inside a page don't carry them (reading one raises NaceError), so read the page's own; UploadedFile and JobEvent have neither. DocumentJob, UploadedFile, UploadSession, UploadSessionStatus, JobEvent and JobRequest keep any other field the document service sends as an attribute and in model_extra, such as job.progress.

DocumentJobCopy Anchor Link

DocumentJobList has items: tuple[DocumentJob, ...] and next_cursor: str | None, None on the last page.

Upload typesCopy Anchor Link

Job file typesCopy Anchor Link

Schema typesCopy Anchor Link

ConstantsCopy Anchor Link

Client constantsCopy Anchor Link

nace_sdk.constants holds the environment variable names and defaults the client uses:

LEGACY_ENV maps each variable to its name before the packages were renamed from drex-*. read_env(name) returns the stripped value of name, else of its LEGACY_ENV name, and None when both are unset or blank; the SDK, CLI and MCP server read every variable through it.

Document constantsCopy Anchor Link

nace_sdk exports the document constants and literal types itself:

nace_sdk.__version__ is the installed version, as a string.

Handle errorsCopy Anchor Link

Every API, connection and job failure raises a NaceError. An error response raises a NaceAPIError subclass:

Every NaceAPIError has:

Invalid arguments the SDK checks itself raise a plain NaceError before anything is sent. A few errors aren't NaceErrors: ValueError for both transport and http_client, or for uploading bytes without a file_name; pydantic.ValidationError for an unknown field on a question or source object; and OSError when a file can't be read or written.

from nace_sdk import NaceAPIError, NaceClient, NaceRateLimitError, Noul

with NaceClient() as client:
    try:
        client.system_one(state="Hello", questions={"greeting": Noul(instructions="Is this a greeting?")})
    except NaceRateLimitError as error:
        print("slow down", error.retry_after_ms, error.request_id)
    except NaceAPIError as error:
        print(error.status, error.type, error.message, error.issues, error.request_id)

The document routes raise the same classes as system_one. On them, error.code is the exact cause, and error.detail carries the document service's detail, such as detail["errors"] listing the fields a refused body got wrong. It's None when there is no detail, as for unsupported_file_type.

A failed multi-part upload raises NaceError, and jobs.wait raises NaceJobFailedError and NaceJobTimeoutError from the table above (see Wait for a job). See Errors for every code and the rate limits.

from nace_sdk import NaceConflictError, NaceUnprocessableEntityError

try:
    file = client.documents.upload("invoice.pdf", path="invoices/invoice.pdf")
except NaceConflictError as error:
    if error.code == "path_conflict":
        file = client.documents.upload("invoice.pdf", path="invoices/invoice.pdf", on_conflict="new_version")
    else:
        raise
try:
    client.documents.parse(file.as_source(), page_ranges=[{"start": 0, "end": 1}])
except NaceUnprocessableEntityError as error:
    print(error.code, error.message, error.detail)

See Errors and retries for what each error means.

RetriesCopy Anchor Link

The client retries 408, 429 and 5xx responses, dropped connections and timeouts, with exponential backoff, and honors retry-after-ms and retry-after. It doesn't retry an error whose body says "retryable": false. Every retry of a document create or upload reuses the same Idempotency-Key, so it never starts or bills a second job.

RetryPolicy is a frozen dataclass:

The budget counts from the first attempt. Before each retry the client adds the wait to the time already spent, and if that reaches timeout, it stops and raises the last error. So with the defaults, an attempt that fails after 30 seconds or more is not retried: a request that runs into the 60-second timeout, or a 529 that came after Drex's 55-second wait for the model. A retry-after longer than what is left of the budget isn't waited out either: the error is raised at once. To retry those, raise the budget above your timeout:

from nace_sdk import NaceClient, RetryPolicy

client = NaceClient(timeout=60, retry=RetryPolicy(max_retries=3, timeout=200))

A retry= on one call replaces the client's policy for that call. The client doesn't expose its policy, so build one from the policy you passed to the client (or from RetryPolicy()) with dataclasses.replace(policy, max_retries=0). A negative max_retries, a negative or infinite delay, a jitter outside 0 to 1, or a nonpositive timeout raises NaceError. Retried attempts carry an X-Nace-Retry-Count header. The job event stream and a download cut off midway are never retried, and saved-schema creates aren't retried unless you pass retry; see Progress events, Download a file and Saved schemas.

LoggingCopy Anchor Link

The SDK logs through the standard logging module, to the nace_sdk logger, which has only a NullHandler. Nothing prints until your app configures a handler, and you choose the level with NACE_LOG_LEVEL or setLevel:

import logging

logging.basicConfig()  # a stderr handler
logging.getLogger("nace_sdk").setLevel(logging.INFO)

NACE_LOG_LEVEL is read once, when nace_sdk is first imported. It takes debug, info, warn, warning, error or off, in any case; anything else is ignored.

Streamed calls log nothing, not even their retries, except an info line when their connection fails. They are jobs.events, jobs.file_link, and jobs.download with its fetch of a signed link.

Before any record reaches a handler, the SDK masks with *** the headers Authorization, Proxy-Authorization, X-Api-Key, Api-Key, X-Upload-Token, Cookie and Set-Cookie, any header whose name contains token or secret, and the body fields token, session_token and password. Connection errors are logged and raised with your credentials masked. URLs are logged as sent, query included, so a signed link's failed fetch logs the link.

Saved credentialsCopy Anchor Link

nace_sdk.credentials reads and writes the key that nace login saves, so the CLI and your own scripts share it. The key is resolved from NACE_API_KEY, then from the config file's api_key. The base URL comes from NACE_BASE_URL, then the file's base_url. The file is ~/.nace/config.toml, or the path in NACE_CONFIG_PATH.

The names from before the packages were renamed from drex-* still work when the new ones are absent: DREX_API_KEY, DREX_BASE_URL and DREX_CONFIG_PATH, and the file drex login wrote, ~/.drex/config.toml, which is read when ~/.nace/config.toml doesn't exist and no path variable is set. login always writes the new file.

from nace_sdk import NaceClient
from nace_sdk.credentials import resolve_credentials

credentials = resolve_credentials()
with NaceClient(api_key=credentials.api_key, base_url=credentials.base_url) as client:
    print(client.models.list().request_id)