NDI Python SDK

Python client for NDI (Nace Document Intelligence): parse, split, classify, extract and ground documents, and build searchable workspaces over a corpus.

pip install ndi-sdk

Requires Python 3.11+. Depends only on httpx and pydantic. The MCP server is a separate MIT package: pip install ndi-mcp (or uvx ndi-mcp). ndi-sdk[mcp] still resolves to that package as a compatibility alias.

Quickstart

from ndi_sdk import NdiClient

with NdiClient(api_key="ndi_sk_…") as client:
    upload = client.documents.create_upload("report.pdf")
    job = client.jobs.wait(client.documents.parse(upload, wait_seconds=60).job_id)
    print(job.result.markdown)

api_key defaults to $NDI_API_KEY and the host to $NDI_BASE_URL, so a configured environment needs only NdiClient(). Every method exists identically on AsyncNdiClient, awaited:

import asyncio

from ndi_sdk import AsyncNdiClient


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


asyncio.run(main())

Share one async client across tasks — asyncio.gather(...) over several documents reuses its connection pool, and each jobs.wait polls independently.

Both clients are context managers. If you pass your own http_client (for proxies, mTLS, or a mock transport in tests), you own closing it.

Use it from an AI agent

NDI speaks MCP, so Claude Code, Cursor, Codex, or opencode can call it as tools. There are two ways in, and they serve the same tools:

Hosted: paste a URL

{
  "mcpServers": {
    "ndi": {
      "type": "http",
      "url": "https://ndi-api.nace.ai/v1/mcp",
      "headers": { "X-API-Key": "ndi_sk_…" }
    }
  }
}

The hosted server runs in our cloud, so it cannot see your filesystem: upload_document and upload_file refuse a local path there. Pass a public URL, or run the local server.

Local: one process, your disk

uvx ndi-mcp login
claude mcp add ndi -- uvx ndi-mcp

ndi-mcp login opens the NDI console in your browser. Approve there; the CLI stores the minted key in ~/.ndi/config.toml. Pass --api-key (or set $NDI_API_KEY) to skip the browser.

Cursor — add to .cursor/mcp.json:

{
  "mcpServers": {
    "ndi": {
      "command": "uvx",
      "args": ["ndi-mcp"]
    }
  }
}

opencode — add to opencode.json:

{
  "mcp": {
    "ndi": {
      "type": "local",
      "command": ["uvx", "ndi-mcp"]
    }
  }
}

$NDI_API_KEY is read first; otherwise the key saved by ndi-mcp login. Local files go through upload_document, then parse_document / extract_data with the returned upload_id. A parse of a long PDF is spilled to a temp file so it does not fill the context window; the tool returns the path.

Workspace flows use create_workspace, upload_file, ingest_workspace, then deep_search — or upload_and_ingest_file to upload one file and make it searchable in a single call. Once a corpus is ingested, fact_search answers in one retrieval pass, qa_file answers about one file, and query_tables answers across several spreadsheets. list_jobs shows what has run. Call get_documentation with a topic (parse, extract, auth, …) before writing integration code — it returns this SDK's current surface.

The two APIs

NDI has two surfaces and this SDK covers both.

Everything slow is a job

Ingestion, parse, extract, search and workspace deletion all return a Job rather than a result, because any of them can outlast a request. There are two ways to wait, and they compose:

# Ask the server to hold the response open, up to its ceiling.
job = client.ingestion.ingest(workspace_id, path_prefix="reports/", wait_seconds=120)

# Poll from the client. Returns as soon as the job is terminal.
job = client.jobs.wait(job.job_id, timeout=600)

wait_seconds saves a round trip for work that finishes quickly; jobs.wait covers the rest. It raises JobFailedError if the job failed (pass raise_on_failure=False to get the failed job back instead) and JobTimeoutError if your budget runs out — the job keeps running server-side either way.

job.result is a discriminated union keyed on result_type, so the result of a parse is a ParseResult and the result of an extract is an ExtractResult, with no casting:

job = client.jobs.wait(client.documents.extract(source, json_schema=schema).job_id)
for field in job.result.fields:
    print(field.path, field.value, field.status, field.citations)

A result type this SDK version does not know arrives as UnknownResult with its payload intact, so a server-side addition never breaks a client.

Job progress is also available as Server-Sent Events (client.jobs.events(job_id)). Treat it as a latency convenience: the stream can end early, so nothing that must be correct should depend on receiving a frame.

