Deep search

search.deep runs an agent over a workspace and returns a Job. The agent picks its own tools, authors evidence, and can take minutes. Files must already be ingested.

Deep search is not chat. Each call is one question. Continue a thread only with that turn’s session_id. After the search result, poll every id in result.grounding_job_ids — do not call Ground for those evidences.

There is no tier argument. effort is the quality and cost knob, here and on search.je_testing.

Basic usage

from ndi_sdk import NdiClient
from ndi_sdk.models.jobs import DeepSearchV2Result, IntelligentSearchResult

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    queued = client.search.deep(
        workspace_id,
        query="Which subsidiaries missed covenant tests?",
        wait_seconds=30,
    )
    job = client.jobs.wait(queued.job_id, timeout=600)
    # Returns: succeeded Job; job.result is DeepSearchV2Result or IntelligentSearchResult

    result = job.result
    if isinstance(result, (DeepSearchV2Result, IntelligentSearchResult)):
        print(result.answer, result.session_id)

Method signature

def deep(
    workspace_id: UUID | str,
    *,
    query: str,
    context: str | None = None,
    effort: "low" | "medium" | "high" | None = None,
    allow_clarification: bool = False,
    path_prefix: str | None = None,
    paths: list[str] | None = None,
    session_id: UUID | None = None,
    surface: "deep_search" | "qa" | "workbook" | None = None,
    use_kg: bool | None = None,
    wait_seconds: int = 0,
    idempotency_key: str | None = None,
) -> Job

Parameters

Parameter	Type	Required	Description
`workspace_id`	`UUID` or `str`	Yes	Workspace that already holds ingested files
`query`	`str`	Yes	The question for this turn
`context`	`str`	No	Extra background the agent should treat as given
`effort`	`"low"` / `"medium"` / `"high"`	No	Quality and cost. Server default when omitted is `low`
`allow_clarification`	`bool`	No	Let a genuinely ambiguous query return one question instead of an answer. Default `False`
`path_prefix`	`str`	No	Limit tools to files under this prefix
`paths`	`list[str]`	No	Limit tools to these uploaded paths (max 64). Intersects with `path_prefix`
`session_id`	`UUID`	No	Continue an earlier turn’s thread
`surface`	`"deep_search"` / `"qa"` / `"workbook"`	No	Console history bucket. Omit for the public default. Does not pick the agent
`use_kg`	`bool`	No	Let an unscoped run consult the workspace’s knowledge graph. Server default when omitted is on; `False` makes the agent work from files and search alone
`wait_seconds`	`int`	No	Hold the submission request open (max 300)
`idempotency_key`	`str`	No	Override the key the SDK normally mints

Returns

A Job. Which result model you get depends on scope:

Scope	`job.result`	`result_type`
No `paths` and no `path_prefix`	`DeepSearchV2Result`	`deep_search_v2`
`paths` and/or `path_prefix`	`IntelligentSearchResult`	`intelligent_search`

A follow-up stays on the thread’s original implementation even if this call adds or drops scope.

Effort

effort maps to thinking budget. The job echoes the level as job.effort.

Value	When to use
`low` (default)	Routine questions; typically tens of seconds
`medium`	Multi-document questions. On a scoped run this acts as `low`
`high`	Analytical questions that justify extra wall clock

from ndi_sdk import NdiClient

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    queued = client.search.deep(
        workspace_id,
        query="Reconcile year-end cash to the bank statements.",
        effort="high",
        wait_seconds=30,
    )
    job = client.jobs.wait(queued.job_id, timeout=600)
    # Returns: succeeded Job; job.effort echoes the requested level
    print(job.effort)

Scope files

Omit both paths and path_prefix for a corpus-wide, source-level run. Pass paths to pin every tool to those files (unknown or inaccessible paths are refused before a job is created). Combine with path_prefix as an intersection.

from ndi_sdk import NdiClient
from ndi_sdk.models.jobs import IntelligentSearchResult

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    queued = client.search.deep(
        workspace_id,
        query="What is the closing cash balance?",
        path_prefix="reports/",
        paths=["reports/cash-flow.pdf", "reports/notes.pdf"],
        wait_seconds=30,
    )
    job = client.jobs.wait(queued.job_id, timeout=600)
    # Returns: succeeded Job; scoped runs land IntelligentSearchResult
    if isinstance(job.result, IntelligentSearchResult):
        print(job.result.interpretation, job.result.answer)

Follow-up threads

Pass the previous result’s session_id. Only the API key that started the thread may continue it. A thread whose turn is still running returns 409 session_busy. Do not reuse one session across unrelated questions.

