End-to-end integration

Every other SDK page is a reference: it tells you what a method does once you know you need it. This page is the opposite — one continuous path from an API key to a verified result, with nothing assumed. Work through it once and you have a working integration; go to the reference pages after, for the parts you want to change.

Java is at 0.4.2; Python is at 0.4.1. Your account manager will confirm the coordinates and access for your environment.

1. Install and authenticate

Python

pip install 9thsense

Java (Maven)

<dependency>
  <groupId>in.digio.ninthsense</groupId>
  <artifactId>ninthsense-sdk</artifactId>
  <version>0.4.2</version>
</dependency>

The PyPI package is 9thsense; the import is ninthsense, because Python module names cannot start with a digit. In Java the groupId is in.digio.ninthsense while the package is ai.ninthsense — that mismatch is deliberate and permanent.

Your key encodes its environment: zk_test_… keys only ever see test cases, zk_live_… only live ones. They are not interchangeable, and a key cannot reach a case in the other environment — you will get a 404, not a 403, because we do not confirm that a case exists somewhere you cannot see.

Python

import asyncio
from ninthsense import Client

async def main():
    async with Client(api_key="zk_live_...", base_url="https://api.9thsense.ai") as client:
        ...

asyncio.run(main())

Java

import ai.ninthsense.Client;

try (Client client = new Client.Builder()
        .apiKey("zk_live_...")
        .baseUrl("https://api.9thsense.ai")
        .build()) {
    // ...
}

Use the context manager / try-with-resources. Both SDKs hold a connection pool; leaking it leaks sockets.

2. Open a case

A case is one unit of work — one applicant, one video call, one loan file. Documents attach to it, and it produces one verdict.

Python

case = await client.cases.create(
    "kyc_verification",
    client_reference_id="LOAN-2291",       # your own id
)
print(case.case_id)                         # == case.session_id

Java

CaseInfo case_ = client.cases()
        .create("kyc_verification", null, "LOAN-2291")
        .join();
System.out.println(case_.sessionId());

Always pass client_reference_id. It makes creation idempotent — repeating the call with the same reference returns the same case rather than a duplicate — and it is echoed on every webhook, so you can correlate without storing our id. If a create response is ever lost in flight, this is what lets you recover:

Python

existing = await client.cases.get_by_client_reference("LOAN-2291")

Java

Optional<CaseInfo> existing = client.cases()
        .getByClientReference("LOAN-2291").join();

If your agent declares pipeline profiles, name one here to pick among them — pipeline_profile="transcribe_only" in Python, a fourth argument to create in Java. Omit it for the agent's default.

3. Grant read access to your bucket

One-time setup, done once per environment, not per case.

Give us a role that can s3:GetObject on the prefix you will drop files into. s3:ListBucket is not required — that is the difference from the polling route, and it is usually what makes this acceptable to a security review.

Then configure it once under Admin → Connectors → Inbound (S3): the bucket, the prefix, and the role. That configuration becomes the default read boundary for step 4.

4. Attach the document

Python

doc = await client.cases.add_document_from_source(
    case.case_id,
    "s3://your-bucket/inbound/LOAN-2291/pan.jpg",
)

if doc.status == "rejected":
    print(doc.error)     # a rejected document never counts toward a required type

Java

DocumentInfo doc = client.cases()
        .addDocumentFromSource(case_.sessionId(),
                               "s3://your-bucket/inbound/LOAN-2291/pan.jpg")
        .join();

if (doc.rejected()) {
    System.out.println(doc.error());
}

With no credential named we use the one on your configured inbound connector and bound the read to that connector's bucket and prefix — so the short call is also the narrow one. Name a credential explicitly only to read outside it.

Call it again for each further document on the same case. Each call attaches a separate document; nothing is deduplicated on the object key, so if you need at-most-once, deduplicate your side.

Rejections happen before any bytes move — we HEAD the object first — so an oversized or wrong-typed file costs one small request, not a transfer.

Pinning the type. If you already know what the file is, say so — the model is then shown that type's output schema and returns the agreed field names instead of inventing them. Pass document_type= in Python, or the extra argument in Java. It does not force the document through: the classifier still runs, and a confident disagreement is still rejected or recorded as a mismatch.

If the file is not in S3, replace this step and change nothing else:

Python

# Your app holds the bytes. Negotiates presigned PUT, falls back to multipart.
doc = await client.cases.upload(case.case_id, "pan.jpg")

Java

DocumentInfo doc = client.cases()
        .uploadFile(case_.sessionId(), Path.of("pan.jpg"), "pan_card")
        .join();

There is also a fully hands-off route where you never call the API at all: drop files in your bucket and we poll it. See §6.8 of the integration contract.

5. Complete, and wait for the verdict

Completing is explicit: it tells us no more documents are coming.

Python

# if_needed=True: some agents auto-complete on their final upload and finish
# before you get here, which answers 409 — the desired end state, not a failure.
await client.cases.complete(case.case_id, if_needed=True)

status = await client.cases.wait(case.case_id, interval=3, timeout=300)
print(status.status)          # completed / review / denied / failed

Java

// ifNeeded=true: some agents auto-complete on their final upload and finish
// before you get here, which answers 409 — the desired end state, not a failure.
client.cases().complete(case_.sessionId(), true).join();

CaseStatus status = client.cases().awaitCompletion(
        case_.sessionId(), Duration.ofSeconds(3), Duration.ofMinutes(5));

Polling is fine for a few cases. For production, use the webhook in step 7 and treat polling as the fallback.

6. Read the results

Python

artifacts = await client.cases.outputs(case.case_id)
for a in artifacts:
    print(a.type, a.filename)

await client.cases.download_all(case.case_id, to="./results")

Java

OutputsManifest outputs = client.cases().listOutputs(case_.sessionId()).join();
for (ArtifactInfo a : outputs.artifacts()) {
    client.cases().downloadArtifact(case_.sessionId(), a, Path.of("results")).join();
}

The manifest is the same set whichever route delivered the input: the verdict, the extracted fields, any transcript, the redacted media, and the report if one was rendered.

If you would rather have results pushed into a bucket you own, configure an s3_push destination — we write every artifact and then manifest.json last, so the appearance of manifest.json is your signal that the case is complete. Trigger off that key, never off the individual artifacts.

7. Receive and verify the webhook

Register once:

Python

await client.admin_webhooks.register(
    "https://your-app.example.com/9thsense/webhook",
    events=["case.completed"],
)

Java

client.adminWebhooks()
      .register("https://your-app.example.com/9thsense/webhook",
                List.of("case.completed"))
      .join();

Then verify every delivery before trusting it. Never parse an unverified body — the endpoint is public, and the signature is the only thing that makes it ours:

Python

from ninthsense import webhooks
from ninthsense.models import SignatureVerificationError

@app.post("/9thsense/webhook")
async def receive(request):
    body = await request.body()
    try:
        event = webhooks.parse_case_completed(
            body, dict(request.headers), secret=WEBHOOK_SECRET)
    except SignatureVerificationError:
        return Response(status_code=400)

    # Your own id, echoed back — correlate without storing ours.
    my_id = event.client_reference_id
    return Response(status_code=200)

Java

WebhookEvent event = Webhooks.constructEvent(
        rawBody, headers, webhookSecret);   // throws SignatureVerificationException

The signature is HMAC-SHA256 over "{timestamp}.{body}", sent as X-Digio-Checksum with X-Digio-Timestamp. Deliveries older than 5 minutes are rejected, so replaying a captured request does not work. Verify against the raw bytes — re-serializing the JSON first changes them and the check fails.

Respond 2xx quickly and do the work asynchronously; a slow endpoint is retried, which means duplicate deliveries. Make your handler idempotent on client_reference_id.

from ninthsense import outbound_webhooks

event = outbound_webhooks.parse_event(
    raw_body, dict(request.headers), secret=SECRET)

if event.outputs and event.outputs.s3_export:
    print(event.outputs.s3_export.manifest_uri)   # written last = complete

CaseCompletedEvent event = OutboundWebhooks.parseEvent(
        rawBody, headers, secret);

Putting it together

Python

import asyncio
from ninthsense import Client

async def verify_applicant(reference: str, s3_uri: str) -> str:
    async with Client(api_key="zk_live_...") as client:
        case = await client.cases.create(
            "kyc_verification", client_reference_id=reference)

        doc = await client.cases.add_document_from_source(case.case_id, s3_uri)
        if doc.status == "rejected":
            raise RuntimeError(f"document rejected: {doc.error}")

        await client.cases.complete(case.case_id)
        status = await client.cases.wait(case.case_id, timeout=300)

        await client.cases.download_all(case.case_id, to=f"./results/{reference}")
        return status.status

print(asyncio.run(verify_applicant(
    "LOAN-2291", "s3://your-bucket/inbound/LOAN-2291/pan.jpg")))

Java

try (Client client = new Client.Builder().apiKey("zk_live_...").build()) {
    CaseInfo c = client.cases()
            .create("kyc_verification", null, "LOAN-2291").join();

    DocumentInfo doc = client.cases()
            .addDocumentFromSource(c.sessionId(),
                    "s3://your-bucket/inbound/LOAN-2291/pan.jpg").join();
    if (doc.rejected()) {
        throw new IllegalStateException("document rejected: " + doc.error());
    }

    client.cases().complete(c.sessionId()).join();
    CaseStatus status = client.cases().awaitCompletion(
            c.sessionId(), Duration.ofSeconds(3), Duration.ofMinutes(5));

    System.out.println(status.status());
}

Before you go live

• Test in the test environment first. zk_test_… keys are free of quota and fully isolated; nothing crosses over.
• Make the webhook handler idempotent. Retries mean duplicate deliveries.
• Decide retention. By default we keep our copy of your files. If the original stays in your bucket — which it does on Route D — turn on purge after export under Admin → Organization → Retention, and our copy is destroyed once results are delivered. The verdict and the audit record are kept either way.
• Handle rejected documents. A rejected document never counts toward a required type, so the case will sit incomplete rather than fail loudly.
• Do not hardcode our case id anywhere your own reference would do.

Where to go next