Python SDK - NDI Platform API

The NDI Python SDK is a typed interface to the /v1 Platform API. It resolves credentials, mints idempotency keys, retries transient failures, and parses every response into a Pydantic model.

Quick start

from pathlib import Path

from ndi_sdk import NdiClient

with NdiClient() as client:
    upload = client.documents.create_upload(Path("report.pdf"))
    job = client.jobs.wait(
        client.documents.parse(upload, wait_seconds=30).job_id,
        timeout=600,
    )
    print(job.result.markdown)
    # Returns: Job; job.result is a ParseResult on success

Key features

Typed end to end

Pydantic models for every request and response. Unknown fields and unknown enum members load instead of raising, so an additive server change never breaks a client.

Sync and async

NdiClient and AsyncNdiClient expose identical namespaces. The sync client runs on httpx.Client directly rather than wrapping an event loop.

Jobs, not polling loops

Slow operations return a Job. Wait with jobs.wait, follow progress with jobs.events, or finish inline with wait_seconds.

Automatic retries

Backoff with full jitter for 429, 5xx, and dropped connections. Job-creating calls carry an idempotency key, so a retry collapses onto the original job instead of billing twice.

Typed error hierarchy

Every failure is an NdiError subclass carrying the status code, error code, request ID, and retryability.

Two thin dependencies

httpx and pydantic only. The wire models are hand-written, so the package pulls in nothing from the NDI backend.

Installation

pip install ndi-sdk

Requires Python 3.11+. The package is ndi-sdk on PyPI. To upgrade, see Update the SDKs, CLI, and MCP.

Authentication

Set your API key as an environment variable:

export NDI_API_KEY="ndi_sk_..."

The SDK reads it automatically:

from ndi_sdk import NdiClient

client = NdiClient()  # reads $NDI_API_KEY
client.close()

Or pass it explicitly:

from ndi_sdk import NdiClient

client = NdiClient(api_key="ndi_sk_...")
client.close()

With neither, the constructor raises ValueError rather than letting the first call fail as a 401. base_url defaults to $NDI_BASE_URL, then https://ndi-api.nace.ai; set it only to reach another host.

Client options

NdiClient(
    *,
    api_key: str | None = None,
    base_url: str | None = None,
    timeout: float = 60,
    retry_policy: RetryPolicy | None = None,
    default_headers: dict[str, str] | None = None,
    http_client: httpx.Client | None = None,
)

Core methods

Upload

Stage bytes for a granular operation. The handle is single-use and expires.

from pathlib import Path

from ndi_sdk import NdiClient

with NdiClient() as client:
    upload = client.documents.create_upload(Path("report.pdf"))
    # Returns: UploadResponse with upload_id and expires_at fields
    print(upload.upload_id, upload.expires_at)

Upload documentation

Content types, TTL, and when to use a workspace instead.

Parse

Turn a file into markdown, blocks, and optional chunks.

from ndi_sdk import NdiClient

with NdiClient() as client:
    upload = client.documents.create_upload("report.pdf")
    job = client.jobs.wait(
        client.documents.parse(upload, wait_seconds=30).job_id
    )
    # Returns: Job; job.result is a ParseResult with document, ocr_applied, units
    print(job.result.document.markdown)

Parse documentation

Parse modes, OCR, figures, chunking, and spreadsheet options.

Extract

Fill a JSON Schema from a document. Pass json_schema or a stored schema_id.

from ndi_sdk import NdiClient

schema = {
    "type": "object",
    "properties": {
        "invoice_number": {"type": "string"},
        "total": {"type": "number"},
    },
    "required": ["invoice_number", "total"],
}

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    job = client.jobs.wait(
        client.documents.extract(upload, json_schema=schema, wait_seconds=30).job_id
    )
    # Returns: Job; job.result is an ExtractResult with data and fields
    print(job.result.data)
    for field in job.result.fields:
        print(field.path, field.status, field.value)

Extract documentation

Schema design, field status, and citations.