from ndi_sdk import NdiClient
from ndi_sdk.models.jobs import DeepSearchV2Result, IntelligentSearchResult

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    first = client.jobs.wait(
        client.search.deep(
            workspace_id,
            query="Which subsidiaries missed covenant tests?",
            wait_seconds=30,
        ).job_id,
        timeout=600,
    )
    # Returns: succeeded Job; read session_id from the result model

    result = first.result
    if not isinstance(result, (DeepSearchV2Result, IntelligentSearchResult)):
        raise TypeError(type(result))
    session_id = result.session_id

    follow = client.jobs.wait(
        client.search.deep(
            workspace_id,
            query="Show the supporting numbers for the first one.",
            session_id=session_id,
            wait_seconds=30,
        ).job_id,
        timeout=600,
    )
    # Returns: succeeded Job in the same thread
    print(follow.result.answer)

Clarifications

With allow_clarification=True, a genuinely ambiguous query can finish with clarification set, answer null, and empty evidences. Reply on the same session_id. Filtered search has no session; deep search does.

from ndi_sdk import NdiClient
from ndi_sdk.models.jobs import DeepSearchV2Result, IntelligentSearchResult

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    job = client.jobs.wait(
        client.search.deep(
            workspace_id,
            query="How did we do?",
            allow_clarification=True,
            wait_seconds=30,
        ).job_id,
        timeout=600,
    )
    result = job.result
    if isinstance(result, (DeepSearchV2Result, IntelligentSearchResult)) and result.clarification:
        print(result.clarification)
        # Re-ask on result.session_id

Context and surface

context is caller-supplied background, not a hidden system prompt replacement. surface only groups console history (qa, workbook). Omit it from SDK integrations. Follow-ups inherit the thread head.

from ndi_sdk import NdiClient

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    queued = client.search.deep(
        workspace_id,
        query="What is the stated reporting currency?",
        context="Prefer the audited consolidated statements over drafts.",
        wait_seconds=30,
    )
    job = client.jobs.wait(queued.job_id, timeout=600)
    # Returns: succeeded Job
    print(job.job_id)

Result structure

exhausted=True means the step budget ran out. The job still succeeded; the answer may rest on less evidence. That is not JobFailedError.

Unscoped: `DeepSearchV2Result`

Field	Description
`answer`	Direct answer. Null on clarification or if the run exhausted before authoring one
`reasoning`	How receipted statements chain into the answer
`evidences`	Quotes with source path, page, and relevance
`sql_receipts`	Statements the run executed; `rows` are a bounded preview
`figures`	Criteria tested, each pointing at receipt `result_id`s
`claims`	Machine-checkable claims that cite those receipts
`coverage`	Per-deliverable status (`answered` / `partial` / `unanswerable`)
`discrepancies`	Empty list is an answer, not an omission
`session_id`	Pass this on the next turn
`clarification`	Set only when `allow_clarification` rejected the query
`degraded`	Run broke after producing working papers; `sql_receipts` may still be useful
`grounding_job_ids`	Ground jobs planned for this answer’s evidences. Poll each with `jobs.wait` / `GET /v1/jobs/{id}` after this result. Empty when postprocessing is off or nothing is groundable

Scoped: `IntelligentSearchResult`

Field	Description
`answer`	Direct answer
`interpretation`	How the agent read the query
`reasoning`	Evidence-authoring notes
`evidences`	Quotes with source path and page
`coverage`	Access-label restrictions (`restricted_candidates`, `labels_required`)
`session_id`	Pass this on the next turn
`clarification`	Set only when `allow_clarification` rejected the query
`exhausted`	Step budget ran out
`grounding_job_ids`	Ground jobs planned for this answer’s evidences. Poll each with `jobs.wait` / `GET /v1/jobs/{id}` after this result. Empty when postprocessing is off or nothing is groundable

Search does not wait for quote locations. After the result, poll every id in grounding_job_ids. An empty list means nothing was planned. A missing field is a job from before this field existed — only then call client.documents.ground.

from ndi_sdk import NdiClient
from ndi_sdk.models.jobs import DeepSearchV2Result, IntelligentSearchResult

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    job = client.jobs.wait(
        client.search.deep(
            workspace_id,
            query="What were total liabilities at year end?",
            wait_seconds=30,
        ).job_id,
        timeout=600,
    )
    result = job.result
    if isinstance(result, (DeepSearchV2Result, IntelligentSearchResult)):
        for ground_id in result.grounding_job_ids or []:
            grounded = client.jobs.wait(ground_id, timeout=300)
            print(grounded.result)

Complete example

from ndi_sdk import NdiClient
from ndi_sdk.models.jobs import DeepSearchV2Result, IntelligentSearchResult

