Kanoniv Cloud

Kanoniv Cloud is the managed identity resolution platform. It persists your identity graph, scales to millions of records, provides health monitoring, and exposes a real-time resolution API. Everything the Core SDK does locally, Kanoniv Cloud does remotely with the capabilities that only a server can provide.

Why Cloud?

The Core SDK runs reconciliation in-process via a Rust engine. It is fast, requires no infrastructure, and is ideal for development, testing, and CI. The cloud SDK exists for when you need more.

What the cloud SDK adds:

Persistent identity graph. Canonical entities, links, and audit history are stored in the database. Query them at any time, not just during a reconciliation run.
Incremental resolution. Resolve records one at a time as they arrive via POST /v1/resolve/realtime. The endpoint uses Fellegi-Sunter scoring against the live graph - no full re-reconciliation needed. Bulk lookups resolve up to 1,000 pairs in a single request via the Redis reverse index.
AutoTune. Automatically optimize your identity spec. AutoTune runs candidate mutations through the reconciliation engine, scores results with z-score normalization, and accepts improvements that pass a conflict-rate guardrail. Works unsupervised or with labeled pairs for F1-driven optimization.
Cloud IDE. Spec Studio for editing and validating specs in the browser, an entity browser for exploring the identity graph, and a run explorer for inspecting job history and health.
Run health and diagnostics. Every job returns a health assessment (healthy / degraded / unhealthy) with signals and recommendations. The local engine returns raw results; the cloud interprets them.
Real-time resolution. Resolve an external ID to its canonical identity in a single API call (client.resolve()), backed by a live graph that updates with every reconciliation.
Warehouse connectors. Native Snowflake and Databricks connectors stream data via Arrow for ~6x faster uploads. dbt adapter reads directly from compiled models.
Entity lifecycle. Lock entities from merging, revert to prior states, view full change history, and manually override decisions through a review queue.
Job orchestration. Submit reconciliation jobs, monitor progress, cancel in flight. Or use kanoniv.cloud.reconcile() to do it all in one call with the same sources and spec you use locally.
Audit trail. Every merge, split, override, and revert is recorded with full provenance.

When to use which:

	Core (`kanoniv.reconcile`)	Cloud (`kanoniv.cloud.reconcile`)
Runs where	In-process (Rust via PyO3)	Kanoniv API server
State	Ephemeral; results live in memory	Persistent; queryable via the API at any time
Scale	Bounded by local memory	Server-managed, horizontally scalable
Auth required	No	Yes (API key)
Incremental resolve	N/A	Real-time + bulk via Redis reverse index
AutoTune	`kanoniv.autotune()` (local)	Cloud IDE + API
IDE	N/A	Spec Studio, entity browser, run explorer
Connectors	CSV, JSON, pandas	+ Snowflake, Databricks, dbt (Arrow fast path)
Health monitoring	N/A	Health status, flags, recommendations
Entity queries	N/A	Resolve, search, history, lock, revert
Best for	Dev, testing, CI, small datasets	Production, large datasets, ongoing resolution

Platform Architecture

Same Package, Different Install

This is the API client portion of the kanoniv package. Install with pip install kanoniv[cloud] to get Client, AsyncClient, and kanoniv.cloud. For offline spec tooling, pip install kanoniv is sufficient. See Python SDK (Core).

Installation

bash

pip install kanoniv[cloud]

Requirements: Python 3.9+. Installs httpx for HTTP transport and pydantic for response models.

Connectors

Kanoniv Cloud reads from your warehouse using native connectors for fast, columnar Arrow transfers. Install the connector for your platform:

Warehouse	Install	Connector	Connection String
Snowflake	`pip install kanoniv[cloud,dataplane]`	`snowflake-connector-python`	`snowflake://user:pass@account/db`
Databricks	`pip install kanoniv[cloud,databricks]`	`databricks-sql-connector`	`databricks://token:dapi...@host?http_path=...`

Each extra installs pyarrow, duckdb, and the native connector. When present, kanoniv.cloud.reconcile() automatically detects the connection string scheme and uses the Arrow fast path (~6x faster than row-by-row).

python

# Snowflake
source = Source.from_warehouse(
    "crm", "ANALYTICS.CUSTOMERS",
    connection_string="snowflake://user:pass@account/ANALYTICS",
    primary_key="customer_id",
)

# Databricks
source = Source.from_warehouse(
    "billing", "main.default.invoices",
    connection_string="databricks://token:[email protected]?http_path=/sql/1.0/warehouses/abc",
    primary_key="invoice_id",
)

Without a connector extra installed, kanoniv.cloud.reconcile() still works but falls back to row-by-row iteration via SQLAlchemy.

Quick Start

Go from raw data to a live identity graph in one script.

Sample Data

Grab the sample datasets from the test-dataset/ directory — 100 Salesforce contacts and 100 Stripe customers with ~40 overlapping identities.

python

import kanoniv
from kanoniv import Source, Spec

# 1. Define your sources — second arg is the path to the CSV file on disk
crm = Source.from_csv("salesforce", "test-dataset/sf_contacts.csv", primary_key="sf_id")
billing = Source.from_csv("stripe", "test-dataset/stripe_customers.csv", primary_key="cus_id")

# 2. Load your identity spec
spec = Spec.from_file("specs/customer.yaml")

# 3. Reconcile on the cloud
with kanoniv.cloud.reconcile([crm, billing], spec, api_key="kn_...") as result:
    print(result.summary())
    # Cloud Reconciliation — job a1b2c3d4-...
    #   Status:          completed
    #   Health:          healthy
    #   Input entities:  12,400
    #   Canonicals:      8,230
    #   Merge rate:      33.6%

The identity graph is now live. Query it at any time:

python

from kanoniv import Client

with Client(api_key="kn_...") as client:
    # Resolve a Salesforce contact to its canonical identity
    result = client.resolve(system="salesforce", external_id="003A000001abc")
    print(result["canonical_id"])  # "c7e2f9a1-..."

    # See every linked record across all systems
    linked = client.entities.get_linked(result["canonical_id"])
    for record in linked["linked"]:
        print(f"  {record['source_name']}: {record['external_id']}")
    # salesforce: 003A000001abc
    # stripe: cus_N4x8Kj

    # Search by email
    results = client.entities.search(q="[email protected]")
    for entity in results["data"]:
        print(entity["id"], entity["data"]["email"])

Real-World Scenarios

These are production patterns. Each scenario uses the full cloud SDK to solve a real identity resolution problem at scale.

Multi-Source Identity Stitching

The most common use case: you have customer data scattered across CRM, billing, support, and marketing systems. Each system has its own ID. The cloud builds a persistent identity graph that unifies them.

yaml

# specs/customer.yaml
api_version: kanoniv/v1
identity_version: customer-v1
entity:
  name: customer
sources:
  - name: salesforce
    attributes:
      email_address: email
      full_name: name
      phone: phone
  - name: stripe
    attributes:
      billing_email: email
      customer_name: name
  - name: zendesk
    attributes:
      requester_email: email
      requester_name: name
rules:
  - name: email_exact
    type: exact
    field: email
    weight: 1.0
  - name: name_fuzzy
    type: similarity
    field: name
    algorithm: jaro_winkler
    threshold: 0.88
    weight: 0.6
decision:
  thresholds:
    match: 0.85
    review: 0.65

python

import kanoniv
from kanoniv import Source, Spec

spec = Spec.from_file("specs/customer.yaml")

sources = [
    Source.from_csv("salesforce", "test-dataset/sf_contacts.csv", primary_key="sf_id"),
    Source.from_csv("stripe", "test-dataset/stripe_customers.csv", primary_key="cus_id"),
    Source.from_csv("zendesk", "data/zd_tickets.csv", primary_key="ticket_id"),
]

with kanoniv.cloud.reconcile(sources, spec, api_key="kn_...") as result:
    print(result.summary())
    # Cloud Reconciliation — job a1b2c3d4-...
    #   Status:          completed
    #   Duration:        12400ms
    #   Health:          healthy
    #   Input entities:  84,230
    #   Canonicals:      31,847
    #   Links:           84,230
    #   Merge rate:      62.2%

    # Every source entity now maps to a canonical customer
    df = result.to_pandas()
    print(f"{len(df)} unified customers from {84_230} source records")

    # Which canonical has the most linked records?
    linked = df.sort_values("link_count", ascending=False).head(1)
    print(f"Largest cluster: {linked.iloc[0]['id']} with {linked.iloc[0]['link_count']} records")

After reconciliation, the identity graph is live. Resolve any source ID to its canonical customer at any time:

python

from kanoniv import Client

with Client(api_key="kn_...") as client:
    # Salesforce contact → canonical customer
    result = client.resolve(system="salesforce", external_id="003A000001abc")
    print(result["canonical_id"])  # "c7e2f9a1-..."

    # See all linked records across all systems
    linked = client.entities.get_linked(result["canonical_id"])
    for record in linked["linked"]:
        print(f"  {record['source_name']}: {record['external_id']}")
    # salesforce: 003A000001abc
    # stripe: cus_N4x8Kj
    # zendesk: req_88291

Batch Processing at Scale

When you have millions of records, the cloud handles the heavy lifting. The cloud.reconcile() function batches data uploads automatically (5,000 records per request), but for very large datasets you control the pipeline directly for progress tracking and resilience.

python

import time
from kanoniv import Client, Source, Spec

spec = Spec.from_file("specs/customer.yaml")
client = Client(api_key="kn_...", timeout=60.0, max_retries=4)

# 1. Upload the spec
client.specs.ingest(spec.raw, compile=True)

# 2. Pull 5M records from the warehouse in streaming batches
source = Source.from_warehouse(
    "crm",
    "analytics.crm_contacts",
    connection_string="snowflake://user:pass@account/db/schema",
)

# 3. Upload in controlled batches with progress
BATCH = 1000
entities = source.to_entities("customer")
total = len(entities)
uploaded = 0

for i in range(0, total, BATCH):
    batch = entities[i : i + BATCH]
    client.ingest("crm", batch)
    uploaded += len(batch)
    if uploaded % 50_000 == 0:
        print(f"  Uploaded {uploaded:,} / {total:,} ({uploaded/total:.0%})")

print(f"Uploaded {total:,} records. Submitting reconciliation job...")

# 4. Submit the job
job = client.jobs.run("reconciliation")
job_id = job["id"]

# 5. Poll with progress
while True:
    status = client.jobs.get(job_id)
    state = status["status"]
    if state in ("completed", "failed", "cancelled"):
        break
    print(f"  Job {job_id}: {state}")
    time.sleep(5)

if state == "failed":
    print(f"Job failed: {status.get('error')}")
elif state == "completed":
    stats = status.get("result", {}).get("stats", {})
    print(f"Done. {stats.get('canonicals_created', 0):,} canonical identities created.")

client.close()

Data Warehouse to Kanoniv

Connect directly to Snowflake, BigQuery, Postgres, Databricks, or any SQLAlchemy-compatible warehouse. The warehouse adapter streams rows without loading the entire table into memory.

python

import kanoniv
from kanoniv import Source, Spec

spec = Spec.from_file("specs/patient.yaml")

# Snowflake
patients = Source.from_warehouse(
    "hospital_emr",
    "raw.emr_patients",
    connection_string="snowflake://user:pass@acct/HEALTHCARE/RAW",
)

# Databricks
claims = Source.from_warehouse(
    "insurance_claims",
    "main.default.member_records",
    connection_string="databricks://token:[email protected]?http_path=/sql/1.0/warehouses/abc",
)

# Postgres (your own operational DB)
support = Source.from_warehouse(
    "support_tickets",
    "public.tickets",
    connection_string="postgresql://user:pass@db-host:5432/support",
)

with kanoniv.cloud.reconcile(
    [patients, claims, support],
    spec,
    api_key="kn_...",
    timeout=600.0,      # large datasets need more time
    poll_interval=10.0,  # check less frequently
) as result:
    print(result.summary())

    if result.health_status != "healthy":
        print(f"Warnings: {result.health_flags}")

    df = result.to_pandas()
    print(f"{len(df)} canonical patients resolved from 3 warehouse sources")

dbt users: If your models are in dbt, pull directly from your compiled models:

python

from kanoniv import Source

# Reads the dbt manifest to find the compiled SQL, then streams results
customers = Source.from_dbt(
    "dbt_customers",
    model="stg_customers",
    manifest_path="target/manifest.json",
    connection_string="snowflake://user:pass@acct/ANALYTICS/PUBLIC",
)

Reverse ETL: Write Canonical IDs Back

After reconciliation, the canonical IDs need to flow back into your systems. Pull the identity graph from Kanoniv and write it to your warehouse, CRM, or any downstream system.

Write to your data warehouse:

python

import kanoniv
from kanoniv import Source, Spec
from sqlalchemy import create_engine, text

spec = Spec.from_file("specs/customer.yaml")
sources = [
    Source.from_warehouse("crm", "raw.sf_contacts", connection_string="snowflake://..."),
    Source.from_warehouse("billing", "raw.stripe_customers", connection_string="snowflake://..."),
]

# 1. Reconcile
with kanoniv.cloud.reconcile(sources, spec, api_key="kn_...") as result:
    df = result.to_pandas()

# 2. Write canonical mappings back to the warehouse
engine = create_engine("snowflake://user:pass@acct/ANALYTICS/PUBLIC")
with engine.begin() as conn:
    conn.execute(text("TRUNCATE TABLE identity.canonical_mappings"))

# source_name, external_id, canonical_id — one row per link
mapping_df = df[["source_name", "external_id", "id"]].rename(columns={"id": "canonical_id"})
mapping_df.to_sql("canonical_mappings", engine, schema="identity", if_exists="append", index=False)

print(f"Wrote {len(mapping_df):,} canonical mappings to Snowflake")

Update your CRM via API:

python

from kanoniv import Client
import requests

with Client(api_key="kn_...") as client:
    # For each Salesforce contact, stamp the canonical ID
    sf_entities = client.entities.search(entity_type="customer", limit=100)
    for entity in sf_entities["data"]:
        linked = client.entities.get_linked(entity["id"])
        for record in linked["linked"]:
            if record["source_name"] == "salesforce":
                # Write canonical_id back to Salesforce
                requests.patch(
                    f"https://your-instance.salesforce.com/services/data/v58.0/sobjects/Contact/{record['external_id']}",
                    json={"Canonical_ID__c": entity["id"]},
                    headers={"Authorization": "Bearer <sf_token>"},
                )

Real-Time Resolution Pipeline

Resolve identities as events arrive. This pattern works with webhooks, message queues, or any event-driven architecture.

The POST /v1/resolve/realtime endpoint uses Fellegi-Sunter probabilistic scoring - the same compiled FS plan from your batch spec - to match incoming records against the existing identity graph. It generates blocking keys, retrieves candidates, and applies the full scoring pipeline with normalizers (email, phone, name, nickname, domain) and comparators (exact, jaro_winkler, levenshtein, soundex, metaphone, cosine).

Two-threshold decision logic:

match_threshold (from your spec) - for linking a new record to an existing entity (lower bar)
merge_threshold (defaults to match * 1.5) - for merging existing entities when a new record bridges them (higher bar)

If no FS plan exists yet, the endpoint falls back to exact email/phone matching (backwards compatible).

FastAPI webhook endpoint:

python

from fastapi import FastAPI, Request
from kanoniv import Client

app = FastAPI()
client = Client(api_key="kn_...")

@app.post("/webhook/crm")
async def handle_crm_event(request: Request):
    payload = await request.json()

    # Real-time resolve with Fellegi-Sunter scoring.
    # Ingests the record, scores against existing canonical entities,
    # and links to an existing entity or creates a new one.
    result = client.resolve_rt.realtime(
        source_name="salesforce",
        external_id=payload["contact_id"],
        data={
            "email": payload["email"],
            "name": payload["name"],
            "phone": payload.get("phone", ""),
        },
    )

    return {
        "entity_id": result["entity_id"],
        "is_new": result["is_new"],
        "confidence": result["confidence"],
    }


@app.post("/webhook/stripe")
async def handle_stripe_event(request: Request):
    event = await request.json()
    if event["type"] == "customer.created":
        customer = event["data"]["object"]

        result = client.resolve_rt.realtime(
            source_name="stripe",
            external_id=customer["id"],
            data={
                "email": customer["email"],
                "name": customer.get("name", ""),
            },
        )

        if not result["is_new"]:
            linked = client.entities.get_linked(result["entity_id"])
            print(f"Stripe {customer['id']} linked to {len(linked['linked_entities'])} records")
            print(f"  Confidence: {result['confidence']}")

    return {"ok": True}

Bulk resolve - look up many source IDs at once:

python

from kanoniv import Client

with Client(api_key="kn_...") as client:
    result = client.resolve_rt.bulk([
        {"source": "salesforce", "id": "003A000001abc"},
        {"source": "stripe", "id": "cus_N4x8Kj"},
        {"source": "zendesk", "id": "req_88291"},
    ])

    print(f"Resolved: {result['resolved']}, Not found: {result['not_found']}")
    for r in result["results"]:
        if r["found"]:
            print(f"  {r['source']}:{r['id']} -> {r['entity_id']}")

Async batch resolution with concurrency:

python

import asyncio
from kanoniv import AsyncClient

async def resolve_realtime_batch(records: list[dict]):
    """Resolve a batch of records concurrently. Useful for processing
    a queue of events that arrived while offline."""
    async with AsyncClient(api_key="kn_...") as client:
        tasks = [
            client.resolve_rt.realtime(
                source_name=item["source"],
                external_id=item["id"],
                data=item["data"],
            )
            for item in records
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        resolved = 0
        for item, result in zip(records, results):
            if isinstance(result, Exception):
                print(f"  Failed {item['source']}:{item['id']}: {result}")
            else:
                resolved += 1

        print(f"Resolved {resolved}/{len(records)} identities")
        return results

# Process a backlog of events
backlog = [
    {"source": "salesforce", "id": "003A000001abc", "data": {"email": "[email protected]"}},
    {"source": "stripe", "id": "cus_N4x8Kj", "data": {"email": "[email protected]"}},
    {"source": "zendesk", "id": "req_88291", "data": {"email": "[email protected]"}},
]
asyncio.run(resolve_realtime_batch(backlog))

For large batches where you already have source+id pairs (and don't need to ingest new data), resolve_rt.bulk() is more efficient than calling resolve_rt.realtime() in a loop - it resolves up to 1000 lookups in a single request via the Redis reverse index.

Human-in-the-Loop Review Queue

When the engine finds a match that scores between the auto-merge and rejection thresholds, it lands in the review queue. Build a review workflow for your operations team.

python

from kanoniv import Client

with Client(api_key="kn_...") as client:
    # 1. Fetch pending reviews
    pending = client.reviews.list(limit=50)
    print(f"{len(pending)} reviews pending")

    for review in pending:
        entity_a = client.entities.get(review["entity_a_id"])
        entity_b = client.entities.get(review["entity_b_id"])

        print(f"\n--- Review (score: {review['score']:.2f}) ---")
        print(f"  A: {entity_a['data'].get('name', '?')} <{entity_a['data'].get('email', '?')}>")
        print(f"     Source: {entity_a.get('source_name')}")
        print(f"  B: {entity_b['data'].get('name', '?')} <{entity_b['data'].get('email', '?')}>")
        print(f"     Source: {entity_b.get('source_name')}")

        # In a real app, this comes from your UI or Slack bot
        decision = input("  Merge or reject? [m/r/s(kip)]: ").strip().lower()
        if decision == "m":
            client.reviews.decide(
                entity_a_id=review["entity_a_id"],
                entity_b_id=review["entity_b_id"],
                decision="merge",
                reason="Confirmed same person by ops team",
            )
            print("  -> Merged")
        elif decision == "r":
            client.reviews.decide(
                entity_a_id=review["entity_a_id"],
                entity_b_id=review["entity_b_id"],
                decision="reject",
                reason="Different people despite similar data",
            )
            print("  -> Rejected")
        else:
            print("  -> Skipped")

Force-merge or force-split when you know better than the engine:

python

with Client(api_key="kn_...") as client:
    # These two records are definitely the same person
    client.overrides.create(
        override_type="force_merge",
        entity_a_id="uuid-a",
        entity_b_id="uuid-b",
    )

    # These were wrongly merged — split them apart
    client.overrides.create(
        override_type="force_split",
        entity_a_id="uuid-c",
        entity_b_id="uuid-d",
    )

    # Check the audit trail to see what happened
    trail = client.audit.entity_trail("uuid-a")
    for event in trail:
        print(f"  {event['event_type']} at {event['created_at']}")

Cloud Reconciliation

kanoniv.cloud.reconcile() accepts the same sources and spec as the local reconcile(), but runs the job on the Kanoniv API. One function call handles everything: spec upload, data upload, job submission, polling, and result extraction.

python

import kanoniv
from kanoniv import Source, Spec

spec = Spec.from_file("specs/customer.yaml")
sources = [
    Source.from_csv("crm", "data/crm_contacts.csv"),
    Source.from_csv("billing", "data/billing_accounts.csv"),
]

# Local (runs in-process via Rust engine)
local_result = kanoniv.reconcile(sources, spec)

# Cloud (runs on the Kanoniv API, same inputs)
cloud_result = kanoniv.cloud.reconcile(sources, spec, api_key="kn_...")

Parameters

Parameter	Type	Default	Description
`sources`	`list[Source]`	required	Data sources to reconcile
`spec`	`Spec`	required	Identity spec defining rules and thresholds
`client`	`Client \| None`	`None`	Existing client instance (skips creating one)
`api_key`	`str \| None`	`None`	API key (used when `client` is not provided)
`base_url`	`str`	`"https://api.kanoniv.com"`	API base URL
`poll_interval`	`float`	`2.0`	Seconds between job-status polls
`timeout`	`float`	`300.0`	Maximum seconds to wait for job completion

What Happens Under the Hood

When you call kanoniv.cloud.reconcile(), five steps run automatically:

1. Build or reuse client. If you pass a client, it is used directly. Otherwise a new Client is created from api_key and base_url.

2. Upload and compile the spec. The raw YAML from spec.raw is sent to the server via client.specs.ingest(spec.raw, compile=True). The server validates, compiles, and stores the spec version.

3. Map attributes and collect entities. For each source, source.to_entities(entity_type) converts rows into entity dicts. If the spec defines attribute mappings (e.g. email_address in source maps to email canonical), they are applied so all sources share a common schema.

4. Upload entities in batches. Entities are uploaded per source in batches of 500 via client.ingest(source_name, batch). A source with 2,000 rows sends 4 requests.

5. Submit job, poll until done. A reconciliation job is submitted via client.jobs.run("reconciliation"). The function polls client.jobs.get(job_id) every poll_interval seconds until the job completes, fails, is cancelled, or timeout is reached.

python

# You can do the same steps manually if you need more control:
from kanoniv import Client, Source, Spec

spec = Spec.from_file("specs/customer.yaml")
source = Source.from_csv("crm", "data/contacts.csv")

with Client(api_key="kn_...", base_url="https://api.kanoniv.com") as client:
    # Step 2: upload spec
    client.specs.ingest(spec.raw, compile=True)

    # Step 4: upload entities
    entities = source.to_entities("contact")
    for i in range(0, len(entities), 500):
        client.ingest("crm", entities[i:i+500])

    # Step 5: run and poll
    job = client.jobs.run("reconciliation")
    # ... poll client.jobs.get(job["id"]) until completed

CloudReconcileResult

The cloud result contains aggregate summaries returned by the server. Raw clusters and golden records are persisted server-side and accessible at any time via the entity APIs.

Fields

Field	Type	Description
`job_id`	`str`	Unique identifier of the reconciliation job
`status`	`str`	Final job status (`"completed"`)
`canonicals_created`	`int`	Number of canonical identities produced
`links_created`	`int`	Number of entity-to-canonical links created
`duration_ms`	`int`	Server-side execution time in milliseconds
`identity_summary`	`dict`	Full summary dict from the server (input counts, output counts, clustering, match quality, survivorship, stability, health flags)
`run_health`	`dict`	Health assessment dict (status, signals, recommendations)

python

result = kanoniv.cloud.reconcile(sources, spec, api_key="kn_...")

print(result.job_id)             # "a1b2c3d4-..."
print(result.canonicals_created) # 1234
print(result.links_created)      # 5678
print(result.duration_ms)        # 4200

# Access the full server response dicts
print(result.identity_summary.keys())
# dict_keys(["input", "output", "clustering", "match_quality", "survivorship", "stability", "health_flags"])

print(result.run_health)
# {"status": "healthy", "signals": [...], "recommendations": [...]}

Properties

Computed from identity_summary and run_health for convenience.

Property	Type	Source	Description
`cluster_count`	`int`	`identity_summary.output.canonical_identities`	Number of canonical identities
`merge_rate`	`float`	`identity_summary.output.merge_rate`	Fraction of entities merged (0.0 to 1.0)
`match_quality`	`dict`	`identity_summary.match_quality`	Breakdown of accepted, rejected, and uncertain matches
`health_status`	`str`	`run_health.status`	`"healthy"`, `"degraded"`, or `"unhealthy"`
`health_flags`	`list[str]`	`identity_summary.health_flags`	Warning labels (e.g. `"low_match_rate"`, `"high_cluster_size"`)

python

result = kanoniv.cloud.reconcile(sources, spec, api_key="kn_...")

print(result.cluster_count)  # 1234
print(result.merge_rate)     # 0.42
print(result.health_status)  # "healthy"
print(result.health_flags)   # ["low_match_rate"]
print(result.match_quality)  # {"accepted_matches": 120, "rejected_matches": 8, ...}

# Use in conditional logic
if result.health_status != "healthy":
    print(f"Run degraded: {result.health_flags}")

Methods

summary() returns a human-readable string of the run.

python

print(result.summary())
# Cloud Reconciliation — job a1b2c3d4-...
#   Status:          completed
#   Duration:        4200ms
#   Health:          healthy
#   Input entities:  2000
#   Canonicals:      1234
#   Links:           5678
#   Merge rate:      42.0%
#   Match quality:   {"accepted_matches": 120, "rejected_matches": 8}
#   Health flags:    low_match_rate

to_pandas() fetches canonical entities from the API with automatic pagination and returns a pandas DataFrame. Requires pip install pandas.

python

df = result.to_pandas()
print(df.shape)       # (1234, 8)
print(df.columns)     # ["id", "entity_type", "data", "source_name", ...]
print(df.head())

# Filter or export
df[df["entity_type"] == "contact"].to_csv("contacts.csv")

close() closes the underlying HTTP connection if the result created it. Called automatically when used as a context manager. No-op if you passed your own Client.

Context Manager

CloudReconcileResult is a context manager. When you use api_key (no explicit client), the result owns the underlying connection and close() cleans it up:

python

import kanoniv

with kanoniv.cloud.reconcile(sources, spec, api_key="kn_...") as result:
    print(result.summary())
    df = result.to_pandas()
# Connection closed automatically

For scripts where cleanup is not critical, you can skip the with block:

python

result = kanoniv.cloud.reconcile(sources, spec, api_key="kn_...")
df = result.to_pandas()

Reusing an Existing Client

Pass a Client to share one connection across multiple calls. The result does not own or close a client you provide:

python

from kanoniv import Client
import kanoniv

with Client(api_key="kn_...", base_url="https://api.kanoniv.com") as client:
    result = kanoniv.cloud.reconcile(sources, spec, client=client)
    df = result.to_pandas()

    # Run a second reconciliation on the same connection
    result2 = kanoniv.cloud.reconcile(other_sources, other_spec, client=client)

    # Query entities directly
    entity = client.entities.get(df.iloc[0]["id"])

Error Handling

Exception	When
`ValueError`	Neither `client` nor `api_key` provided
`TimeoutError`	Job did not complete within `timeout` seconds. The error message includes the `job_id` so you can check status later.
`RuntimeError`	Job failed or was cancelled on the server. The error message includes the server error or cancellation notice.

python

import kanoniv

try:
    result = kanoniv.cloud.reconcile(sources, spec, api_key="kn_...")
except TimeoutError as e:
    # Extract the job_id from the message and check later
    print(e)
    # "Cloud reconciliation job abc-123 did not complete within 300.0s.
    #  Check status with client.jobs.get('abc-123')."
except RuntimeError as e:
    # Server-side failure
    print(e)
    # "Cloud reconciliation job abc-123 failed: out of memory"

For long-running jobs, increase timeout or lower poll_interval:

python

result = kanoniv.cloud.reconcile(
    sources, spec,
    api_key="kn_...",
    timeout=600.0,       # wait up to 10 minutes
    poll_interval=5.0,   # check every 5 seconds
)

Autodetect

The autodetect endpoint profiles your ingested data and returns detected identity signals, blocking keys, and an inferred entity type. Use it to understand what the engine sees in your data before writing a spec or bootstrapping an identity plan.

Endpoint: POST /v1/autodetect

Request

Field	Type	Required	Default	Description
`sample_size`	`int`	No	`500`	Number of rows to sample for profiling
`bootstrap`	`bool`	No	`false`	If `true`, generate and persist an identity plan from detected signals
`identity_version`	`string`	No	-	Version label for the bootstrapped plan (used when `bootstrap` is `true`)
`entity_type`	`string`	No	-	Entity type hint: `"person"`, `"company"`, `"product"`, `"transaction"`, or `"healthcare"`. If omitted, the profiler infers the entity type from the data.

python

import httpx

resp = httpx.post(
    "https://api.kanoniv.com/v1/autodetect",
    headers={"X-API-Key": "kn_..."},
    json={"sample_size": 1000},
)
result = resp.json()
print(result["inferred_entity_type"])  # "product"
print(result["identity_signals"])       # ["sku", "upc", "product_name", "brand"]

Response

Field	Type	Description
`total_records`	`int`	Total records in the tenant's data
`rows_sampled`	`int`	Number of rows actually sampled
`sources`	`list[str]`	Source names found in the sampled data
`columns`	`list[object]`	Per-column profiling results (signal, uniqueness, null_rate, cardinality, sample_values)
`blocking_keys`	`list[list[str]]`	Inferred blocking key combinations
`identity_signals`	`list[str]`	Detected identity-relevant signals (e.g. `["email", "phone", "full_name"]`)
`inferred_entity_type`	`string`	The entity type inferred from the data: `"person"`, `"company"`, `"product"`, `"transaction"`, or `"healthcare"`
`bootstrapped_version`	`string \| null`	Version label of the bootstrapped plan, if `bootstrap` was `true`

The inferred_entity_type field determines which signal catalog the profiler uses. Each entity type has its own set of expected signals and blocking key templates. See Entity Type Detection for more on how inference works.

Entity Type Hint

When you provide an entity_type in the request, the profiler skips inference and uses that type's signal catalog directly. This is useful when:

Your data contains signals for multiple entity types and you want to force a specific interpretation
The inferred type is wrong (e.g. a dataset with both person names and product SKUs)
You want to use a specific set of blocking key templates

python

resp = httpx.post(
    "https://api.kanoniv.com/v1/autodetect",
    headers={"X-API-Key": "kn_..."},
    json={"entity_type": "healthcare", "sample_size": 500},
)

Bootstrap

When bootstrap is true, the endpoint generates a full identity plan from the detected signals and persists it. The plan includes matching rules, blocking keys, and Fellegi-Sunter scoring parameters calibrated to the data distribution. The bootstrapped_version field in the response contains the version label.

python

resp = httpx.post(
    "https://api.kanoniv.com/v1/autodetect",
    headers={"X-API-Key": "kn_..."},
    json={"bootstrap": true, "identity_version": "v1"},
)
print(resp.json()["bootstrapped_version"])  # "v1"

Authentication

python

from kanoniv import Client

# API key (recommended for programmatic use)
client = Client(api_key="kn_abc123", base_url="https://api.kanoniv.com")

# JWT bearer token (for interactive sessions)
client = Client(access_token="eyJ...", base_url="https://api.kanoniv.com")

Context Manager

Both sync and async clients support context managers for automatic cleanup:

python

from kanoniv import Client

with Client(api_key="kn_...", base_url="https://api.kanoniv.com") as client:
    result = client.resolve(system="stripe", external_id="cus_123")

Async Usage

All methods are available in both sync and async variants:

python

import asyncio
from kanoniv import AsyncClient

async def main():
    async with AsyncClient(api_key="kn_...", base_url="https://api.kanoniv.com") as client:
        result = await client.resolve(system="crm", external_id="sf_123")
        entities = await client.entities.search(q="jane")
        await client.ingest("src-id", records=[{"id": "1", "name": "Jane"}])  # upload for resolution

asyncio.run(main())

Configuration

python

from kanoniv import Client

client = Client(
    api_key="kn_...",                            # API key authentication
    base_url="https://api.kanoniv.com",          # default; override for self-hosted
    timeout=30.0,                                # request timeout in seconds (default)
    max_retries=2,                               # retry on 429/5xx errors (default)
)

Parameter	Type	Default	Description
`api_key`	`str \| None`	`None`	API key (sets `X-API-Key` header)
`access_token`	`str \| None`	`None`	JWT bearer token (alternative to `api_key`)
`base_url`	`str`	`"https://api.kanoniv.com"`	API base URL. Override for self-hosted deployments.
`timeout`	`float`	`30.0`	Request timeout in seconds
`max_retries`	`int`	`2`	Number of retries on transient errors

The client retries on transient errors (408, 429, 502, 503, 504) with exponential backoff. The Retry-After header is respected for 429 responses.

API Reference

Every method below is available on both Client (sync) and AsyncClient (async, prefix calls with await). All methods return parsed JSON as dict or list[dict] unless noted otherwise. HTTP errors raise typed exceptions (see Error Handling).

Top-Level Methods

These are convenience methods on the client itself, not on a sub-resource.

`client.resolve()`

Resolve an identity by system + external ID, or search by free-text query.

Parameter	Type	Required	Description
`system`	`str \| None`	no	Source system name (e.g. `"salesforce"`)
`external_id`	`str \| None`	no	External ID within the system
`query`	`str \| None`	keyword-only	Free-text search (e.g. an email address)

Returns: dict with canonical_id, matched entity data, and confidence score.

Endpoint: GET /v1/resolve

python

# By system + external ID
result = client.resolve(system="salesforce", external_id="003xxx")
print(result["canonical_id"])

# By free-text query
result = client.resolve(query="[email protected]")

`client.ingest()`

Upload records for cloud reconciliation. Used internally by cloud.reconcile() or directly for batch pipelines.

Parameter	Type	Required	Description
`source_id`	`str`	yes	Source UUID or name
`records`	`list[dict]`	yes	List of record dicts to upload

Returns: dict with upload status and count.

Endpoint: POST /v1/ingest/webhook/{source_id}

python

client.ingest("source-uuid", records=[
    {"id": "ext_1", "name": "John", "email": "[email protected]"},
    {"id": "ext_2", "name": "Jane", "email": "[email protected]"},
])

`client.ingest_file()`

Upload a file for cloud reconciliation.

Parameter	Type	Required	Description
`source_id`	`str`	yes	Source UUID or name
`path`	`str \| Path`	yes	Path to the file (string or `pathlib.Path`)

Returns: dict with processing status and record count.

Endpoint: POST /v1/ingest/file/process

python

client.ingest_file("source-uuid", path="data/contacts.csv")

from pathlib import Path
client.ingest_file("source-uuid", path=Path("data/contacts.csv"))

`client.ingest_parquet()`

Upload a Parquet file for bulk ingestion. Uses content hashing for idempotent ingest - unchanged records are skipped. This is the upload method used by the Arrow data plane.

Parameter	Type	Required	Description
`source_name`	`str`	yes	Source system name
`path`	`str \| Path`	yes	Path to a local `.parquet` file
`entity_type`	`str`	no	Entity type label (default: `"entity"`)

entity_type is keyword-only.

Returns: dict with new, updated, and unchanged counts.

Endpoint: POST /v1/ingest/parquet

python

result = client.ingest_parquet("crm", "data/crm.parquet", entity_type="customer")
print(f"New: {result['new']}, Updated: {result['updated']}, Unchanged: {result['unchanged']}")

from pathlib import Path
client.ingest_parquet("billing", Path("data/billing.parquet"))

`client.stats()`

Get dashboard statistics.

Returns: dict with entity counts, source counts, and job summaries.

Endpoint: GET /v1/stats

python

stats = client.stats()
print(f"{stats['total_canonical_entities']} canonical entities")

Entities: `client.entities`

Manage canonical entities, search the identity graph, view history, and control entity lifecycle.

`entities.search()`

Search entities with optional filters and pagination.

Parameter	Type	Required	Description
`q`	`str \| None`	no	Free-text search query
`entity_type`	`str \| None`	no	Filter by entity type (e.g. `"contact"`)
`limit`	`int \| None`	no	Max results to return
`offset`	`int \| None`	no	Number of results to skip (for pagination)

All parameters are keyword-only.

Returns: dict with data (list of entities) and pagination metadata.

Endpoint: GET /v1/entities

python

page = client.entities.search(q="[email protected]", entity_type="contact", limit=20, offset=0)
for entity in page["data"]:
    print(entity["id"], entity["data"]["email"])

`entities.get()`

Get a canonical entity by ID.

Parameter	Type	Required	Description
`id`	`str`	yes	Canonical entity UUID

Returns: dict with entity fields, data, source links, and metadata.

Endpoint: GET /v1/canonical/{id}

python

entity = client.entities.get("canonical-uuid")
print(entity["data"]["name"])

`entities.get_linked()`

Get a canonical entity with all linked external entities.

Parameter	Type	Required	Description
`id`	`str`	yes	Canonical entity UUID

Returns: dict with the canonical entity and a linked array of source entities.

Endpoint: GET /v1/canonical/{id}/linked

python

result = client.entities.get_linked("canonical-uuid")
for linked in result["linked"]:
    print(f"  {linked['source_name']}: {linked['external_id']}")

`entities.get_linked_bulk()`

Fetch linked entities for multiple canonical IDs in a single request. Accepts up to 1000 IDs. Used by the Arrow data plane to build crosswalk tables efficiently.

Parameter	Type	Required	Description
`entity_ids`	`list[str]`	yes	Canonical entity UUIDs (max 1000)

Returns: dict with results mapping each entity ID to its linked records.

Endpoint: POST /v1/entities/linked/bulk

python

result = client.entities.get_linked_bulk(["uuid-1", "uuid-2", "uuid-3"])
for entity_id, linked in result["results"].items():
    for record in linked:
        print(f"  {entity_id} -> {record['source_name']}: {record['external_id']}")

`entities.history()`

Get the change history of an entity.

Parameter	Type	Required	Description
`id`	`str`	yes	Entity UUID

Returns: dict with a list of audit events (merges, splits, reverts, data changes).

Endpoint: GET /v1/entities/{id}/history

python

history = client.entities.history("entity-uuid")
for event in history["events"]:
    print(f"{event['event_type']} at {event['created_at']}")

`entities.lock()`

Lock an entity to prevent further merging.

Parameter	Type	Required	Description
`id`	`str`	yes	Entity UUID

Returns: dict confirming the lock status.

Endpoint: POST /v1/entities/{id}/lock

python

client.entities.lock("entity-uuid")

`entities.revert()`

Revert an entity to a prior state.

Parameter	Type	Required	Description
`id`	`str`	yes	Entity UUID
`event_id`	`str`	yes	Audit event UUID to revert to

Returns: dict with the reverted entity state.

Endpoint: POST /v1/entities/{id}/revert/{event_id}

python

# Revert to the state before a specific merge
client.entities.revert("entity-uuid", "event-uuid")

Real-Time Resolution: `client.resolve_rt`

Resolve records against the live identity graph in real time. These are Cloud-only endpoints - the local SDK uses kanoniv.reconcile() for batch processing instead.

`resolve_rt.realtime()`

Resolve a single record against the identity graph. Ingests the record, matches it against existing canonical entities using the compiled Fellegi-Sunter plan, and returns the resolved entity. Creates a new canonical entity if no match is found.

Parameter	Type	Required	Description
`source_name`	`str`	yes	Source system name (e.g. `"crm"`)
`external_id`	`str`	yes	Unique ID of this record in the source system
`data`	`dict`	yes	Record fields to match on (e.g. `{"email": "...", "name": "..."}`)

All parameters are keyword-only.

Returns: dict with entity_id, canonical_data, is_new, matched_source, and confidence.

Endpoint: POST /v1/resolve/realtime

python

result = client.resolve_rt.realtime(
    source_name="crm",
    external_id="sf_123",
    data={"email": "[email protected]", "name": "Alice Smith", "phone": "555-0101"},
)
print(result["entity_id"])      # "c7e2f9a1-..."
print(result["is_new"])         # True if new entity created
print(result["confidence"])     # 1.0 for exact match, lower for probabilistic
print(result["canonical_data"]) # {"email": "[email protected]", "name": "alice smith"}

`resolve_rt.bulk()`

Batch resolve multiple source+id pairs against the identity graph. Uses the Redis reverse index for sub-millisecond lookups, falling back to the database. This is a read-only lookup - it does not ingest new data.

Parameter	Type	Required	Description
`lookups`	`list[dict]`	yes	List of `{"source": "...", "id": "..."}` dicts. Maximum 1000 per request.

Returns: dict with results (list of resolved entries), resolved count, and not_found count.

Endpoint: POST /v1/resolve/bulk

python

result = client.resolve_rt.bulk([
    {"source": "salesforce", "id": "003A000001abc"},
    {"source": "stripe", "id": "cus_N4x8Kj"},
    {"source": "zendesk", "id": "req_88291"},
])
print(f"Resolved: {result['resolved']}, Not found: {result['not_found']}")

for r in result["results"]:
    if r["found"]:
        print(f"  {r['source']}:{r['id']} -> {r['entity_id']}")

Sources: `client.sources`

Create, configure, and manage data sources. Sources define where entity data comes from and how fields map to the canonical schema.

`sources.list()`

List all sources.

Returns: list[dict] of source objects.

Endpoint: GET /v1/sources

python

sources = client.sources.list()
for src in sources:
    print(f"{src['name']} ({src['source_type']})")

`sources.get()`

Get a source by ID.

Parameter	Type	Required	Description
`id`	`str`	yes	Source UUID

Returns: dict with source config, status, and metadata.

Endpoint: GET /v1/sources/{id}

python

source = client.sources.get("source-uuid")
print(source["name"], source["config"])

`sources.create()`

Create a new source.

Parameter	Type	Required	Description
`name`	`str`	yes	Display name
`source_type`	`str`	yes	Source type (e.g. `"webhook"`, `"file"`)
`config`	`dict \| None`	no	Source-specific configuration
`**kwargs`	`Any`	no	Additional fields passed to the API body

All parameters are keyword-only.

Returns: dict with the created source, including its id.

Endpoint: POST /v1/sources

python

source = client.sources.create(
    name="CRM Import",
    source_type="webhook",
    config={"dedupe_key": "email"},
)
print(source["id"])

`sources.update()`

Update an existing source.

Parameter	Type	Required	Description
`id`	`str`	yes	Source UUID (positional)
`name`	`str \| None`	no	New display name
`config`	`dict \| None`	no	New configuration
`**kwargs`	`Any`	no	Additional fields passed to the API body

Returns: dict with the updated source.

Endpoint: PUT /v1/sources/{id}

python

client.sources.update("source-uuid", name="CRM v2", config={"dedupe_key": "id"})

`sources.delete()`

Delete a source.

Parameter	Type	Required	Description
`id`	`str`	yes	Source UUID

Returns: None

Endpoint: DELETE /v1/sources/{id}

python

client.sources.delete("source-uuid")

`sources.sync()`

Trigger a data sync for a source.

Parameter	Type	Required	Description
`id`	`str`	yes	Source UUID

Returns: dict with sync job status.

Endpoint: POST /v1/sources/{id}/sync

python

result = client.sources.sync("source-uuid")
print(result["status"])

`sources.preview()`

Preview source data and detected schema.

Parameter	Type	Required	Description
`id`	`str`	yes	Source UUID

Returns: dict with sample rows and detected schema.

Endpoint: GET /v1/ingest/sources/{id}/preview

python

preview = client.sources.preview("source-uuid")
for row in preview["rows"][:5]:
    print(row)

`sources.get_mapping()`

Get field mapping configuration for a source.

Parameter	Type	Required	Description
`id`	`str`	yes	Source UUID

Returns: dict with field-to-canonical mappings.

Endpoint: GET /v1/ingest/sources/{id}/mapping

python

mapping = client.sources.get_mapping("source-uuid")
print(mapping["mappings"])  # {"email_address": "email", "full_name": "name"}

`sources.upsert_mapping()`

Create or update field mapping for a source.

Parameter	Type	Required	Description
`mapping`	`dict`	yes	Mapping object with `source_id` and `mappings`

Returns: dict with the saved mapping.

Endpoint: POST /v1/ingest/sources/mapping

python

client.sources.upsert_mapping({
    "source_id": "source-uuid",
    "mappings": {"email_address": "email", "full_name": "name"},
})

Rules: `client.rules`

Manage matching rules. Rules are versioned; creating a rule with the same name creates a new version.

`rules.list()`

List active rules (latest version of each).

Returns: list[dict] of rule objects.

Endpoint: GET /v1/rules

python

rules = client.rules.list()
for rule in rules:
    print(f"{rule['name']} (v{rule['version']}): weight={rule['weight']}")

`rules.create()`

Create a new rule version.

Parameter	Type	Required	Description
`name`	`str`	yes	Rule name (creates a new version if the name already exists)
`rule_type`	`str`	yes	Match type (e.g. `"exact"`, `"similarity"`, `"range"`)
`config`	`dict`	yes	Rule-specific config (e.g. `{"field": "email"}`)
`weight`	`float \| None`	no	Scoring weight for this rule
`**kwargs`	`Any`	no	Additional fields passed to the API body

All parameters are keyword-only.

Returns: dict with the created rule version.

Endpoint: POST /v1/rules

python

rule = client.rules.create(
    name="email_exact",
    rule_type="exact",
    config={"field": "email"},
    weight=1.0,
)
print(f"Created rule v{rule['version']}")

`rules.history()`

Get all versions of a rule.

Parameter	Type	Required	Description
`name`	`str`	yes	Rule name

Returns: list[dict] of rule versions, newest first.

Endpoint: GET /v1/rules/{name}/history

python

versions = client.rules.history("email_exact")
for v in versions:
    print(f"v{v['version']} created {v['created_at']}")

Jobs: `client.jobs`

Submit, monitor, and cancel reconciliation jobs.

`jobs.list()`

List jobs with optional filters.

Parameter	Type	Required	Description
`limit`	`int \| None`	no	Max results to return
`job_type`	`str \| None`	no	Filter by type (e.g. `"reconciliation"`)

All parameters are keyword-only.

Returns: list[dict] of job objects with status, timestamps, and results.

Endpoint: GET /v1/jobs

python

jobs = client.jobs.list(limit=10, job_type="reconciliation")
for job in jobs:
    print(f"{job['id']}: {job['status']}")

`jobs.get()`

Get job details and status.

Parameter	Type	Required	Description
`id`	`str`	yes	Job UUID

Returns: dict with status, timestamps, result data (if completed), and error (if failed).

Endpoint: GET /v1/jobs/{id}

python

job = client.jobs.get("job-uuid")
print(job["status"])  # "pending", "running", "completed", "failed", "cancelled"

`jobs.run()`

Submit a new job for execution.

Parameter	Type	Required	Description
`job_type`	`str`	yes	Job type (e.g. `"reconciliation"`)
`payload`	`dict \| None`	no	Additional job parameters

Returns: dict with id, initial status, and timestamps.

Endpoint: POST /v1/jobs/run

python

job = client.jobs.run("reconciliation")
print(f"Started job {job['id']}")

# With options
job = client.jobs.run("reconciliation", payload={"dry_run": True})

`jobs.cancel()`

Cancel a running job.

Parameter	Type	Required	Description
`id`	`str`	yes	Job UUID

Returns: dict with updated job status.

Endpoint: POST /v1/jobs/{id}/cancel

python

client.jobs.cancel("job-uuid")

Reviews: `client.reviews`

Human-in-the-loop review queue for uncertain matches. When the engine produces matches below the auto-merge threshold but above the rejection threshold, they land here for manual review.

`reviews.list()`

List pending review items.

Parameter	Type	Required	Description
`limit`	`int \| None`	no	Max results to return
`offset`	`int \| None`	no	Number of results to skip

All parameters are keyword-only.

Returns: list[dict] of review items, each containing the two candidate entities and match metadata.

Endpoint: GET /v1/resolve/pending

python

reviews = client.reviews.list(limit=20, offset=0)
for review in reviews:
    print(f"{review['entity_a_id']} <-> {review['entity_b_id']} ({review['score']})")

`reviews.decide()`

Submit a merge/reject decision on a review item.

Parameter	Type	Required	Description
`entity_a_id`	`str`	yes	First entity UUID
`entity_b_id`	`str`	yes	Second entity UUID
`decision`	`str`	yes	`"merge"` or `"reject"`
`reason`	`str \| None`	no	Optional human-readable justification

All parameters are keyword-only.

Returns: dict confirming the decision.

Endpoint: POST /v1/resolve/quick

python

client.reviews.decide(
    entity_a_id="uuid-a",
    entity_b_id="uuid-b",
    decision="merge",
    reason="Same person, confirmed by support team",
)

Overrides: `client.overrides`

Manual overrides let you force-merge or force-split entities, bypassing the engine's automated decisions.

`overrides.list()`

List all active overrides.

Returns: list[dict] of override objects.

Endpoint: GET /v1/overrides

python

overrides = client.overrides.list()
for o in overrides:
    print(f"{o['override_type']}: {o['entity_a_id']} <-> {o['entity_b_id']}")

`overrides.create()`

Create a new override.

Parameter	Type	Required	Description
`override_type`	`str`	yes	`"force_merge"` or `"force_split"`
`entity_a_id`	`str`	yes	First entity UUID
`entity_b_id`	`str`	yes	Second entity UUID
`**kwargs`	`Any`	no	Additional fields (e.g. `reason`) passed to the API body

All parameters are keyword-only.

Returns: dict with the created override, including its id.

Endpoint: POST /v1/overrides

python

override = client.overrides.create(
    override_type="force_merge",
    entity_a_id="uuid-a",
    entity_b_id="uuid-b",
)
print(f"Override {override['id']} created")

`overrides.delete()`

Delete an override.

Parameter	Type	Required	Description
`id`	`str`	yes	Override UUID

Returns: None

Endpoint: DELETE /v1/overrides/{id}

python

client.overrides.delete("override-uuid")

Audit: `client.audit`

Query the immutable audit trail. Every merge, split, override, revert, and data change is recorded.

`audit.list()`

List audit events with optional filters.

Parameter	Type	Required	Description
`limit`	`int \| None`	no	Max results to return
`offset`	`int \| None`	no	Number of results to skip
`event_type`	`str \| None`	no	Filter by type (e.g. `"merge"`, `"split"`, `"override"`, `"revert"`)

All parameters are keyword-only.

Returns: list[dict] of audit events with timestamps, actors, and event data.

Endpoint: GET /v1/audit

python

events = client.audit.list(limit=50, event_type="merge")
for event in events:
    print(f"{event['event_type']} at {event['created_at']}: {event['entity_id']}")

`audit.entity_trail()`

Get the complete audit trail for a specific entity.

Parameter	Type	Required	Description
`entity_id`	`str`	yes	Entity UUID

Returns: list[dict] of all events affecting this entity, in chronological order.

Endpoint: GET /v1/audit/entity/{entity_id}

python

trail = client.audit.entity_trail("entity-uuid")
for event in trail:
    print(f"{event['event_type']}: {event['details']}")

Feedback: `client.feedback`

Store labeled match/no-match pairs for active learning. Labels persist server-side and can be retrieved for audit or future use.

Labels are stored but not yet auto-applied

Feedback labels uploaded via this API are persisted for record-keeping but are not yet automatically applied during cloud reconciliation. To use feedback labels for scoring refinement, run the active learning loop locally with kanoniv.reconcile(sources, spec, feedback=labels), then upload the labels to the cloud for record-keeping.

`feedback.list()`

List feedback labels for the current tenant.

Parameter	Type	Required	Description
`limit`	`int \| None`	no	Max results to return
`offset`	`int \| None`	no	Number of results to skip

All parameters are keyword-only.

Returns: list[dict] of feedback label objects.

Endpoint: GET /v1/feedback

python

labels = client.feedback.list(limit=100)
for label in labels:
    print(f"{label['entity_a_id']} <-> {label['entity_b_id']}: {label['label']}")

`feedback.create()`

Create feedback labels in batch. Each label describes a human judgment about whether two records are the same entity.

Parameter	Type	Required	Description
`labels`	`list[dict]`	yes	List of label dicts (see below)

labels is keyword-only.

Each label dict should contain:

Field	Type	Description
`entity_a_id`	`str`	Record ID from the first source
`entity_b_id`	`str`	Record ID from the second source
`source_a`	`str`	Source system name for record A
`source_b`	`str`	Source system name for record B
`label`	`str`	`"match"` or `"no_match"`
`reason`	`str`	Optional human-readable justification

Returns: list[dict] of created label objects with server-assigned IDs.

Endpoint: POST /v1/feedback

python

created = client.feedback.create(labels=[
    {
        "entity_a_id": "cust_123",
        "entity_b_id": "cust_456",
        "source_a": "crm",
        "source_b": "billing",
        "label": "match",
        "reason": "Same person, confirmed by support team",
    },
    {
        "entity_a_id": "cust_789",
        "entity_b_id": "cust_012",
        "source_a": "crm",
        "source_b": "billing",
        "label": "no_match",
    },
])
print(f"Created {len(created)} feedback labels")

`feedback.delete()`

Delete a feedback label by ID.

Parameter	Type	Required	Description
`id`	`str`	yes	Feedback label UUID

Returns: None

Endpoint: DELETE /v1/feedback/{id}

python

client.feedback.delete("label-uuid")

Specs: `client.specs`

Upload, validate, and version identity specs. Each upload creates a new version; the server always uses the latest compiled version for reconciliation.

`specs.list()`

List all spec versions.

Returns: list[dict] of spec versions with version numbers, timestamps, and validity status.

Endpoint: GET /v1/identity/specs

python

versions = client.specs.list()
for v in versions:
    print(f"v{v['version']} ({v['status']}): uploaded {v['created_at']}")

`specs.get()`

Get a specific spec version.

Parameter	Type	Required	Description
`version`	`int \| str`	yes	Version number (positional)

Returns: dict with the spec YAML content, validation results, and metadata.

Endpoint: GET /v1/identity/specs/{version}

python

spec = client.specs.get(1)
print(spec["spec_yaml"])

`specs.ingest()`

Upload and validate a spec. Optionally compile it for immediate use.

Parameter	Type	Required	Description
`yaml_content`	`str`	yes	Raw YAML string of the identity spec
`compile`	`bool`	no	If `True`, compile the spec for use in reconciliation. Default: `False`.

compile is keyword-only.

Returns: dict with valid (bool), plan_hash, validation errors (if any), and the new version number.

Endpoint: POST /v1/identity/specs

python

result = client.specs.ingest(open("spec.yaml").read(), compile=True)
if result["valid"]:
    print(f"Spec v{result['version']} compiled, plan hash: {result['plan_hash']}")
else:
    for err in result["errors"]:
        print(f"  {err}")

Error Handling

Every HTTP error from the API is mapped to a typed Python exception. All exceptions inherit from KanonivError, so you can catch broadly or narrowly.

python

from kanoniv.exceptions import (
    KanonivError,
    AuthenticationError,
    ForbiddenError,
    NotFoundError,
    ValidationError,
    ConflictError,
    RateLimitError,
    ServerError,
)

Exception Hierarchy

KanonivError                 (base for all SDK errors)
├── ValidationError          (400 Bad Request)
├── AuthenticationError      (401 Unauthorized)
├── ForbiddenError           (403 Forbidden)
├── NotFoundError            (404 Not Found)
├── ConflictError            (409 Conflict)
├── RateLimitError           (429 Too Many Requests)
└── ServerError              (5xx Server Error)

Exception Attributes

All exceptions carry context from the HTTP response.

`KanonivError` (base class)

Attribute	Type	Description
`message`	`str`	Human-readable error message (also the string representation)
`status_code`	`int \| None`	HTTP status code (`None` for connection errors)
`body`	`Any`	Raw response body (parsed JSON dict or string)

python

try:
    client.entities.get("nonexistent")
except KanonivError as e:
    print(e)              # "Not found"
    print(e.status_code)  # 404
    print(e.body)         # {"error": "Not found", "detail": "..."}

`RateLimitError` (extends `KanonivError`)

Has one additional attribute:

Attribute	Type	Description
`retry_after`	`float \| None`	Seconds to wait before retrying (from the `Retry-After` header)

python

try:
    client.entities.search(q="john")
except RateLimitError as e:
    print(e.retry_after)  # 2.0
    time.sleep(e.retry_after or 5)
    # retry...

HTTP Status Mapping

Exception	HTTP Status	When
`ValidationError`	400	Invalid request body, missing required fields, malformed parameters
`AuthenticationError`	401	Missing API key, expired token, invalid credentials
`ForbiddenError`	403	Valid credentials but insufficient permissions for the resource
`NotFoundError`	404	Entity, source, job, or other resource does not exist
`ConflictError`	409	Duplicate resource, state conflict (e.g. cancelling a completed job)
`RateLimitError`	429	Too many requests; `retry_after` tells you when to retry
`ServerError`	5xx	Server-side error; safe to retry

Automatic Retries

The transport layer retries automatically on transient errors. You do not need to implement retry logic for these status codes.

Retried Status	Meaning
408	Request Timeout
429	Too Many Requests (respects `Retry-After` header)
502	Bad Gateway
503	Service Unavailable
504	Gateway Timeout

Retries use exponential backoff (0.5s, 1s, 2s, ...) up to max_retries (default: 2). For 429 responses, the Retry-After header overrides the backoff delay.

Connection errors (ConnectError, ReadTimeout) are also retried with the same backoff. After all retries are exhausted, a KanonivError is raised with status_code=None.

python

# Increase retries for unreliable networks
client = Client(api_key="kn_...", max_retries=4, timeout=60.0)

Catching Errors

Catch a specific error:

python

from kanoniv.exceptions import NotFoundError

try:
    entity = client.entities.get("nonexistent")
except NotFoundError:
    print("Entity not found")

Catch rate limits with retry:

python

import time
from kanoniv.exceptions import RateLimitError

try:
    results = client.entities.search(q="john")
except RateLimitError as e:
    print(f"Rate limited, retrying in {e.retry_after}s")
    time.sleep(e.retry_after or 5)
    results = client.entities.search(q="john")

Catch all API errors:

python

from kanoniv.exceptions import KanonivError

try:
    client.sources.delete("source-uuid")
except KanonivError as e:
    print(f"API error {e.status_code}: {e}")
    if e.body and isinstance(e.body, dict):
        print(f"Detail: {e.body.get('detail', 'none')}")

Distinguish auth errors:

python

from kanoniv.exceptions import AuthenticationError, ForbiddenError

try:
    client.entities.search(q="john")
except AuthenticationError:
    print("Bad credentials. Check your API key.")
except ForbiddenError:
    print("Valid credentials, but you lack access to this resource.")

Kanoniv Cloud ​

Why Cloud? ​

Platform Architecture ​

Installation ​

Connectors ​

Quick Start ​

Real-World Scenarios ​

Multi-Source Identity Stitching ​

Batch Processing at Scale ​

Data Warehouse to Kanoniv ​

Reverse ETL: Write Canonical IDs Back ​

Real-Time Resolution Pipeline ​

Human-in-the-Loop Review Queue ​

Cloud Reconciliation ​

Parameters ​

What Happens Under the Hood ​

CloudReconcileResult ​

Fields ​

Properties ​

Methods ​

Context Manager ​

Reusing an Existing Client ​

Error Handling ​

Autodetect ​

Request ​

Response ​

Entity Type Hint ​

Bootstrap ​

Authentication ​

Context Manager ​

Async Usage ​

Configuration ​

API Reference ​

Top-Level Methods ​

client.resolve() ​

client.ingest() ​

client.ingest_file() ​

client.ingest_parquet() ​

client.stats() ​

Entities: client.entities ​

entities.search() ​

entities.get() ​

entities.get_linked() ​

entities.get_linked_bulk() ​

entities.history() ​

entities.lock() ​

entities.revert() ​

Real-Time Resolution: client.resolve_rt ​

resolve_rt.realtime() ​

resolve_rt.bulk() ​

Sources: client.sources ​

sources.list() ​

sources.get() ​

sources.create() ​

sources.update() ​

sources.delete() ​

sources.sync() ​

sources.preview() ​

sources.get_mapping() ​

sources.upsert_mapping() ​

Rules: client.rules ​

rules.list() ​

rules.create() ​

rules.history() ​

Jobs: client.jobs ​

jobs.list() ​

jobs.get() ​

jobs.run() ​

jobs.cancel() ​

Reviews: client.reviews ​

reviews.list() ​

reviews.decide() ​

Overrides: client.overrides ​

overrides.list() ​

overrides.create() ​

overrides.delete() ​

Audit: client.audit ​

audit.list() ​

audit.entity_trail() ​

Feedback: client.feedback ​

Kanoniv Cloud

Why Cloud?

Platform Architecture

Installation

Connectors

Quick Start

Real-World Scenarios

Multi-Source Identity Stitching

Batch Processing at Scale

Data Warehouse to Kanoniv

Reverse ETL: Write Canonical IDs Back

Real-Time Resolution Pipeline

Human-in-the-Loop Review Queue

Cloud Reconciliation

Parameters

What Happens Under the Hood

CloudReconcileResult

Fields

Properties

Methods

Context Manager

Reusing an Existing Client

Error Handling

Autodetect

Request

Response

Entity Type Hint

Bootstrap

Authentication

Context Manager

Async Usage

Configuration

API Reference

Top-Level Methods

`client.resolve()`

`client.ingest()`

`client.ingest_file()`

`client.ingest_parquet()`

`client.stats()`

Entities: `client.entities`

`entities.search()`

`entities.get()`

`entities.get_linked()`

`entities.get_linked_bulk()`

`entities.history()`

`entities.lock()`

`entities.revert()`

Real-Time Resolution: `client.resolve_rt`

`resolve_rt.realtime()`

`resolve_rt.bulk()`

Sources: `client.sources`

`sources.list()`

`sources.get()`

`sources.create()`

`sources.update()`

`sources.delete()`

`sources.sync()`

`sources.preview()`

`sources.get_mapping()`

`sources.upsert_mapping()`

Rules: `client.rules`

`rules.list()`

`rules.create()`

`rules.history()`

Jobs: `client.jobs`

`jobs.list()`

`jobs.get()`

`jobs.run()`

`jobs.cancel()`

Reviews: `client.reviews`

`reviews.list()`

`reviews.decide()`

Overrides: `client.overrides`

`overrides.list()`

`overrides.create()`

`overrides.delete()`

Audit: `client.audit`

`audit.list()`

`audit.entity_trail()`

Feedback: `client.feedback`