Python SDK

The published client is ndi-sdk on PyPI. It covers the same /v1 surface as this site. Requires Python 3.11+. Depends only on httpx and pydantic.

pip install ndi-sdk

Mint a key on the API keys page in the NDI console, then:

export NDI_API_KEY="ndi_sk_..."
# Defaults to production. Set this only for another host:
# export NDI_BASE_URL="http://localhost:8003"

NdiClient() reads $NDI_API_KEY and $NDI_BASE_URL. You can also pass api_key= / base_url= explicitly.

What you can do

Job	Call
Parse a file into markdown	`client.documents.parse`
Extract JSON against a schema	`client.documents.extract`
Split a packet / classify a document	`client.documents.split` / `.classify`
Pin quoted text to a location	`client.documents.ground`
Build a searchable corpus	`workspaces` + `files` + `ingestion` + `tools.hybrid_search`

One-shot calls on client.documents do not write to a workspace. Uploads for those calls are single-use: the bytes are dropped when the job that reads them finishes.

Parse a local file

from pathlib import Path
from ndi_sdk import NdiClient

with NdiClient() as client:
    upload = client.documents.create_upload(Path("report.pdf"))
    job = client.documents.parse(
        upload,
        wait_seconds=30,
    )
    print(job.result.markdown)

wait_seconds holds the HTTP response open when the work finishes quickly. For everything else, poll:

job = client.jobs.wait(job.job_id, timeout=600)
print(job.result.markdown)

Extract structured JSON

Pass a JSON Schema. Extract from an upload, or from a prior parse with ParseResultSource so the parse is not paid for twice. not_found on a field is an answer about the document, not an error.

from ndi_sdk import NdiClient, ParseResultSource

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")
    parsed = client.jobs.wait(client.documents.parse(upload, wait_seconds=30).job_id)
    job = client.jobs.wait(
        client.documents.extract(
            ParseResultSource(job_id=parsed.job_id),
            json_schema=schema,
            wait_seconds=30,
        ).job_id
    )
    print(job.result.data)

Split and classify

split cuts a scanned packet into logical documents. classify labels one document against classes you define. Classify reads the original file, so it rejects a parse-result source.

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

with NdiClient() as client:
    packet = client.documents.create_upload("packet.pdf")
    client.documents.split(
        packet,
        classes=[
            SplitCategory(id="invoice", label="Invoice", description="A supplier invoice"),
            SplitCategory(id="receipt", label="Receipt", description="A payment receipt"),
        ],
        wait_seconds=30,
    )

    document = client.documents.create_upload("invoice.pdf")
    client.documents.classify(
        document,
        classes=[
            ClassifyClass(id="invoice", label="Invoice", description="A supplier invoice"),
        ],
        wait_seconds=30,
    )

Ground quoted text

Ground pins a quote back to a location in the source.

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")],
            wait_seconds=30,
        ).job_id
    )
    for target in job.result.targets:
        for match in target.matches:
            print(target.id, match.matched_text, match.location)

Search a workspace

A workspace is a durable corpus: upload files, ingest them once, then search and ask questions repeatedly. Uploading does not make a file searchable — ingestion does.

from ndi_sdk import NdiClient

with NdiClient() as client:
    workspace = client.workspaces.create(name="fy25-audit")
    client.jobs.wait(
        client.files.upload(
            workspace.workspace_id,
            "report.pdf",
            path="reports/report.pdf",
        ).job_id
    )

    ingestion = client.ingestion.ingest(workspace.workspace_id, path_prefix="reports/")
    client.jobs.wait(ingestion.job_id, timeout=600)

    hits = client.tools.hybrid_search(
        workspace.workspace_id,
        query="total liabilities at year end",
        k=5,
    )
    for hit in hits.hits:
        print(hit.path, hit.snippet)

    answer = client.tools.qa_file(
        workspace.workspace_id,
        path="reports/report.pdf",
        query="What were total liabilities at year end?",
    )
    print(answer.answer)

For a single file, client.upload_and_ingest(...) collapses the upload and ingest calls. Across several spreadsheets, client.tools.query_tables(...) answers one question over the set.

Async

Every method exists on AsyncNdiClient, awaited:

from ndi_sdk import AsyncNdiClient

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

Both clients are context managers. If you pass your own http_client, you own closing it.

Next steps

Authentication — minting and rotating keys
TypeScript SDK
MCP — use NDI from Claude Code, Cursor, or opencode
Document operations
Workspaces
Search tools
File tools
Jobs and idempotency

Getting started

SDKs and agents

Concepts

What you can do

Parse a local file

Extract structured JSON

Split and classify

Ground quoted text

Search a workspace

Async

Next steps

​What you can do

​Parse a local file

​Extract structured JSON

​Split and classify

​Ground quoted text

​Search a workspace

​Async

​Next steps

What you can do

Parse a local file

Extract structured JSON

Split and classify

Ground quoted text

Search a workspace

Async

Next steps