with NdiClient() as client:
    workspace = client.workspaces.create(name="fy25-audit")
    uploaded = client.upload_and_ingest(
        workspace.workspace_id,
        "report.pdf",
        path="reports/report.pdf",
    )
    client.jobs.wait(uploaded.ingestion_job.job_id, timeout=600)

    first = client.jobs.wait(
        client.search.deep(
            workspace.workspace_id,
            query="What were total liabilities at year end?",
            effort="medium",
            wait_seconds=30,
        ).job_id,
        timeout=600,
    )
    result = first.result
    if isinstance(result, DeepSearchV2Result):
        print(result.answer)
        for receipt in result.sql_receipts:
            print(receipt.result_id, receipt.sql, receipt.truncated)
    elif isinstance(result, IntelligentSearchResult):
        print(result.interpretation, result.answer)

    if isinstance(result, (DeepSearchV2Result, IntelligentSearchResult)) and result.session_id:
        follow = client.jobs.wait(
            client.search.deep(
                workspace.workspace_id,
                query="Break that down by current vs non-current.",
                session_id=result.session_id,
                wait_seconds=30,
            ).job_id,
            timeout=600,
        )
        # Returns: succeeded Job in the same thread
        print(follow.result.answer)

Async client

import asyncio

from ndi_sdk import AsyncNdiClient
from ndi_sdk.models.jobs import DeepSearchV2Result, IntelligentSearchResult


async def main() -> None:
    async with AsyncNdiClient() as client:
        workspace_id = "550e8400-e29b-41d4-a716-446655440001"
        queued = await client.search.deep(
            workspace_id,
            query="What is the reporting period?",
            wait_seconds=30,
        )
        job = await client.jobs.wait(queued.job_id, timeout=600)
        result = job.result
        if isinstance(result, (DeepSearchV2Result, IntelligentSearchResult)):
            print(result.answer)


asyncio.run(main())

Error handling

from ndi_sdk import ConflictError, InvalidRequestError, JobFailedError, NdiClient

with NdiClient() as client:
    workspace_id = "550e8400-e29b-41d4-a716-446655440001"
    try:
        queued = client.search.deep(
            workspace_id,
            query="What were total liabilities at year end?",
            session_id=None,
        )
        client.jobs.wait(queued.job_id, timeout=600)
        # Returns: succeeded Job, or raises for request/job failure
    except ConflictError as exc:
        print(exc.code, exc.message)  # session_busy
    except InvalidRequestError as exc:
        print(exc.code, exc.message)
    except JobFailedError as exc:
        print(exc.job.status, exc.job.error)

Code	When
`session_busy` (409)	That thread still has a running turn
`session_not_found` (404)	`session_id` is unknown or belongs to another key
`invalid_path_prefix`	`path_prefix` is not a usable workspace prefix
`workspace_not_found`	Workspace id is unknown

Best practices

Ingest first

Search reads derived representations. Upload plus ingest, then ask.

One question per turn

Start a new session for a new investigation. Follow-ups belong on that turn’s session_id.

Choose effort on purpose

Leave effort unset for lookups. Use high when the extra minutes are worth it.

Read exhaustion

exhausted=True is a successful, budget-capped answer — not a failed job.

Need	Method
Ranked snippets, inline	`client.tools.hybrid_search`
One-pass lookup job	`client.search.fact`
Filter/rank/count the catalog	`client.search.filtered` (Python SDK only)
Test a ledger package, with the SQL published	`client.search.je_testing`

See Search methods and the Search guide.

Next steps

Workspaces

Create a corpus, upload, and ingest before searching.

Job management

Wait, stream events, and cancel long-running jobs.

Journal-entry testing

Test a ledger package and read the SQL behind each figure.

Get started

Core methods

Async and utilities

Basic usage

Method signature

Parameters

Returns

Effort

Scope files

Follow-up threads

Clarifications

Context and surface

Result structure

Unscoped: `DeepSearchV2Result`

Scoped: `IntelligentSearchResult`

Complete example

Async client

Error handling

Best practices

Ingest first

One question per turn

Choose effort on purpose

Read exhaustion

Next steps

Workspaces

Job management

Journal-entry testing

​Basic usage

​Method signature

​Parameters

​Returns

​Effort

​Scope files

​Follow-up threads

​Clarifications

​Context and surface

​Result structure

​Unscoped: DeepSearchV2Result

​Scoped: IntelligentSearchResult

​Complete example

​Async client

​Error handling

​Best practices

Ingest first

One question per turn

Choose effort on purpose

Read exhaustion

​Related search methods

​Next steps

Workspaces

Job management

Journal-entry testing

Basic usage

Method signature

Parameters

Returns

Effort

Scope files

Follow-up threads

Clarifications

Context and surface

Result structure

Unscoped: `DeepSearchV2Result`

Scoped: `IntelligentSearchResult`

Complete example

Async client

Error handling

Best practices

Related search methods

Next steps