Python SDK

The ninthsense Python SDK is the primary way to run 9thSense verifications from your application. It is async-first, fully typed with Pydantic models, and ships a LocalSimulator so you can write tests without a running server.

The SDK is organised around the same primitives as the platform: Cases for full verification workflows, one-shot verification for single documents and selfies, and Faces for direct biometric operations. Every result carries the check verdicts and Goal Rule outcomes that decided it.

Installation

pip install 9thsense

The PyPI package is 9thsense. The import name is ninthsense — Python module names cannot start with a digit.

python -c "import ninthsense; print(ninthsense.__version__)"

Client

from ninthsense import Client

Constructor

Client(
    api_key: str | None = None,
    base_url: str = "https://api.9thsense.ai",
    tenant_id: str = "",
    timeout: float = 60.0,
)

Context manager (recommended)

The client holds an HTTP connection pool. Using it as an async context manager ensures the pool is closed on exit.

import asyncio
import os
from ninthsense import Client

async def main():
    async with Client(api_key=os.environ["NINTHSENSE_API_KEY"]) as client:
        health = await client.healthz()
        print(health)  # {"status": "ok"}

asyncio.run(main())

Methods

---

Cases namespace

A Case is one verification run for one subject, executed against a deployed agent. All case operations live under client.cases.

cases.create(goal, *, client_reference_id, pipeline_profile)

Always pass client_reference_id — it makes creation idempotent (repeating the call with the same reference returns the same case) and is echoed on every webhook.

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

# Recover a case if a create response is ever lost in flight:
existing = await client.cases.get_by_client_reference("LOAN-2291")

cases.upload(case_id, path, *, doc_type)

Attach a document or selfie. Negotiates a presigned upload and falls back to multipart; uploads are auto-classified when doc_type is omitted.

doc = await client.cases.upload(case.case_id, "pan_card.jpg")
await client.cases.upload(case.case_id, "selfie.jpg", doc_type="selfie")

# Or pull directly from your bucket without moving bytes through your app:
doc = await client.cases.add_document_from_source(case.case_id, s3_uri)

cases.get(case_id)

Fetch current status, extracted data, goal progress, and check verdicts.

case = await client.cases.get(case.case_id)
print(case.status)          # collecting | processing | review | completed | denied
print(case.goal_progress)   # per-rule pass/fail/pending
print(case.checks)          # list[CheckResult]

cases.complete(case_id, *, if_needed) / cases.wait(case_id, *, interval, timeout)

Completing is explicit — it tells the platform no more documents are coming. Then poll until the case reaches a terminal state.

# if_needed=True: agents that auto-complete on their final upload answer 409 —
# the desired end state, not a failure.
await client.cases.complete(case.case_id, if_needed=True)

final = await client.cases.wait(case.case_id, interval=3, timeout=300)

---

One-shot verification

For flows that don't need a multi-document case, client.verify runs extraction plus the configured checks on a single submission and returns everything in one call.

verify.document(content, filename, *, doc_type, allow_cloud)

result: VerifyResult = await client.verify.document(
    content=pdf_bytes,
    filename="statement_jan2026.pdf",
    doc_type="fin_bank_statement",
)

print(result.extracted)          # typed fields
print(result.checks)             # e.g. document integrity verdict
print(result.rules)              # goal-rule outcomes

verify.selfie(content, *, reference)

Runs liveness, deepfake detection, and — when reference names an ID photo already in the case or request — face match.

result = await client.verify.selfie(
    content=selfie_bytes,
    reference="kyc_pan_card.photo",
)
for check in result.checks:
    print(check.name, check.verdict, check.confidence)
# liveness   pass  0.97
# deepfake   pass  0.99
# face_match pass  0.91

---

Faces namespace

Direct biometric operations, mirroring the Face Biometrics API.

# 1:1 likeness
match = await client.faces.match(image_a=selfie_bytes, image_b=id_photo_bytes)
print(match.score, match.verdict)

