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, UrlSource

with NdiClient(api_key="ndi_sk_…") as client:
    job = client.documents.parse(
        UrlSource(url="https://example.com/report.pdf", file_name="report.pdf"),
        wait_seconds=60,
    )
    job = client.jobs.wait(job.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:

from ndi_sdk import AsyncNdiClient

async with AsyncNdiClient() as client:
    job = await client.documents.parse(source)
    job = await client.jobs.wait(job.job_id)

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, hybrid_search returns passages to read yourself, 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)

    # Search the corpus.
    search = client.tools.hybrid_search(workspace.workspace_id, query="total liabilities", k=10)
    for hit in search.hits:
        print(hit.path, hit.locator, hit.snippet)

    # Or 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?",
            include_answer=True,
        ).job_id
    )
    print(answer.result.answer)
    for evidence in answer.result.evidences:
        print(evidence.source_path, evidence.page, evidence.quote)

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=600,
)
# 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)

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.

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 read a workspace's structure and content directly. They answer inline — no jobs.

client.tools.folder_metadata(workspace_id, directory="reports/")   # what is in here
client.tools.file_metadata(workspace_id, path="reports/model.xlsx")  # what is in this file
client.tools.read_file(workspace_id, path="reports/minutes.pdf", pages=[3, 4])
client.tools.qa_file(workspace_id, path="reports/model.xlsx", query="What is the EBITDA margin?")
client.tools.query_tables(                                        # one question, several tables
    workspace_id, paths=["reports/q1.xlsx", "reports/q2.xlsx"], query="Compare quarterly revenue"
)

client.tools.kg_info(workspace_id)                                # is there a graph, and its shape
client.tools.kg_search(workspace_id, query="the parent holding company")
client.tools.kg_walk(workspace_id, start_node_ids=["entity:acme"], hops=2)

The knowledge graph is built per workspace, not per file — entity resolution links entities across documents:

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

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, NdiError, RateLimitError

try:
    client.files.upload(workspace_id, "local/report.pdf", path="reports/report.pdf")
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.

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.

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

Development

This package lives in the audit-app monorepo as a uv workspace member and depends on nothing else in it — its wire models are hand-written copies, which is what lets it ship standalone.

uv sync --all-extras --all-packages --dev

uv run pytest sdk/ndi-python/tests            # SDK unit tests, mock transport
uv run ruff format && uv run ruff check
uv run pyrefly check

The copies are kept honest by a contract test on the server side, where importing both is allowed: services/ndi_service/tests/platform_api/test_sdk_contract.py compares every wire field and enum member against ndi_service's own schemas and round-trips payloads through both. If you change a /v1 schema, that test tells you what the SDK still needs.

Note that CI does not run that test for a PR touching only this package, because ndi_service declares no dependency on it — run it locally. The whole update loop (mirror the drift, bump the version, write the changelog entry) is scripted as the ndi-sdk-update skill in .cursor/skills/.

Releasing

A release is a tag push. pyproject.toml, __version__, and the newest CHANGELOG.md heading must already agree — tests/test_version.py enforces that, and the release refuses a tag naming a different version:

git tag ndi-sdk-v0.2.0
git push origin ndi-sdk-v0.2.0

.github/workflows/ndi-sdk-publish.yml then runs the SDK suite and the contract guard above — the one place that guard is enforced for an SDK-only change — builds the wheel and sdist, and uploads them with PyPI Trusted Publishing, so no PyPI credential is stored anywhere. Running the workflow manually does everything except the upload, which is how to rehearse. A published version can never be replaced, only superseded.