didit-sdk

Unofficial, community-maintained Python client for the Didit Identity Verification & KYC API.

Highlights

• Ergonomic Sync & Async: Dual-client architecture built on top of high-performance httpx.
• Strictly Typed & Validated: 100% type annotations (PEP 561 compliant with py.typed) and robust Pydantic v2 domain models.
• Didit V3 API Alignment: Full fidelity to upstream V3 schemas (id_verifications[], liveness_checks[], face_matches[], aml_screenings[], reviews[], warnings[]) with forward compatibility (extra="allow"), verification warnings helper (decision.has_warning(...)), and backward-compatible property accessors.
• Complete Session Lifecycle: Covers all 10 documented Didit session statuses (Not Started, In Progress, In Review, Approved, Declined, Expired, Abandoned, Kyc Expired, Resubmitted, Awaiting User) with granular state inspection (is_decided, is_closed, requires_review, requires_user_action, requires_resubmission).
• Resubmission & Manual Review: Full resubmission support (client.sessions.resubmit()) and review status transitions (client.sessions.update_status()).
• Official Compliance PDF Reports: Download verification audit reports (client.sessions.generate_pdf_report() / download_pdf_report()) with binary header (%PDF-), MIME format validation, and atomic disk writes with private permissions.
• Production CLI (didit): Inspect sessions, download PDF reports, verify webhook signatures, query sandbox scenarios, and run diagnostic health checks with zero-credential exposure in command arguments.
• Cryptographic Security & DoS Defense: Constant-time HMAC-SHA256 signature verification (X-Signature-V2 & X-Signature), full UTF-8 Unicode canonical JSON support (ensure_ascii=False, allow_nan=False), authoritative signed body anti-replay verification with strict header matching, and bounded body limits (HTTP 413) to prevent memory exhaustion.
• Webhook Deduplication: Thread-safe in-memory and atomic Redis deduplication stores (InMemoryWebhookDedupStore, RedisWebhookDedupStore, AsyncRedisWebhookDedupStore) with configurable duplicate actions (respond_ok, pass, raise).
• Multi-Framework Integrations: Native adapters for FastAPI (DiditWebhookGuard), Django (@didit_webhook_view), and Flask (@didit_webhook).
• Resilient HTTP Engine: Deterministic jittered exponential backoff retries with fail-fast budget timers for transient errors (HTTP 429, 502, 503, 504).
• High-Fidelity Sandbox Parity: 16 official Didit sandbox outcome simulation slugs (approve, decline_document_expired, decline_face_match_low_similarity, decline_aml_hit, review_face_match_borderline, review_aml_possible_match, etc.) and calibrated 0–100 biometric confidence scoring.

Installation

# Core SDK (httpx + pydantic)
pip install didit-sdk

# With framework integrations
pip install "didit-sdk[fastapi]"
pip install "didit-sdk[django]"
pip install "didit-sdk[flask]"

# With Redis webhook deduplication store
pip install "didit-sdk[redis]"

# All optional dependencies
pip install "didit-sdk[all]"

Quickstart

1. Synchronous Client

from didit import Didit, SessionStatus

# Initialize client (falls back to DIDIT_API_KEY environment variable if omitted)
client = Didit(api_key="your_api_key", webhook_secret="your_webhook_secret")

# Create a verification session with sandbox outcome scenario
session = client.sessions.create(
    vendor_data="user_12345",
    workflow_id="wf_kyc_standard",
    callback="https://yourapp.com/kyc/complete",
    language="es",
    sandbox_scenario="approve",
)

print(f"Verification URL: {session.url}")
print(f"Session ID: {session.session_id}")

# Fetch verification decision immediately:
decision = client.sessions.get_decision(session.session_id)

# Or poll until a terminal outcome (Approved or Declined) is reached:
decision = client.sessions.poll_decision(session.session_id, timeout=60.0, interval=2.0)
if decision.status == SessionStatus.APPROVED:
    print(f"User approved! Document: {decision.document.document_number}")
    if decision.has_warning("DOCUMENT_POOR_QUALITY"):
        print("Note: Document quality was flagged.")

2. Asynchronous Client (asyncio)

import asyncio
from didit import AsyncDidit, SessionStatus


async def main():
    async with AsyncDidit(api_key="your_api_key") as client:
        session = await client.sessions.create(
            vendor_data="user_987",
            workflow_id="wf_kyc_standard",
        )
        print(f"Session URL: {session.url}")

        decision = await client.sessions.poll_decision(session.session_id, timeout=30.0)
        if decision.status.is_terminal:
            print(f"Final outcome: {decision.status}")


asyncio.run(main())

3. Session Listing & Pagination

Query verification sessions from Didit V3 (GET /v3/sessions/) with typed pagination and multi-attribute filtering:

from didit import Didit, SessionStatus

client = Didit()

# Fetch paginated verification sessions
page = client.sessions.list(
    status=SessionStatus.APPROVED,
    vendor_data="user_12345",
    country="ESP",  # Normalized to uppercase and validated as 3-letter alpha-3-shaped code
    limit=50,
    offset=0,
)

print(f"Total matching sessions: {page.count}")
for item in page.results:
    print(f"- Session {item.session_id}: {item.status.value} (Workflow: {item.workflow_id})")

4. Snapshot Reconciliation (Drift Detection)

Audit your local database state against upstream Didit API snapshots to detect state divergence that may result from delivery gaps, stale local state, or other synchronization failures without exposing PII:

from didit import Didit, ObservedSessionState, SessionStatus

client = Didit()

# Reconcile a single session snapshot
report = client.sessions.reconcile(
    "sess_12345",
    observed=ObservedSessionState(
        session_id="sess_12345",
        status=SessionStatus.IN_REVIEW,
        warning_codes=["SUSPECTED_FRAUD"],
    ),
)

if not report.is_in_sync:
    print(f"Drift detected! Remote status is {report.remote_status} (Local: {report.local_status})")
    print(f"Warnings added remotely: {report.warning_codes_added}")
    print(f"Warnings removed remotely: {report.warning_codes_removed}")


# Batch reconcile over a time window against your database with auto-pagination
class DatabaseSessionSource:
    def get(self, session_id: str) -> ObservedSessionState | None:
        row = db.find_session(session_id)
        if not row:
            return None
        return ObservedSessionState(
            session_id=row.id, status=row.status, warning_codes=row.warnings
        )


batch_report = client.sessions.reconcile_range(
    since="2026-01-01T00:00:00Z",
    until="2026-01-02T00:00:00Z",
    source=DatabaseSessionSource(),
    page_size=50,
    max_sessions=1000,
)
print(
    f"Audited {batch_report.total_evaluated} sessions (remote: {batch_report.remote_count}, "
    f"truncated: {batch_report.truncated}): {batch_report.drift_count} drifts found."
)

5. PII-Minimized Telemetry Event Sink

Attach a passive DiditEventSink to collect structured, privacy-safe SDK operational metrics (backoff retries, rate limits, duplicate webhooks, lease degradations/losses, and reconciliation drift). Event delivery is synchronous and fail-isolated (exceptions raised inside emit() are safely caught and suppressed by safe_emit); sink implementations MUST return promptly and should enqueue external I/O:

from didit import Didit, DiditEventSink, DiditSDKEvent


class MetricsEventSink:
    def emit(self, event: DiditSDKEvent) -> None:
        # PII-minimized: contains only structured identifiers, status enums, and timing metrics
        statsd.increment(f"didit.sdk.{event.event_type}")


client = Didit(event_sink=MetricsEventSink())

6. Session Resubmission & Lifecycle Updates

Handle customer resubmission workflows when a verification session requires document or biometric resubmission, or perform manual reviewer status transitions:

from didit import Didit, SessionStatus

client = Didit()

# Inspect whether a decision requires user resubmission
decision = client.sessions.get_decision("sess_12345")
if decision.requires_resubmission and decision.resubmit_info:
    print(f"Resubmit steps: {decision.resubmit_info.nodes}")
    if decision.resubmit_info.available_attempts is not None:
        print(f"Attempts remaining: {decision.resubmit_info.available_attempts}")

    from didit import ResubmitFeature, ResubmitNode

    # Request resubmission for specific failed check nodes
    result = client.sessions.resubmit(
        "sess_12345",
        nodes_to_resubmit=[
            ResubmitNode(
                node_id="feature_ocr",
                feature=ResubmitFeature.OCR,
            )
        ],
    )
    print(f"Resubmission acknowledged for session: {result.session_id}")

    # Query decision endpoint to retrieve the updated session state
    current_decision = client.sessions.get_decision(result.session_id)
    print(f"Current decision status: {current_decision.status}")

# Update session status (manual review transitions: 'Approved', 'Declined', 'Resubmitted')
updated = client.sessions.update_status("sess_12345", new_status=SessionStatus.APPROVED)
print(f"Session status update acknowledged: {updated.session_id}")

7. Compliance PDF Reports

Download official verification audit PDF reports with strict binary header (%PDF-) and MIME (application/pdf) format validation:

from didit import Didit

client = Didit()

# Option A: Direct-to-disk secure download (atomic swap, POSIX 0600 private permissions)
report_path = client.sessions.download_pdf_report(
    "sess_12345",
    "reports/verification_report.pdf",
    force=True,
)

# Option B: In-memory raw binary bytes (60s default timeout tailored to upstream generation latencies)
pdf_bytes = client.sessions.generate_pdf_report("sess_12345")

Webhook Integrations & Tokenized Reservation Protocol

Didit dispatches signed HTTP POST events upon verification completion. didit-sdk provides native, production-grade adapters with streaming body limits (HTTP 413) and a crash-recoverable tokenized reservation protocol using Redis Lua CAS state machines:

• Active Lease (PROCESSING): Acquired via a unique cryptographic worker token with a short lease (default lease_ttl_seconds=30). If a worker crashes (SIGKILL, OOM, reboot), the lease automatically expires so Didit's retries can be recovered by another worker.
• Terminal Retention (COMPLETED): Upon a 2xx response, the reservation atomically transitions to COMPLETED with long retention (default completed_ttl_seconds=86400).
• Concurrent Protection (processing_action="retry"): If a second delivery arrives while processing is in flight, the adapter returns HTTP 503 with Retry-After: 5 so Didit retries after the lease expires.

FastAPI

from fastapi import FastAPI, APIRouter
from didit import WebhookPayload, SessionStatus, RedisWebhookReservationStore
from didit.integrations.fastapi import didit_webhook, DiditWebhookRoute

app = FastAPI()
store = RedisWebhookReservationStore.from_url("redis://localhost:6379/0")
router = APIRouter(route_class=DiditWebhookRoute)


@router.post("/webhooks/didit")
@didit_webhook(
    secret="whsec_...",
    dedup_store=store,
    lease_ttl_seconds=30,  # Lease while processing (auto-recovered on crash)
    completed_ttl_seconds=86400,  # 24h retention after 2xx completion
    duplicate_action="pass",  # At-least-once delivery to DB transaction boundary
    processing_action="retry",  # HTTP 503 with Retry-After: 5 for concurrent in-flight deliveries
)
async def handle_webhook(payload: WebhookPayload):
    # Enforce idempotency at the database/transaction boundary:
    async with db.transaction():
        if not await is_event_unprocessed(payload.event_id):
            return {"status": "already_processed"}

        if payload.status == SessionStatus.APPROVED:
            # Idempotently process approved KYC verification
            await mark_user_verified(payload.session_id)

    return {"status": "ok"}


app.include_router(router)

Alternatively, use DiditWebhookGuard as a FastAPI Depends(...) dependency. When using DiditWebhookGuard with a reservation store, the application is responsible for explicitly completing or releasing the reservation, or configuring route_class=DiditWebhookRoute on your router for automated lifecycle management:

from fastapi import APIRouter, Depends, Request
from didit.integrations.fastapi import DiditWebhookGuard, DiditWebhookRoute

router = APIRouter(route_class=DiditWebhookRoute)
guard = DiditWebhookGuard(secret="whsec_...", dedup_store=store)


@router.post("/webhooks/didit")
async def handle_webhook_guard(payload: WebhookPayload = Depends(guard)):
    return {"status": "ok"}


app.include_router(router)

When using DiditWebhookGuard on standard routers without DiditWebhookRoute, manage the lifecycle explicitly via await guard.complete_reservation(request) or await guard.release_claim(request).

Django

from django.http import HttpRequest, HttpResponse
from didit import WebhookPayload, SessionStatus, RedisWebhookReservationStore
from didit.integrations.django import didit_webhook_view

store = RedisWebhookReservationStore.from_url("redis://localhost:6379/0")


@didit_webhook_view(
    secret="whsec_...",
    dedup_store=store,
    lease_ttl_seconds=30,
    completed_ttl_seconds=86400,
)
def my_webhook_view(request: HttpRequest, payload: WebhookPayload) -> HttpResponse:
    if payload.status == SessionStatus.APPROVED:
        ...
    return HttpResponse(status=200)

Flask

from flask import Flask
from didit import WebhookPayload, SessionStatus, RedisWebhookReservationStore
from didit.integrations.flask import didit_webhook

app = Flask(__name__)
store = RedisWebhookReservationStore.from_url("redis://localhost:6379/0")


@app.route("/webhooks/didit", methods=["POST"])
@didit_webhook(
    secret="whsec_...",
    dedup_store=store,
    lease_ttl_seconds=30,
    completed_ttl_seconds=86400,
)
def handle_didit(payload: WebhookPayload):
    if payload.status == SessionStatus.APPROVED:
        ...
    return {"status": "ok"}, 200

Testing & Offline Simulation

Developing without live credentials or mock servers? Use the built-in simulator:

from didit import SimulatedDidit, SessionStatus

# Create in-memory client
sim = SimulatedDidit(webhook_secret="test_secret")

# 1. Create a simulated session
session = sim.sessions.create(vendor_data="dev_user", workflow_id="wf_dev")
assert session.session_id.startswith("sim_")

# 2. Simulate user passing identity check
sim.approve_session(session.session_id)

decision = sim.sessions.get_decision(session.session_id)
assert decision.status == SessionStatus.APPROVED

# 3. Generate signed webhook payloads to test your local receiver
raw_body, headers = sim.generate_webhook_event(session.session_id)
# headers contain valid X-Signature-V2 and fresh X-Timestamp!

Developer CLI (didit)

The SDK includes a security-hardened, script-friendly command-line interface (didit) for development, operations, and diagnostic audits.

Zero Argv Credentials

To prevent credential leakage through process inspection (ps aux, /proc), shell histories, or CI logs, the CLI never accepts API keys or webhook secrets via positional arguments or command flags. Pass credentials via environment variables or file paths:

• DIDIT_API_KEY or --api-key-file /path/to/key.txt
• DIDIT_WEBHOOK_SECRET or --secret-file /path/to/secret.txt

Universal --json Contract

Pass --json to any command for deterministic, machine-readable output:

• Success Envelope: {"status": "ok", ...} (for session resources, verification status is available under "session_status").
• Error Envelope: {"status": "error", "error": {"code": "...", "message": "..."}}.
• Non-zero exit code (1) on any error condition.

Commands

Diagnostics & Healthcheck (didit doctor)

Audit your network connectivity, API availability, and authentication credentials without querying or exposing customer KYC sessions:

didit doctor
# Or with structured JSON output
didit doctor --json

# In CI/CD pipelines (fails with non-zero exit code if healthcheck or latency probes are unavailable)
didit doctor --strict

Session Inspection & Lifecycle

By default, session output redacts sensitive bearer tokens and URLs. Supply --include-sensitive only when necessary:

# Get session details
didit session get sess_12345

# Create a session
didit session create --workflow-id wf_standard --vendor-data user_42

# List sessions with pagination
didit session list --status Approved --limit 20
didit session list --all --max-sessions 500

# Resubmit a session with explicit node and feature
didit session resubmit sess_12345 --node feature_ocr:OCR

# Download PDF report with atomic POSIX 0600 file permissions and overwrite guard
didit session pdf sess_12345 --output report.pdf
didit session pdf sess_12345 --output report.pdf --force

Webhook Signature Verification (didit webhook verify)

Verify Didit webhook signatures without exposing raw payload bytes in command arguments:

# Verify using body file
didit webhook verify \
  --secret-file /etc/secrets/didit_secret.txt \
  --signature "d8e8fca2dc64a51e6d1b7a2d4b6c8e9f0123456789abcdef0123456789abcdef" \
  --timestamp 1700000000 \
  --body-file payload.json

# Or verify from stdin in Unix pipelines
cat payload.json | didit webhook verify \
  --secret-file /etc/secrets/didit_secret.txt \
  --signature "d8e8fca2dc64a51e6d1b7a2d4b6c8e9f0123456789abcdef0123456789abcdef" \
  --timestamp 1700000000 \
  --stdin

# Bypass timestamp freshness check for historic replay audits (emits warning to stderr)
cat payload.json | didit webhook verify \
  --secret-file /etc/secrets/didit_secret.txt \
  --signature "d8e8fca2dc64a51e6d1b7a2d4b6c8e9f0123456789abcdef0123456789abcdef" \
  --timestamp 1600000000 \
  --stdin \
  --skip-freshness-check

Sandbox Scenarios Explorer (didit sandbox scenarios)

List official Didit sandbox simulation slugs with optional category filtering:

didit sandbox scenarios
didit sandbox scenarios --category decline

Exception Handling

All client errors inherit from DiditError:

from didit import Didit
from didit.errors import (
    DiditAuthenticationError,
    DiditNotFoundError,
    DiditRateLimitError,
    DiditServerError,
)

client = Didit()

try:
    decision = client.sessions.get_decision("non_existent_session")
except DiditNotFoundError:
    print("Session not found")
except DiditRateLimitError as exc:
    print(f"Rate limited. Retry after {exc.retry_after} seconds.")
except DiditAuthenticationError:
    print("Invalid API key")
except DiditServerError as exc:
    print(f"Didit server error ({exc.status_code}): {exc.response_body}")

Environment Variables

Development & Testing

# Clone the repository
git clone https://github.com/agustin18/didit-sdk.git
cd didit-sdk

# Install dependencies with uv
uv sync --extra dev

# Run unit tests with 100% coverage gate
uv run pytest --cov=didit --cov-report=term-missing

# Lint & Format check
uv run ruff check .
uv run ruff format --check .

# Strict type check
uv run mypy src/didit

License

This project is licensed under the terms of the MIT License.