Split

Cut a multi-document packet into classified segments. Give every class you expect to find; pages that match none are governed by unknown_policy.

from ndi_sdk import NdiClient
from ndi_sdk.models.document_ops import SplitCategory

with NdiClient() as client:
    packet = client.documents.create_upload("vendor-packet.pdf")
    job = client.jobs.wait(
        client.documents.split(
            packet,
            classes=[
                SplitCategory(id="invoice", label="Invoice", description="A supplier invoice"),
                SplitCategory(id="receipt", label="Receipt", description="A payment receipt"),
                SplitCategory(id="purchase_order", label="Purchase order", description="A signed purchase order"),
                SplitCategory(id="bank_statement", label="Bank statement", description="A monthly bank statement"),
            ],
            wait_seconds=30,
        ).job_id
    )
    # Returns: Job; job.result is a SplitResult with page_count and segments
    for segment in job.result.segments:
        print(segment.split_class, segment.start_page, segment.end_page)

Split documentation

Classes, unknown and overlap policy, and split rules.

Classify

Label a whole document, or each page, against classes you define.

from ndi_sdk import NdiClient
from ndi_sdk.models.document_ops import ClassifyClass

with NdiClient() as client:
    document = client.documents.create_upload("mailroom-scan.pdf")
    job = client.jobs.wait(
        client.documents.classify(
            document,
            classes=[
                ClassifyClass(id="invoice", label="Invoice", description="A supplier invoice"),
                ClassifyClass(id="receipt", label="Receipt", description="A payment receipt"),
                ClassifyClass(id="contract", label="Contract", description="A signed agreement"),
                ClassifyClass(id="bank_statement", label="Bank statement", description="A monthly bank statement"),
            ],
            granularity="page",
            wait_seconds=30,
        ).job_id
    )
    # Returns: Job; job.result is a ClassifyResult with one unit per page
    for unit in job.result.units:
        print(unit.unknown, unit.labels)

Classify documentation

Granularity, unknown policy, and per-page labels.

Ground

Pin quoted text to a location in the source document.

from ndi_sdk import NdiClient
from ndi_sdk.models.document_ops import GroundTarget

with NdiClient() as client:
    upload = client.documents.create_upload("report.pdf")
    job = client.jobs.wait(
        client.documents.ground(
            upload,
            targets=[
                GroundTarget(id="total", text="1,200.50"),
                GroundTarget(id="fiscal_year", text="year ended December 31, 2025"),
            ],
            wait_seconds=30,
        ).job_id
    )
    # Returns: Job; job.result is a GroundResult with one entry per target
    for target in job.result.targets:
        print(target.id, target.status, len(target.matches))

Ground documentation

Targets, match methods, and location shapes.

Workspaces

Build a durable corpus, then search it. Uploading registers a ledger row; ingestion is a separate, billable call.

from ndi_sdk import NdiClient

with NdiClient() as client:
    workspace = client.workspaces.create(name="fy25-audit")
    # Returns: Workspace with workspace_id

    result = client.upload_and_ingest(
        workspace.workspace_id,
        "report.pdf",
        path="reports/report.pdf",
    )
    # Returns: UploadAndIngestResult with file, deduplicated, ingestion_job
    client.jobs.wait(result.ingestion_job.job_id, timeout=600)

    answer = client.jobs.wait(
        client.search.automatic(
            workspace.workspace_id,
            query="What were total liabilities at year end?",
            mode="fact",
            wait_seconds=30,
        ).job_id,
        timeout=300,
    )
    # Returns: succeeded Job; job.result is an AutomaticSearchResult
    print(answer.result)

Workspaces documentation

Workspaces, files, ingestion, tools, and search.

Automatic search and QA

Ask one question with search.automatic, or hold a conversation with search.qa. Both return a Job.

from ndi_sdk import NdiClient
from ndi_sdk.models.jobs import AutomaticAgentResult, AutomaticSearchResult

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    job = client.jobs.wait(
        client.search.automatic(
            workspace_id,
            query="Which subsidiaries missed covenant tests, and by how much?",
            mode="agent",
            wait_seconds=30,
        ).job_id,
        timeout=600,
    )
    result = job.result
    if isinstance(result, AutomaticSearchResult) and isinstance(result.output, AutomaticAgentResult):
        print(result.output.answer.answer)
    # Returns: Job; the result is AutomaticSearchResult, whose output is the route's result

search.fact, search.deep, and search.filtered are deprecated: they warn and keep working until they are retired. See Search.

Automatic search

Routes, modes, structured answers, and catalog tables.

QA

Conversations, follow-ups, and clarifying questions.

Journal-entry testing

An audit procedure over an ingested ledger package, not a search preset. Returns a Job whose result is a JetResult. effort is the one knob; omit it to take the deployment’s own floor.

from ndi_sdk import NdiClient

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    job = client.jobs.wait(
        client.search.je_testing(
            workspace_id,
            query="test whether any posted journal line was approved by its own preparer",
            paths=["ledger/journal.xlsx"],
        ).job_id
    )
    print(job.result.answer)
    # Returns: Job; job.result is a JetResult on success

Journal-entry testing

Effort, file scope, SQL receipts, and figures.

Ledger understanding

Measures the ingested journal package — table roles, integrity, and optional GL parsing — and stores a report on the workspace. Returns a Job whose result is a LedgerUnderstandingResult.

from ndi_sdk import NdiClient

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    job = client.jobs.wait(
        client.workspaces.start_ledger_understanding(workspace_id).job_id,
        timeout=3600,
    )
    print(job.result.report.stale, job.result.report.blocking)
    report = client.workspaces.get_ledger_understanding(workspace_id)
    print(len(report.tables), report.parsing)
    # Returns: Job; job.result is a LedgerUnderstandingResult on success

Ledger understanding

Policy, reuse, the report, and in-progress conflicts.

Async client

AsyncNdiClient has the same surface with await. Share one client across tasks.

import asyncio

from ndi_sdk import AsyncNdiClient


async def main() -> None:
    async with AsyncNdiClient() as client:
        upload = await client.documents.create_upload("report.pdf")
        queued = await client.documents.parse(upload, wait_seconds=30)
        job = await client.jobs.wait(queued.job_id, timeout=600)
        print(job.result.markdown)


asyncio.run(main())

Async and concurrency guide

Bounded concurrency, return_exceptions, and client lifetime.

Response types

documents.parse returns a Job, not a ParseResult — the work outlives the HTTP request, so the result arrives on the job. Once the job succeeds, job.result is a ParseResult. Each method has one result model, tagged by result_type:

Method	`job.result` model	`result_type`
`documents.parse`	`ParseResult`	`parse`
`documents.extract`	`ExtractResult`	`extract`
`documents.split`	`SplitResult`	`split`
`documents.classify`	`ClassifyResult`	`classify`
`documents.ground`	`GroundResult`	`ground`
`search.automatic`	`AutomaticSearchResult`	`automatic_search`
`search.qa`	`AutomaticAgentResult`	`automatic_agent`
`search.deep` (deprecated)	`DeepSearchV2Result`	`deep_search_v2`
`search.je_testing`	`JetResult`	`je_testing`
`workspaces.start_ledger_understanding`	`LedgerUnderstandingResult`	`ledger_understanding`
`ingestion.ingest`	`IngestionResult`	`ingestion`

from ndi_sdk import NdiClient

with NdiClient() as client:
    upload = client.documents.create_upload("report.pdf")
    queued = client.documents.parse(upload, wait_seconds=30)
    job = client.jobs.wait(queued.job_id, timeout=600)

    print(job.result.markdown)      # ParseResult.markdown
    print(job.result.ocr_applied)   # ParseResult.ocr_applied
    print(job.result.units)         # billed units

Because Job is one model for every kind of work, its result field is typed as the union of all result models. That is invisible at runtime, but a type checker will ask you to narrow before reading variant fields:

from ndi_sdk.models.jobs import ParseResult

result = job.result
if isinstance(result, ParseResult):
    print(result.document.markdown)

A result whose tag this SDK version does not know arrives as an UnknownResult holding the raw payload, so a newly added job kind is never an outage.

Pagination

iter_all walks cursor pages for workspaces, files, jobs, and domains; access labels use iter_access_labels. See Pagination.

Error handling

Catch NdiError for anything the SDK raises, or a specific subclass when you plan to act on it.

from ndi_sdk import NdiClient
from ndi_sdk.errors import AuthenticationError, JobFailedError, NdiStatusError

with NdiClient() as client:
    try:
        client.documents.parse(client.documents.create_upload("report.pdf"))
    except AuthenticationError as exc:
        print(exc.status_code, exc.request_id)
    except JobFailedError as exc:
        print(exc.job.status, exc.job.error)
    except NdiStatusError as exc:
        print(exc.code, exc.retryable, exc.message)

Error handling guide

The full exception hierarchy, retry policy, and request IDs.

Advanced usage

Timeouts and retries

timeout bounds one HTTP request, not a job. Waiting for work is jobs.wait(...).

from ndi_sdk import NdiClient, RetryPolicy

with NdiClient(timeout=120, retry_policy=RetryPolicy(max_attempts=5)) as client:
    print(client.base_url)

Custom HTTP client

Bring your own httpx.Client for proxies, mTLS, connection tuning, or a mock transport in tests. When you pass one, you close it.

import httpx

from ndi_sdk import NdiClient

http_client = httpx.Client(proxy="http://proxy.internal:8080")
client = NdiClient(http_client=http_client)
print(client.base_url)
http_client.close()

Default headers

from ndi_sdk import NdiClient

with NdiClient(default_headers={"X-Team": "audit-platform"}) as client:
    print(client.base_url)

Streaming job progress

jobs.events yields Server-Sent Events until the job is terminal. Treat it as a latency convenience and confirm the final state with jobs.get or jobs.wait.

from ndi_sdk import NdiClient

with NdiClient() as client:
    queued = client.documents.parse(client.documents.create_upload("report.pdf"))
    for event in client.jobs.events(queued.job_id):
        # Returns: an iterator of JobEvent
        print(event.status, event.message)

Legacy API surface

client.legacy exposes the older /api/v1 product API. It is imported on first use, so a process that never touches it does not pay for its models.

Next steps

Get an API key

Create a key and export it as NDI_API_KEY. See Authentication.

Run the quickstart

Parse your first document end to end in the Quickstart.

Explore the methods

Start with Parse, then Job management. The same operations also run from the terminal — see NDI CLI.

​Quick start

​Key features

Typed end to end

Sync and async

Jobs, not polling loops

Automatic retries

Typed error hierarchy

Two thin dependencies

​Installation

​Authentication

​Client options

​Core methods

​Upload

Upload documentation

​Parse

Parse documentation

​Extract

Extract documentation

​Split

Split documentation

​Classify

Classify documentation

​Ground

Ground documentation

​Workspaces

Workspaces documentation

​Automatic search and QA

Automatic search

QA

​Journal-entry testing

Journal-entry testing

​Ledger understanding

Ledger understanding

​Async client

Async and concurrency guide

​Response types

​Pagination

​Error handling

Error handling guide

​Advanced usage

​Timeouts and retries

​Custom HTTP client

​Default headers

​Streaming job progress

​Legacy API surface

​Next steps

Quick start

Key features

Installation

Authentication

Client options

Core methods

Upload

Parse

Extract

Split

Classify

Ground

Workspaces

Automatic search and QA

Journal-entry testing

Ledger understanding

Async client

Response types

Pagination

Error handling

Advanced usage

Timeouts and retries

Custom HTTP client

Default headers

Streaming job progress

Legacy API surface

Next steps