What you'll build: the layer that answers "which campaigns actually created pipeline?" on one DataK3 bucket. Campaigns, their members, and every marketing touch are merge-keyed tables; a won deal's revenue is split across the campaigns that touched its buyer; and campaign ROI — credited revenue ÷ budget — falls out of a single JOIN. Turn it on for path-based multi-touch credit by walking a graph, or let kimi-k2.6 weight the split when flat rules are too blunt. Same bucket, same rows as the rest of your CRM (crm/core owns leads / contacts / opportunities) — no export, no second attribution tool.

What you'll learn:

  • Model campaigns → members → touchpoints as merge-keyed DataK3 tables — idempotent upserts.
  • Flip a member to responded and drop a touchpoint in the same write — the event is a row.
  • Split won-deal revenue across the touching campaigns (first / last / linear) — weights that sum to 1, amounts that sum to the deal.
  • Compute campaign ROI — SUM(credited) ÷ budget — over attribution JOIN campaigns.
  • Turn on multi-touch path attribution by projecting campaign nodes into the shared crm_graph and walking influenced / member_of back from a won opportunity.
  • Deploy a scale-to-zero crm-attributor engine on Ignite that recomputes credit on demand.
  • Optionally let a Models gate (kimi-k2.6) weight the multi-touch split by buying intent.
  • Run every step two ways — by prompting an agent over the MCP, or the dodil data CLI.

The problem — and why it matters

Marketing spends real money — a webinar, a paid-search line, a field event — and the CFO asks the one question the SaaS CRM never answers cleanly: which of those actually produced the $12,000 deal that just closed? The usual answer is a separate attribution tool, a nightly export of opportunities and campaign members, and a spreadsheet nobody trusts. Meanwhile "campaign influenced" in the CRM is a checkbox a rep ticks, not a number you can divide by budget.

Here it's just the CRM. The deal, the buyer, the campaigns, and the touches are all rows in one DataK3 bucket, so attribution is a JOIN — not an integration. Credit the two campaigns that touched Carol before Acme EU closed, and the model tells you the webinar returned 1.2× its budget while the paid-search line returned 0.75× — the number that decides next quarter's spend, computed over the live rows.

PieceLands inPillar
Campaigns + budgettable campaignsSQL
Who's in a campaigntable campaign_members (responded flag)SQL
Every marketing touchtable touchpointsSQL
Credit per campaign per dealtable attributionSQL
Multi-touch path (optional)campaign nodes + member_of / influenced edges in crm_graphGraph
The recompute enginewrites attributionIgnite app (scale-to-zero)
Intent-weighted split (optional)attribution.weightIgnite Models (kimi-k2.6)

NOTE

This is the crm/campaign-to-lead skill. It consumes leads, contacts, and opportunities from crm/core — in the full suite those masters already exist. Standalone, seed a few rows (Step 1) so you can follow along. Every step shows an Ask your agent tab and the CLI.

Prerequisites

  • The dodil CLI (dodil auth login) or the DODIL MCP connected to your agent (Claude Code, Cursor, VS Code, Codex).
  • export BUCKET=crm — one bucket is the whole CRM's data plane.
  • The crm/core masters (contacts, leads, opportunities) — or seed the handful below to run solo.

Step 1 — Seed the masters you consume (from crm/core)

Attribution is only as real as the deal it credits. crm/core owns contacts, leads, and opportunities; this skill reads them. In the suite they're already populated — standalone, upsert one won opportunity and its buyer so the rest of the walk has something to attribute. A --merge-key (PRIMARY KEY) is required on every table — writes are keyed, so re-runs upsert idempotently and reads are read-your-writes (fresh rows are visible immediately, JOINs included — no compaction step).

Each data step below carries a third tab — ORM: the exact SQLAlchemy class from the downloadable package's models.py (see Get the code), the same table a third way. Natural keys and Numeric(18,2) money map 1:1 to the CLI with no generated-PK round-trip.

You

In bucket crm, make sure the crm/core masters exist, then upsert a won opportunity to attribute: contacts (key email) with [email protected] (Carol Ng, acme.eu, customer); leads (key lead_id) with lead-carol; and opportunities (key opportunity_id) with opp-acme-eu — Acme EU platform, primary_contact_email [email protected], stage closed_won, status won, amount 12000, close_date 2026-08-20.

DODIL MCP tools called
data_bucket_create→data_table_create→data_table_upsert
Agent

Bucket crm ready. Upserted contact [email protected], lead lead-carol (converted), and the won opportunity opp-acme-eu ($12,000, status=won) — the deal campaign credit will be split across.

Step 2 — Model campaigns and their members

A campaign is a row with a budget; a member ties a contact (and its lead) to a campaign. channel comes from the channels list you seed — email, webinar, paid_search, event. campaign_members carries a responded boolean: joined-but-silent versus actually engaged. Keys are campaign_id and member_id, so imports and re-syncs upsert instead of duplicating.

You