Workspace flow, end to end

A workspace is a durable corpus: upload files, ingest them once, then search and query them repeatedly.

from ndi_sdk import NdiClient

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

    # Upload: a path, raw bytes, or an open binary file.
    client.files.upload(
        workspace.workspace_id,
        "local/balance-sheet.xlsx",
        path="reports/balance-sheet.xlsx",
        labels={"engagement": "fy25"},
    )

    # Or have NDI fetch the bytes itself, e.g. from a presigned URL.
    client.files.upload_from_url(
        workspace.workspace_id,
        path="reports/minutes.pdf",
        url=presigned_url,
        file_name="minutes.pdf",
    )

    # Ingest. Uploading does not make a file searchable; this does.
    ingestion = client.ingestion.ingest(workspace.workspace_id, path_prefix="reports/")
    ingestion = client.jobs.wait(ingestion.job_id, timeout=1800)

    # Outcomes are per file: one corrupt document does not fail the run.
    for outcome in ingestion.result.outcomes:
        if outcome.error:
            print("skipped", outcome.path, outcome.error.code, outcome.error.message)

    # Ask a question and get cited evidence back.
    answer = client.jobs.wait(
        client.search.deep(
            workspace.workspace_id,
            query="What were total liabilities at year end, and where is that stated?",
        ).job_id
    )
    print(answer.result.answer)
    for evidence in answer.result.evidences:
        print(evidence.source_path, evidence.page, evidence.quote)

client.search.automatic lets the service answer one query with a one-pass fact search or an agent that reads documents, queries tables, and selects documents from the metadata catalog.

For a single file, client.upload_and_ingest(...) collapses the two calls — it uploads, then ingests the returned file_id, and hands back the file together with the ingestion job. It is a client-side convenience over the same two API calls, so prefer the explicit form when batching several uploads into one ingestion run:

result = client.upload_and_ingest(
    workspace.workspace_id,
    "local/balance-sheet.xlsx",
    path="reports/balance-sheet.xlsx",
)
client.jobs.wait(result.ingestion_job.job_id)

Direct end-client uploads

When your own users' files should reach NDI without a round trip through your backend — and without your API key ever reaching their browser — mint a short-lived upload grant and hand its token to the client:

grant = client.files.create_upload_grant(
    workspace.workspace_id,
    path="inbox/statement.pdf",  # optional: pin the destination
    max_bytes=10 * 1024 * 1024,  # optional: cap the size
    ttl_seconds=3600,
)
# Give grant.token to the browser. It uploads directly:
#   POST {grant.upload_url}   (multipart: file + metadata parts)
#   X-Upload-Token: {grant.token}

The grant authorizes exactly one thing — a multipart upload into that workspace, within the constraints above — until expires_at. It is single-use on deployments running Redis, and it stops working immediately if the API key that minted it is revoked.

Re-ingest after files change with stale_only=True, which narrows the run to files whose bytes moved:

client.ingestion.ingest(workspace_id, path_prefix="reports/", stale_only=True)

client.workspaces.stats(workspace_id) is the completeness instrument:

from ndi_sdk import IngestionStatus

stats = client.workspaces.stats(workspace_id)
stats.files_by_ingestion_status.get(IngestionStatus.STALE, 0)    # > 0 means re-ingest
stats.files_by_ingestion_status.get(IngestionStatus.EXPIRED, 0)  # > 0 means retention dropped derivatives

Deletion is irreversible and needs the name back as confirmation:

job = client.workspaces.delete(workspace_id, confirm_name="fy25-audit")
client.jobs.wait(job.job_id)

One-shot document operations

client.documents is stateless: nothing is written to a workspace, and each call names its own source. Four kinds of source are accepted —

from ndi_sdk import ParseResultSource, UploadSource, UrlSource, WorkspaceFileSource

UrlSource(url=presigned_url, file_name="invoice.pdf")     # NDI fetches it
UploadSource(upload_id=upload.upload_id)                  # staged bytes
WorkspaceFileSource(workspace_id=ws_id, file_id=file_id)  # a file already in a workspace
ParseResultSource(job_id=parse_job.job_id)                # reuse a parse, do not pay twice

For local bytes, stage them once and reuse the handle across operations. The handle from create_upload is accepted directly wherever a source is:

upload = client.documents.create_upload("local/invoice.pdf")

parse = client.jobs.wait(client.documents.parse(upload).job_id)

Pass parse_mode="medium" to route to the best parser, or "high" to add page verification and correction. The default is "low".

Extract

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

# Check the schema first — free, and reports every violation at once.
validation = client.documents.validate_extract_schema(json_schema=schema)
assert validation.valid, validation.errors

job = client.jobs.wait(client.documents.extract(upload, json_schema=schema).job_id)

print(job.result.data)                  # the schema-shaped payload
for field in job.result.fields:         # per field: status and provenance
    print(field.path, field.value, field.status, field.confidence)

status == "not_found" is a real answer about the document, not an error.

Ground

Ground pins quoted text back to a location in its source — the step that turns an answer into something auditable.

Experimental GroundOptions.use_word_level_grounding matches targets against word-level OCR bboxes (PDF only). Targets shorter than 2 normalized characters return status failed with error needle_too_short.

from pathlib import Path

from ndi_sdk.models.document_ops import GroundOptions, GroundTarget

job = client.jobs.wait(
    client.documents.ground(
        upload,
        targets=[GroundTarget(id="total", text="1,200.50")],
        options=GroundOptions(include_previews=True),
    ).job_id
)

for target in job.result.targets:
    for match in target.matches:
        print(target.id, match.matched_text, match.location)
        if match.cropped_image_url:
            png = client.jobs.ground_crop(job.job_id, match.cropped_image_url)
            Path(f"{target.id}.png").write_bytes(png)

Split and classify

from ndi_sdk.models.document_ops import ClassifyClass, SplitCategory

# Separate a scanned packet into its logical documents.
client.documents.split(upload, classes=[
    SplitCategory(id="invoice", label="Invoice", description="A supplier invoice"),
    SplitCategory(id="receipt", label="Receipt", description="A payment receipt"),
])

# Label one document against classes you define.
client.documents.classify(upload, classes=[
    ClassifyClass(id="invoice", label="Invoice", description="A supplier invoice"),
])

Workspace tools

Beyond search, the tools ask one file or a set of tables a question. They answer inline — no jobs.

client.tools.qa_file(workspace_id, path="reports/report.pdf", query="What is the invoice total?")
client.tools.query_tables(                                        # one question, several tables
    workspace_id, paths=["reports/q1.xlsx", "reports/q2.xlsx"], query="Compare quarterly revenue"
)

The knowledge graph is built per workspace, not per file — entity resolution links entities across documents, and fact and deep search use it:

build = client.ingestion.build_knowledge_graph(workspace_id)
client.jobs.wait(build.job_id, timeout=3600)

Ledger understanding measures the ingested journal package — table roles, integrity, and optional GL parsing — and stores a report on the workspace:

from ndi_sdk.models.workspaces import LedgerUnderstandingConfig

client.workspaces.update(
    workspace_id,
    ledger_understanding=LedgerUnderstandingConfig(trigger="auto", gl_parsing=True),
)
job = client.workspaces.start_ledger_understanding(workspace_id)
client.jobs.wait(job.job_id, timeout=3600)
report = client.workspaces.get_ledger_understanding(workspace_id)
print(report.stale, report.blocking, report.parsing)
# list_ledger_sources / select_ledger_sources confirm a population;
# pass the returned selection_id to start and get.

System One (Decision Model)

client.systemone.evaluate answers typed questions about a state — text or any JSON value — in one synchronous call. No workspace, no job:

from ndi_sdk.models.systemone import ChoiceQuestion, NoulCriteria, NoulQuestion, ScoreQuestion

result = client.systemone.evaluate(
    "Vendor invoice of $48,000 approved by the requester, no PO on file.",
    {
        "needs_review": NoulQuestion(criteria=NoulCriteria(true="Needs a second approver")),
        "category": ChoiceQuestion(criteria={"services": None, "goods": None, "travel": None}),
        "risk": ScoreQuestion(criteria=["low", "medium", "high"]),
    },
)
print(result.answers["needs_review"].noul)     # probability it is true
print(result.answers["category"].choice)       # likeliest option
print(result.answers["risk"].score)            # expected level index, 0 = "low"

Each answer is keyed by its question's name. A refused question raises InvalidRequestError whose body.detail["issues"] names it by path, such as questions.risk.criteria.

Pagination

Every listing is cursor-paginated. Take a page at a time, or let the SDK follow the cursor:

from ndi_sdk import JobKind

page = client.files.list(workspace_id, path_prefix="reports/")
print(page.items, page.next_cursor, page.total_count)

for file in client.files.iter_all(workspace_id, path_prefix="reports/"):
    print(file.path, file.ingestion_status)

for job in client.jobs.iter_all(workspace_id=workspace_id, kind=[JobKind.INGESTION]):
    print(job.job_id, job.status)

On a files.list page, coverage says what the access gate withheld, so a short page is distinguishable from a filtered one.

Errors

Every failure is an NdiError. HTTP failures carry the server's typed error code, so you can branch without matching on message text.

from ndi_sdk import ConflictError, ErrorCode, InvalidRequestError, NdiError, RateLimitError

try:
    client.files.upload(workspace_id, "local/report.pdf", path="reports/report.pdf")
except InvalidRequestError as exc:
    if exc.code != ErrorCode.UNSUPPORTED_FILE_TYPE:
        raise
    # No workspace file was created, so there is no ingestion job to poll.
    print(exc.message)
except ConflictError as exc:
    if exc.code != ErrorCode.PATH_CONFLICT:
        raise
    # Something is already at that path — keep both as versions instead.
    client.files.upload(
        workspace_id, "local/report.pdf", path="reports/report.pdf", on_conflict="new_version"
    )
except RateLimitError as exc:
    print(exc.retryable, exc.request_id)
except NdiError:
    raise

exc.request_id is worth logging: it is what NDI support needs to find your request.

Format rejection is synchronous and non-retryable. documents.create_upload rejects a format that no granular document operation supports; a document-operation call rejects an otherwise stageable source when that operation does not support it. Workspace uploads reject unsupported inputs before creating a file, and upload_and_ingest does not queue ingestion. The service also rejects a filename that conflicts with a conclusive signature in the bounded byte prefix; multipart Content-Type is advisory and the full file is not buffered for detection. Use the server's unsupported_file_type response as the authority instead of copying a format allowlist into the client.

Retries and idempotency

Transient failures (429, 5xx, dropped connections) are retried automatically with exponential backoff, honouring Retry-After. A 408 is not one of them: on NDI it only ever means a legacy synchronous endpoint's wait window closed while the job runs on, so it raises SyncWaitTimeoutError and you poll get_job. Every job-creating call carries an Idempotency-Key, so a retry collapses onto the original job instead of starting — and billing — a second one. Pass your own idempotency_key to extend that guarantee across process restarts:

from datetime import date

from ndi_sdk import RetryPolicy

client = NdiClient(retry_policy=RetryPolicy(max_attempts=5, initial_backoff=1.0))

client.ingestion.ingest(workspace_id, idempotency_key=f"nightly-ingest-{date.today()}")

A key you pass is sent exactly as given. An empty string raises ValueError rather than being quietly replaced with a fresh key: the server reads it as a key like any other, so every call carrying one would replay the first such job instead of doing its own work.

Legacy /api/v1

Kept whole under client.legacy for existing integrations. It is stateless and has no workspaces: each call names its document by URL or by an ndi://file/<uuid> handle from upload().

Each action has two forms. The plain form waits inline; the _async form starts the job and hands back its id:

handle = client.legacy.upload("local/invoice.pdf")

# Wait inline. Small documents only — raises SyncWaitTimeoutError if the window elapses,
# and the job keeps running.
result = client.legacy.parse(handle, timeout_seconds=60)

# Or start it and poll.
accepted = client.legacy.extract_async(handle, json_schema=schema)
job = client.legacy.get_job(accepted.job_id)

Both forms are typed as JobStatus | JobAccepted, because either can come back from either: a replayed idempotency key answers an async start with the existing job, and a sync call with no wait capacity answers with a handle to poll.

Forward compatibility

The SDK is built to survive a server that grows:

• Unknown fields on a response are kept, not rejected.
• Unknown enum members (a new JobKind, a new ErrorCode) arrive as their string value and still compare equal to it, rather than failing validation.
• Unknown job results arrive as UnknownResult with the payload intact, and unknown System One answer types as UnknownAnswer.

So a new NDI capability does not require an SDK upgrade before your integration keeps working.

Versioning

Semantic versioning, with the usual pre-1.0 caveat: while the major version is 0, a minor bump may change the API surface. Anything that breaks a caller is listed under a Breaking heading in CHANGELOG.md.

Pin accordingly:

ndi-sdk>=0.2,<0.3