Metadata-Version: 2.4
Name: ndi-sdk
Version: 0.7.0
Summary: Python SDK for NDI (Nace Document Intelligence)
Keywords: ndi,document-intelligence,document-parsing,extraction
Author: Nace AI
Author-email: Nace AI <engineering@nace.ai>
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic>=2.11.5,<3.0.0
Requires-Dist: ndi-mcp>=0.1,<1 ; extra == 'mcp'
Requires-Python: >=3.11, <3.14
Project-URL: Homepage, https://github.com/nace-ai/audit-app/tree/main/sdk/ndi-python
Project-URL: Repository, https://github.com/nace-ai/audit-app
Project-URL: Changelog, https://github.com/nace-ai/audit-app/blob/main/sdk/ndi-python/CHANGELOG.md
Project-URL: Issues, https://github.com/nace-ai/audit-app/issues
Provides-Extra: mcp
Description-Content-Type: text/markdown

# NDI Python SDK

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

```bash
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

```python
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:

```python
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 (`https://ndi-api.nace.ai/v1/mcp`) | Local (`uvx ndi-mcp`) |
|---|---|---|
| Install | none | Python 3.11+, `uvx` |
| Auth | `X-API-Key` or `Authorization: Bearer` header | `ndi-mcp login`, or `$NDI_API_KEY` |
| Files on your machine | public URLs only | uploaded straight from disk |
| Best for | quick setup, shared and CI agents | coding agents working on local files |

### Hosted: paste a URL

```json
{
  "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

```bash
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`:

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

opencode — add to `opencode.json`:

```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.

| | Platform `/v1` | Legacy `/api/v1` |
|---|---|---|
| Where | `client.workspaces`, `client.files`, `client.ingestion`, `client.documents`, `client.tools`, `client.search`, `client.jobs`, `client.domains` | `client.legacy` |
| State | Workspaces persist; files are ingested once and searched many times | Stateless, one-shot, nothing retained past the job |
| Use it for | Anything new | Maintaining an integration already written against it |

## 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:

```python
# 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:

```python
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.

```python
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:

```python
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:

```python
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:

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

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

```python
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:

```python
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 —

```python
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:

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

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

### Extract

```python
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.

```python
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

```python
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.

```python
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:

```python
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:

```python
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.

```python
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:

```python
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:

```python
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](https://semver.org/spec/v2.0.0.html), 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](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.

```bash
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:

```bash
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.
