What you'll build: the master-data core of a CRM — the system of record every other workflow (lead-to-opportunity, scoring, forecasting, quote-cpq, attribution, account-360) reads — on one DataK3 bucket, fronted by a small FastAPI app you can download and run. Four merge-keyed master tables (contacts, deals, activities, accounts) are the transactional spine; an account graph rolls up pipeline by corporate family; and a VECTOR(2048) dedup table catches "same person, different spelling" before a duplicate contact corrupts every number downstream. The same rows answer by content (SQL), by relationship (graph), and by meaning (vector) over one copy — no ETL, no Postgres + Neo4j + Pinecone to keep in sync.

What you'll learn:

  • Model CRM master data three ways at once — an agent prompt, the dodil data CLI, and a plain SQLAlchemy model — over merge-keyed DataK3 tables with idempotent upserts.
  • Project accounts into a graph and traverse it — roll up a corporate family's pipeline in one call.
  • Stand up a dedup vector — a VECTOR(2048) column, jina-embeddings-v4, cosine KNN.
  • Serve it all behind a FastAPI app (routes.py) — CRUD plus the two questions a flat CRM can't answer.
  • (Optional) Deploy a crm-dedup engine that flags merge candidates and adjudicates the ambiguous ones on kimi-k2.6.

The problem — and why it matters

Ask a RevOps leader what their "CRM" is and the honest answer is a stack: Salesforce (or HubSpot) for the records, ZoomInfo / Clearbit to enrich them, Outreach / Apollo to sequence them, a warehouse (Snowflake / BigQuery) for reporting, and reverse-ETL to sync the warehouse back into the CRM. You pay per seat for the CRM, per record for enrichment, per seat again for the sequencer, and per credit for the warehouse — and every one of them holds its own copy of the same customer, drifting out of sync between syncs.

After all that spend, the stack still can't answer the three questions that actually move deals:

  • "What's the whole corporate family's pipeline?" Acme Corp owns Acme Labs and Acme EU. A rep working the account wants the number for the whole tree, not one legal entity — and WHERE org_domain = can't express "and all its subsidiaries." You need a graph.
  • "Is this new contact a duplicate?" "Jane Doe" and "Jane M. Doe" at the same domain are one person, entered twice. A keyword match on email misses it; the addresses differ. Every downstream report then double-counts her pipeline and splits her history. You need semantic search.
  • "Can I qualify 200k leads without paying to enrich every one?" Enrichment is billed per record. You want to spend the credits only on the leads worth spending them on. You need a cost gate — which starts with a clean, deduped, graph-aware system of record.