In crm, create campaigns (key campaign_id: name, channel, type, start_date, end_date, status, budget double) and campaign_members (key member_id: campaign_id, contact_email, lead_id, status, responded boolean, joined_at). Upsert two campaigns — cmp-webinar (EU Data Residency Webinar, channel webinar, budget 5000) and cmp-paid (Sovereign Cloud Search, channel paid_search, budget 8000) — and add Carol to both: mem-1 on cmp-webinar (responded true) and mem-2 on cmp-paid (responded false, not engaged yet).

DODIL MCP tools called
data_table_create→data_table_upsert
Agent

Created campaigns and campaign_members. Seeded cmp-webinar ($5,000) and cmp-paid ($8,000), and enrolled [email protected] in both — mem-1 already responded, mem-2 not yet. responded is the engagement signal attribution keys on.

TIP

Never "" / null in a merge-key. An empty string or JSON null in campaign_id / member_id reads back null and the row silently drops on the next read. Use a real id (or a sentinel like "none"), never "".

Step 3 — Log a response as a touchpoint

A touchpoint is one marketing touch — Carol attending the webinar, the paid ad she clicked. When a member engages, you flip responded=true and append the touchpoint in the same beat: the event is a row, and the two writes are ordinary keyed upserts. mem-2 was silent; mark it engaged now, and record both touches so the won deal has a trail to attribute.

You

In crm, create touchpoints (key touchpoint_id: contact_email, lead_id, campaign_id, channel, occurred_at, value double). Record Carol's two touches — tp-1 webinar attendance on cmp-webinar (2026-06-01) and tp-2 a paid-search click on cmp-paid (2026-05-20) — and flip campaign_members mem-2 to responded=true (status responded).

DODIL MCP tools called
data_table_create→data_table_upsert
Agent

Created touchpoints and logged tp-1 (webinar) + tp-2 (paid_search) for [email protected]. Flipped mem-2 to responded=true via a partial merge — the paid-search member is now engaged, and both campaigns have a touch on the buyer of opp-acme-eu.

Confirm the flip landed:

You

In crm, show member_id and responded for every campaign_members row.

DODIL MCP tools called
data_sql
Agent

mem-1 → true, mem-2 → true. Both members have responded — the partial merge flipped mem-2 without disturbing its campaign_id or joined_at.

Step 4 — Split the credit: linear attribution over a won deal

Here's the money. A won opportunity's amount is split across the distinct campaigns that touched its buyer. The default model is linear — equal credit per campaign: two campaigns touched Carol, so each gets weight = 0.5 and amount = $12,000 × 0.5 = $6,000. The split is derived from the touchpoints, not typed by hand — join the won opp to its buyer's touches and let the window function count them.

You

In crm, for the won opportunity opp-acme-eu, split its amount linearly across the distinct campaigns that touched its primary contact — weight = 1/N, amount = opp.amount/N per campaign — using its touchpoints.

DODIL MCP tools called
data_sql
Agent

opp-acme-eu ($12,000) was touched by 2 campaigns (cmp-webinar, cmp-paid) → weight 0.5 each, amount $6,000 each. The split is computed straight from touchpoints: N = COUNT(*) OVER () = 2.

Now persist the credit into attribution — one row per (opportunity, campaign), keyed attribution_id so a recompute upserts in place (idempotent). Keep the attribution_id deterministic (att-<opp>-<campaign>) so re-runs never fork a second row.

You

In crm, create attribution (key attribution_id: opportunity_id, campaign_id, model, weight double, amount double, computed_at) and write the linear split for opp-acme-eu — att-opp-acme-eu-cmp-webinar and att-opp-acme-eu-cmp-paid, model linear, weight 0.5, amount 6000 each.

DODIL MCP tools called
data_table_create→data_table_upsert
Agent

Created attribution and wrote two rows for opp-acme-eu (model=linear, 0.5 / $6,000 each). Deterministic ids (att-opp-acme-eu-cmp-webinar / att-opp-acme-eu-cmp-paid) mean a recompute upserts the same two rows — never duplicates.

The split is sound only if the weights sum to 1 and the credited amounts sum back to the deal:

You

In crm, check the attribution split for opp-acme-eu — count the rows, sum the weights, sum the amounts, and compare the amount sum to the opportunity's amount.

DODIL MCP tools called
data_sql
Agent

2 rows, weight_sum = 1.0, amount_sum = $12,000, and opp_amount = $12,000 — the split is complete and conserves the deal value exactly.