# 1:N search across configured databases
hits = await client.faces.search(image=selfie_bytes, top_k=5)
for hit in hits:
    print(hit.identity_id, hit.database, hit.confidence)

---

Models

CheckResult

class CheckResult(BaseModel):
    name: str                    # "liveness", "deepfake", "face_match", ...
    verdict: str                 # "pass" | "fail" | "flag"
    confidence: float
    markers: list[Marker] = []   # timestamped markers for video checks

RuleOutcome

class RuleOutcome(BaseModel):
    rule_id: str
    passed: bool
    severity: str                # "hard_stop" | "require"
    message: str = ""

VerifyResult

class VerifyResult(BaseModel):
    success: bool
    extracted: dict[str, Any] = {}
    checks: list[CheckResult] = []
    rules: list[RuleOutcome] = []
    error: str = ""
    latency_ms: int = 0
    from_self_hosted: bool = True

    def raise_on_error(self) -> "VerifyResult": ...

raise_on_error() returns self when success is True, and raises VerificationError otherwise. Use it to integrate cleanly with try/except.

try:
    result = await client.verify.document(content=raw_bytes, filename="pan.jpg")
    result.raise_on_error()
    fields = result.extracted
except VerificationError as e:
    print(f"Verification failed: {e.message}")

---

Error classes

All SDK errors inherit from NinthSenseError.

from ninthsense import AuthError, CaseNotFoundError, VerificationError, NinthSenseError

try:
    result = await client.verify.document(content=raw_bytes, filename="pan.jpg")
    result.raise_on_error()
except AuthError:
    # Regenerate or rotate the API key
    ...
except VerificationError as e:
    print(f"Verification failed: {e.message}")
except NinthSenseError as e:
    # Catch-all for any other SDK error
    print(e)

---

LocalSimulator

LocalSimulator is a drop-in replacement for Client that requires no server or API key. It is useful in unit tests and CI environments.

from ninthsense.simulator import LocalSimulator

async def test_kyc_flow():
    async with LocalSimulator() as client:
        case = await client.cases.create(goal="kyc_individual")
        await client.cases.upload(case.case_id, b"<bytes>", "pan_card.jpg")
        final = await client.cases.wait(case.case_id, interval=3, timeout=300)
        assert final.status == "completed"
        assert all(r.passed for r in final.rules)

LocalSimulator implements the same interface as Client — cases, verify, faces, and healthz() — so you can swap it in without changing your application code.

The simulator ships deterministic stub responses:

---

Full example workflow

The following script opens a case, uploads a PAN card and a selfie, waits for the checks and Goal Rules to evaluate, and reads the final verdict.

import asyncio
import os
from ninthsense import Client, VerificationError

async def run_kyc(pan_bytes: bytes, selfie_bytes: bytes) -> dict:
    async with Client(
        api_key=os.environ["NINTHSENSE_API_KEY"],
        tenant_id=os.environ.get("TENANT_ID", ""),
    ) as client:

        # 1. Open a case against the KYC agent
        case = await client.cases.create(goal="kyc_individual")

        # 2. Upload evidence — auto-classified and routed
        await client.cases.upload(case.case_id, pan_bytes, "pan_card.jpg")
        await client.cases.upload(case.case_id, selfie_bytes, "selfie.jpg", doc_type="selfie")

        # 3. Wait for extraction, checks, and rule evaluation
        final = await client.cases.wait(case.case_id, timeout=180.0)

        # 4. Inspect check verdicts and goal progress
        for check in final.checks:
            print(f"{check.name}: {check.verdict} ({check.confidence:.2f})")
        for rule in final.rules:
            print(f"rule {rule.rule_id}: {'PASS' if rule.passed else 'FAIL'}")

        return {"status": final.status, "extracted": final.extracted}

if __name__ == "__main__":
    with open("pan_card.jpg", "rb") as p, open("selfie.jpg", "rb") as s:
        result = asyncio.run(run_kyc(p.read(), s.read()))
    print(result)