One DataK3 bucket collapses the stack: the master record is the graph node is the dedup index, over one copy of the rows. No enrichment seat to hold the graph, no Pinecone to hold the vectors, no warehouse to hold the reporting copy — SQL, graph, and vector are three views of the same table. The core is four tables, one edge table, one graph, one vector table, and (optionally) one Ignite app. This is the anchor every other crm/* skill composes onto.

PieceLands inPillar
Contacts / deals / activities / accountstables (merge-keyed)SQL
Account hierarchyaccounts (account_id, parent_id) + account_edges → graph crm_accountsGraph
Contact dedupcontact_vectors (VECTOR(2048), jina-embeddings-v4)Vector
Dedup engine (optional)flags merge candidates → activitiesIgnite + Models

NOTE

Connect the DODIL MCP once — then every data step shows an Ask your agent tab (the default — DODIL is agent-native), a CLI tab, and an ORM tab (the SQLAlchemy model that ships in the downloadable package). Three front-ends, one set of rows.

Prerequisites

  • A DODIL organization with the dodil CLI authenticated (dodil auth login) or the DODIL MCP connected to your agent (Claude Code, Cursor, VS Code, Codex). Headless? Check auth_status first — an agent can't do the browser login for you.
  • export BUCKET=crm — one bucket is the whole CRM's data plane. (This is the bucket param, default crm — and crm is a real, persistent bucket: the whole suite was re-validated live on it 2026-09-06.)

Step 1 — Stand up the master tables (the transactional core)

The CRM is the schema: merge-keyed tables in one bucket. email is the contact key, deal_id keys deals, activity_id keys the activity log, and account_id (an integer you assign) keys accounts. A --merge-key (PRIMARY KEY) is required — writes are keyed, so re-runs and shard retries upsert idempotently.

Every table has a natural primary key — an email, a deal_id, an account_id you assign — never a SERIAL/autoincrement. DataK3 has no sequences, and the wire doesn't return a staged write via RETURNING, so natural keys are what let the ORM tab map 1:1 to the CLI with no generated-PK round-trip. Each data step below carries that third tab: it's the exact class from the package's models.py — plain SQLAlchemy, the same op a third way.

You

Create a DataK3 bucket crm, then three merge-keyed master tables: contacts (key email) with user_id, full_name, org_domain, lifecycle_stage, source, subscribed(boolean), created_at, updated_at; deals (key deal_id) with title, contact_email, org_domain, pipeline, stage, status, value_num(double), owner, source, created_at; and activities (key activity_id) with deal_id, contact_email, kind, subject, body, direction, status, ts.

DODIL MCP tools called
data_bucket_create→data_table_create
Agent

Created bucket crm and 3 master tables — contacts (pk email), deals (pk deal_id), activities (pk activity_id). Upserts are idempotent; the same bucket also holds the accounts graph and the dedup vectors.

Now seed the demo dataset (seed_data=demo): the Acme corporate family (Acme Corp + two subsidiaries), independent Greyparrot, their contacts, and three open deals. Everything downstream — the family rollup, the dedup KNN — asserts against exactly these rows.

You

In crm, upsert 5 contacts: [email protected] (Jane Doe, acme.io, customer), [email protected] (Bob Lin, labs.acme.io), [email protected] (Carol Ng, acme.eu), [email protected] (Dan Roe, greyparrot.ai), and a near-duplicate [email protected] (Jane M. Doe, acme.io). Upsert 3 open deals: deal-labs-1 on labs.acme.io for 24000, deal-eu-1 on acme.eu for 12000, deal-grey-1 on greyparrot.ai for 30000.

DODIL MCP tools called
data_table_upsert
Agent

Upserted 5 contacts (incl. the planted near-dup [email protected]) and 3 open deals ($24k Labs, $12k EU, $30k Greyparrot). Merge-keyed, so re-running lands each row once.

TIP

Keyed upserts drop null/empty keys. A JSON null — or an empty string "" — in a merge-key column reads back as null and the row silently disappears. Give every keyed row a real, non-empty key (an email, a deal_id), never "".

Step 2 — Build the account graph (hierarchy=true)

Companies aren't flat: Acme Corp owns Acme Labs and Acme EU. To roll up the whole family's pipeline you project accounts into a graph and traverse the ownership tree. In this core the account is the graph node — accounts is keyed on an integer account_id precisely because graph_khop (and the cypher() call the app uses) take an integer-literal start node. The single account_edges table carries the child→parent edges, tagged by rel (subsidiary_of). Populate the edges fully, commit, then CREATE GRAPH — the graph snapshots its edges at creation, so edges added afterwards aren't traversable until you re-create it. And the commit matters: a CREATE GRAPH in the same transaction as the edge writes snapshots none of them (DataK3 has no read-your-writes inside an open transaction) — a bug that actually bit the suite's graph-assembly route until it committed first, then DROP+CREATE.

You

In crm, create accounts (key account_id bigint) with org_domain, name, parent_id(bigint), tier, employees(bigint), and project the 4 accounts: 1 Acme Corp (acme.io, no parent), 2 Acme Labs (labs.acme.io, parent 1), 3 Acme EU (acme.eu, parent 1), 4 Greyparrot (greyparrot.ai, no parent). Create account_edges (src, dst, rel) with subsidiary_of edges 2->1 and 3->1. Then CREATE GRAPH crm_accounts over accounts (KEY account_id) and account_edges (SRC src DST dst).

DODIL MCP tools called
data_table_create→data_table_upsert→data_pg
Agent

accounts has 4 rows (account_id 1-4, parent_id set on the two subsidiaries); account_edges has 2 subsidiary_of edges (2->1, 3->1). CREATE GRAPH crm_accounts NODES (accounts KEY account_id) EDGES (account_edges SRC src DST dst) succeeded — edges were snapshotted at creation.

NOTE

In the suite (all seven crm/* skills in one bucket) core creates accounts/account_edges and the base edges but defers CREATE GRAPH — campaign-to-lead and account-360 add more edges, and the graph is created once, last (the snapshot rule). Standalone, core creates crm_accounts here.

Step 3 — Roll up pipeline by corporate family (the payoff traversal)

Every subsidiary points up toward Acme Corp (account 1), so a reverse k-hop from node 1 collects the whole corporate family — subsidiaries at hop 1 — in one traversal. JOIN that to deals and you have the family's total open pipeline no flat query gives you. Greyparrot ($30k) is correctly excluded — it's not in Acme's tree.

You

In crm, from Acme Corp (graph node 1) traverse crm_accounts inward to collect the whole corporate family, then roll up open deal pipeline across it.

DODIL MCP tools called
data_pg→data_bolt
Agent

Family = Acme Corp (1) + Acme Labs (2) + Acme EU (3). Pipeline rollup: Acme Labs $24,000 + Acme EU $12,000 = $36,000 across the family. Greyparrot ($30k) is excluded — not in Acme's tree.

NOTE

graph_khop('crm_accounts', 1, 5, 'in') walks incoming edges, climbing down the ownership tree from the parent, and returns node/hop_distance columns. The literal start key (1) lets the traversal JOIN straight to your relational tables in one SQL statement — graph and SQL over one copy of the rows. (Cypher's id(...) can't appear in a Bolt RETURN — aggregate in the enclosing SQL, as above.)

Step 4 — The dedup vector (VECTOR(2048), one copy of the rows)

Master data's silent killer is the duplicate: "Jane Doe" and "Jane M. Doe" at acme.io are one person, entered twice. A keyword match on email misses it (the addresses differ). So you embed the identity text of each contact with jina-embeddings-v4 into a VECTOR(2048) column and KNN by meaning — near-dups land next to each other before a blind upsert creates a second record. The vector is keyed on email, the same key as contacts, so it's the same entity, one copy: contact_vectors holds nothing but the key and the embedding.

You

In crm, create contact_vectors (key email, embedding VECTOR(2048)) and index the vector column. Then for each contact embed its identity text (full_name at org_domain) with jina-embeddings-v4 and upsert one row per call.

DODIL MCP tools called
data_table_create→data_pg→ignite_models_embed→data_table_upsert
Agent

Created contact_vectors (pk email, embedding VECTOR(2048)) and built the vector index. Embedded each contact's identity text with jina-embeddings-v4 (2048-dim) and upserted one row per call — batching many 2048-dim vectors in a single upsert can hit a gRPC frame limit.

Now catch the near-duplicate by meaning — the two "Jane" records land as the two nearest neighbors:

You

In crm, is there a likely duplicate of Jane Doe (acme.io) in contact_vectors? Vector-search by meaning and show the nearest matches with cosine distance.

DODIL MCP tools called
data_vsearch
Agent

The row itself ranks first; the nearest OTHER contact is [email protected] at cosine 0.0364 — the planted near-duplicate — well ahead of the next contact at 0.335. Below the 0.20 dedup_threshold, [email protected] is a merge candidate.

Routes

The download (see Get the code) fronts this bucket with a small FastAPI app, routes.py — CRUD over the models plus the two questions a flat CRM can't answer. This is what you deploy. The routes live on an APIRouter — the suite app mounts all seven CRM components on one FastAPI under per-component prefixes — while app = FastAPI(...) at the bottom of the file keeps the package independently runnable (uvicorn routes:app). Every route follows the same three DataK3 rules the package bakes in, so quoting it is documenting them.

The connection and the one write helper live in db.py. A DataK3 bucket is a Postgres endpoint — db name = the bucket, user = the literal token, password = your DODIL token — so there's no data connect step in code, just fixed region constants. upsert() is the only writer every route uses:

# db.py — INSERT ... ON CONFLICT DO UPDATE (idempotent keyed write); DO NOTHING for pure edge rows
def upsert(session, model, rows, key):
    keys = [key] if isinstance(key, str) else list(key)
    table = model.__table__
    # normalise to a uniform column set — a multi-row VALUES needs every row to name the
    # same columns; fill any a caller omitted with None.
    cols = {c for r in rows for c in r}
    rows = [{c: r.get(c) for c in cols} for r in rows]
    stmt = pg_insert(table).values(rows)
    update_cols = [c.name for c in table.columns if c.name not in keys and c.name in cols]
    if update_cols:
        stmt = stmt.on_conflict_do_update(
            index_elements=keys,
            set_={c: getattr(stmt.excluded, c) for c in update_cols},
        )
    else:
        # every column is part of the key (a pure edge/junction row) — nothing to update.
        stmt = stmt.on_conflict_do_nothing(index_elements=keys)
    session.execute(stmt)

Why it matters: on DataK3 a bare re-INSERT of an already-committed primary key raises duplicate-key 23505 — a plain INSERT is not an upsert on re-write. upsert makes a retry, a shard replay, or a re-import land the row once. That's the whole reason the master tables are safe to re-run.

CRUD — each write is an upsert then a commit (DataK3 has no read-your-writes inside an open transaction; the engine is expire_on_commit=False, so routes commit before they return):

# routes.py — CRUD over the models, keyed on the natural PK
@router.post("/contacts")
def upsert_contact(c: ContactIn, s: Session = Depends(db)):
    upsert(s, Contact, [c.model_dump()], key="email")
    s.commit()
    return {"ok": True, "email": c.email}
 
 
@router.get("/contacts/{email}")
def get_contact(email: str, s: Session = Depends(db)):
    row = s.get(Contact, email)
    if not row:
        raise HTTPException(404, "no such contact")
    return row.__dict__ | {"_sa_instance_state": None}
 
 
@router.post("/deals")
def upsert_deal(d: DealIn, s: Session = Depends(db)):
    upsert(s, Deal, [d.model_dump()], key="deal_id")
    s.commit()
    return {"ok": True, "deal_id": d.deal_id}
 
 
@router.post("/activities")
def log_activity(a: ActivityIn, s: Session = Depends(db)):
    # activities is an append-only event log keyed on activity_id — upsert makes a replayed
    # webhook / at-least-once delivery land the event once.
    upsert(s, Activity, [a.model_dump()], key="activity_id")
    s.commit()
    return {"ok": True, "activity_id": a.activity_id}

Workflow op 1 — the corporate-family pipeline rollup (GRAPH). GET /accounts/{account_id}/family traverses crm_accounts from this account to every subsidiary, then sums the deals of the whole family. DataK3 runs a Cypher subset embedded in SQL — cypher('<graph>', 'MATCH …') — and it comes with three rules the code obeys: it's a top-level table function (no UNION/subquery/CTE), the anchor id must be an integer literal (so the FastAPI-validated account_id is inlined, not bound), and you feed the returned node ids into a SQL IN (…) (DuckDB has no = ANY(array)):

# routes.py — workflow op 1: corporate-family pipeline rollup (GRAPH)
@router.get("/accounts/{account_id}/family")
def family_pipeline(account_id: int, s: Session = Depends(db)):
    """Roll up a whole corporate family's pipeline. graph_khop walks the crm_accounts
    graph from this account down to every subsidiary (the 'in' direction: children point
    at their parent), then we sum the deals of every account in the family."""
    # DataK3 runs a Cypher subset embedded in SQL: cypher('<graph>', 'MATCH ...'). It's a
    # top-level table function (can't sit in a UNION/subquery) and the anchor id must be an
    # integer literal, so we inline the FastAPI-validated account_id and add the root in
    # Python. Edges are child->parent, so from the root we expand the 'in' direction (`<-`).
    kids = s.execute(
        text(
            "SELECT node FROM cypher('crm_accounts', "
            f"'MATCH (root)<-[*1..5]-(child) WHERE id(root) = {account_id} RETURN child')"
        )
    ).scalars().all()
    fam = list(dict.fromkeys([account_id, *kids]))  # root first, de-duped
    if len(fam) == 1 and not s.get(Account, account_id):
        raise HTTPException(404, "no such account")
    # DuckDB has no `= ANY(:array)` (UNNEST), so sum over an inlined IN (...) of ints.
    ids = ",".join(str(int(i)) for i in fam)
    total = s.execute(
        text(
            "SELECT COALESCE(SUM(d.value_num), 0) FROM deals d JOIN accounts a "
            f"ON a.org_domain = d.org_domain WHERE a.account_id IN ({ids})"
        )
    ).scalar()
    return {"root": account_id, "family_account_ids": fam, "pipeline_value": float(total)}

Live-verified 2026-09-06 on the persistent crm bucket: GET /accounts/1/family returns {"root": 1, "family_account_ids": [1, 3, 2], "pipeline_value": 36000.0} — the root first, then the subsidiaries in traversal order, $24k Labs + $12k EU rolled up in one call, with Greyparrot excluded because it isn't in the tree. No flat query gives you that number.

Workflow op 2 — contact dedup / semantic match (VECTOR). POST /contacts/similar takes a query embedding and returns the nearest contacts by pgvector cosine distance — the same "is this a duplicate?" check as Step 4, now callable from your app before it creates a contact:

# routes.py — workflow op 2: contact dedup / semantic match (VECTOR)
@router.post("/contacts/similar")
def similar_contacts(q: SimilarIn, s: Session = Depends(db)):
    """Nearest contacts to a query embedding — pgvector cosine distance over the same
    bucket. Used to dedup a new contact against existing ones before you create it."""
    rows = s.execute(
        select(ContactVector.email, ContactVector.embedding.cosine_distance(q.embedding).label("d"))
        .order_by("d")
        .limit(q.top_k)
    ).all()
    return {"matches": [{"email": e, "distance": float(dist)} for e, dist in rows]}

Adding a new business operation touches only routes.py (and maybe models.py) — the plumbing in db.py is fixed. The pattern is one Pydantic *In schema + one @app.<verb> function: write via upsert, graph via cypher(…), vector via cosine_distance(…) (see EXTENDING.md in the package).

Auth — config at the edge, a role gate in the app

On Ignite, end-user login is configuration, not code. The suite deploys with the crm-suite dodil-appid pool attached (user_pool: crm-suite in .dodil/deploy.yaml) and the per-cluster Ignite gateway runs the entire browser login at the edge — PKCE S256 against the pool issuer (https://appid.dodil.io/ihdiash/crm-suite), an AEAD-sealed host-only session cookie, JWT verification — then injects the verified identity into every request it forwards: X-Dodil-User (sub, email, app_roles), X-Dodil-User-Jwt (the raw verified token, carrying the catalog-expanded permissions claim) and X-Dodil-Auth-Source. Any inbound copy of those headers is stripped first, so a caller can never forge them.

What survives in the package is a small auth.py that ships no verifier — no JWKS client, no issuer/audience env, no crypto dependency. It reads the injected header and keeps the one job the app still owns: role-based gating.

# auth.py — a header-trust reader, not a verifier (the gateway already checked the JWT)
def current_user(request: Request) -> dict:
    raw = request.headers.get("x-dodil-user")   # {"sub","email","connection","app_roles"}
    ...                                          # + permissions read off x-dodil-user-jwt
    raise HTTPException(401, "end-user login required — no X-Dodil-User from the gateway")
 
def require_permission(perm: str):
    """Gate a route on a pool permission: Depends(require_permission("quotes:approve"))."""
    def _dep(user: dict = Depends(current_user)) -> dict:
        if perm not in user["permissions"] and perm not in user["roles"]:
            raise HTTPException(403, f"missing permission: {perm}")
        return user
    return _dep

Across the seven CRM components, exactly four routes kept a require_permission gate after the audit — the ones that spend model money or move the number: orgs:qualify (lead-to-opportunity's paid qualify batch), leads:score (qualification-scoring's paid gate), quotes:approve (quote-cpq's discount judgement) and forecast:override (pipeline-forecast's risk downgrade). Core has none — nothing in this component's routes.py imports auth. The gateway's authentication is the whole story here: the app deploys private, so an unauthenticated request is redirected to /.dodil/auth/login and never reaches the pod.

The pool is created once for the whole suite, with the role catalog the gates check (sales / analyst / manager):

You

Create a dodil-appid pool crm-suite with email+password, and set its role catalog: analyst can run the paid qualify and scoring batches; manager can also approve quotes and override the forecast.

Agent

Pool crm-suite created — issuer https://appid.dodil.io/ihdiash/crm-suite, audience pool:crm-suite, email+password (local) enabled. Catalog set: analyst = orgs:qualify,leads:score; manager adds quotes:approve,forecast:override. A user's next token carries app_roles plus the catalog-expanded permissions claim the gates read.

The two-plane rule is unchanged: the pool identifies the user; the app still reaches DataK3 through its own service account — an app-user is never a bucket principal. Locally (no gateway in front of uvicorn routes:app) opt in to a stub identity with DEV_ALLOW_ANON=1; the stub carries no permissions unless you grant them (DEV_USER_PERMISSIONS=quotes:approve,…), so the gated routes stay gated even on a laptop. The full flow — creating the pool, the redirect_uris allowlist, what the gateway injects, and the off-gateway path where you do verify the pool JWT yourself (iss and aud mandatory) — is App authentication; the catalog mechanics are App roles.

UI

The app a business user actually opens ships in the package under web/ — Vite + React + TypeScript + Tailwind + shadcn-style components + TanStack Query + TanStack Table. It's a static SPA that talks only to the routes above — never to DataK3 directly. That's the two-plane rule made physical: deployed, the browser holds nothing but the gateway's session cookie (no token in the SPA — the gateway did the login); the API is still the only thing with a bucket credential (its service account).

The speed unlock is a generated OpenAPI client: routes.py already publishes an OpenAPI schema, so openapi-typescript turns it into typed TypeScript and openapi-fetch calls it — zero hand-written fetch. Request bodies (ContactIn, SimilarIn) and path params are checked against the real routes at compile time, so renaming a route surfaces as a TypeScript error, not a runtime 404. Regenerate the client whenever the routes change — the FastAPI app emits its schema without a database connection (the SQLAlchemy engine is lazy), so no bucket is touched to build the UI:

# from the repo root — emit the schema, then generate the typed client
python3 -c "import json,sys; sys.path.insert(0,'code/crm-core/v1'); from routes import app; print(json.dumps(app.openapi()))" \
  > /tmp/crm-openapi.json
cd code/crm-core/v1/web
npx openapi-typescript /tmp/crm-openapi.json -o src/api/schema.ts
 
# then run it
npm install
cp .env.example .env       # set VITE_API_BASE (the API; DEV_ALLOW_ANON=1 on it locally)
npm run dev                # http://localhost:5173

Deployed, sign-in is the gateway's: an unauthenticated request is redirected to /.dodil/auth/login, the PKCE round-trip runs against the crm-suite pool, and the browser comes back holding only the sealed session cookie — the SPA ships no login form and stores no token. Full flow: App authentication. Three pages sit behind that gate: Contacts (a TanStack Table plus a create form onto POST /contacts), and Account family (enter a parent account id, call the family-rollup route, and render the rolled-up pipeline_value — the graph payoff, as a number on a card). It's served as a static build, or as its own Ignite app; npm run build type-checks and bundles it to dist/.

Get the code

The package is a real download — code/crm-core/v1.tar. This post is a walkthrough of exactly those files; the tarball is source only (no Dockerfile — deploy is in Ship it):

models.py          # SQLAlchemy — contacts, deals, activities, accounts (+ account_edges), contact_vectors
routes.py          # FastAPI    — an APIRouter the suite mounts + a standalone app (uvicorn routes:app)
db.py              # lazy engine (openapi() builds with no creds) + the ON CONFLICT upsert helper
auth.py            # gateway header-trust: current_user + require_permission — no verifier
sa_token.py        # deployed: mints + refreshes the service-account token for the pg-wire password
.env.example       # BUCKET + PG_HOST/PG_PORT + DODIL_TOKEN (or the SA pair) + DEV_ALLOW_ANON
requirements.txt   # sqlalchemy, psycopg[binary], pgvector, fastapi, uvicorn, pydantic, httpx

Run it — point .env at your bucket, create the tables from the models, serve:

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
 
cp .env.example .env      # then set DODIL_TOKEN (your `dodil auth login` token)
# BUCKET defaults to "crm"; create it once (Step 1), then create the tables from the models:
python -c "import db, models; models.Base.metadata.create_all(db.engine)"
 
uvicorn routes:app --reload
# POST /contacts, /deals, /activities · GET /accounts/{id}/family · POST /contacts/similar

models.Base.metadata.create_all is the ORM tab's every class at once — the same tables Step 1–4 built by CLI, created from the natural-key models with no migration tool.

In the suite, you don't run seven servers: the seven crm/* packages compose into one Ignite app — crm-suite-app mounts each component's APIRouter under a per-component prefix (/core, /lead-to-opportunity, /qualification-scoring, /campaign-to-lead, /quote-cpq, /account-360, /pipeline-forecast) on one FastAPI process, over one engine to the one crm bucket — one connection pool instead of seven, which is what respects the tables reader's concurrency cap. It ships through the git cycle with user_pool: crm-suite in .dodil/deploy.yaml; the deploy walkthrough is Ship a DODIL app.

Step 5 — The crm-dedup engine (optional, dedup_engine=true)

Core is mostly declarative — schema, graph, vector, and a stateless API. The one stateful piece is optional: an Ignite app that, on each new contact write, embeds it, KNN-checks contact_vectors, and if the nearest cosine distance is below dedup_threshold (default 0.20) flags a merge candidate (an activities row kind=dedup_candidate) instead of blindly upserting a duplicate. In the ambiguous band it asks kimi-k2.6 to adjudicate. It's a separate workload, so it gets its own service account with least-privilege roles: k3.editor (write the flag + vector) + ignite.model-user (embed + adjudicate). (The identity that deploys it also needs ignite.app-developer.)

The engine ships as an image-mode Ignite app — a plain HTTP server behind a Dockerfile, built by the platform on deploy (Kaniko build-on-deploy, Lane B). It exposes GET /healthz (the probe) and POST /dedup (a new-contact JSON). The data plane is the drop-in Postgres wire (pg.uk-lon-1.dodil.io:5432, dbname=<bucket>, user=token, password=<the SA access token>) — the KNN is a pgvector <=> query, and an idempotent re-write of a contact uses INSERT … ON CONFLICT (email) DO UPDATE (a bare re-INSERT of an already-committed PK raises duplicate-key 23505; DuckDB pg-wire supports ON CONFLICT, or use the managed data_table_upsert). Models is OpenAI-compatible at https://api.dodil.io/v1 — an explicit User-Agent is required (stdlib urllib's default is Cloudflare-banned), kimi-k2.6 needs max_tokens: 4096 (reasoning tokens else empty content), and both Models responses come wrapped in data.

# dedup/server.py — IMAGE-mode Ignite app (HTTP server on $PORT).
#   GET  /healthz -> 200 {"status":"ready"}      (probe path; no auth)
#   POST /dedup   -> body = a new contact JSON    (embed -> KNN -> flag / adjudicate / store)
# Only psycopg is third-party (the pg driver); everything else is stdlib.
 
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
from psycopg import errors as pg_errors
 
# --- hoisted policy knobs: mirror the dedup policy, injected as env at deploy ---
DEDUP_THRESHOLD = float(os.environ.get("DEDUP_THRESHOLD", "0.20"))   # cosine distance; mirrors dedup_threshold
AMBIG_LO        = float(os.environ.get("AMBIG_LO", "0.15"))          # inside [LO,HI] -> ask the model
AMBIG_HI        = float(os.environ.get("AMBIG_HI", "0.30"))
EMBED_MODEL     = os.environ.get("EMBED_MODEL", "jina-embeddings-v4")  # 2048-dim
CHAT_MODEL      = os.environ.get("CHAT_MODEL", "kimi-k2.6")
BUCKET          = os.environ["BUCKET"]
SA_ID           = os.environ["DODIL_SERVICE_ACCOUNT_ID"]   # the cli-… serviceAccountId, NOT the uuid
SA_SECRET       = os.environ["DODIL_SERVICE_ACCOUNT_SECRET"]
PG_HOST         = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
PG_PORT         = int(os.environ.get("PG_PORT", "5432"))
 
ID_URL    = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
CHAT_URL  = "https://api.dodil.io/v1/chat/completions"
EMBED_URL = "https://api.dodil.io/v1/embeddings"
# an explicit User-Agent is REQUIRED — stdlib urllib's default "Python-urllib/x" is
# banned by Cloudflare at id/api.dodil.io (HTTP 403 "error code: 1010").
UA = "crm-dedup/1.0"
 
ADJUDICATE_SYS = ('Decide if two CRM contact rows are the same person. Return ONLY JSON: '
                  '{"same": boolean, "confidence": number, "reason": string}.')
 
 
def _now():
    return datetime.now(timezone.utc).isoformat()
 
 
def _http_post(url, data, headers, form=False):
    headers = {"User-Agent": UA, **headers}
    if form:
        body = urllib.parse.urlencode(data).encode()
        headers["Content-Type"] = "application/x-www-form-urlencoded"
    else:
        body = json.dumps(data).encode()
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
    with urllib.request.urlopen(req, timeout=90) as r:
        return json.loads(r.read().decode())
 
 
def _token():
    out = _http_post(ID_URL, {"grant_type": "client_credentials",
                              "client_id": SA_ID, "client_secret": SA_SECRET},
                     headers={}, form=True)
    return out["access_token"]
 
 
def _embed(token, text):
    out = _http_post(EMBED_URL, {"model": EMBED_MODEL, "input": text},
                     headers={"Authorization": f"Bearer {token}"})
    env = out.get("data", out)            # response is wrapped in "data" on this platform
    return env["data"][0]["embedding"]    # jina-embeddings-v4 -> 2048-dim float list
 
 
def _chat(token, a, b, distance):
    # kimi-k2.6 is a reasoning model: reasoning tokens eat the budget first — set
    # max_tokens high (4096) or message.content comes back empty.
    out = _http_post(CHAT_URL,
                     {"model": CHAT_MODEL, "max_tokens": 4096,
                      "messages": [{"role": "system", "content": ADJUDICATE_SYS},
                                   {"role": "user",
                                    "content": f"Row A: {a}. Row B: {b}. "
                                               f"Vector cosine distance {distance:.3f}."}]},
                     headers={"Authorization": f"Bearer {token}"})
    env = out.get("data", out)            # response is wrapped in "data" on this platform
    return env["choices"][0]["message"]["content"]
 
 
def _pg(token):
    # drop-in Postgres wire: DB name = bucket, user "token", password = the SA access token.
    return psycopg.connect(host=PG_HOST, port=PG_PORT, dbname=BUCKET,
                           user="token", password=token, sslmode="require",
                           connect_timeout=20, autocommit=False)
 
 
def _extract_json(text):
    if not text:
        raise ValueError("empty model content")
    text = text.strip()
    if text.startswith("```"):
        text = re.sub(r"^```[a-zA-Z]*\n?", "", text)
        text = re.sub(r"\n?```$", "", text).strip()
    m = re.search(r"\{.*\}", text, re.DOTALL)
    return json.loads(m.group(0) if m else text)
 
 
def _vec_literal(vec):
    return "[" + ",".join(repr(float(x)) for x in vec) + "]"   # pgvector text form
 
 
def _retry(fn):
    # the tables engine is serializable — retry a write on a transient serialization/deadlock.
    for attempt in range(4):
        try:
            return fn()
        except (pg_errors.SerializationFailure, pg_errors.DeadlockDetected):
            if attempt == 3:
                raise
            time.sleep(0.4 * (attempt + 1))
 
 
def _knn(token, email, vec):
    # nearest OTHER contact by cosine distance — pgvector `<=>` over the pg wire.
    lit = _vec_literal(vec)
    with _pg(token) as conn, conn.cursor() as cur:
        cur.execute("SELECT email, embedding <=> %s::vector AS distance "
                    "FROM contact_vectors WHERE email <> %s "
                    "ORDER BY distance LIMIT 1", (lit, email))
        r = cur.fetchone()
    return None if not r else {"email": r[0], "distance": float(r[1])}
 
 
def _flag(token, contact, near):
    # Each flag is a fresh activity_id (a new PRIMARY KEY every call), so this is an append-only
    # first-write — the plain INSERT is correct here. (Re-writing the SAME key would instead need
    # INSERT … ON CONFLICT DO UPDATE: a bare re-INSERT of a committed PK raises duplicate-key 23505.)
    row = {"activity_id": str(uuid.uuid4()), "deal_id": None,
           "contact_email": contact["email"], "kind": "dedup_candidate",
           "subject": f"merge? {near['email']}", "body": f"cosine={near['distance']:.3f}",
           "direction": "system", "status": "open", "ts": _now()}
    cols = list(row)
    sql = (f"INSERT INTO activities ({', '.join(cols)}) "
           f"VALUES ({', '.join(['%s'] * len(cols))})")
 
    def _w():
        with _pg(token) as conn, conn.cursor() as cur:
            cur.execute(sql, [row[c] for c in cols])
            conn.commit()
    _retry(_w)
 
 
def _store(token, contact, vec):
    # unique enough — write the contact AND its dedup vector (both keyed on email). Re-POSTing the
    # same contact re-writes the SAME email PK, so both writes are ON CONFLICT (email) DO UPDATE:
    # a bare re-INSERT of a committed PK raises duplicate-key 23505 (DuckDB pg-wire supports ON CONFLICT).
    lit, c = _vec_literal(vec), contact
 
    def _w():
        with _pg(token) as conn, conn.cursor() as cur:
            cur.execute("INSERT INTO contacts (email, full_name, org_domain, "
                        "lifecycle_stage, source, updated_at) "
                        "VALUES (%s,%s,%s,%s,%s,%s) "
                        "ON CONFLICT (email) DO UPDATE SET full_name = EXCLUDED.full_name, "
                        "org_domain = EXCLUDED.org_domain, "
                        "lifecycle_stage = EXCLUDED.lifecycle_stage, source = EXCLUDED.source, "
                        "updated_at = EXCLUDED.updated_at",
                        (c["email"], c["full_name"], c["org_domain"],
                         c.get("lifecycle_stage"), c.get("source"), _now()))
            cur.execute("INSERT INTO contact_vectors (email, embedding) "
                        "VALUES (%s,%s::vector) "
                        "ON CONFLICT (email) DO UPDATE SET embedding = EXCLUDED.embedding",
                        (c["email"], lit))
            conn.commit()
    _retry(_w)
 
 
def dedup_contact(contact):
    token = _token()
    text  = f'{contact["full_name"]} at {contact["org_domain"]}'
    vec   = _embed(token, text)                       # jina-embeddings-v4 -> 2048-dim
    near  = _knn(token, contact["email"], vec)        # nearest OTHER contact
    if near and near["distance"] < DEDUP_THRESHOLD:
        same, verdict = True, None
        if AMBIG_LO <= near["distance"] <= AMBIG_HI:  # ambiguous band -> ask the model
            verdict = _extract_json(_chat(token, contact, near, near["distance"]))
            same = bool(verdict["same"])
        if same:
            _flag(token, contact, near)
            return {"flagged": True, "match": near["email"],
                    "distance": near["distance"], "adjudication": verdict}
    _store(token, contact, vec)                       # unique enough — write it
    return {"flagged": False}
 
 
class Handler(BaseHTTPRequestHandler):
    def _send(self, code, body):
        payload = json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)
 
    def do_GET(self):
        if self.path == "/healthz":
            return self._send(200, {"status": "ready"})
        return self._send(404, {"error": "no_route", "path": self.path})
 
    def do_POST(self):
        if self.path != "/dedup":
            return self._send(404, {"error": "no_route", "path": self.path})
        try:
            n = int(self.headers.get("Content-Length") or 0)
            contact = json.loads(self.rfile.read(n) or b"{}")
            if not contact.get("email") or not contact.get("full_name"):
                return self._send(400, {"error": "missing email/full_name"})
            return self._send(200, dedup_contact(contact))
        except urllib.error.HTTPError as e:
            return self._send(502, {"error": "upstream", "code": e.code,
                                    "body": e.read().decode(errors="replace")[:600]})
        except Exception as e:
            return self._send(500, {"error": type(e).__name__, "detail": str(e)[:600]})
 
    def log_message(self, *a):
        pass
 
 
if __name__ == "__main__":
    port = int(os.environ.get("PORT", "8080"))
    print(f"crm-dedup serving on 0.0.0.0:{port} "
          f"threshold={DEDUP_THRESHOLD} bucket={BUCKET}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

The engine's image is a two-file build alongside server.py — the Dockerfile and its one pip dep:

# dedup/Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
ENV PORT=8080
EXPOSE 8080
CMD ["python", "server.py"]
# dedup/requirements.txt
psycopg[binary]==3.2.3

The adjudication gate, proven live against the two Jane rows (cosine 0.0364):

NOTE

kimi-k2.6 is a reasoning model. Called via ignite models chat (MCP/CLI) there's no max_tokens knob, so it can return empty content. End the prompt with Return ONLY compact JSON, no reasoning or preamble and retry once on an empty reply. (The crm-dedup handler sets max_tokens: 4096 on the raw api.dodil.io/v1 call — the interactive CLI/MCP path can't.)

You

Are these two CRM contact rows the same person? Row A [email protected] Jane Doe acme.io; Row B [email protected] Jane M. Doe acme.io; vector cosine distance 0.0364. Answer on kimi-k2.6 as JSON {same, confidence, reason}.

DODIL MCP tools called
ignite_models_chat
Agent

{"same": true, "confidence": 0.97, "reason": "Same org, matching name variants, very low embedding distance."}

Give the engine its own least-privilege identity, then deploy it as an image — the platform builds the Dockerfile on deploy (Kaniko build-on-deploy), so --allow-unauthenticated needs no pull secret and the app comes up at a public FQDN. Not --runtime python (that's compile mode). The runtime DODIL_SERVICE_ACCOUNT_ID is the cli-… serviceAccountId that auth service-account create prints — the internal uuid fails client_credentials with invalid_client:

You

Create a service account for crm-dedup, grant it k3.editor and ignite.model-user, then deploy my ./dedup app to Ignite as crm-dedup in image mode (build the Dockerfile on deploy) with the service-account creds as runtime env. Use its cli- serviceAccountId as DODIL_SERVICE_ACCOUNT_ID.

DODIL MCP tools called
auth_service-account_create→auth_service-account_grant-role→ignite_app_deploy
Agent

Created service account crm-dedup-sa, granted k3-authorization-service k3.editor + ignite-authorization-service ignite.model-user, and deployed crm-dedup in image mode (Dockerfile built on deploy) serving on /dedup, health /healthz — request-invoked, scale-to-zero. Runtime DODIL_SERVICE_ACCOUNT_ID is the cli- serviceAccountId (not the uuid). It embeds each new contact, KNN-checks contact_vectors over pg-wire, and flags a dedup_candidate below the 0.20 threshold (adjudicating the 0.15-0.30 band on kimi-k2.6).

NOTE

Deploy: image mode (Lane B), validated live 2026-09-02. crm-dedup ships in image mode — a Dockerfile + --dockerfile-path, Kaniko build-on-deploy (not --runtime python). Validated live 2026-09-02: this pattern deploys, serves /healthz + its route unauthenticated, and writes durably — confirmed end-to-end via the sibling crm-lead-scorer engine (written rows survived +154s, re-confirmed at +95s); crm-dedup reuses that identical handler/deploy pattern. Its embed + KNN + adjudication logic is each also proven live.

If dodil ignite app deploy returns IAM resource registration failed: broken pipe, deploy under a fresh app name — the half-created app can't be updated or deleted (UMA can't authorize an unregistered resource).

How the pillars map

One bucket, one bill, one auth context — the master record is the graph node is the dedup index, over one copy of the rows. What this core would otherwise be:

JobThe usual stackOn DataK3
Contacts / deals / activities / accountsSalesforce/HubSpot seat + a warehouse copySQL master tables in the bucket
Roll up a corporate family's pipelineNeo4j + a sync jobcrm_accounts — graph_khop JOINed to deals
"Is this a duplicate contact?"ZoomInfo/Clearbit + an MDM match-merge toolVECTOR(2048) column + data vsearch (cosine KNN)
Serve it to your appPer-system SDKs & reverse-ETLone FastAPI app (routes.py) over pg-wire
Point your own tools at itPer-system drivers & credsdata connect — psql / bolt / pgvector, DB = bucket

No ETL, no second copy, no drift between the record and its search index or its graph node — because it's all one bucket. This is the anchor the rest of the CRM suite (crm/lead-to-opportunity, crm/qualification-scoring, crm/pipeline-forecast, crm/quote-cpq, crm/campaign-to-lead, crm/account-360) composes onto by consuming these masters.

Customize — the decisions this skill asks you

Q1 · seed_data — demo rows or empty schemas?

"Load the Acme demo dataset, or ship empty schemas?"

  • demo (default) → loads the Acme family + Greyparrot + the planted near-dup, so ## Test asserts exact counts and the $36,000 family rollup.
  • empty → schemas + graph tables + vectors only; ## Test switches to structural assertions (tables exist, the graph traverses 0 rows without error, vsearch returns []).

Q2 · hierarchy — build the account graph?

"Do you want family/subsidiary rollups?" → true (default, recommended) builds accounts/account_edges + the subsidiary_of edges + crm_accounts (Step 2). false skips them — the module is flat CRM with no family rollup. The family pipeline rollup is core's single highest-value read, so keep it on unless you genuinely have no corporate hierarchies.

Q3 · dedup_engine + dedup_threshold — deploy the dedup engine?

"Flag duplicate contacts automatically, and at what similarity?"

  • dedup_engine=false (default) → the vector table ships as search-only — you run vsearch on demand (Step 4) or call POST /contacts/similar. No Ignite app.
  • dedup_engine=true → deploys crm-dedup (Step 5). dedup_threshold (default 0.20 cosine distance) is written into the handler constant DEDUP_THRESHOLD and a policy row (single source): nearer than that = a flagged merge candidate; inside the 0.15–0.30 band, kimi-k2.6 adjudicates. A lower threshold means fewer false merges but more real duplicates slip through.

TIP

Industry overlays. This core is the cross-industry base (industry: software). Industry variants (saas, manufacturing, finserv, real-estate) compose this core with a small additive diff — see the per-industry CRM pages for each build.

Test

The package functions (models.py / routes.py / db.py) are live-verified against DataK3 on 2026-09-06, on the persistent crm bucket the whole suite runs on; the commands below mirror them and assert the seeded demo's exact results. The optional crm-dedup deploy in Step 5 uses the image-mode pattern validated live 2026-09-02 on crm-lead-scorer (deploys, serves /healthz unauthenticated, writes durably — see the note), and its embed + KNN + adjudication logic is each proven live. (empty branch swaps assertions 2–4 for: tables exist, the graph traverses 0 rows without error, vsearch returns [].)

# 1) the master + vector tables exist (4 master + 1 vector), plus the graph edge table
dodil data table list -b "$BUCKET"
#  contacts, deals, activities, accounts, contact_vectors  (+ account_edges)
 
# 2) seeded rows land
dodil data sql -b "$BUCKET" "SELECT count(*) AS n FROM contacts"   # n = 5  (incl. the near-dup)
dodil data sql -b "$BUCKET" "SELECT count(*) AS n FROM deals"      # n = 3
 
# 3) the graph rolls up the Acme corporate family — Greyparrot excluded
dodil data pg -b "$BUCKET" "
  SELECT count(DISTINCT d.deal_id) AS deals, coalesce(sum(d.value_num),0) AS pipeline
  FROM graph_khop('crm_accounts', 1, 5, 'in') g
  JOIN accounts a ON a.account_id = g.node
  JOIN deals d ON d.org_domain = a.org_domain AND d.status='open'"   # deals = 2, pipeline = 36000
 
# 4) dedup vector search surfaces the planted near-duplicate as the nearest OTHER row
dodil data vsearch -b "$BUCKET" -t contact_vectors --column embedding \
  --text "Jane Doe at Acme Corp (acme.io)" \
  --model jina-embeddings-v4 --metric cosine --top-k 2      # [email protected], then [email protected] 0.0364
 
# 5) drop-in clients: same bucket, your own psql / cypher-shell / pgvector driver
dodil data connect "$BUCKET" -o psql
#  postgresql://token:[email protected]:5432/crm?sslmode=require   -> paste into psql

One-shot

With the DODIL MCP connected, paste this to scaffold the whole master-data core at once:

Scaffold the CRM master-data core on DataK3 (one bucket = SQL + graph + vector). Confirm each step.
 
1. Create a DataK3 bucket `crm`, then three merge-keyed master tables:
   - contacts (key email): user_id, full_name, org_domain, lifecycle_stage, source, subscribed(boolean), created_at, updated_at.
   - deals (key deal_id): title, contact_email, org_domain, pipeline, stage, status, value_num(double), owner, source, created_at.
   - activities (key activity_id): deal_id, contact_email, kind, subject, body, direction, status, ts.
2. Seed the demo dataset: 5 contacts incl. a near-dup [email protected] of [email protected]; 3 open deals
   ($24k Labs, $12k EU, $30k Greyparrot).
3. Create accounts (key account_id bigint: org_domain, name, parent_id, tier, employees) with 4 rows
   (1 Acme Corp acme.io, 2 Acme Labs labs.acme.io parent 1, 3 Acme EU acme.eu parent 1, 4 Greyparrot greyparrot.ai);
   account_edges (src,dst,rel) with subsidiary_of 2->1 and 3->1; THEN CREATE GRAPH crm_accounts over them.
   Roll up open pipeline for Acme Corp's whole family with graph_khop('crm_accounts',1,5,'in') JOINed to
   accounts + deals — expect $36,000, Greyparrot excluded.
4. Create contact_vectors (key email, embedding VECTOR(2048)); embed each contact's identity text
   with jina-embeddings-v4 (one row per upsert); vsearch contact_vectors for "Jane Doe at Acme Corp (acme.io)"
   and confirm the near-dup [email protected] is the nearest OTHER row (cosine < 0.20).
5. (Optional, dedup_engine=true) Deploy an Ignite app `crm-dedup` (own SA: k3.editor + ignite.model-user) that
   embeds each new contact, KNN-checks contact_vectors, and flags a dedup_candidate below the 0.20 threshold,
   adjudicating the 0.15-0.30 band on kimi-k2.6.

Ship it

The base core is a small FastAPI app (routes.py) over a declarative bucket — tables, a graph, and a vector. You download it (Get the code), point .env at your bucket, and either run it locally (uvicorn routes:app) or deploy it the same way the sibling engines ship: an image-mode Ignite app the platform builds on deploy and runs scale-to-zero. The one stateful extra is the optional crm-dedup engine (Step 5). DODIL_SERVICE_ACCOUNT_ID is the cli-… serviceAccountId auth service-account create printed (not the uuid):

# IMAGE mode (Dockerfile built on deploy) — NOT --runtime python (compile mode).
dodil ignite app deploy crm-dedup \
  --code ./dedup --dockerfile-path Dockerfile \
  --port 8080 --health-path /healthz --allow-unauthenticated \
  --env BUCKET="$BUCKET" \
  --env DODIL_SERVICE_ACCOUNT_ID="$SA_ID" --env DODIL_SERVICE_ACCOUNT_SECRET="$SA_SECRET" \
  --env DEDUP_THRESHOLD=0.20

The full supply chain — DODIL git → CI checks → a scanned registry image → versioning and rollback — is walked end to end in Ship a DODIL App.

Connect your tools

Everything this build wrote lives in one DataK3 bucket, reachable by your own stack — not just the CLI or the FastAPI app. data connect crm prints the endpoints; point your tools straight at the same rows:

  • SQL over Postgres wire — psql, sqlx/diesel (Rust), psycopg/asyncpg (Python), node-postgres (TS).
  • Graph over Bolt — a Neo4j driver or cypher-shell against crm_accounts.
  • Vector — pgvector (<=>) over the same wire, or a Qdrant/Pinecone client against the same rows.

Full, live-validated walkthrough: Connect your tools.

Conclusion

You now have the master-data core of a CRM on one DataK3 bucket, behind a FastAPI app you can download and run: four merge-keyed master tables you upsert idempotently, an account graph that rolls up pipeline by corporate family, and a VECTOR(2048) table that catches duplicates by meaning before they corrupt a single downstream number — optionally guarded by a kimi-k2.6 dedup engine. One bucket, one bill, three pillars over one copy of the rows — replacing the Salesforce + enrichment + sequencer + warehouse stack with one system of record. This is what the rest of the CRM suite reads.

Next steps:

  • Build a CRM on DataK3 in an Hour — the sequence engine + cost-gated lead pipeline this core feeds.
  • Compose the workflow skills onto this core: lead-to-opportunity, qualification-scoring, pipeline-forecast, quote-cpq, campaign-to-lead, account-360.