Step 5 — Campaign ROI (the number that sets next quarter's spend)

Credit divided by budget. attribution JOIN campaigns gives you, per campaign, the pipeline it created against what it cost — a finite ratio you can rank. Here the $5,000 webinar earns $6,000 of credit (1.2×) while the $8,000 paid-search line earns $6,000 (0.75×): fund the webinar, question the ads.

You

In crm, compute campaign ROI — credited revenue divided by budget — per campaign, over attribution joined to campaigns.

DODIL MCP tools called
data_sql
Agent

cmp-webinar: $6,000 credited ÷ $5,000 budget = 1.2×. cmp-paid: $6,000 ÷ $8,000 = 0.75×. The webinar returned above its cost; the paid line did not — the spend decision, straight from the live rows.

Step 6 — Multi-touch, the graph way (optional — multi_touch_graph)

Linear says "both campaigns, equally." Multi-touch path attribution asks a graph question instead: from the won opportunity, walk backward through the campaigns that influenced it and the members who carried that influence. Project campaign nodes and two edge kinds into the shared crm_graph (owned by crm/core): member_of (contact → campaign) and influenced (campaign → the opportunity's anchor node). A DataK3 graph is table-backed and snapshots its edges at CREATE GRAPH — so populate crm_node and crm_edge fully first, then create the graph.

You

In crm, extend the shared crm_graph for multi-touch. In crm_node (id bigint KEY, kind, biz_key, name) add the opportunity anchor 300001 (opp-acme-eu), Carol 10001 (contact), and campaign nodes 200001 (cmp-webinar) + 200002 (cmp-paid). In crm_edge (src, dst, rel) add member_of edges Carol→each campaign and influenced edges each campaign→the opp anchor. Populate both tables, then CREATE GRAPH crm_graph.

DODIL MCP tools called
data_pg
Agent

crm_node holds the opp anchor (300001), contact (10001), and 2 campaign nodes (200001/200002); crm_edge holds 2 member_of + 2 influenced edges. CREATE GRAPH crm_graph NODES (crm_node KEY id) EDGES (crm_edge SRC src DST dst) snapshotted them — campaigns now sit on the influence path to the deal.

Now the path query: from the opportunity anchor, a reverse k-hop reaches the campaigns that influenced it (hop 1) and the members behind them (hop 2). Filter to kind='campaign' for the set that earns multi-touch credit — the same two campaigns, now proven by a graph path, not a flat join.

You

In crm, from the opportunity anchor (graph node 300001) traverse crm_graph inward up to 2 hops and return the campaigns on the influence path.

DODIL MCP tools called
data_pg→data_bolt
Agent

graph_khop('crm_graph', 300001, 2, 'in') reaches cmp-webinar (200001) and cmp-paid (200002) at hop 1 (influenced edges) and Carol (10001) at hop 2 (member_of). Filtered to kind='campaign' → both campaigns are on the path to the won deal.

NOTE

multi_touch_graph skips gracefully when crm/core was scaffolded with hierarchy=false (no crm_node / crm_edge) — attribution falls back to the SQL rollups of Step 4. The graph path is the upgrade, not a dependency.

Routes

The download (see Get the code) fronts this bucket with a small FastAPI app, routes.py — CRUD over the models plus the ops that turn marketing touches into a defensible spend number: an attribute split, a campaign roi rollup, and a multi-touch influence-path walk. This is the app layer of Steps 1–6. 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 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 recompute land the row once. That's the whole reason the attribution split is safe to re-run. (Verified live: re-running the split kept attribution at exactly 2 rows; a bare re-INSERT of a committed attribution_id returned SQLSTATE 23505.)

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). Logging a touchpoint flips its member to responded in the same beat — the event is a row:

# routes.py — CRUD, keyed on the natural PK; the touch and the engagement flag move together
@router.post("/campaigns")
def upsert_campaign(c: CampaignIn, s: Session = Depends(db)):
    upsert(s, Campaign, [c.model_dump()], key="campaign_id")
    s.commit()
    return {"ok": True, "campaign_id": c.campaign_id}
 
 
@router.post("/touchpoints")
def log_touchpoint(t: TouchpointIn, s: Session = Depends(db)):
    """A marketing touch IS a row. Log it (keyed on touchpoint_id — a replayed webhook lands
    once) and, in the same beat, flip its member to responded via a partial merge."""
    row = t.model_dump()
    respond = row.pop("respond")
    upsert(s, Touchpoint, [row], key="touchpoint_id")
    if respond and t.campaign_id and t.contact_email:
        s.execute(
            text("UPDATE campaign_members SET responded = true, status = 'responded' "
                 "WHERE campaign_id = :cid AND contact_email = :email"),
            {"cid": t.campaign_id, "email": t.contact_email},
        )
    s.commit()
    return {"ok": True, "touchpoint_id": t.touchpoint_id}

Workflow op 1 — split the credit (SQL, optional Models gate). POST /opportunities/{id}/attribute refuses anything but a won deal, derives the distinct touching campaigns from its buyer's touchpoints (in first-touch order, so first_touch/last_touch pick the right end), splits the amount, and upserts one attribution row per campaign — keyed att-<opp>-<campaign>, so a recompute upserts the same rows in place. Money is DECIMAL, written over the pg wire so it keeps its precision:

# routes.py — workflow op 1: split a won deal's amount across the campaigns that touched its buyer
@router.post("/opportunities/{opportunity_id}/attribute")
def attribute(opportunity_id: str, a: AttributeIn, s: Session = Depends(db)):
    opp = s.get(Opportunity, opportunity_id)
    if not opp:
        raise HTTPException(404, "no such opportunity")
    if opp.status != "won":
        raise HTTPException(409, f"opportunity is status={opp.status!r}, not 'won' — nothing to attribute")
    camps = _touching_campaigns(s, opp)          # distinct, first-touch order
    if not camps:
        raise HTTPException(409, "no touching campaigns for this deal's buyer")
    if a.influence_gate and len(camps) > 1:
        credited, model = _gate_split(opp, camps), "multi_touch"   # kimi-k2.6 re-weights by intent
    else:
        credited, model = _split(camps, opp.amount, a.attribution_model), a.attribution_model
    rows = [{
        "attribution_id": f"att-{opportunity_id}-{cid}", "opportunity_id": opportunity_id,
        "campaign_id": cid, "model": model, "weight": w, "amount": amt,
        "computed_at": a.computed_at,
    } for cid, w, amt in credited]
    upsert(s, Attribution, rows, key="attribution_id")   # deterministic ids -> idempotent recompute
    s.commit()                                            # commit before any read-back
    return {"opportunity_id": opportunity_id, "model": model,
            "credited": [{"campaign_id": cid, "weight": w, "amount": float(amt)} for cid, w, amt in credited]}

Live-verified 2026-09-06 on the persistent crm bucket, over the suite's seeded funnel: attributing the won opp opp:greyparrot.ai ($28,200) linearly wrote 2 rows — camp-webinar-datak3 / camp-paid-search-q3, weight 0.5, amount 14100.00 each — SUM(weight) = 1.0 and SUM(amount) = 28200.00 = opp.amount. With influence_gate=true, kimi-k2.6 returned weights [0.75, 0.25] and the same recompute re-wrote those two rows in place as model=multi_touch (amounts 21150.00 / 7050.00, sum still 28200.00) — the DECIMAL value was preserved on the pg wire, never dropped to 0.

Workflow op 2 — campaign ROI (SQL). GET /campaigns/roi divides credited revenue by budget per campaign — one JOIN of attribution to campaigns, a finite ratio you can rank. One FastAPI rule this route teaches the hard way: it must be registered before GET /campaigns/{campaign_id} — FastAPI matches routes in registration order, so the param route would otherwise swallow roi as a campaign id. That's not hypothetical: composing the suite, GET /campaigns/roi returned 404 no such campaign until the static route moved above the param route.

# routes.py — workflow op 2: credited revenue / budget per campaign.
# NOTE: registered BEFORE /campaigns/{campaign_id} — FastAPI matches in registration
# order, and the param route would otherwise swallow "roi" as a campaign_id
# (found live composing the suite: GET /campaigns/roi returned 404 "no such campaign").
@router.get("/campaigns/roi")
def campaign_roi(s: Session = Depends(db)):
    rows = s.execute(text(
        "SELECT c.campaign_id, SUM(a.amount) AS credited, MAX(c.budget) AS budget, "
        "ROUND(SUM(a.amount) / MAX(c.budget), 4) AS roi "
        "FROM attribution a JOIN campaigns c ON c.campaign_id = a.campaign_id "
        "GROUP BY c.campaign_id ORDER BY c.campaign_id"
    )).all()
    return {"roi": [{"campaign_id": cid, "credited": float(cr), "budget": float(b), "roi": float(r)}
                    for cid, cr, b, r in rows]}

Live-verified 2026-09-06: camp-webinar-datak3 $14,100 credited ÷ $5,000 budget = 2.82×; camp-paid-search-q3 $14,100 ÷ $8,000 = 1.7625× — both returned above cost here, and the webinar returned $1.60 more per pound; fund the webinar first.

Workflow op 3 — multi-touch influence path (GRAPH). GET /opportunities/anchor/{opp_node}/influence-path walks the shared crm_graph inward from the deal's anchor node and keeps the campaign nodes. DataK3 runs a Cypher subset embedded in SQL — cypher('<graph>', 'MATCH …') — with three rules the code obeys: it's a top-level table function (no UNION/subquery), the anchor id is an integer literal (so the FastAPI-validated opp_node is inlined, not bound), and you feed the returned node ids into a SQL IN (…) (DuckDB has no = ANY(array)):

# routes.py — workflow op 3: which campaigns are on the influence path to a won deal (GRAPH)
@router.get("/opportunities/anchor/{opp_node}/influence-path")
def influence_path(opp_node: int, s: Session = Depends(db)):
    nodes = s.execute(
        text("SELECT node FROM cypher('crm_graph', "
             f"'MATCH (root)<-[*1..2]-(child) WHERE id(root) = {int(opp_node)} RETURN child')")
    ).scalars().all()
    if not nodes:
        return {"opp_node": opp_node, "campaigns": []}
    ids = ",".join(str(int(n)) for n in dict.fromkeys(nodes))
    camps = s.execute(text(
        f"SELECT id, biz_key, name FROM crm_node WHERE id IN ({ids}) AND kind = 'campaign' ORDER BY id"
    )).all()
    return {"opp_node": opp_node,
            "campaigns": [{"id": i, "campaign_id": bk, "name": nm} for i, bk, nm in camps]}

Live-verified 2026-09-06: from the opp anchor 300001, the cypher() walk returned the campaign nodes 200001 (camp-webinar-datak3, DataK3 Deep-Dive Webinar) and 200002 (camp-paid-search-q3, Q3 Paid Search) — the same two the flat join found, now proven by a graph path.

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, graph via a top-level cypher(…) fed into a SQL IN (…) (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 against the pool issuer, 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 therefore a header-trust reader, not a verifier — no JWKS client, no issuer/audience env — exposing current_user and require_permission for role gating.

Campaign-to-lead kept no role gate after the audit. The attribute recompute is idempotent (deterministic ids — running it twice changes nothing) and the Models gate is off by default, so the gateway's authentication alone guards these routes; nothing here imports auth. 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.

The two-plane rule is unchanged: the pool identifies the user; the app reaches DataK3 through its own service account — 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 role catalog: App roles.

Get the code

The package is a real download — code/crm-campaign-to-lead/v1.tar. This post is a walkthrough of exactly those files; the tarball is source only (no Dockerfile — the crm-attributor recompute engine and its deploy live in Step 7):

models.py          # SQLAlchemy — campaigns / campaign_members / touchpoints / attribution, the
                   #              crm/core masters this consumes (stubbed), and the crm_graph tables
routes.py          # FastAPI    — an APIRouter the suite mounts + a standalone app; CRUD +
                   #              attribute (split) + roi + influence-path (graph)
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], 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 /campaigns · POST /campaign_members · POST /touchpoints
# POST /opportunities/{id}/attribute · GET /campaigns/roi · GET /opportunities/anchor/{node}/influence-path

models.Base.metadata.create_all is the ORM tab's every class at once — the same tables Steps 1–6 built by CLI, created from the natural-key models 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 /campaign-to-lead) 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.

Step 7 — Deploy the attributor engine (Ignite)

The recompute belongs in a small Ignite app so it runs on demand (or on a tick) instead of by hand. Ship it as a container image (Dockerfile + --dockerfile-path) — the app is a tiny HTTP server: GET /healthz for the probe and POST /attribute for the work. It's a separate workload, so it needs its own service account to reach DataK3 over the drop-in Postgres wire (pg.uk-lon-1.dodil.io:5432, dbname=<bucket>, user=token, password=<the SA access token>), injected as runtime env. crm-attributor is pure SQL for the first / last / linear / multi_touch split, so its identity needs only k3.editor (plus ignite.app-developer for the deploy identity) — no ignite.model-user unless you enable the influence gate (Step 8).

# attributor/server.py — an IMAGE-mode Ignite app (HTTP server on $PORT).
#   GET  /healthz    -> 200 {"status":"ready"}      (probe path; no auth)
#   POST /attribute  -> recompute credit for every won opp from its buyer's touchpoints
# The data plane is the DROP-IN POSTGRES WIRE (psycopg) — there is no K3 HTTP API.
 
from datetime import date
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
from psycopg import errors as pg_errors
 
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"))
MODEL_ID   = os.environ.get("MODEL_ID", "kimi-k2.6")
DEF_MODEL  = os.environ.get("ATTRIBUTION_MODEL", "linear")   # mirrors the attribution_model param
DEF_GATE   = os.environ.get("INFLUENCE_GATE", "false").lower() == "true"
 
ID_URL     = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
MODELS_URL = "https://api.dodil.io/v1/chat/completions"
# an explicit User-Agent is REQUIRED — stdlib urllib's default "Python-urllib/x" is
# banned by Cloudflare at id.dodil.io / api.dodil.io (HTTP 403 "error code: 1010").
UA = "crm-attributor/1.0"
ATTR_COLS = ["attribution_id", "opportunity_id", "campaign_id", "model",
             "weight", "amount", "computed_at"]
GATE_SYS = ("You weight marketing campaign influence on a won opportunity. Given a contact's "
            "touchpoints around a won deal, return ONLY JSON: {\"influential_campaign_ids\":[string],"
            "\"weights\":[number that sum to 1],\"reason\":string}. Weight higher-intent touches (an "
            "attended webinar) above lower-intent ones (a paid-search impression).")
 
 
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 _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 _chat(token, user):
    # kimi-k2.6 is a reasoning model: reasoning tokens eat the budget first — set max_tokens
    # high (4096) or `content` comes back empty. The reply is wrapped in "data" on this platform.
    out = _http_post(MODELS_URL,
                     {"model": MODEL_ID, "max_tokens": 4096,
                      "messages": [{"role": "system", "content": GATE_SYS},
                                   {"role": "user", "content": user}]},
                     headers={"Authorization": f"Bearer {token}"})
    env = out.get("data", out)
    return env["choices"][0]["message"]["content"]
 
 
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):
    # the pg 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 _won_opps(token):
    # every won opp + the campaigns that touched its primary contact, ordered by touch time
    # so first_touch / last_touch pick the right end.
    sql = ("SELECT o.opportunity_id, o.amount, o.primary_contact_email, t.campaign_id "
           "FROM opportunities o "
           "JOIN touchpoints t ON t.contact_email = o.primary_contact_email "
           "WHERE o.status = 'won' "
           "ORDER BY o.opportunity_id, t.occurred_at")
    with _pg(token) as conn, conn.cursor() as cur:
        cur.execute(sql)
        rows = cur.fetchall()
    by_opp = {}
    for oid, amount, email, cid in rows:
        e = by_opp.setdefault(oid, {"amount": float(amount or 0), "email": email, "camps": []})
        if cid not in e["camps"]:
            e["camps"].append(cid)          # distinct, in first-touch order
    return by_opp
 
 
def _split(camps, amount, model):
    n = len(camps)
    if model == "first_touch":
        return [(camps[0], 1.0, amount)]
    if model == "last_touch":
        return [(camps[-1], 1.0, amount)]
    return [(c, 1.0 / n, amount / n) for c in camps]      # linear | multi_touch: equal floor
 
 
def _gate_split(token, oid, info):
    # influence_gate: kimi-k2.6 re-weights the floor by buying intent; model becomes multi_touch.
    bundle = (f"Won opp {oid}, amount {info['amount']}, contact {info['email']}. "
              f"Touching campaigns: {', '.join(info['camps'])}.")
    v = _extract_json(_chat(token, bundle))
    ids, ws = v["influential_campaign_ids"], v["weights"]
    total = sum(ws) or 1.0
    return [(cid, w / total, info["amount"] * w / total) for cid, w in zip(ids, ws)]
 
 
def _write(token, oid, credited, model):
    # attribution_id is deterministic (att-<opp>-<campaign>), so a recompute re-writes the SAME PK — this
    # MUST be an ON CONFLICT upsert. A bare re-INSERT of an already-committed PK raises duplicate-key 23505;
    # DuckDB pg-wire supports ON CONFLICT (verified live), or use the managed data_table_upsert.
    setc = ", ".join(f"{c} = EXCLUDED.{c}" for c in ATTR_COLS if c != "attribution_id")
    upsert = (f"INSERT INTO attribution ({', '.join(ATTR_COLS)}) "
              f"VALUES ({', '.join(['%s'] * len(ATTR_COLS))}) "
              f"ON CONFLICT (attribution_id) DO UPDATE SET {setc}")
    today = date.today().isoformat()
 
    def _w():
        with _pg(token) as conn, conn.cursor() as cur:
            for cid, w, amt in credited:
                cur.execute(upsert, [f"att-{oid}-{cid}", oid, cid, model, w, amt, today])
            conn.commit()
    _retry(_w)
 
 
def attribute(req):
    model = req.get("attribution_model", DEF_MODEL)
    gate  = bool(req.get("influence_gate", DEF_GATE))
    token = _token()
    by_opp = _won_opps(token)
    rows = 0
    for oid, info in by_opp.items():
        if gate and len(info["camps"]) > 1:
            credited, used = _gate_split(token, oid, info), "multi_touch"
        else:
            credited, used = _split(info["camps"], info["amount"], model), model
        _write(token, oid, credited, used)
        rows += len(credited)
    return {"model": ("multi_touch" if gate else model),
            "attributed_opps": len(by_opp), "attribution_rows": rows}
 
 
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 != "/attribute":
            return self._send(404, {"error": "no_route", "path": self.path})
        try:
            n = int(self.headers.get("Content-Length") or 0)
            req = json.loads(self.rfile.read(n) or b"{}")
            return self._send(200, attribute(req))
        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-attributor serving on 0.0.0.0:{port} bucket={BUCKET}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

Two files ship alongside it — the container recipe and its one third-party dep (psycopg, the pg driver):

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

Give the engine its own least-privilege identity, then deploy the image (Lane B — the platform builds the container on deploy, so --allow-unauthenticated works with no pull secret):

You

Create a service account for crm-attributor, grant it k3.editor (write attribution) and ignite.app-developer (deploy identity), then build-and-deploy my ./attributor image to Ignite as crm-attributor (Dockerfile, port 8080, health /healthz) with the SA creds + the bucket as runtime env. Then hit POST /attribute once to recompute attribution.

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

Created crm-attributor-sa (serviceAccountId cli-crm-attributor-sa), granted k3-authorization-service k3.editor + ignite-authorization-service ignite.app-developer, and image-built + deployed crm-attributor (request-invoked, scale-to-zero) at crm-attributor-<org>-8080.ignite.dodil.cloud. POST /attribute {attribution_model:linear} recomputes credit for every won opp and upserts attribution — returns model linear, attributed_opps 1, attribution_rows 2.

NOTE

Deploy: image mode (Lane B), validated live 2026-09-02. crm-attributor 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-attributor reuses that identical handler/deploy pattern. DODIL_SERVICE_ACCOUNT_ID is the cli-… serviceAccountId (the uuid fails client_credentials).

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).

Step 8 — Intent-weighted split with a Models gate (optional — influence_gate)

Equal credit is often too blunt: a webinar Carol attended signals more buying intent than a paid-search impression she scrolled past. Turn on influence_gate and the handler hands the buyer's touchpoints to kimi-k2.6, which returns a weighted split you write into attribution.weight. The deterministic linear split is the baseline; the gate only re-weights it, with a reason.

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

Here it is live (2026-09-06) on the suite's seeded funnel deal — opp:greyparrot.ai, $28,200, whose buyer was touched by the DataK3 webinar and the Q3 paid-search line:

You

Weight campaign influence on the won opp opp:greyparrot.ai ($28,200) for contact [email protected] on kimi-k2.6. Touchpoints: camp-webinar-datak3 (attended the DataK3 Deep-Dive Webinar) and camp-paid-search-q3 (a paid-search click). Return ONLY JSON: influential_campaign_ids, weights (sum to 1), reason.

DODIL MCP tools called
ignite_models_chat
Agent

{"influential_campaign_ids":["camp-webinar-datak3","camp-paid-search-q3"],"weights":[0.75,0.25],"reason":"Webinar attendance indicates higher buying intent than a paid-search click."}

With the gate on, the same recompute re-wrote the two rows in place as model='multi_touch' with weight 0.75 / 0.25 — amounts $21,150 / $7,050, still summing to the deal's $28,200 — and the ROI in Step 5 shifts accordingly (verified live 2026-09-06). The gate is a per-deal judgement; the rules remain the floor. One operational note: a kimi-k2.6 call runs anywhere from ~12s to nearly 3 minutes — the route reads the deal, computes, and only then opens its write; never hold a pg connection across the model call.

How the pillars map

One bucket answers three ways over one copy of the rows — no export to an attribution tool, no second copy to drift.

JobPillarHow
Campaigns, members, touchpoints, creditSQLmerge-keyed tables; the linear split + ROI are one JOIN each
Multi-touch path — which campaigns influenced a dealGraphcampaign nodes + member_of / influenced edges in crm_graph; graph_khop(...,'in') from the opp anchor
Recompute engineIgnitecrm-attributor — scale-to-zero, own SA, k3.editor
Intent-weighted splitModelskimi-k2.6 returns weights that re-shape the credit

Attribution has no vector pillar — credit is relational and path-based, not semantic. (Similar-deal retrieval lives in crm/core's opportunity_vectors.)

Customize — the decisions this skill asks you

Q1 · attribution_model — how is credit split?

"First touch, last touch, equal (linear), or path-based multi-touch?"

  • first_touch → 100% to the earliest touching campaign (demand-gen bias).
  • last_touch → 100% to the latest (closing-motion bias).
  • linear (default) → equal split across every touching campaign — the neutral baseline this walk builds; sets attribution.model and the handler's compute branch.
  • multi_touch → path-weighted credit; requires multi_touch_graph=true (Step 6).

Q2 · multi_touch_graph — build the influence path?

"Project campaign nodes/edges into crm_graph for path attribution?"

  • false (default) → SQL rollups only (Steps 4–5); no graph writes.
  • true → adds campaign nodes + member_of / influenced edges to the shared crm_graph (Step 6). Skips gracefully if crm/core was built with hierarchy=false.

Q3 · influence_gate — model-weighted split?

"Let kimi-k2.6 weight the multi-touch split by buying intent, or keep pure rules?"

  • false (default) → deterministic first/last/linear only; the handler needs just k3.editor.
  • true → the handler calls Models (Step 8) and its SA also needs ignite.model-user (token-billed). Recommended when touches vary wildly in intent (attended webinar vs. impression).

Q4 · channels — which marketing channels exist?

"What channels do your campaigns run on?" → Default [email, webinar, paid_search, event]. Seeds the valid campaigns.channel values and the demo campaigns; extend it (e.g. + [in_product] for a PLG motion) and the seed + validation follow.

Test

Every command below ran live against DataK3 (org IHDIASH) with the results inline — the standalone demo dataset on 2026-09-01, and the package routes re-validated 2026-09-06 on the persistent crm bucket over the suite's composed funnel (linear 14100.00 × 2, ROI 2.82× / 1.7625×, gate 21150.00 / 7050.00). The Ignite deploy in Step 7 uses the image-mode pattern validated live 2026-09-02 on crm-lead-scorer (deploys, serves /healthz unauthenticated, writes durably — see the note after Step 7).

# 1 · the four owned tables exist
dodil data table list -b "$BUCKET"      # campaigns, campaign_members, touchpoints, attribution (+ consumed masters)
 
# 2 · a response flips the member
dodil data sql -b "$BUCKET" "SELECT member_id, responded FROM campaign_members ORDER BY member_id"
#  mem-1 true | mem-2 true
 
# 3 · linear split — weights sum to 1, amounts sum to the deal
dodil data sql -b "$BUCKET" "
  SELECT COUNT(*) AS rows, SUM(weight) AS w, SUM(amount) AS amt,
         (SELECT amount FROM opportunities WHERE opportunity_id='opp-acme-eu') AS opp
  FROM attribution WHERE opportunity_id='opp-acme-eu'"
#  rows=2  w=1.0  amt=12000  opp=12000
 
# 4 · campaign ROI is a finite ratio
dodil data sql -b "$BUCKET" "
  SELECT c.campaign_id, SUM(a.amount)/MAX(c.budget) AS roi
  FROM attribution a JOIN campaigns c ON c.campaign_id=a.campaign_id GROUP BY c.campaign_id"
#  cmp-webinar 1.2 | cmp-paid 0.75
 
# 5 · multi_touch — the graph path from the opp anchor reaches both campaigns
dodil data pg -b "$BUCKET" "
  SELECT n.id FROM graph_khop('crm_graph', 300001, 2, 'in') g
  JOIN crm_node n ON n.id=g.node WHERE n.kind='campaign' ORDER BY n.id"
#  200001 | 200002
 
# 6 · influence gate returns a valid weighted split (JSON parses, weights sum to 1)
dodil ignite models chat kimi-k2.6 --system 'Return ONLY JSON: {influential_campaign_ids, weights, reason}.' \
  --message 'opp:greyparrot.ai; touches: camp-webinar-datak3 (attended webinar), camp-paid-search-q3 (click).'
#  {"influential_campaign_ids":["camp-webinar-datak3","camp-paid-search-q3"],"weights":[0.75,0.25],"reason":"…"}
 
# 7 · idempotent recompute — re-upserting the split keeps exactly 2 rows
dodil data sql -b "$BUCKET" "SELECT COUNT(*) FROM attribution"    # 2 (unchanged on re-run)

Tested branches (live, re-validated 2026-09-06): {attribution_model: linear} (primary), {attribution_model: multi_touch, multi_touch_graph: true} (graph path), and {influence_gate: true} (one live kimi-k2.6 call, weights [0.75, 0.25]). Every assertion above returned the value shown.

One-shot: build it by prompting your agent

With the DODIL MCP connected, paste this to scaffold campaign attribution end to end:

Scaffold crm/campaign-to-lead on one DataK3 bucket `crm`. Confirm each step.
 
1. Ensure the crm/core masters exist (contacts, leads, opportunities); seed one WON opp opp-acme-eu
   ($12,000, primary_contact_email [email protected]) and its contact/lead to attribute.
2. Create campaigns (key campaign_id: name, channel, type, start_date, end_date, status, budget double)
   and campaign_members (key member_id: campaign_id, contact_email, lead_id, status, responded boolean,
   joined_at). Seed cmp-webinar (webinar, $5,000) + cmp-paid (paid_search, $8,000); enroll Carol in both.
3. Create touchpoints (key touchpoint_id: contact_email, lead_id, campaign_id, channel, occurred_at,
   value double). Log tp-1 (webinar) + tp-2 (paid_search); flip campaign_members mem-2 responded=true.
4. Create attribution (key attribution_id: opportunity_id, campaign_id, model, weight double, amount
   double, computed_at). LINEAR split of opp-acme-eu across its 2 touching campaigns (0.5 / $6,000 each);
   verify weights sum to 1 and amounts sum to $12,000.
5. Campaign ROI: SUM(amount)/budget per campaign (webinar 1.2×, paid 0.75×).
6. (optional, multi_touch_graph) Project campaign nodes (200001/200002) + an opp anchor (300001) into
   crm_node/crm_edge with member_of + influenced edges; CREATE GRAPH crm_graph; graph_khop(...,300001,2,'in')
   filtered to kind='campaign' returns both campaigns.
7. Deploy an Ignite app `crm-attributor` (own SA, k3.editor + ignite.app-developer) as a container image
   (Dockerfile, port 8080, health /healthz): an HTTP server whose POST /attribute recomputes attribution
   for every won opp from touchpoints over the pg wire; call POST /attribute {"attribution_model":"linear"}.
8. (optional, influence_gate) kimi-k2.6 returns an intent-weighted split; SA also gets ignite.model-user.

Ship it — crm-attributor as a service

crm-attributor is request-invoked and scale-to-zero — wire it to a tick or call it after a deal closes. Ship it as a container image through the real DODIL supply chain (git → CI → registry → deploy), exactly as Ship a DODIL App walks end to end:

# image mode (Dockerfile + --dockerfile-path); DODIL_SERVICE_ACCOUNT_ID is the cli-… serviceAccountId,
# NOT the uuid — the uuid fails client_credentials with invalid_client.
dodil ignite app deploy crm-attributor \
  --code ./attributor --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

It carries only DataK3 write access (k3.editor) — add ignite.model-user only if influence_gate is on. Add --auto-min-instances 1 to avoid a cold-start 502 on the first hit (bills continuously). Recurring recompute is your own scheduler or an always-on poll loop pinned --reserved 1 --max-replicas 1 (DODIL is request-invoked — there's no server-side scheduler).

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 for the multi-touch path.

Full, live-validated walkthrough: Connect your tools.

Conclusion

Campaign attribution that's just your CRM — campaigns, members, and touchpoints you upsert; a won deal's revenue split across the campaigns that created it; ROI that's one JOIN; an optional graph path for true multi-touch; and a Models gate that weights the split by intent. One DataK3 bucket, one bill, the same rows the rest of the CRM already holds — no export, no second attribution system to reconcile.

You can do this today: Create a free account, connect your agent over MCP in under two minutes, and attribute your first closed deal this afternoon.

Next steps: