What you'll build: the Account 360 — the graph layer of the DODIL CRM suite. Your crm bucket already holds accounts (with a subsidiary hierarchy), contacts, opportunities, and activities as merge-keyed tables, plus the crm_node / crm_edge tables and account_vectors that crm/core owns. This skill assembles the final graph — it adds partner / competitor / supplier edges between companies and owned-product edges from your deals, then snapshots crm_graph — and turns that graph into three answers a flat CRM can't give you:

  • Roll up the whole corporate family — pipeline, headcount, open opps, and a health score for Acme Corp and every subsidiary, in one traversal.
  • Walk the relationship neighborhood — who partners with, competes with, or supplies whom, N hops out.
  • Find the whitespace — which products the family already owns versus your full catalog, ranked by a kimi-k2.6 next-best-action gate, with semantic lookalike targeting over account_vectors.

One bucket answers by content (SQL), by relationship (graph), and by meaning (vector) over one copy of the rows. No Neo4j sync, no Pinecone index, no warehouse.

The problem — and why it matters

You're a named-account rep (or the RevOps team behind them), and you don't sell to acme.io. You sell to the Acme family — Acme Corp, Acme Labs, Acme EU — and they need to know the family pipeline ($36k across two subsidiaries here), which family member has gone quiet, who Acme partners and competes with, and which of your products the family has not bought yet. In a seat-priced CRM that's three integrations and a nightly ETL. On DataK3 it's a graph you assemble once and query three ways.

PieceLands inPillar
Inter-account relationshipstable relationshipsSQL → graph edges
The assembled account graphcrm_node / crm_edge → crm_graphGraph
Family rollup + healthtable account_summarySQL (graph-JOIN)
Cross-sell whitespace gridtable whitespaceSQL + Models gate
Lookalike account targetingaccount_vectors (VECTOR(2048))Vector
The 360 compute loopcrm-account360Ignite (own service account)

NOTE

This is skill 7 of 7 in the CRM suite and the last graph contributor — it runs the single CREATE GRAPH. Because CREATE GRAPH snapshots its edges, every edge from every skill must be in crm_edge before the graph is (re-)created. Standalone, this skill DROP+CREATEs crm_graph itself.

Prerequisites

  • The dodil CLI (dodil auth login) and an agent connected over MCP (Claude Code, Cursor, VS Code, Codex).
  • export BUCKET=crm — the one bucket the whole suite shares.
  • crm/core already scaffolded: accounts (with parent_domain hierarchy), contacts, opportunities, activities, the crm_node / crm_edge tables with base subsidiary_of + works_at edges, and account_vectors. (Standalone, the One-shot below stubs a minimal slice so you can run this skill by itself.)
  • A product catalog — crm/quote-cpq's products, or the minimal one this skill seeds when it's absent.

Step 1 — Stand up the Account-360 tables

Three merge-keyed tables: relationships is the inter-company edge data (it becomes graph edges), account_summary is the per-account 360 rollup, and whitespace is the product×account cross-sell grid. A --merge-key (PRIMARY KEY) is required — writes are keyed, so the nightly recompute upserts idempotently instead of duplicating.

Every table has a natural primary key — never a SERIAL/autoincrement (DataK3 has no sequences). This table-creation step carries a third ORM tab: the exact SQLAlchemy classes from the downloadable package's models.py — money as Numeric(18,2) (DECIMAL), the 0-1 scores as Float. Three front-ends (agent · CLI · ORM), one set of rows.

You

In the crm bucket, create three merge-keyed tables: relationships (key relationship_id: from_domain, to_domain, rel_type, strength(double), note, created_at); account_summary (key org_domain: family_pipeline(double), family_headcount(int), open_opps(int), health_score(double), whitespace_json, last_touched, computed_at); and whitespace (key whitespace_id: org_domain, product_id, has_it(boolean), opportunity_potential(double), rationale).

DODIL MCP tools called
data_table_create
Agent

Created relationships (key relationship_id), account_summary (key org_domain), and whitespace (key whitespace_id) — each with a PRIMARY KEY, so upserts are idempotent and re-runs are safe.

Now record the relationships themselves. rel_type is one of partner / competitor / supplier / customer; strength (0–1) weights the edge. Never write an empty-string key — a "" in a merge-key column reads back null and the row silently drops; use a real relationship_id.

You

In crm, upsert three relationships: rel-1 acme.io is a partner of initech.com (strength 0.8, 'Co-sell alliance'); rel-2 acme.io competes with greyparrot.ai (strength 0.6); rel-3 initech.com is a supplier to acme.io (strength 0.5).

DODIL MCP tools called
data_table_upsert
Agent

Upserted 3 relationships (wal_written: true): acme.io—partner→initech.com, acme.io—competitor→greyparrot.ai, initech.com—supplier→acme.io. These become partner_of / competes_with / supplies graph edges in Step 2.

Step 2 — Assemble the graph (the snapshot rule, load-bearing)

Core's base crm_graph already has subsidiary_of (child→parent) and works_at (contact→account) edges. This skill adds two more edge kinds into the same crm_edge table:

  • relationship edges — partner_of / competes_with / supplies, account→account, projected from relationships.
  • owns_product edges — account→product, i.e. which accounts already bought which product, read from won/open opportunities (or quote_lines). Product nodes live in the 100000–109999 id range.

Then — because CREATE GRAPH snapshots its edges at creation time — you must DROP and re-CREATE crm_graph so the new edges become traversable. Edges added after a graph exists are invisible until you re-snapshot. This is the single most important rule in the graph skill.

You

In crm, insert this skill's edges into crm_edge: partner_of acme.io(1)->initech.com(5); competes_with acme.io(1)->greyparrot.ai(4); supplies initech.com(5)->acme.io(1); and owns_product edges for the products each family account already bought (Acme Labs(2)->prod-core(100001), Acme EU(3)->prod-core(100001)). Then DROP GRAPH crm_graph and CREATE it again so the new edges are snapshotted.

DODIL MCP tools called
data_pg
Agent

Inserted 5 edges (3 relationship + 2 owns_product). Before re-create, a MATCH from node 1 returned 0 outgoing neighbors (the snapshot was stale). After DROP GRAPH + CREATE GRAPH crm_graph, node 1 has 2 outgoing neighbors — Initech (partner_of) and Greyparrot (competes_with).

TIP

Prove the snapshot rule to yourself. Insert the partner_of edge, then before re-creating, run MATCH (a)-[:crm_edge]->(b) WHERE id(a)=1 RETURN b over crm_graph — it returns 0 rows. Re-create the graph and the same query returns 2. Edges are frozen at CREATE GRAPH time.

Step 3 — Traverse it: relationship neighbors + the family rollup

Two traversals fall out of the assembled graph. First, the relationship neighborhood — Acme's partners and competitors — over Bolt/Cypher (the graph plane speaks the Neo4j protocol; the Cypher subset returns a single target node):

You

In crm, over crm_graph, who are Acme Corp's direct relationship neighbors (partners / competitors / suppliers)? Return the neighbor nodes, then hydrate them with the rel_type from crm_edge.

DODIL MCP tools called
data_bolt→data_pg
Agent

From Acme Corp (node 1): 2 outgoing neighbors — Initech (partner_of) and Greyparrot (competes_with). Hydrating crm_edge: Initech=partner_of, Greyparrot=competes_with. The supplies edge (Initech->Acme) is incoming, surfaced by flipping the arrow.

Now the family rollup. Core rolls up a hierarchy-only graph with graph_khop(...,'in'). But the graph is now multi-relational — a supplies edge (Initech→Acme) is also an incoming edge to node 1, so an untyped k-hop would wrongly pull Initech into the Acme family. The corporate family is exactly the accounts reachable by subsidiary_of edges, so filter to that relationship with a typed recursive traversal — then JOIN to opportunities, accounts, and activities for pipeline, headcount, and the health score.

You

In crm, roll up the Acme corporate family from node 1 by following ONLY subsidiary_of edges: total open pipeline, open-opp count, family headcount (from employee_band), and the most recent activity timestamp. Exclude partners/suppliers.

DODIL MCP tools called
data_pg
Agent

Family = Acme Corp + Acme Labs + Acme EU (Initech excluded — it's a supplier, not a subsidiary). family_pipeline = $36,000 across 2 open opps; family_headcount = 4200; last_touched = 2026-08-30. Greyparrot ($30k) is out of the tree.

NOTE

graph_khop is only referenceable in a top-level FROM, not inside a CTE. The recursive-CTE form above traverses crm_edge directly with an explicit rel = 'subsidiary_of' filter — the right tool once the graph carries more than one relationship kind.

Step 4 — Find the whitespace (owned products vs the catalog)

The whitespace grid is products the family already owns (the owns_product edges) versus your full active catalog. Every gap — a product the family lacks — is a cross-sell candidate, valued at the product's list price.

You

In crm, build the whitespace grid for the Acme family (head acme.io): for each active product, is it owned by anyone in the subsidiary_of family, and if not, what's the list-price opportunity potential? Return one row per product.

DODIL MCP tools called
data_pg
Agent

3 products, 2 gaps: the Acme family owns DataK3 Core (from its deals) but NOT Ignite Compute ($18,000 potential) or Models Gateway ($12,000). One row per product; has_it=false marks each whitespace gap.

Step 5 — Lookalike targeting (Vector)

Which other accounts look like your best family — so you can pitch them the same product? Reuse core's account_vectors (VECTOR(2048), jina-embeddings-v4, indexed — CREATE INDEX ON account_vectors (embedding), because an unindexed VECTOR is an exact scan on every KNN): embed each account's profile once, then KNN by meaning. An enterprise-software peer ranks far above an unrelated vision-analytics company.

You

In crm, embed each account's profile with jina-embeddings-v4 into account_vectors (one row per upsert), then KNN-search for accounts most like an enterprise-software firm consolidating Postgres/Neo4j/Pinecone into one unified data platform.

DODIL MCP tools called
ignite_models_embed→data_table_upsert→data_vsearch
Agent

Embedded 5 account profiles (2048-dim, one vector per upsert). Nearest to the ideal-buyer query: Initech (cosine 0.20) — an enterprise-software lookalike of Acme — then Acme itself (0.32); Greyparrot (waste-analytics vision, 0.57) ranks last. Initech is the whitespace target.

Routes

The download (see Get the code) fronts this bucket with a small FastAPI app, routes.py — relationship + summary CRUD plus the ops that turn the graph into a 360. 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 (this one at /account-360) — while app = FastAPI(...) at the bottom keeps the package independently runnable (uvicorn routes:app). Every route follows the same DataK3 rules the package bakes in, so quoting it is documenting them. Every value below is live-captured — first against a throwaway bucket (2026-09-04), then re-validated 2026-09-06 on the persistent crm bucket, whose suite-seeded funnel extends the standalone demo (same Acme family; the relationship set adds partner Globex, and the product catalog is crm/quote-cpq's).

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 the nightly recompute land each row once (re-running the family rollup keeps account_summary at exactly 1 row for acme.io; the crm_edge all-key rows take the DO NOTHING branch). Money columns (family_pipeline, opportunity_potential) are Numeric(18,2) written over the pg wire — the gRPC upsert path can silently drop an integer 0 into a DECIMAL, so the app writes DECIMAL with psycopg's proper numeric type (confirmed live: the columns land as numeric).

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). The relationship route guards the empty-string key that would silently drop the row:

# routes.py — relationship CRUD, keyed on the natural PK
@router.post("/relationships")
def upsert_relationship(r: RelationshipIn, s: Session = Depends(db)):
    if not r.relationship_id:
        raise HTTPException(422, "relationship_id must be non-empty (a '' merge-key drops the row)")
    upsert(s, Relationship, [r.model_dump()], key="relationship_id")
    s.commit()
    return {"ok": True, "relationship_id": r.relationship_id}

Workflow op 1 — assemble the graph (the snapshot rule, load-bearing). POST /graph/assemble projects this skill's edges into the one crm_edge table — relationship edges from the relationships rows (partner→partner_of, …), owns_product edges from the request body — then DROP GRAPH IF EXISTS + CREATE GRAPH. Because CREATE GRAPH snapshots its edges, every edge must exist before the (re-)create:

# routes.py — workflow op 1: assemble the graph, then re-snapshot (GRAPH)
@router.post("/graph/assemble")
def assemble_graph(a: AssembleIn, s: Session = Depends(db)):
    edges: list[dict] = []
    # relationship edges: resolve each domain to its node id, map rel_type -> rel kind
    for r in s.execute(select(Relationship)).scalars().all():
        rel = REL_KIND.get((r.rel_type or "").lower())
        src, dst = _node_id(s, r.from_domain), _node_id(s, r.to_domain)
        if rel and src is not None and dst is not None:
            edges.append({"src": src, "dst": dst, "rel": rel})
    # owns_product edges: account node -> product node
    for op in a.owns_product:
        src = _node_id(s, op.org_domain)
        dst = s.execute(
            select(CrmNode.id).where(CrmNode.biz_key == op.product_id, CrmNode.kind == "product")
        ).scalar()
        if src is not None and dst is not None:
            edges.append({"src": src, "dst": dst, "rel": "owns_product"})
    if edges:
        upsert(s, CrmEdge, edges, key=("src", "dst", "rel"))  # DO NOTHING (pure edge rows)
    # COMMIT the projected edges BEFORE the re-snapshot: DataK3 has no read-your-writes
    # inside an open transaction, so a CREATE GRAPH in the SAME txn snapshots the table
    # WITHOUT the rows just upserted (found live on the suite bucket: /neighbors came back
    # empty while every table-reading route saw the edges). Two transactions, in order.
    s.commit()
    # re-snapshot: edges added AFTER a graph exists are invisible until re-create.
    s.execute(text("DROP GRAPH IF EXISTS " + GRAPH))
    s.execute(text(
        "CREATE GRAPH " + GRAPH + " NODES (crm_node KEY id) EDGES (crm_edge SRC src DST dst)"
    ))
    s.commit()
    return {"ok": True, "edges_projected": len(edges), "graph": GRAPH}

That mid-function s.commit() is a hard-won line. The first version upserted the edges and DROP+CREATEd the graph in the same transaction — and every /neighbors call came back empty, while every table-reading route saw the edges perfectly. The snapshot rule had collided with a second DataK3 rule: no read-your-writes inside an open transaction. CREATE GRAPH reads crm_edge to build its snapshot, so in the same txn it snapshots the table without the rows just upserted — silently. The fix is two transactions, in order: commit the edges, then DROP+CREATE. (Standalone — each CLI statement its own transaction — you never hit this; composed into one route, you do.)

Live-verified: assemble_graph projected the relationship + owns_product edges onto the base subsidiary_of edges; after the commit-then-re-snapshot, a cypher() from node 1 returns its outgoing relationship neighbors — 0 before the DROP+CREATE.

Workflow op 2 — the corporate-family rollup (GRAPH). The corporate family is exactly the accounts reachable by subsidiary_of edges — a supplies edge (Initech→Acme) is also incoming to node 1, so an untyped traversal would wrongly pull Initech in. Here a load-bearing DataK3 fact surfaced live: the embedded cypher() subset supports only an id(<var>) = <key> anchor — it can't filter on an edge's rel (cypher subset: unsupported WHERE condition at 'ALL'). So the typed traversal uses a recursive CTE over crm_edge (a top-level SELECT, integer-literal root, node ids → IN (…)), while cypher() is reserved for the untyped neighborhood below:

# routes.py — the subsidiary_of family via a typed recursive CTE (cypher() can't filter edge rel)
def _subsidiary_family(s: Session, root_id: int) -> list[int]:
    fam = s.execute(
        text(
            "WITH RECURSIVE family(node) AS ("
            "  SELECT " + str(int(root_id)) + " "
            "  UNION "
            "  SELECT e.src FROM crm_edge e JOIN family f ON e.dst = f.node "
            "  WHERE e.rel = 'subsidiary_of') "
            "SELECT node FROM family"
        )
    ).scalars().all()
    return list(dict.fromkeys([int(root_id), *[int(k) for k in fam]]))  # root first, de-duped
 
 
@router.get("/accounts/{org_domain}/family")
def family_rollup(org_domain: str, s: Session = Depends(db)):
    root = _node_id(s, org_domain)
    if root is None:
        raise HTTPException(404, "no such account node")
    fam = _subsidiary_family(s, root)
    ids = ",".join(str(i) for i in fam)  # inlined int IN(…) — DuckDB has no `= ANY(array)`
    row = s.execute(text(
        "WITH fam AS ("
        "  SELECT n.biz_key AS org_domain, a.employee_band"
        "  FROM crm_node n JOIN accounts a ON a.org_domain = n.biz_key"
        f"  WHERE n.id IN ({ids}) AND n.kind = 'account') "
        "SELECT"
        "  (SELECT COALESCE(SUM(o.amount),0) FROM opportunities o"
        "     WHERE o.status='open' AND o.account_domain IN (SELECT org_domain FROM fam)) AS family_pipeline,"
        "  (SELECT COUNT(*) FROM opportunities o"
        "     WHERE o.status='open' AND o.account_domain IN (SELECT org_domain FROM fam)) AS open_opps,"
        "  (SELECT COALESCE(SUM(CASE employee_band WHEN '1000-5000' THEN 3000"
        "            WHEN '200-1000' THEN 600 WHEN '50-200' THEN 125 ELSE 0 END),0)"
        "     FROM fam) AS family_headcount,"
        "  (SELECT MAX(a.ts) FROM activities a JOIN opportunities o ON o.opportunity_id = a.opportunity_id"
        "     WHERE o.account_domain IN (SELECT org_domain FROM fam)) AS last_touched"
    )).one()
    family_pipeline, open_opps, family_headcount, last_touched = row
    health = _health_score(last_touched)
    upsert(s, AccountSummary, [{
        "org_domain": org_domain, "family_pipeline": family_pipeline,
        "family_headcount": int(family_headcount or 0), "open_opps": int(open_opps or 0),
        "health_score": health, "last_touched": str(last_touched) if last_touched else None,
        "computed_at": _now(),
    }], key="org_domain")
    s.commit()
    return {"org_domain": org_domain, "family_node_ids": fam,
            "family_pipeline": float(family_pipeline or 0), "open_opps": int(open_opps or 0),
            "family_headcount": int(family_headcount or 0), "health_score": health,
            "last_touched": str(last_touched) if last_touched else None}

Live-verified 2026-09-06 on the crm bucket: GET /accounts/acme.io/family returned family_node_ids [1, 2, 3] (partner/supplier Globex and unrelated Greyparrot correctly excluded), family_pipeline 36000, open_opps 2, family_headcount 4200, health_score 0.94. No flat query gives you that number — and the typed traversal is what keeps it honest: pipeline-forecast learned the hard way that an untyped family walk pulls a supplier in the moment these multi-relational edges land in the shared graph (see Pipeline forecast).

Workflow op 2b — the relationship neighborhood (GRAPH via cypher()). This is where cypher() earns its keep: an untyped outgoing-neighbor walk (any relationship edge is a valid neighbor), a top-level SELECT with an integer-literal anchor, then a SQL IN (…) to hydrate the kind from crm_edge:

# routes.py — workflow op 2b: relationship neighbors (cypher top-level SELECT, int-literal anchor)
@router.get("/accounts/{org_domain}/neighbors")
def relationship_neighbors(org_domain: str, s: Session = Depends(db)):
    root = _node_id(s, org_domain)
    if root is None:
        raise HTTPException(404, "no such account node")
    nbrs = s.execute(text(
        "SELECT * FROM cypher('" + GRAPH + "', "
        "'MATCH (a)-[]->(b) WHERE id(a) = " + str(int(root)) + " RETURN b')"
    )).scalars().all()
    inter = [int(n) for n in nbrs]
    if not inter:
        return {"org_domain": org_domain, "neighbors": []}
    ids = ",".join(str(i) for i in inter)  # inlined int IN(…) — DuckDB has no `= ANY(array)`
    rows = s.execute(text(
        "SELECT n.biz_key AS org_domain, n.name, e.rel "
        "FROM crm_edge e JOIN crm_node n ON n.id = e.dst "
        f"WHERE e.src = {int(root)} AND e.dst IN ({ids}) "
        "AND e.rel IN ('partner_of','competes_with','supplies') ORDER BY e.rel"
    )).all()
    return {"org_domain": org_domain,
            "neighbors": [{"org_domain": d, "name": nm, "rel": rel} for d, nm, rel in rows]}

cypher() yields a single column named after the RETURN variable — SELECT * + .scalars() reads it without hard-coding the name. Live-verified 2026-09-06 on the suite bucket: GET /accounts/acme.io/neighbors returned Globex (partner_of) — the outgoing relationship edge; the supplies edge points the other way (Globex→Acme), which is exactly why it doesn't appear here but would leak into an untyped inward family walk.

Workflow op 3 — the whitespace grid. The family's owned products (owns_product edges of the subsidiary_of family) LEFT JOINed to the active catalog — one row per product, has_it=false marks a gap valued at list price, upserted idempotently:

# routes.py — workflow op 3: owned products vs the catalog (SQL over the assembled graph)
@router.post("/accounts/{org_domain}/whitespace")
def build_whitespace(org_domain: str, s: Session = Depends(db)):
    root = _node_id(s, org_domain)
    if root is None:
        raise HTTPException(404, "no such account node")
    fam = _subsidiary_family(s, root)
    ids = ",".join(str(i) for i in fam)
    grid = s.execute(text(
        "WITH owned AS ("
        "  SELECT DISTINCT pn.biz_key AS product_id"
        "  FROM crm_edge e JOIN crm_node pn ON pn.id = e.dst AND pn.kind = 'product'"
        f"  WHERE e.src IN ({ids}) AND e.rel = 'owns_product') "
        "SELECT p.product_id, p.name, p.list_price,"
        "       CASE WHEN o.product_id IS NULL THEN false ELSE true END AS has_it "
        "FROM products p LEFT JOIN owned o ON o.product_id = p.product_id "
        "WHERE p.active ORDER BY has_it, p.product_id"
    )).all()
    rows = [{
        "whitespace_id": f"ws-{org_domain}-{pid}", "org_domain": org_domain, "product_id": pid,
        "has_it": bool(has), "opportunity_potential": (0 if has else list_price), "rationale": "",
    } for pid, _name, list_price, has in grid]
    if rows:
        upsert(s, Whitespace, rows, key="whitespace_id")
    s.commit()
    gaps = [{"product_id": r["product_id"], "opportunity_potential": float(r["opportunity_potential"] or 0)}
            for r in rows if not r["has_it"]]
    return {"org_domain": org_domain, "products": len(rows), "gaps": gaps}

Live-verified 2026-09-06 against the suite's real crm/quote-cpq catalog: 3 products, 2 gaps — prod-ignite (opportunity_potential 900.00, its list price) and prod-models (600.00); the family owns prod-datak3. (The standalone stub catalog in Step 4 carries its own demo prices — the shape of the answer is the same.)

Workflow op 4 — lookalike targeting (VECTOR). POST /accounts/lookalike takes a query embedding and returns the nearest accounts by pgvector cosine distance over account_vectors — the same rows, by meaning:

# routes.py — workflow op 4: lookalike targeting (pgvector cosine)
@router.post("/accounts/lookalike")
def lookalike_accounts(q: LookalikeIn, s: Session = Depends(db)):
    rows = s.execute(
        select(AccountVector.org_domain, AccountVector.embedding.cosine_distance(q.embedding).label("d"))
        .order_by("d")
        .limit(q.top_k)
    ).all()
    return {"matches": [{"org_domain": d, "distance": float(dist)} for d, dist in rows]}

Live-verified 2026-09-06 against real jina-embeddings-v4 vectors over the suite's account_vectors (the indexed VECTOR(2048) column): the nearest lookalike to the ideal-buyer query came back at cosine 0.2719, with the unrelated account ranking last — the enterprise-software peers cluster, the odd one out ranks furthest.

Workflow op 5 — the NBA gate (MODELS). POST /accounts/{org_domain}/rank-nba bundles the gaps, owned products, and pipeline and asks kimi-k2.6 to rank + explain, then writes the verdict back onto the winning whitespace row and account_summary.whitespace_json. It's the only route that leaves the bucket — it mints a service-account token and calls https://api.dodil.io/v1:

# routes.py — workflow op 5: rank the next best action on kimi-k2.6, write the verdict back
def _rank_nba(token: str, bundle: str) -> dict:
    """kimi-k2.6 is a reasoning model: set max_tokens high (4096) or `content` comes back empty;
    the response is wrapped in `data`."""
    client = httpx.Client(base_url=MODELS_BASE, headers={**_UA, "Authorization": f"Bearer {token}"})
    body = {"model": CHAT_MODEL, "max_tokens": 4096,
            "messages": [{"role": "system", "content": NBA_SYS},
                         {"role": "user", "content": bundle}]}
    out = client.post("/chat/completions", json=body, timeout=90).json()
    env = out.get("data", out)                      # this platform wraps the response in `data`
    content = env["choices"][0]["message"]["content"]
    m = re.search(r"\{.*\}", content or "", re.DOTALL)
    return json.loads(m.group(0) if m else content)
 
 
@router.post("/accounts/{org_domain}/rank-nba")
def rank_nba(org_domain: str, s: Session = Depends(db)):
    gaps = s.execute(
        select(Whitespace.product_id, Whitespace.opportunity_potential)
        .where(Whitespace.org_domain == org_domain, Whitespace.has_it.is_(False))
    ).all()
    if not gaps:
        raise HTTPException(409, "no whitespace gaps — run /whitespace first")
    owned = s.execute(
        select(Whitespace.product_id)
        .where(Whitespace.org_domain == org_domain, Whitespace.has_it.is_(True))
    ).scalars().all()
    summary = s.get(AccountSummary, org_domain)
    pipeline = float(summary.family_pipeline) if summary and summary.family_pipeline else 0.0
    bundle = (f"Family head: {org_domain}. Owns: {', '.join(owned) or 'none'}. "
              f"Whitespace gaps: {', '.join(f'{p} (potential {float(v or 0)})' for p, v in gaps)}. "
              f"Open pipeline: {pipeline}.")
    nba = _rank_nba(_models_token(), bundle)
    win = nba.get("next_best_product")
    row = s.get(Whitespace, f"ws-{org_domain}-{win}")
    if row:
        merged = {c.name: getattr(row, c.name) for c in Whitespace.__table__.columns}
        merged["rationale"] = f"NBA {nba.get('priority')}: {nba.get('rationale')}"
        upsert(s, Whitespace, [merged], key="whitespace_id")
    if summary:
        merged = {c.name: getattr(summary, c.name) for c in AccountSummary.__table__.columns}
        merged["whitespace_json"] = json.dumps({"owned": list(owned),
                                                 "gaps": [g[0] for g in gaps], "next_best": win})
        upsert(s, AccountSummary, [merged], key="org_domain")
    s.commit()
    return {"org_domain": org_domain, "nba": nba}

Live-verified 2026-09-06: kimi-k2.6 returned strict JSON — next_best_product: prod-ignite, priority: high, rationale "Highest whitespace potential; active pipeline indicates strong current engagement." The write-back then landed whitespace.rationale = "NBA high: …" and account_summary.whitespace_json = {owned: [prod-datak3], gaps: [prod-ignite, prod-models], next_best: prod-ignite}. (The rationale write-back reads the row and upserts the full merged dict, so the other columns are kept. And a kimi-k2.6 call runs 12s–170s — the route reads its inputs first and writes only after the reply lands; never hold a pg connection open across the model call.)

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 @router.<verb> function: write via upsert, typed graph via a recursive CTE, untyped graph via cypher(…), vector via cosine_distance(…) (see EXTENDING.md in the package).

Auth — config at the edge, no gate in this component

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, issuer https://appid.dodil.io/ihdiash/crm-suite) and the per-cluster Ignite gateway runs the entire browser login at the edge — PKCE S256, an AEAD-sealed session cookie, JWT verification — then injects the verified identity as X-Dodil-User (plus X-Dodil-User-Jwt, carrying the catalog-expanded permissions claim). Inbound copies of those headers are stripped, so they can't be forged. The package's auth.py is a header-trust reader, not a verifier — no JWKS client, no issuer/audience env — exposing current_user and require_permission for role gating.

Account-360 kept no role gate after the audit — nothing in this component's routes.py imports auth. Its reads are safe for any signed-in rep, its writes are idempotent recomputes, and the NBA gate's verdict is advisory (a rationale on a row, not a state change) — so the gateway's authentication alone guards these routes. The four gates that did survive live in the sibling components: orgs:qualify (lead-to-opportunity), leads:score (qualification-scoring), quotes:approve (quote-cpq), forecast:override (pipeline-forecast) — checked against the pool's sales / analyst / manager role catalog. Want to make the model spend manager-only here too? It's one line: Depends(require_permission("whitespace:rank")) on rank_nba, plus the catalog entry.

The two-plane rule is unchanged: the pool identifies the user; the app reaches DataK3 through its own service account (the same one the crm-account360 compute loop uses) — an app-user is never a bucket principal. Locally, DEV_ALLOW_ANON=1 opts into a stub identity. Pool creation, the redirect_uris allowlist, and the off-gateway verify-it-yourself path (iss and aud mandatory): App authentication; the catalog: App roles.

Get the code

The package is a real download — code/crm-account-360/v1.tar. This post is a walkthrough of exactly those files; db.py + auth.py are shared byte-identical with the rest of the suite, and the tarball is source only (no Dockerfile — deploy is in Ship it):

models.py          # SQLAlchemy — relationships, account_summary, whitespace + crm_node/crm_edge + stubbed masters
routes.py          # FastAPI    — an APIRouter the suite mounts + a standalone app; CRUD + assemble-graph
                   #              + family rollup + neighbors + whitespace + lookalike + NBA gate
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 /graph/assemble · GET /accounts/{id}/family · GET /accounts/{id}/neighbors
# POST /accounts/{id}/whitespace · POST /accounts/lookalike · POST /accounts/{id}/rank-nba

models.Base.metadata.create_all is the ORM tab's every class at once — the same tables Step 1 built by CLI, created from the natural-key models (money as DECIMAL over the pg wire) with no migration tool.

In the suite, this component doesn't run alone: the seven crm/* packages compose into one Ignite app — crm-suite-app mounts each component's APIRouter under a per-component prefix (this one at /account-360) on one FastAPI process, over one engine to the one crm bucket, deployed through the git cycle with user_pool: crm-suite — see Ship a DODIL app. The prefix is also what lets core and account-360 both expose a GET /accounts/…/family without colliding.

Step 6 — Rank the next best action (Models gate) + the compute loop

Deterministic gap detection gives you the list of whitespace; the kimi-k2.6 gate ranks and explains it. Given the family, its owned products, the gap list with potential values, the peer lookalike, and open pipeline, it returns strict JSON the handler writes back into whitespace.rationale and account_summary.whitespace_json.

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. (An Ignite handler sets max_tokens: 4096 on the raw api.dodil.io/v1 call — the interactive CLI/MCP path can't.)

You

Rank the next best product to cross-sell into the Acme family on kimi-k2.6. It owns DataK3 Core; gaps are prod-ignite (Ignite Compute, potential 18000) and prod-models (Models Gateway, 12000); peer lookalike is Initech; open pipeline is 36000. Return ONLY JSON: next_best_product, priority (high|med|low), rationale.

DODIL MCP tools called
ignite_models_chat→data_table_update
Agent

{"next_best_product":"prod-ignite","priority":"high","rationale":"Highest whitespace aligned to owned Core; active 36k pipeline and peer modernization signal."}

The compute loop is one small Ignite app — crm-account360. Each hit it traverses crm_graph per top-level account, computes account_summary, fills the whitespace grid, and calls the gate. It's a separate workload, so it runs under its own service account — k3.editor to write tables, plus ignite.model-user to call the gate (drop ignite.model-user and set nba_gate=false for a pure-SQL, no-Models build) and ignite.app-developer as the deploy identity.

This ships as an image-mode Ignite app: a plain HTTP server (GET /healthz for the probe, POST /account360 for the work), packaged by a Dockerfile and built on deploy — not a handler(payload, ctx) compile-mode function. Four things matter, each a line below:

  • Graph reads go over Bolt. The family traversal — the root plus every node reachable by subsidiary_of edges only (typed, so partner/supplier edges never leak in) — and the owns_product read run over the Bolt/Cypher plane (bolt+s://bolt.uk-lon-1.dodil.io:7687), authed with the service-account token, via the neo4j driver.
  • Relational reads + writes go over the drop-in Postgres wire (pg.uk-lon-1.dodil.io:5432, dbname=<bucket>, user=token, password=<the SA access token>) via psycopg — there is no K3 HTTP API. The family rollup, the pgvector <=> lookalike over account_vectors, and the whitespace grid are SQL; the writes are too. The recompute re-writes the same whitespace_id / org_domain each run, so the writes are INSERT … ON CONFLICT (<pk>) DO UPDATE (a bare re-INSERT of an already-committed PK raises duplicate-key 23505 — a plain INSERT is not an upsert on re-write; DuckDB pg-wire supports ON CONFLICT, or use the managed data_table_upsert). Writes retry on SerializationFailure.
  • The Models gate is the real endpoint (api.dodil.io/v1), authed with the same token, max_tokens: 4096 (kimi-k2.6 is a reasoning model — a low budget returns empty content), response wrapped in data (out["data"]["choices"]…).
  • Every call to id.dodil.io / api.dodil.io sets an explicit User-Agent — stdlib urllib's default is Cloudflare-banned (HTTP 403 "error code: 1010").
# server.py — crm-account360, an IMAGE-mode Ignite app (HTTP server on $PORT).
#   GET  /healthz    -> {"status":"ready"}    (probe; no auth)
#   POST /account360 -> {"root_node_id","org_domain"} -> rolls up the family, fills whitespace, ranks the NBA
 
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
from psycopg import errors as pg_errors
from neo4j import GraphDatabase
 
# --- hoisted knobs, injected as env at deploy (== skill param defaults) ---
MODEL_ID           = os.environ.get("MODEL_ID", "kimi-k2.6")
NBA_GATE           = os.environ.get("NBA_GATE", "true").lower() == "true"   # == param nba_gate default
HEALTH_WINDOW_DAYS = int(os.environ.get("HEALTH_WINDOW_DAYS", "90"))        # recency window for health_score
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"))
BOLT_URL           = os.environ.get("BOLT_URL", "bolt+s://bolt.uk-lon-1.dodil.io:7687")
GRAPH              = os.environ.get("GRAPH", "crm_graph")
 
ID_URL     = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
MODELS_URL = "https://api.dodil.io/v1/chat/completions"
UA         = "crm-account360/1.0"   # explicit UA — stdlib urllib's default is Cloudflare-banned (403 1010)
 
NBA_SYS = ("Rank the next best product to cross-sell into an account family. Given the family, its owned "
           "products, whitespace gaps (product_id + potential value), a peer lookalike, and open pipeline, "
           'return ONLY JSON: {"next_best_product":"<product_id>","priority":"high|med|low","rationale":"<=20 words"}.')
 
SUMMARY_COLS = ["org_domain", "family_pipeline", "family_headcount", "open_opps", "health_score",
                "whitespace_json", "last_touched", "computed_at"]
WS_COLS      = ["whitespace_id", "org_domain", "product_id", "has_it", "opportunity_potential", "rationale"]
 
def _now(): return datetime.now(timezone.utc).isoformat()
 
def _http_post(url, data, headers, form=False):
    headers = {"User-Agent": UA, **headers}          # the UA is required (see above)
    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():                                        # OIDC client_credentials -> access 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 _chat(token, bundle):                            # kimi-k2.6: max_tokens 4096, response wrapped in "data"
    out = _http_post(MODELS_URL,
                     {"model": MODEL_ID, "max_tokens": 4096,
                      "messages": [{"role": "system", "content": NBA_SYS},
                                   {"role": "user", "content": bundle}]},
                     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=bucket, user=token, pw=SA token
    return psycopg.connect(host=PG_HOST, port=PG_PORT, dbname=BUCKET,
                           user="token", password=token, sslmode="require",
                           connect_timeout=20, autocommit=False)
 
def _bolt(token):                                    # graph reads speak the Neo4j/Bolt protocol; auth = SA token
    return GraphDatabase.driver(BOLT_URL, auth=("token", token))
 
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 _retry(fn):                                      # pg engine is serializable — retry transient conflicts
    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 _days_since(iso):
    if not iso: return HEALTH_WINDOW_DAYS
    try:
        t = datetime.fromisoformat(str(iso).replace("Z", "+00:00"))
    except ValueError:
        return HEALTH_WINDOW_DAYS
    if t.tzinfo is None: t = t.replace(tzinfo=timezone.utc)
    return (datetime.now(timezone.utc) - t).days
 
# --- GRAPH reads over Bolt: the subsidiary_of family + what it already owns ---------------------
def _family_over_bolt(token, root):
    # Family = root + every node reachable by subsidiary_of edges ONLY (typed; immune to partner/supplier).
    fam_cy = ("MATCH (m)-[rels:crm_edge*0..8]->(root) "
              "WHERE id(root) = $root AND ALL(e IN rels WHERE e.rel = 'subsidiary_of') "
              "RETURN DISTINCT id(m) AS nid, m.biz_key AS org_domain")
    owned_cy = ("MATCH (fam)-[o:crm_edge]->(p) "
                "WHERE id(fam) IN $ids AND o.rel = 'owns_product' "
                "RETURN DISTINCT p.biz_key AS product_id")
    with _bolt(token) as drv, drv.session(database=GRAPH) as ses:
        fam = ses.run(fam_cy, root=root).data()
        ids = [r["nid"] for r in fam]
        domains = [r["org_domain"] for r in fam if r["org_domain"]]
        owned = [r["product_id"] for r in ses.run(owned_cy, ids=ids).data()]
    return domains, owned
 
# --- SQL + VECTOR reads over pg-wire -----------------------------------------------------------
FAM_ROLLUP = """
  SELECT
    (SELECT coalesce(sum(o.amount),0) FROM opportunities o
       WHERE o.status='open' AND o.account_domain = ANY(%(d)s))                 AS family_pipeline,
    (SELECT count(*) FROM opportunities o
       WHERE o.status='open' AND o.account_domain = ANY(%(d)s))                 AS open_opps,
    (SELECT coalesce(sum(CASE a.employee_band WHEN '1000-5000' THEN 3000
              WHEN '200-1000' THEN 600 WHEN '50-200' THEN 125 ELSE 0 END),0)
       FROM accounts a WHERE a.org_domain = ANY(%(d)s))                         AS family_headcount,
    (SELECT max(a.ts) FROM activities a JOIN opportunities o
       ON o.opportunity_id = a.opportunity_id
       WHERE o.account_domain = ANY(%(d)s))                                     AS last_touched"""
# lookalike = the family head's nearest peer BY MEANING — pgvector <=> over account_vectors (vector pillar)
LOOKALIKE = """
  SELECT av2.org_domain AS peer, (av1.embedding <=> av2.embedding) AS dist
  FROM account_vectors av1, account_vectors av2
  WHERE av1.org_domain = %(head)s AND av2.org_domain <> %(head)s
  ORDER BY dist ASC LIMIT 1"""
GRID = """
  SELECT p.product_id, p.name, p.list_price, (p.product_id = ANY(%(owned)s)) AS has_it
  FROM products p WHERE p.active ORDER BY has_it, p.product_id"""
 
def _rank_nba(token, org, owned, gaps, pipeline, peer):
    bundle = (f"Family head: {org}. Owns: {', '.join(owned) or 'none'}. "
              f"Whitespace gaps: {', '.join(f'{p} (potential {v})' for p, v in gaps) or 'none'}. "
              f"Peer lookalike: {peer or 'none'}. Open pipeline: {pipeline}.")
    return _extract_json(_chat(token, bundle))
 
def _write_rows(token, table, cols, rows, pk=None):
    # The nightly recompute re-writes the SAME PK each run (whitespace_id / org_domain), so this is an
    # ON CONFLICT upsert: a bare re-INSERT of a committed PK raises duplicate-key 23505. DuckDB pg-wire
    # supports ON CONFLICT (verified live); the managed data_table_upsert is the alternative.
    pk  = pk or cols[0]
    setc = ", ".join(f"{c} = EXCLUDED.{c}" for c in cols if c != pk)
    sql = (f"INSERT INTO {table} ({', '.join(cols)}) VALUES ({', '.join(['%s'] * len(cols))}) "
           f"ON CONFLICT ({pk}) DO UPDATE SET {setc}")
    def _w():
        with _pg(token) as conn, conn.cursor() as cur:
            for r in rows:
                cur.execute(sql, [r[c] for c in cols])
            conn.commit()
    _retry(_w)
 
def compute_360(payload):
    root = int((payload or {}).get("root_node_id", 1))
    org  = (payload or {}).get("org_domain", "acme.io")
    token = _token()
 
    domains, owned = _family_over_bolt(token, root)          # GRAPH plane (Bolt)
    if org not in domains: domains = [org] + domains
 
    with _pg(token) as conn, conn.cursor() as cur:           # SQL + VECTOR plane (pg-wire)
        cur.execute(FAM_ROLLUP, {"d": domains})
        pipeline, open_opps, headcount, last_touched = cur.fetchone()
        cur.execute(LOOKALIKE, {"head": org})
        peer_row = cur.fetchone()
        peer = peer_row[0] if peer_row else None
        cur.execute(GRID, {"owned": owned})
        grid = cur.fetchall()                                # (product_id, name, list_price, has_it)
 
    gaps   = [(pid, price) for pid, _n, price, has in grid if not has]
    ownedp = [pid for pid, _n, _p, has in grid if has]
    health = round(1 - min(_days_since(last_touched), HEALTH_WINDOW_DAYS) / HEALTH_WINDOW_DAYS, 2)
    nba    = _rank_nba(token, org, ownedp, gaps, pipeline, peer) if (NBA_GATE and gaps) else None   # MODELS gate
 
    ws_rows = [{"whitespace_id": f"ws-{org}-{pid}", "org_domain": org, "product_id": pid,
                "has_it": bool(has), "opportunity_potential": 0 if has else price,
                "rationale": (f"NBA {nba['priority']}: {nba['rationale']}"
                              if nba and pid == nba.get("next_best_product") else "")}
               for pid, _n, price, has in grid]
    _write_rows(token, "whitespace", WS_COLS, ws_rows)       # WRITE over pg-wire
    _write_rows(token, "account_summary", SUMMARY_COLS, [{
        "org_domain": org, "family_pipeline": float(pipeline or 0), "family_headcount": int(headcount or 0),
        "open_opps": int(open_opps or 0), "health_score": health,
        "whitespace_json": json.dumps({"owned": ownedp, "gaps": [g[0] for g in gaps],
                                       "next_best": (nba or {}).get("next_best_product"), "peer_lookalike": peer}),
        "last_touched": last_touched, "computed_at": _now()}])
 
    return {"org_domain": org, "family_pipeline": float(pipeline or 0), "family_headcount": int(headcount or 0),
            "open_opps": int(open_opps or 0), "health_score": health, "peer_lookalike": peer,
            "nba": nba, "written": True}
 
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 != "/account360": return self._send(404, {"error": "no_route", "path": self.path})
        try:
            n = int(self.headers.get("Content-Length") or 0)
            body = json.loads(self.rfile.read(n) or b"{}")
            return self._send(200, compute_360(body))
        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-account360 serving on 0.0.0.0:{port} bucket={BUCKET} nba_gate={NBA_GATE}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

psycopg (Postgres wire) and neo4j (Bolt) are the only third-party deps; everything else is stdlib. The two sibling files that make it an image — ./account360-app/Dockerfile and ./account360-app/requirements.txt:

# ./account360-app/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"]
# ./account360-app/requirements.txt
psycopg[binary]==3.2.3
neo4j==5.24.0

Give it its own least-privilege identity, then deploy:

You

Create a service account crm-account360-sa, grant it k3.editor plus ignite.model-user (NBA gate) plus ignite.app-developer, then deploy my ./account360-app (image mode — its Dockerfile builds on deploy) to Ignite as crm-account360 on port 8080 with health path /healthz, passing the service-account creds, BUCKET, and NBA_GATE=true as runtime env. Use the cli- serviceAccountId as DODIL_SERVICE_ACCOUNT_ID, not the uuid. Then POST Acme to /account360 to smoke-test.

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

Created crm-account360-sa (serviceAccountId cli-crm-account360-sa), granted k3.editor + ignite.model-user + ignite.app-developer, built + deployed crm-account360 (image:build, public FQDN on :8080, scale-to-zero). POST /account360 for root_node_id=1 returned family_pipeline 36000, open_opps 2, health_score 0.98, nba prod-ignite, written=true.

NOTE

Deploy: image mode (Lane B), validated live 2026-09-02. The crm-account360 deploy + /account360 smoke-test above use image mode — a Dockerfile + --dockerfile-path, Kaniko build-on-deploy. 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, independently re-confirmed at +95s). crm-account360 reuses that identical handler/deploy pattern. Its own operations are each also proven one call at a time — the Bolt family traversal, the pg-wire rollup/lookalike/grid, the gate, and the keyed writes are validated in Steps 2–6 and ## Test.

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, three pillars, one copy of the rows — this is where the graph plane earns its keep.

JobThe usual stackOn DataK3
Inter-company relationships + hierarchyNeo4j + a sync jobcrm_graph — crm_node / crm_edge, one CREATE GRAPH
Roll up a corporate family's pipelineGraph DB traversal + a warehouse JOINrecursive subsidiary_of over crm_edge, JOINed to opportunities
Relationship neighborhood ("who competes with Acme?")Neo4j Cypherdata bolt MATCH over the same bucket
Lookalike account targetingPinecone + an embedding pipelineaccount_vectors VECTOR(2048) + data vsearch
Rank the whitespaceA bespoke ML serviceone kimi-k2.6 gate call, verdict → a SQL column

No ETL, no second copy, no drift between the graph and the record — the family rollup, the relationship walk, and the lookalike all read the same rows.

Customize — the decisions this skill asks you

Q1 · rel_types — which relationships beyond hierarchy?

"Which inter-company relationships do you track — partner, competitor, supplier, customer?" → Seeds the allowed relationships.rel_type values and which crm_edge.rel kinds Step 2 projects (partner→partner_of, competitor→competes_with, supplier→supplies). Hierarchy (subsidiary_of) always exists from core. Fewer types = a leaner graph; the family rollup only ever reads subsidiary_of.

Q2 · whitespace — build the cross-sell grid?

"Do you want the product×account whitespace grid (needs a product catalog)?"

  • true (default) → Steps 2 (owns_product edges) + 4 (the grid) run; if crm/quote-cpq isn't installed, a minimal products catalog is seeded so the grid has a right-hand side.
  • false → hierarchy + relationships only; no owns_product edges, no whitespace table. Use this when you want the relationship graph but sell a single product.

Q3 · nba_gate — rank the whitespace with a model?

  • true (default) → deploy crm-account360 with ignite.model-user and call the kimi-k2.6 gate to rank + explain each family's top gap (written to whitespace.rationale).
  • false → deterministic gap list only; the handler needs just k3.editor (no ignite.model-user, no token-billed calls). The gap list is unranked but free.

Q4 · health_score — score account health?

  • true (default) → the handler computes health_score = 1 − min(days_since_last_activity, 90)/90 into account_summary (recent activity = healthy; a family untouched for 90+ days scores 0).
  • false → skip it; account_summary still carries pipeline / headcount / open-opp rollups.

Industry variants (saas / manufacturing / finserv / real-estate) compose this skill with extra graph edges and gate tweaks — see the per-industry CRM pages.

(Suite-shared answers — bucket and seed_data — are asked once at the suite level and not re-asked here.)

Test

The routes.py package was exercised end-to-end live on 2026-09-04 (a throwaway bucket) and re-validated 2026-09-06 on the persistent crm bucket as part of the composed suite: Base.metadata.create_all built all the tables (money columns confirmed numeric/DECIMAL over the pg wire), then every route ran — assemble_graph (edges projected, committed, then the re-snapshot — the commit-first fix), family_rollup (subsidiary_of only → [1,2,3] / 36000 / 2 / 4200 / health 0.94), relationship_neighbors (Globex partner_of), build_whitespace (2 gaps: prod-ignite 900 / prod-models 600), lookalike_accounts (real jina-embeddings-v4, nearest 0.2719), and the kimi-k2.6 NBA gate + write-back (prod-ignite, high) — with a re-run proving idempotency (account_summary stays 1 row). The equivalent CLI/agent queries below are the same operations over the same rows; the Ignite deploy in Step 6 uses the image-mode pattern validated live 2026-09-02 on crm-lead-scorer (deploys, serves /healthz + its route unauthenticated, writes durably — see the note). tested_branches: full (whitespace: true, nba_gate: true, health_score: true) and hierarchy-only (whitespace: false).

# 1) the snapshot re-create works — relationship neighbors become traversable
dodil data bolt -b "$BUCKET" -g crm_graph "MATCH (a)-[:crm_edge]->(b) WHERE id(a)=1 RETURN b"
#  -> 2 rows: 5 (Initech, partner_of), 4 (Greyparrot, competes_with)   [0 rows BEFORE the DROP+CREATE]
 
# 2) family pipeline rolls up by corporate family (subsidiary_of only) = $36,000
dodil data pg -b "$BUCKET" "
  WITH RECURSIVE family(node) AS (SELECT 1 UNION
    SELECT e.src FROM crm_edge e JOIN family f ON e.dst=f.node WHERE e.rel='subsidiary_of')
  SELECT coalesce(sum(o.amount),0) AS family_pipeline, count(*) AS open_opps
  FROM opportunities o JOIN crm_node n ON n.biz_key=o.account_domain AND n.kind='account'
  JOIN family f ON f.node=n.id WHERE o.status='open'"     # family_pipeline = 36000, open_opps = 2
 
# 3) the whitespace grid has a row per product with at least one has_it=false gap
dodil data sql -b "$BUCKET" "SELECT product_id, has_it, opportunity_potential FROM whitespace WHERE org_domain='acme.io' ORDER BY has_it"
#  -> prod-ignite false 18000 ; prod-models false 12000 ; prod-core true 0     (2 gaps)
 
# 4) lookalike KNN returns an enterprise-software peer nearest, unrelated account last
dodil data vsearch -b "$BUCKET" -t account_vectors --column embedding \
  --text "enterprise software consolidating Postgres Neo4j and Pinecone into one data platform" \
  --model jina-embeddings-v4 --metric cosine --top-k 5     # initech.com 0.20 nearest; greyparrot.ai 0.57 last
 
# 5) the NBA gate returns valid JSON
dodil ignite models chat kimi-k2.6 --system 'Return ONLY JSON: {next_best_product, priority, rationale}.' \
  --message 'Owns prod-core; gaps prod-ignite(18000), prod-models(12000); pipeline 36000.'
#  -> {"next_best_product":"prod-ignite","priority":"high","rationale":"…"}
 
# 6) recompute is idempotent (account_summary keyed on org_domain -> one row)
dodil data sql -b "$BUCKET" "SELECT count(*) AS n FROM account_summary WHERE org_domain='acme.io'"   # n = 1

Live-captured values (2026-09-06 suite run): account_summary for acme.io = {family_pipeline: 36000, family_headcount: 4200, open_opps: 2, health_score: 0.94}; the whitespace grid = prod-ignite (900) / prod-models (600) gaps + prod-datak3 owned; the NBA gate returned {"next_best_product":"prod-ignite","priority":"high",…}; data connect on the bucket printed pg.uk-lon-1.dodil.io:5432/crm (drop-in proof). A key live finding: the embedded cypher() subset anchors only on id(<var>) = <key> — no edge-rel filter — so the typed subsidiary rollup uses a recursive CTE (Step 3 / _subsidiary_family), while cypher() drives the untyped neighborhood.

One-shot

With the DODIL MCP connected, paste this to assemble the Account 360 — it stubs a minimal core slice so it runs standalone:

Assemble an Account 360 on DataK3 (bucket crm = SQL + graph + vector). Confirm each step.
 
1. If crm/core is absent, stub it: accounts (key org_domain: name, parent_domain, tier, country,
   industry, employee_band, annual_revenue) with Acme Corp (acme.io), Acme Labs (labs.acme.io, parent
   acme.io), Acme EU (acme.eu, parent acme.io), Greyparrot (greyparrot.ai), Initech (initech.com);
   opportunities (open: Acme Labs $24k, Acme EU $12k, Greyparrot $30k); contacts; activities; products
   (prod-core, prod-ignite, prod-models); account_vectors (VECTOR 2048); and crm_node (BIGINT KEY:
   accounts 1-5, contacts 10001+, products 100001+) + crm_edge with subsidiary_of (2->1,3->1) and
   works_at edges, then CREATE GRAPH crm_graph.
2. Create relationships, account_summary, whitespace (all merge-keyed). Upsert 3 relationships
   (acme.io partner initech.com; acme.io competitor greyparrot.ai; initech.com supplier acme.io).
3. Insert partner_of/competes_with/supplies + owns_product edges into crm_edge, then DROP GRAPH crm_graph
   and CREATE it again (snapshot rule). Prove a MATCH from node 1 returns 2 neighbors (0 before re-create).
4. Roll up the Acme family by subsidiary_of ONLY -> family_pipeline 36000, open_opps 2, headcount 4200;
   write account_summary (health_score from activity recency).
5. Build the whitespace grid (family owned products vs catalog) -> 2 gaps; embed the accounts with
   jina-embeddings-v4 and KNN account_vectors for the best lookalike (Initech).
6. Rank the top gap with kimi-k2.6 (return ONLY JSON next_best_product/priority/rationale) and write it
   onto the whitespace row.

Ship it

crm-account360 is an image-mode Ignite app — an HTTP server you POST an account root to on /account360, scale-to-zero so a nightly scheduler (or an on-demand agent) can hit it. Give it a least-privilege service account (the three roles above, and set DODIL_SERVICE_ACCOUNT_ID to the cli-crm-account360-sa serviceAccountId, not the uuid), then deploy its Dockerfile — the platform builds the image on deploy (Lane B), no --runtime python and no separate build step:

dodil ignite app deploy crm-account360 \
  --code ./account360-app --dockerfile-path Dockerfile \
  --port 8080 --health-path /healthz --allow-unauthenticated \
  --env BUCKET=crm --env NBA_GATE=true \
  --env DODIL_SERVICE_ACCOUNT_ID="$SA_ID" --env DODIL_SERVICE_ACCOUNT_SECRET="$SA_SECRET"
dodil ignite app get crm-account360 --output json    # -> public_urls (…-8080.ignite.dodil.cloud)

The full lifecycle — DODIL git → CI checks → a scanned registry image → versioning and rollback — is 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. 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_graph.
  • Vector — pgvector (<=>) over the same wire, or a Qdrant/Pinecone client against the same account_vectors rows.

Full, live-validated walkthrough: Connect your tools.

Conclusion

You assembled the CRM's relationship graph on one DataK3 bucket — hierarchy, partner/competitor/supplier edges, and owned-product edges in one crm_edge table, snapshotted once into crm_graph — then rolled up the corporate family ($36k across the Acme tree), scored account health, found the cross-sell whitespace, and ranked it with a kimi-k2.6 gate and semantic lookalike targeting. The graph, the record, and the search index are the same rows — no Neo4j sync, no Pinecone index, no drift.

Next steps: