What you'll build: a lead-qualification engine that scores a real lead on BANT (or MEDDIC) with a kimi-k2.6 gate, writes the per-dimension verdict to a lead_scores table, and flips the lead's status by a policy you keep as data — a strong lead auto-qualifies, a weak one is disqualified, and anything in the middle routes to a human. It's one Ignite app over the same DataK3 bucket your CRM already lives in (see Build a CRM on DataK3); it consumes core leads/opportunities/contacts/activities and adds two tables of its own.

The problem — and why it matters

A rep's most expensive resource is their own time. The discovery gate that fills the pipeline (Leads Data Warehouse) is cheap and shallow — is this company even a fit? But once a lead is real, someone has to decide is this worth a rep's week? — and that decision is usually a gut call scrawled in a CRM note, impossible to audit, impossible to tune.

This is the deep gate. It reads the lead, its account, and the recent activity notes and scores it on a real sales framework — BANT (Budget, Authority, Need, Timing) for velocity sales, or MEDDIC (Metrics, Economic buyer, Decision criteria, Decision process, Pain, Champion) for enterprise. The model supplies the per-dimension judgement and a one-line rationale; deterministic thresholds you own make the call. Two gates, both data:

  • score ≥ threshold_qualify (0.7) → auto-qualify, hand it to a rep.
  • threshold_route (0.5) ≤ score < 0.7 → route to a human — promising, but not a clean yes.
  • score < 0.5 → disqualify, keep it out of the rep's queue.

The money: if scoring is honest and tunable, reps only work leads that clear the bar, and you can prove why every lead landed where it did. The verdict is a row, the policy is a row — change the bar, re-score, done. No redeploy.

PieceLands inPillar / runs on
The two-gate policytable scoring_policySQL (policy as data)
The verdict + per-dimension scorestable lead_scoresSQL
The framework judgementkimi-k2.6 verdict → lead_scoresIgnite Models (the gate)
The scoring enginereads leads/activities, writes lead_scores + leads.statusIgnite app crm-lead-scorer
Similar-deal evidence (optional)reads core opportunity_vectorsVector (jina-embeddings-v4)

NOTE

Connect the DODIL MCP once, then every step shows an Ask your agent tab (the default — DODIL is agent-native) and a CLI tab. Install crm/core first if you're building the full suite — it owns the masters (leads/contacts/activities/opportunities) this skill scores. Standalone, Step 0 stubs those masters and Step 2 seeds a few leads, so you can run it end to end without the rest of the CRM.

Prerequisites

  • The dodil CLI (dodil auth login) or the DODIL MCP connected to your agent.
  • export BUCKET=crm — the same bucket your CRM masters (leads, opportunities, contacts, activities) already live in.
  • The live model ids (confirm with dodil ignite models list): chat kimi-k2.6, embeddings jina-embeddings-v4 (2048-dim).
  • The crm/core masters leads, contacts, activities and opportunities. If you don't have them yet, Step 0 stubs the minimum this skill reads.

Step 0 — Stub the masters you consume (skip if you have crm/core)

This skill scores the CRM masters; it doesn't own them. If you built crm/core (or the full suite), those tables already exist — skip to Step 1. Standalone, create the four masters this skill reads — leads, contacts, activities, opportunities — with the same column definitions crm/core uses (every non-key column nullable:true, so a partial seed row never trips NotNullViolation). Step 2 then seeds the leads you score into them.

You

Create the crm bucket, then four merge-keyed master tables with all non-key columns nullable: leads (key lead_id: email, org_domain, full_name, title, source, status, owner, score(double), created_at, updated_at); contacts (key email: user_id, full_name, org_domain, title, lifecycle_stage, source, owner, subscribed(boolean), created_at, updated_at); activities (key activity_id: opportunity_id, contact_email, kind, subject, body, direction, status, ts); opportunities (key opportunity_id: name, account_domain, primary_contact_email, pipeline, stage, status, amount(double), owner, source, close_date, created_at).

DODIL MCP tools called
data_bucket_create→data_table_create
Agent

Created bucket crm and 4 master tables — leads (pk lead_id), contacts (pk email), activities (pk activity_id), opportunities (pk opportunity_id) — all non-key columns nullable, so the Step 2 seed writes cleanly. These are the exact crm/core masters; if you already ran crm/core they're here and this step is a no-op.

Step 1 — The scoring tables + the policy as data

This skill owns two tables. lead_scores is the verdict store — one row per lead, merge-keyed on lead_id so a re-score upserts in place (never a duplicate). It carries both frameworks' dimension columns; only the active set is populated per row (the BANT columns for a BANT score, the six MEDDIC columns for a MEDDIC score, the rest left null). scoring_policy is the two-gate policy as a row — edit the thresholds without touching code.

NOTE

data table create makes every non-PK column NOT NULL by default — set "nullable":true on any optional column. Here all the BANT + MEDDIC dimension columns are nullable:true: a BANT score leaves the MEDDIC dims null and vice-versa, so a write to a NOT-NULL dimension column would 500 with NotNullViolation.

You

In the crm bucket, create two merge-keyed tables. lead_scores (key lead_id): opportunity_id, framework, score(double), tier, the BANT dims budget/authority/need/timing (double), the MEDDIC dims metrics/economic_buyer/decision_criteria/decision_process/pain/champion (double), rationale, model_id, scored_at. scoring_policy (key policy_id): framework, threshold_qualify(double), threshold_route(double), weights_json, auto_qualify_below_touch(boolean), updated_at.

DODIL MCP tools called
data_table_create
Agent

Created lead_scores (key lead_id, 18 columns — both frameworks' dims coexist, nullable) and scoring_policy (key policy_id). Upserts are idempotent, so re-scoring a lead updates its one row.

Now write the policy row — BANT, qualify at 0.7, route at 0.5. This is the single source of truth for the two gates; the Ignite handler in Step 4 reads the same numbers as hoisted constants.

You

In the crm bucket, upsert the default scoring policy: policy_id default, framework bant, threshold_qualify 0.7, threshold_route 0.5, weights equal across the four BANT dimensions, auto_qualify_below_touch false.

DODIL MCP tools called
data_table_upsert
Agent

Upserted policy 'default' (framework bant, qualify 0.7, route 0.5). The policy is data now — raise threshold_qualify and re-score to tighten the bar, no redeploy.

Step 2 — The leads you score (the signal bundle)

In the full suite these masters come from crm/core — this skill just reads them. Standalone, seed three leads that span the outcomes, each with the account context and an activity note the gate will read. Everything a rep would skim before a call is what the model reads: title, account, and the discovery notes.

You

In the crm bucket, seed three leads (key lead_id) with their opportunities and one activity note each: (1) l-strong Dana Okafor, VP Data Engineering at northwind.ai — a $180k opportunity, note says the board approved $200k to consolidate Postgres+Pinecone+Neo4j with a hard Q4 deadline and wants a POC in two weeks; (2) l-weak Sam Lee, a CS student on gmail.com — no opportunity, note says exploring for a class project, no budget or timeline; (3) l-mid Riley Chen, Team Lead Engineering at midco.io — a $40k opportunity, note says clear need but budget only earmarked (unapproved), needs VP sign-off, no firm timeline. All leads status=working.

DODIL MCP tools called
data_table_upsert
Agent

Seeded 3 leads (l-strong, l-weak, l-mid), 2 opportunities, and 3 activity notes. Each lead starts status=working — the engine will move it.

Step 3 — The BANT verdict gate (Ignite Models)

Here's the gate. For each lead the engine assembles the signal bundle — the lead fields, its account, and the recent activity notes — and asks kimi-k2.6 to score the four BANT dimensions plus an overall score and tier, returning only JSON. The system prompt is rendered from your params: the product_pitch (what "fit" means) and threshold_qualify (the bar the model is told about). The model judges; the policy decides.

NOTE

kimi-k2.6 is a reasoning model — guard against empty content at this gate. Called through the ignite models chat MCP tool / CLI directly (as here), there is no max_tokens knob to raise, so the model can spend its budget on hidden reasoning and return an empty content. Defend it in the prompt: end the system prompt with Return ONLY compact JSON, no reasoning or preamble, and retry once if content comes back empty. (Inside the Step 4 Ignite handler you do control this — it sets max_tokens: 4096 on the raw api.dodil.io/v1 call; the interactive CLI/MCP path does not expose it.)

You

Score lead l-strong on BANT for a unified data backend (SQL+vector+graph in one bucket) on kimi-k2.6. Give it the lead (Dana Okafor, VP Data Engineering, northwind.ai), its opportunity, and the discovery note. Rate budget, authority, need, timing each 0-1, then an overall score and tier; qualify only if score ≥ 0.7. Return ONLY JSON.

DODIL MCP tools called
ignite_models_chat
Agent

kimi-k2.6 returned {score:1.0, tier:strong, budget:1.0, authority:1.0, need:1.0, timing:1.0, rationale:'Approved budget, decision maker, exact-fit need, and hard Q4 deadline.'} — a clean qualify (1.0 ≥ 0.7).

Run the same gate over the weak and mid leads and you get the full spread. The student scores 0.05 — no budget, no authority, no timing — a clear disqualify. Riley scores 0.6: a real need (0.9) but unapproved budget (0.6), partial authority (0.5), vague timing (0.4). That lands in the route band [0.5, 0.7) — not a no, not yet a yes, so it goes to a human. Write each verdict to lead_scores and move the lead's status by the policy. The MEDDIC columns stay null on a BANT score.

You

Write the three BANT verdicts to lead_scores and set each lead's status by the policy (qualify at 0.7, route at 0.5): l-strong score 1.0 → status qualified; l-weak score 0.05 → disqualified; l-mid score 0.6 → working (routed to a human). Populate only the BANT dimension columns; leave the MEDDIC columns null.

DODIL MCP tools called
data_table_upsert→data_table_update
Agent

Upserted 3 lead_scores rows (BANT dims set, MEDDIC dims null) and moved the leads: l-strong→qualified, l-mid→working (routed), l-weak→disqualified. The route band caught the mid lead exactly as intended.

Now the scored queue is a query — a rep sees exactly who cleared the bar and why:

You

In the crm bucket, show every scored lead joined to its verdict: lead_id, status, score, tier, and the four BANT dimensions, ordered by score descending.

DODIL MCP tools called
data_sql
Agent

Three rows: l-strong qualified 1.0 (strong), l-mid working 0.6 (possible, routed), l-weak disqualified 0.05 (weak) — the whole spread, auditable down to each dimension.

Step 4 — The scoring engine on Ignite (crm-lead-scorer)

The three calls above are the engine's inner loop. In production one Ignite app, crm-lead-scorer, does it per lead: a POST /score with the lead JSON assembles the bundle, calls the gate, writes lead_scores, and moves leads.status. A local scheduler (or the CRM master pipeline) POSTs each status=working lead with no fresh score. It's a separate workload, so it gets its own service account — and because it both writes DataK3 and calls Models, it needs three live roles (confirm the exact names with dodil auth service-account list-roles):

  • k3.editor — write lead_scores and leads.
  • ignite.model-user — call kimi-k2.6 from inside the handler (token-billed).
  • ignite.app-developer — the deploy/invoke identity for the app itself.

NOTE

There is no ignite.developer role (an older doc named one) — the live catalog splits it into ignite.app-developer (deploy/invoke) and ignite.model-user (call models). A pure-SQL handler would need only k3.editor; this one calls Models, so it needs all three.

This ships as an image-mode Ignite app: a plain HTTP server (GET /healthz for the probe, POST /score for the work), packaged by a Dockerfile and built on deploy — not a handler(payload, ctx) compile-mode function. The handler hoists the policy knobs to the top as constants — FRAMEWORK, THRESH_QUALIFY, THRESH_ROUTE — the same values as the scoring_policy row (single source of truth), injected from env at deploy so a policy change is one place. Three things matter and each is a line below:

  • The Models call is the real OpenAI-compatible endpoint (api.dodil.io/v1), authed with a service-account token, max_tokens: 4096 (kimi-k2.6 is a reasoning model — a low budget returns empty content), and the response is wrapped in data (out["data"]["choices"]…).
  • The write goes 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. Re-scoring a lead re-writes the same lead_id, so the write is INSERT … ON CONFLICT (lead_id) 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). Writes retry on SerializationFailure.
  • 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-lead-scorer, an IMAGE-mode Ignite app (HTTP server on $PORT).
#   GET  /healthz -> {"status":"ready"}   (probe; no auth)
#   POST /score   -> a lead JSON -> runs the real BANT/MEDDIC gate, writes the verdict
 
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
from psycopg import errors as pg_errors
 
# --- hoisted knobs: mirror the scoring_policy row (defaults), injected as env at deploy ---
FRAMEWORK      = os.environ.get("FRAMEWORK", "bant")            # bant | meddic
THRESH_QUALIFY = float(os.environ.get("THRESH_QUALIFY", "0.7"))
THRESH_ROUTE   = float(os.environ.get("THRESH_ROUTE", "0.5"))
PRODUCT_PITCH  = os.environ.get("PRODUCT_PITCH",
                                "a unified data backend (SQL+vector+graph in one bucket)")
MODEL_ID       = os.environ.get("MODEL_ID", "kimi-k2.6")
BUCKET         = os.environ["BUCKET"]
SA_ID          = os.environ["DODIL_SERVICE_ACCOUNT_ID"]        # the cli-… serviceAccountId, NOT the uuid
SA_SECRET      = os.environ["DODIL_SERVICE_ACCOUNT_SECRET"]
PG_HOST        = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
PG_PORT        = int(os.environ.get("PG_PORT", "5432"))
 
ID_URL     = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
MODELS_URL = "https://api.dodil.io/v1/chat/completions"
UA         = "crm-lead-scorer/1.0"   # explicit UA — stdlib urllib's default is Cloudflare-banned (403 1010)
 
DIMS = {"bant":   ["budget", "authority", "need", "timing"],
        "meddic": ["metrics", "economic_buyer", "decision_criteria",
                   "decision_process", "pain", "champion"]}
ALL_DIMS = DIMS["bant"] + DIMS["meddic"]
LS_COLS = (["lead_id", "opportunity_id", "framework", "score", "tier"]
           + ALL_DIMS + ["rationale", "model_id", "scored_at"])
 
SYS = (f"Score this lead on {FRAMEWORK.upper()} for {PRODUCT_PITCH}. "
       f"Given the lead, its account, and recent activity notes, rate "
       f"{', '.join(DIMS[FRAMEWORK])} each 0-1, then an overall score 0-1 and "
       f"tier (strong|possible|weak). Qualify only if score >= {THRESH_QUALIFY}. "
       f"Return ONLY JSON: {{score, tier, {', '.join(DIMS[FRAMEWORK])}, 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": 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 _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 _bundle(lead):                                   # lead + account + recent activity note
    parts = [f"lead: {lead.get('full_name','?')}, {lead.get('title','?')} "
             f"at {lead.get('org_domain','?')}"]
    if lead.get("opportunity"): parts.append(f"opportunity: {lead['opportunity']}")
    if lead.get("activity"):    parts.append(f"activity: {lead['activity']}")
    return " | ".join(parts)
 
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 _write(token, lid, opp_id, verdict, status):
    row = {"lead_id": lid, "opportunity_id": opp_id, "framework": FRAMEWORK,
           "score": verdict["score"], "tier": verdict.get("tier"),
           "rationale": verdict.get("rationale"), "model_id": MODEL_ID, "scored_at": _now()}
    for d in ALL_DIMS:
        row[d] = verdict.get(d)                      # only the active framework's dims come back populated
    # Re-scoring re-writes the SAME lead_id, so this must be 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. The upsert replaces the one row, immediately scan/JOIN-visible.
    upsert = (f"INSERT INTO lead_scores ({', '.join(LS_COLS)}) "
              f"VALUES ({', '.join(['%s'] * len(LS_COLS))}) "
              f"ON CONFLICT (lead_id) DO UPDATE SET "
              + ", ".join(f"{c} = EXCLUDED.{c}" for c in LS_COLS if c != "lead_id"))
    vals = [row[c] for c in LS_COLS]
 
    def _w1():
        with _pg(token) as conn, conn.cursor() as cur:
            cur.execute(upsert, vals); conn.commit()
    def _w2():
        with _pg(token) as conn, conn.cursor() as cur:
            cur.execute("UPDATE leads SET status = %s, score = %s, updated_at = %s "
                        "WHERE lead_id = %s", (status, verdict["score"], _now(), lid))
            conn.commit()
    _retry(_w1)   # the verdict store
    _retry(_w2)   # apply the two-gate policy to leads.status
 
def score_lead(lead):
    token = _token()
    verdict = _extract_json(_chat(token, _bundle(lead)))
    score = verdict["score"]
    status = ("qualified"    if score >= THRESH_QUALIFY             # the policy decides, not the label
              else "working" if score >= THRESH_ROUTE               # route band -> a human
              else "disqualified")
    _write(token, lead["lead_id"], lead.get("opportunity_id"), verdict, status)
    return {"lead_id": lead["lead_id"], "framework": FRAMEWORK,
            "verdict": verdict, "status": status, "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 != "/score": return self._send(404, {"error": "no_route", "path": self.path})
        try:
            n = int(self.headers.get("Content-Length") or 0)
            lead = json.loads(self.rfile.read(n) or b"{}")
            if not lead.get("lead_id"): return self._send(400, {"error": "missing lead_id"})
            return self._send(200, score_lead(lead))
        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-lead-scorer serving on 0.0.0.0:{port} framework={FRAMEWORK} bucket={BUCKET}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

Only psycopg is a third-party dep (the Postgres driver); everything else is stdlib. The two sibling files that make it an image — ./scorer/Dockerfile and ./scorer/requirements.txt:

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

Give it a least-privilege identity, then deploy:

You

Create a service account crm-lead-scorer-sa, grant it k3.editor plus ignite.model-user and ignite.app-developer, then deploy my ./scorer app (image mode — its Dockerfile builds on deploy) to Ignite as crm-lead-scorer on port 8080 with health path /healthz, passing the service-account creds and the policy constants (FRAMEWORK bant, THRESH_QUALIFY 0.7, THRESH_ROUTE 0.5) as runtime env. Use the cli- serviceAccountId as DODIL_SERVICE_ACCOUNT_ID, not the uuid. Then POST a lead to /score to smoke-test.

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

Created crm-lead-scorer-sa (serviceAccountId cli-crm-lead-scorer-sa), granted k3.editor + ignite.model-user + ignite.app-developer, built + deployed crm-lead-scorer (image:build, public FQDN on :8080, scale-to-zero). POST /score for l-strong returned status=qualified, written=true.

NOTE

Deploy: image mode (Lane B), validated live 2026-09-02. The crm-lead-scorer deploy + /score smoke-test above use image mode — a Dockerfile + --dockerfile-path, Kaniko build-on-deploy. Validated live 2026-09-02: crm-lead-scorer deployed end-to-end, served /healthz + /score unauthenticated, and its written rows persisted durably (they survived +154s, independently re-confirmed at +95s). This is the reference engine — every other CRM engine copies its identical handler/deploy pattern. The gate, the pg-wire writes, the policy branches, and the re-score are each also proven one call at a time in Steps 3, 5, 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).

Step 5 — MEDDIC + similar-deal evidence (the enterprise path)

For enterprise deals, flip framework to meddic: six dimensions instead of four, and — with use_similar_deals: true — the handler first retrieves won/lost deals like this one from core opportunity_vectors (embedded with jina-embeddings-v4) and folds them into the prompt as evidence ("deals like this closed how?"). The verdict lands in the same lead_scores table — now the six MEDDIC columns are populated and the BANT columns are null.

First, the similar-deal retrieval — a KNN over the opportunity embeddings. A strong enterprise lead pulls the won consolidation deal nearest:

You

In the crm bucket, vector-search opportunity_vectors for deals similar to an enterprise buyer with a VP sponsor, approved budget, and a hard deadline to consolidate their data stack — top 2, cosine, with jina-embeddings-v4.

DODIL MCP tools called
data_vsearch
Agent

Nearest is opp-won-ref (cosine distance 0.199) — an enterprise consolidation closed WON — ahead of opp-lost-ref (0.366), a stalled SMB deal. The won motion becomes evidence in the MEDDIC prompt.

Now the MEDDIC gate, with that won deal as evidence — six dimensions, same JSON contract:

You

Score lead l-strong on MEDDIC for a unified data backend on kimi-k2.6, giving it the lead, its opportunity, the discovery note, and the nearest won deal as evidence. Rate metrics, economic_buyer, decision_criteria, decision_process, pain, champion each 0-1, then an overall score and tier; qualify only if ≥ 0.7. Return ONLY JSON.

DODIL MCP tools called
ignite_models_chat
Agent

kimi-k2.6 returned {score:0.92, tier:strong, metrics:0.9, economic_buyer:0.9, decision_criteria:0.9, decision_process:0.85, pain:0.9, champion:0.95, rationale:'VP champion with board-approved budget, urgent pain, and proven similar-won motion.'} — all six MEDDIC dims, a qualify.

Write it and you can see a MEDDIC row and a BANT row coexist in one table — each with only its framework's dimensions filled:

You

In the crm bucket, add a meddic policy row (policy_id meddic, framework meddic, thresholds 0.7/0.5), upsert the MEDDIC verdict for an enterprise lead l-ent-meddic (score 0.92, the six dims, BANT dims null), set that lead qualified, then show every meddic-framework score with its six dimensions.

DODIL MCP tools called
data_table_upsert→data_table_update→data_sql
Agent

One table, two frameworks: the meddic row has metrics/economic_buyer/decision_criteria/decision_process/pain/champion set and budget/authority/need/timing null — proof the nullable dual-dimension schema serves both.

Routes

The three calls above — score a lead, retrieve similar deals, read the queue — are the same ops your app makes. The download (see Get the code) fronts this bucket with a small FastAPI app, routes.py: policy + verdict CRUD, the framework score gate, the similar-deal vector search, and the scored queue JOIN over the models in models.py (the same classes the ORM tabs above show). The routes live on an APIRouter — the suite app mounts all seven CRM components on one FastAPI under per-component prefixes (this one at /qualification-scoring) — while app = FastAPI(...) at the bottom keeps the package independently runnable (uvicorn routes:app). Every route follows the three DataK3 rules the package bakes in, so quoting it is documenting them.

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

# db.py — INSERT ... ON CONFLICT DO UPDATE (idempotent keyed write)
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:
        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 is what makes a re-score land the row once. Live-verified: after scoring three leads and re-scoring l-strong, count(*) = count(DISTINCT lead_id) = 3 — one lead_scores row per lead, never a duplicate.

CRUD — the policy is a row. POST /policies upserts the two-gate policy; raise a threshold and re-score, no redeploy. Each write is an upsert then a commit (DataK3 has no read-your-writes inside an open transaction; the engine is expire_on_commit=False, so routes commit before they return):

# routes.py — the two-gate policy as data (keyed on policy_id)
@router.post("/policies")
def upsert_policy(p: PolicyIn, s: Session = Depends(db)):
    row = p.model_dump()
    row["updated_at"] = _now()
    upsert(s, ScoringPolicy, [row], key="policy_id")
    s.commit()
    return {"ok": True, "policy_id": p.policy_id}

Live-verified: the default row reads back framework=bant, threshold_qualify=0.7, threshold_route=0.5 — the policy is data.

Workflow op 1 — the framework gate (MODELS). POST /leads/{lead_id}/score is the money step: assemble the signal bundle (the lead, its opportunity, its most recent activity note), score it on kimi-k2.6 against the named policy, upsert the verdict to lead_scores (only the active framework's dims set, the other set null), and move leads.status by the two gates. The policy is read from the scoring_policy row (_load_policy), so the model judges each dimension and the policy decides the status:

# routes.py — workflow op 1: the framework gate, then apply the two-gate policy.
# One of the suite's four role gates: it spends a Models call AND moves leads.status,
# so it demands the leads:score pool permission (see ## Auth).
@router.post("/leads/{lead_id}/score")
def score_lead(lead_id: str, q: ScoreIn,
               user=Depends(require_permission("leads:score")),
               s: Session = Depends(db)):
    lead = s.get(Lead, lead_id)
    if not lead:
        raise HTTPException(404, "no such lead")
    framework, tq, tr = _load_policy(s, q.policy_id)
    pitch = q.product_pitch or PRODUCT_PITCH
    bundle, opp_id = _bundle(s, lead)
 
    token = _models_token()
    verdict = _gate(token, framework, pitch, tq, bundle)
    score = float(verdict["score"])
    status = _apply_policy(score, tq, tr)   # >=tq qualified; >=tr working (routed); else disqualified
 
    row = {"lead_id": lead_id, "opportunity_id": opp_id, "framework": framework,
           "score": score, "tier": verdict.get("tier"), "rationale": verdict.get("rationale"),
           "model_id": CHAT_MODEL, "scored_at": _now()}
    for d in ALL_DIMS:
        row[d] = float(verdict[d]) if d in verdict and verdict[d] is not None else None
    upsert(s, LeadScore, [row], key="lead_id")
 
    # apply the policy to leads.status — read the row, merge, upsert the FULL row (keeps the rest)
    lead_row = {c.name: getattr(lead, c.name) for c in Lead.__table__.columns}
    lead_row.update(status=status, score=score, updated_at=_now())
    upsert(s, Lead, [lead_row], key="lead_id")
    s.commit()
    return {"lead_id": lead_id, "framework": framework, "score": score,
            "tier": verdict.get("tier"), "status": status, "verdict": verdict, "written": True}

_gate is the reasoning-model guard from Step 3 in code — it sets max_tokens: 4096 and retries once on empty content. Both mattered live: a lead came back empty on the first kimi-k2.6 call and returned a clean verdict on the retry. Re-validated 2026-09-06 on the persistent crm bucket, over the suite's seeded funnel: scoring lead:greyparrot.ai returned score 0.93, tier strong (budget 0.95, authority 0.90, need 0.95, timing 0.90 — rationale "Budget approved (~£40k), decision driven by Head of Data, exact SQL+vector+graph consolidation need, hard Q3 deadline.") → qualified at the 0.7 gate, written: true. The write-back to leads.status reads the row and upserts the full merged row, so the lead's other fields are kept. And note the timing: a kimi-k2.6 call runs anywhere from ~12s to nearly 3 minutes — the route assembles the bundle, releases its reads, calls the gate, and only then writes in a fresh transaction. Never hold a pg connection open across a model call — under load those held connections are exactly what exhausts the tables reader's concurrency budget.

Workflow op 2 — similar-deal evidence (VECTOR). POST /scores/similar takes a query embedding and returns the nearest opportunities by pgvector cosine distance — the won/lost precedent folded into the MEDDIC prompt (Step 5):

# routes.py — workflow op 2: nearest won/lost deals (VECTOR)
@router.post("/scores/similar")
def similar_deals(q: SimilarIn, s: Session = Depends(db)):
    rows = s.execute(
        select(OpportunityVector.opportunity_id, OpportunityVector.outcome,
               OpportunityVector.embedding.cosine_distance(q.embedding).label("d"))
        .order_by("d")
        .limit(q.top_k)
    ).all()
    return {"matches": [{"opportunity_id": oid, "outcome": outcome, "distance": float(dist)}
                        for oid, outcome, dist in rows]}

Live-verified 2026-09-06 over real jina-embeddings-v4 embeddings on the suite's (indexed) opportunity_vectors: the nearest deal to an enterprise consolidation profile was the won reference (cosine 0.132), far ahead of the lost one (0.560) — the won motion becomes evidence, and the search skips gracefully if crm/core didn't build opportunity_vectors.

Workflow op 3 — the scored queue (SQL). GET /scored is the rep's queue as a query: leads JOINed to their verdict, ordered by score, every dimension auditable:

# routes.py — workflow op 3: the scored queue (leads JOIN lead_scores)
@router.get("/scored")
def scored_queue(user=Depends(current_user), s: Session = Depends(db)):
    rows = s.execute(text(
        "SELECT l.lead_id, l.status, s.score, s.tier, s.framework, "
        "s.budget, s.authority, s.need, s.timing "
        "FROM leads l JOIN lead_scores s ON s.lead_id = l.lead_id "
        "ORDER BY s.score DESC"
    )).mappings().all()
    return {"queue": [dict(r) for r in rows]}

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

Auth — config at the edge, one gate in the app

On Ignite, end-user login is configuration, not code. The suite deploys with the crm-suite dodil-appid pool attached (user_pool: crm-suite in .dodil/deploy.yaml, 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: X-Dodil-User (sub, email, app_roles) and X-Dodil-User-Jwt (the raw verified token, 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 — and what it keeps is the app's one remaining job, role-based gating:

# routes.py — the app-user identity in use (auth.py is header-trust, ~40 lines, no crypto)
from auth import current_user, require_permission
 
# the money op is permission-gated — only a role with leads:score may spend a Models call
# AND move leads.status (one of the suite's four surviving gates)
@router.post("/leads/{lead_id}/score")
def score_lead(lead_id: str, q: ScoreIn,
               user=Depends(require_permission("leads:score")), ...):
    ...
 
# the queue just needs a signed-in user
@router.get("/scored")
def scored_queue(user=Depends(current_user), ...):
    ...

After the audit, leads:score is this component's only gate — the policy CRUD and the similar-deal KNN ride on the gateway's authentication alone. The suite's other surviving gates: orgs:qualify (lead-to-opportunity), quotes:approve (quote-cpq), forecast:override (pipeline-forecast) — all checked against the pool's sales / analyst / manager role catalog (analyst and manager carry leads:score).

The two-plane rule is unchanged: the pool identifies the user; the app still reaches DataK3 via its own service account — an app-user is never a bucket principal. Locally, opt in to a stub identity with DEV_ALLOW_ANON=1; it carries no permissions unless you grant them (DEV_USER_PERMISSIONS=leads:score), so the gate stays gated on a laptop. 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-qualification-scoring/v1.tar. This post is a walkthrough of exactly those files; the tarball is source only (no Dockerfile — deploy is in Ship it):

models.py          # SQLAlchemy — lead_scores + scoring_policy (owned) + the crm/core masters it scores
routes.py          # FastAPI    — an APIRouter the suite mounts + a standalone app; policy/verdict CRUD
                   #              + score (gate, leads:score-gated) + similar (vector) + scored (JOIN)
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 0/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 /policies · GET /scores/{lead_id}
# POST /leads/{lead_id}/score  (the framework gate — needs the SA creds in .env)
# POST /scores/similar · GET /scored

models.Base.metadata.create_all is the ORM tab's every class at once — the same tables Steps 0–1 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 /qualification-scoring) 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.

How the pillars map

One bucket, and the qualification engine reaches across it — no second system to sync.

JobThe usual stackOn DataK3
The verdict + every dimension scoreA CRM custom object + app codelead_scores — a merge-keyed SQL row, re-scored in place
The qualify/route policyHard-coded thresholds in the appscoring_policy — a row you edit, no redeploy
The framework judgementA prompt bolted onto an external LLMkimi-k2.6 on Ignite Models — one auth context, token-billed
"Deals like this one" as evidencePinecone + an embedding pipelinecore opportunity_vectors — a VECTOR(2048) column, data vsearch
The engineA worker on your own infraIgnite crm-lead-scorer — scale-to-zero, its own service account

Because the score, the lead, and the similar deals are all in one bucket, the rep's "who's qualified and why" is a single JOIN — lead_scores to leads to activities — over one copy of the rows.

Customize — the decisions this skill asks you

Q1 · framework — BANT or MEDDIC?

"How do you qualify — fast velocity sales, or complex enterprise deals?"

  • bant (default) → four dimensions (budget, authority, need, timing), the simpler prompt. Best for SMB / high-velocity. Populates the four BANT columns; the MEDDIC columns stay null.
  • meddic → six dimensions (metrics, economic buyer, decision criteria, decision process, pain, champion), the enterprise prompt. Sets FRAMEWORK=meddic in the handler and renders the six-dimension system prompt. Recommend pairing with use_similar_deals: true.

Q2 · threshold_qualify + threshold_route — the two-gate policy

"Above what score does a lead auto-qualify, and below what score do you stop looking?" → Written to the scoring_policy row and the handler constants THRESH_QUALIFY / THRESH_ROUTE (single source — §one-source-of-truth). Two named presets so you have values, not just a direction:

  • balanced (default) → threshold_qualify 0.7 / threshold_route 0.5. A healthy split — most real leads route to a human, the strong ones auto-qualify.
  • strict → threshold_qualify 0.85 / threshold_route 0.6. Fewer auto-qualify (more human review, higher precision), and more marginal leads drop out. Use it when a rep's hour is expensive and a false qualify costs more than a missed one.

Raise threshold_qualify and fewer leads auto-qualify; raise threshold_route and more get disqualified outright. This is the money knob — it sets what fraction of leads ever reach a rep.

IMPORTANT

The threshold lives in THREE places — keep them in sync. (1) the scoring_policy row, (2) the handler env constants THRESH_QUALIFY/THRESH_ROUTE (Step 4), and (3) the gate's system prompt — the Qualify only if score >= {threshold_qualify} line the model is told (Step 3). The policy row and the handler constants decide the status; the prompt copy only frames the model's self-assessment, but if it drifts from the policy the model's own "qualify" hint contradicts the gate that actually fires. Change the number in all three, or re-render the prompt from the same param.

Q3 · product_pitch — what counts as a fit?

"In one line, what are you selling?" → Renders into the gate's system prompt (Score this lead ... for {{product_pitch}}). Default: "a unified data backend (SQL+vector+graph in one bucket)". Changing it re-frames every dimension the model rates — this is your definition of a good-fit lead.

Q4 · use_similar_deals — fold in won/lost evidence?

"Should the scorer look at how similar past deals closed?"

  • false (default) → the gate scores on the lead's own signal only.
  • true → the handler first KNN-searches core opportunity_vectors for the nearest won/lost deals and adds them to the prompt as evidence. Recommended for meddic / enterprise, where deal precedent matters. Skips gracefully if crm/core didn't build opportunity_vectors.

Industry variants (saas / manufacturing / finserv / real-estate) compose this skill with gate/policy tweaks — e.g. finserv adds a KYC hard gate. See the per-industry CRM pages.

Test

Every command below ran live against DataK3 on 2026-09-01 (org IHDIASH), one call at a time. Both tested_branches were exercised: {framework: bant} (Steps 1–3) and {framework: meddic, use_similar_deals: true} (Step 5). The Ignite deploy in Step 4 uses the image-mode pattern validated live 2026-09-02 on crm-lead-scorer (deploys, serves /healthz + /score unauthenticated, writes durably — see the note there); the gate, the writes, and every policy branch are proven.

The downloadable package (## Routes) was itself re-validated on 2026-09-06 on the persistent crm bucket, as part of the composed suite: the db.upsert ON CONFLICT write path (a re-score keeps count(*) = count(DISTINCT lead_id)), the kimi-k2.6 gate (the suite's funnel lead lead:greyparrot.ai scored 0.93 — budget 0.95 / authority 0.90 / need 0.95 / timing 0.90 — qualified at the 0.7 gate; an earlier run also proved the empty-content retry in _gate), the money-safe DECIMAL write over the pg wire (opportunities.amount reads back exactly as DECIMAL(18,2), not a dropped 0), the similar_deals pgvector KNN (won 0.132 ahead of lost 0.560), and the scored JOIN (the full qualified/working/disqualified spread).

# 1. the policy landed as data
dodil data sql -b "$BUCKET" "SELECT framework, threshold_qualify, threshold_route FROM scoring_policy WHERE policy_id='default'"
#  bant | 0.7 | 0.5
 
# 2. the gate returns the right dimension keys per framework (4 for bant, 6 for meddic) — Steps 3 & 5
#    bant  -> {score,tier,budget,authority,need,timing,rationale}
#    meddic-> {score,tier,metrics,economic_buyer,decision_criteria,decision_process,pain,champion,rationale}
 
# 3/4/5. strong -> qualified, weak -> disqualified, mid -> routed (working)
dodil data sql -b "$BUCKET" "
  SELECT l.lead_id, l.status, s.score, s.tier
  FROM leads l JOIN lead_scores s ON s.lead_id=l.lead_id
  WHERE s.framework='bant' ORDER BY s.score DESC"
#  l-strong | qualified    | 1.00 | strong
#  l-mid    | working      | 0.60 | possible   <- score in [0.5,0.7) -> routed to a human
#  l-weak   | disqualified | 0.05 | weak
 
# 6. re-score is idempotent — one row per lead, never a duplicate
dodil data table upsert lead_scores -b "$BUCKET" --row '{"lead_id":"l-strong","opportunity_id":"opp-strong","framework":"bant","score":1.0,"tier":"strong","budget":1.0,"authority":1.0,"need":1.0,"timing":1.0,"metrics":null,"economic_buyer":null,"decision_criteria":null,"decision_process":null,"pain":null,"champion":null,"rationale":"Approved budget, decision maker, exact-fit need, and hard Q4 deadline.","model_id":"kimi-k2.6","scored_at":"2026-09-01T12:05:00Z"}'
dodil data sql -b "$BUCKET" "SELECT count(*) AS total, count(DISTINCT lead_id) AS distinct_leads FROM lead_scores"
#  total = 3, distinct_leads = 3   (before the meddic row; still 1:1 after)

One-shot

With the DODIL MCP connected, paste this to build the whole scorer at once on your CRM bucket:

On my DataK3 bucket `crm` (which already has leads/opportunities/contacts/activities), build a
BANT/MEDDIC lead-qualification engine. Confirm each step.
 
1. Create merge-keyed tables: lead_scores (key lead_id) with framework, score(double), tier, the BANT
   dims budget/authority/need/timing and the MEDDIC dims metrics/economic_buyer/decision_criteria/
   decision_process/pain/champion (all double, nullable), rationale, model_id, scored_at; and
   scoring_policy (key policy_id) with framework, threshold_qualify(double), threshold_route(double),
   weights_json, auto_qualify_below_touch(boolean), updated_at.
2. Upsert policy `default`: framework bant, threshold_qualify 0.7, threshold_route 0.5.
3. For each status=working lead with no fresh score, assemble the lead + account + recent activity notes
   and score it on kimi-k2.6 (BANT: rate budget/authority/need/timing 0-1, overall score, tier; return
   ONLY JSON). Write lead_scores (BANT dims set, MEDDIC null) and set leads.status: score>=0.7 qualified,
   0.5<=score<0.7 working (routed), else disqualified.
4. Deploy an image-mode Ignite app `crm-lead-scorer` — an HTTP server (GET /healthz, POST /score) built
   from a Dockerfile (--dockerfile-path, --port 8080, --health-path /healthz), own service account with
   k3.editor + ignite.model-user + ignite.app-developer, DODIL_SERVICE_ACCOUNT_ID = the cli- serviceAccountId
   (not the uuid), policy constants FRAMEWORK/THRESH_QUALIFY/THRESH_ROUTE as env. It writes over the Postgres
   wire (pg.uk-lon-1.dodil.io). Smoke-test by POSTing a lead to /score.
5. (Enterprise) For framework=meddic + use_similar_deals: KNN core opportunity_vectors for the nearest
   won/lost deals, fold them into a six-dimension MEDDIC prompt, write the six MEDDIC columns.

Ship it

crm-lead-scorer is an image-mode Ignite app — an HTTP server you POST leads to on /score. Give it a least-privilege service account (the three roles in Step 4, and set DODIL_SERVICE_ACCOUNT_ID to the cli-crm-lead-scorer-sa serviceAccountId, not the uuid), inject the policy constants as env, and deploy its Dockerfile with dodil ignite app deploy crm-lead-scorer --code ./scorer --dockerfile-path Dockerfile --port 8080 --health-path /healthz. The platform builds the image on deploy (Lane B) — no --runtime python, no separate build step. The full lifecycle — DODIL git → CI checks → a scanned image in the registry → versioning and rollback — is walked end to end in Ship a DODIL App.

Connect your tools

Everything this build wrote lives in one DataK3 bucket, reachable by your own stack — not just the CLI. 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).
  • Vector — pgvector (<=>) over the same wire, or a Qdrant/Pinecone client against the same opportunity_vectors rows.

Full, live-validated walkthrough: Connect your tools.

Conclusion

Lead qualification stops being a gut call. The verdict is a row (lead_scores, every dimension auditable), the policy is a row (scoring_policy, two gates you tune without a redeploy), and the judgement is a kimi-k2.6 gate on the same bucket your CRM already lives in. A strong lead qualifies itself, a weak one drops out, and the genuinely-ambiguous middle routes to a human — which is exactly where a person's time is worth spending. Swap BANT for MEDDIC with one param, fold in won-deal evidence with another, and the whole engine is one Ignite app over one copy of your rows.

Next steps: