The problem — and why it matters

You run RevOps or a deal desk, and every CRM eventually needs the same unglamorous thing: turn a deal into a quote with real prices, real discounts, and a rule about who is allowed to give away margin. Salesforce charges for CPQ as a separate SKU; most teams bolt on a second tool, sync the product catalog nightly, and still email a spreadsheet to a sales manager for sign-off. The money leak isn't the tool licence — it's the week of deal-desk latency on quotes that should self-approve, and the margin that walks out the door on the ones that shouldn't.

The whole thing is six merge-keyed tables and one rule. A products catalog, price_books (one per currency/region), price_book_entries (the negotiated price per product per book), quotes + quote_lines, and — the load-bearing piece — a discount_policy row that is data, not prose: below approval_threshold_pct (10%) a quote self-approves, up to max_discount_pct (25%) it routes for approval, above 25% it's rejected. A small Ignite engine reprices lines and stamps that verdict; a kimi-k2.6 gate handles the judgement calls the threshold can't ("22% is over the auto line, but it's a strategic first logo — route it, don't reject it").

What you'll build: a one-bucket crm CPQ — the SQL catalog + pricing tables, the crm-cpq-pricer Ignite engine that recomputes line_total = quantity × unit_price × (1 − discount_pct) and rolls the header, and a kimi-k2.6 discount-approval gate. Prices resolve from the price book, the policy lives in a row you can edit without a deploy, and re-pricing is idempotent — every quote is merge-keyed, so a retry writes the same total, never a duplicate.

Prerequisites

  • The dodil CLI (dodil auth login) and an agent connected over MCP (Claude Code, Cursor, VS Code, Codex).
  • export BUCKET=crm — one bucket is the whole CPQ data plane. (Standalone it's its own bucket; in the crm suite it lands in the shared crm bucket next to crm/core's masters.)
  • The consumed masters opportunities, accounts, contacts (owned by crm/core). Standalone, seed a couple of stub rows so a quote has an opportunity + account to point at.

Step 1 — Stand up the catalog and price books (the SQL core)

Create the bucket, then the product catalog and the price books. A PRIMARY KEY (merge-key) is required — writes are keyed, so re-runs and shard retries upsert idempotently instead of duplicating a product or a price.

You

Create a DataK3 bucket called $BUCKET, then merge-keyed tables: products (key product_id: name, sku, category, unit, list_price double, active boolean), price_books (key price_book_id: name, currency, region, active boolean), and price_book_entries (key entry_id: price_book_id, product_id, unit_price double).

DODIL MCP tools called
data_bucket_create→data_table_create
Agent

Created bucket $BUCKET. products (PK product_id, 7 cols), price_books (PK price_book_id, 5 cols), and price_book_entries (PK entry_id, 4 cols) created.

Seed a catalog and one USD price book — the default currency. The book price can sit below list (the negotiated rate for this book); the pricer resolves the book price first and only falls back to products.list_price when a product isn't in the book.

You

Seed 3 products (prod-seat 'DataK3 Platform Seat' subscription unit seat list 1000; prod-storage 'DataK3 Storage (TB/mo)' usage unit tb list 200; prod-support 'Premium Support' service unit year list 5000, all active), a USD price book pb-usd (USD, region NA, active), and entries: seat 900, storage 200, support 5000.

DODIL MCP tools called
data_table_upsert
Agent

Upserted 3 products, 1 price book (pb-usd, USD), 3 entries (wal_written: true).

Multi-currency is one param away. With multi_currency: true, add a second book — e.g. pb-eur (currency EUR, region EU) with its own entries (seat 850, storage 190) — and point a quote's price_book_id at it. The pricer resolves that quote's lines against its book, so a EUR quote prices in EUR with no code change. (Validated below.)

Step 2 — The discount policy (as data) and the quote tables

The policy is the money knob, so it lives as a row, not as a constant buried in code — edit it and every future quote re-prices under the new rule with no deploy. Then the quotes header and quote_lines.

You

In $BUCKET create discount_policy (key policy_id: max_discount_pct double, approval_threshold_pct double, role_required, updated_at), quotes (key quote_id: opportunity_id, account_domain, price_book_id, status, currency, subtotal double, discount_total double, total double, valid_until, created_at), and quote_lines (key quote_line_id: quote_id, product_id, quantity int, unit_price double, discount_pct double, line_total double). Seed one policy row: auto-approve at/under 0.10, max 0.25, role sales_manager.

DODIL MCP tools called
data_table_create→data_table_upsert
Agent

Created discount_policy, quotes, quote_lines. Seeded policy 'default' (approval_threshold_pct 0.10, max_discount_pct 0.25).

Now draft a quote against an opportunity and add its lines. A line carries the product_id, quantity, and the discount_pct the rep is asking for; unit_price and line_total start at 0 — the pricer fills them. (Standalone, upsert a stub opportunities/accounts row first so the quote has something to point at; in the suite those come from crm/core.)

You

In $BUCKET, draft quote q-approved (opportunity opp-acme-1, account acme.com, price_book pb-usd, currency USD, status draft, totals 0) with two lines at a 5% discount: 10× prod-seat and 5× prod-storage.

DODIL MCP tools called
data_table_upsert
Agent

Drafted q-approved with 2 lines (qa-1: 10× prod-seat @5%, qa-2: 5× prod-storage @5%). unit_price/line_total left 0 for the pricer.

Routes

The download (see Get the code) fronts this bucket with a small FastAPI app, routes.py — CRUD over the catalog, price books, and quotes, plus the two ops that turn a draft into a priced, policy-gated quote: a deterministic price (the pricer logic) and a discount-band adjudicate (Ignite Models). This is the app layer of Steps 1–4. The routes live on an APIRouter — the suite app mounts all seven CRM components on one FastAPI under per-component prefixes (this one at /quote-cpq) — while app = FastAPI(...) at the bottom keeps the package independently runnable (uvicorn routes:app). Every route follows the same DataK3 rules the package bakes in, so quoting it is documenting them.

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:
        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. Verified live: re-inserting a committed quote_id failed with SQLSTATE 23505, while upsert (→ INSERT … ON CONFLICT DO UPDATE) re-wrote it in place. That's the whole reason re-pricing is safe to re-run.

Catalog / book / quote 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). Money columns are Numeric(18, 2) DECIMALs written over the pg wire — _money() quantizes to cents so a price never lands as a bare float:

# routes.py — CRUD over the catalog + quotes, keyed on the natural PK
@router.post("/products")
def upsert_product(p: ProductIn, s: Session = Depends(db)):
    upsert(s, Product, [p.model_dump()], key="product_id")
    s.commit()
    return {"ok": True, "product_id": p.product_id}
 
 
@router.post("/price_book_entries")
def upsert_price_book_entry(e: PriceBookEntryIn, s: Session = Depends(db)):
    row = e.model_dump()
    row["unit_price"] = _money(row["unit_price"])           # money -> DECIMAL over the pg wire
    upsert(s, PriceBookEntry, [row], key="entry_id")
    s.commit()
    return {"ok": True, "entry_id": e.entry_id}
 
 
@router.post("/quote_lines")
def upsert_quote_line(ln: QuoteLineIn, s: Session = Depends(db)):
    # unit_price/line_total are left 0 for the pricer to resolve.
    row = ln.model_dump()
    row.update(unit_price=Decimal("0.00"), line_total=Decimal("0.00"))
    upsert(s, QuoteLine, [row], key="quote_line_id")
    s.commit()
    return {"ok": True, "quote_line_id": ln.quote_line_id}

Workflow op 1 — price the quote (SQL, deterministic). POST /quotes/{quote_id}/price is the pricer: it resolves each line's unit_price from the quote's price book (fallback products.list_price), computes line_total = quantity × unit_price × (1 − discount_pct), rolls subtotal / discount_total / total onto the header, and stamps status from the effective discount vs the discount_policy row. Every write is a keyed upsert, so repricing is idempotent:

# routes.py — workflow op 1: the pricer (resolve -> roll -> stamp status)
@router.post("/quotes/{quote_id}/price")
def price_quote(quote_id: str, s: Session = Depends(db)):
    quote = s.get(Quote, quote_id)
    if not quote:
        raise HTTPException(404, "no such quote")
    lines = s.execute(_PRICE_SQL, {"qid": quote_id}).mappings().all()
    if not lines:
        raise HTTPException(409, f"quote {quote_id} has no lines to price")
 
    approval, max_disc = _policy(s)                       # the money knob lives as the policy ROW
    subtotal = discount_total = total = Decimal("0")
    for ln in lines:
        unit = _money(ln["unit_price"])
        qty = int(ln["quantity"])
        disc = Decimal(str(ln["discount_pct"]))
        gross = unit * qty
        line_total = _money(gross * (Decimal("1") - disc))
        subtotal += gross
        discount_total += gross * disc
        total += line_total
        upsert(s, QuoteLine, [{
            "quote_line_id": ln["quote_line_id"], "quote_id": quote_id,
            "product_id": ln["product_id"], "quantity": qty,
            "unit_price": unit, "discount_pct": float(disc), "line_total": line_total,
        }], key="quote_line_id")
 
    subtotal, discount_total, total = _money(subtotal), _money(discount_total), _money(total)
    eff = float(discount_total / subtotal) if subtotal else 0.0
    status = ("approved" if eff <= approval
              else "needs_approval" if eff <= max_disc
              else "rejected")
    head = {c.name: getattr(quote, c.name) for c in Quote.__table__.columns}
    head.update(subtotal=subtotal, discount_total=discount_total, total=total, status=status)
    upsert(s, Quote, [head], key="quote_id")              # full-row upsert of the header
    s.commit()
    return {"quote_id": quote_id, "subtotal": float(subtotal),
            "discount_total": float(discount_total), "total": float(total),
            "effective_discount": eff, "status": status}

Live-verified over the pg wire on a throwaway bucket, the three demo quotes priced exactly as the policy dictates — the resolve alone returned qa-1 unit 900.00 / line 8550.00, qa-2 unit 200.00 / line 950.00:

QuoteEffective discountsubtotal / discount_total / totalstatus
q-approved5% (≤ 10%)10000.00 / 500.00 / 9500.00approved
q-needs15% (> 10%, ≤ 25%)14000.00 / 2100.00 / 11900.00needs_approval
q-rejected30% (> 25%)20000.00 / 6000.00 / 14000.00rejected

And repricing is idempotent: re-running the pricer left count(*) = 3 quotes, 6 lines, and SUM(total) = 35400.00 unchanged — the ON CONFLICT DO UPDATE re-wrote the same PKs in place, never a duplicate. Money survived every round-trip as a DECIMAL(18,2) (900.00, not 900).

Re-validated 2026-09-06 on the persistent crm bucket, over the suite's funnel quote q-grey-1 — 20 seats at 15% off the negotiated 1000.00, 10 Ignite apps at 0% off 850.00, and 5 Models bundles at 10% off the list-price fallback 600.00 (deliberately absent from pb-usd): subtotal 31500.00 − discount 3300.00 = total 28200.00, effective discount 10.48% — just over the 10% auto line → needs_approval. Every value crossed the pg wire as an exact DECIMAL(18,2).

Workflow op 2 — adjudicate the discount band (Ignite Models). POST /quotes/{quote_id}/adjudicate is the judgement call the threshold can't settle (22% is over the 10% auto line but under the 25% cap). It reads the quote's effective discount + the policy, then asks kimi-k2.6 — a standalone call, not run inside the pricer. kimi-k2.6 is a reasoning model, so max_tokens is high (else content is empty), the response is wrapped in data, and it retries once on an empty reply:

# routes.py — workflow op 2: the discount gate (kimi-k2.6).
# One of the suite's four role gates: the discount judgement is a deal-desk power,
# so it demands the quotes:approve pool permission (see ## Auth).
@router.post("/quotes/{quote_id}/adjudicate")
def adjudicate_discount(quote_id: str, a: AdjudicateIn,
                        user: dict = Depends(require_permission("quotes:approve")),
                        s: Session = Depends(db)):
    quote = s.get(Quote, quote_id)
    if not quote:
        raise HTTPException(404, "no such quote")
    approval, max_disc = _policy(s)
    subtotal = Decimal(str(quote.subtotal or 0))
    discount_total = Decimal(str(quote.discount_total or 0))
    eff = float(discount_total / subtotal) if subtotal else 0.0
    # ... build the system+user prompt from the policy + account tier + deal size, then:
    token = _models_token()                               # SA client_credentials -> bearer
    body = {"model": CHAT_MODEL, "max_tokens": 4096, "messages": [...]}
    for _ in range(2):                                    # retry once on an empty reasoning reply
        out = client.post("/chat/completions", json=body, timeout=90).json()
        env = out.get("data", out)                        # response wrapped in `data`
        content = env["choices"][0]["message"]["content"]
        m = re.search(r"\{.*\}", content or "", re.DOTALL)
        if m:
            return {"quote_id": quote_id, "effective_discount": eff, **json.loads(m.group(0))}
    raise HTTPException(502, "model returned empty content twice")

Live-verified: the 22% strategic case returned {"decision": "route_review", "reason": "22% exceeds auto-approve threshold; strategic tier and deal size warrant review."} — valid JSON, routed rather than auto-rejected, exactly the band the deterministic threshold hands off.

The 2026-09-06 re-validation added a lesson worth its own paragraph. Adjudicating the funnel quote's 10.48% discount, the pricer had stamped needs_approval (10.48 > 10) — but kimi-k2.6 returned {"decision": "auto_approve", "reason": "At 10% threshold; auto-approve per policy."}. The model rounded the number the policy would not — it was more lenient than the deterministic pricer. That's why the two are wired the way they are: the pricer's status is the floor, and the gate is advisory. The adjudicate route returns the model's verdict for a deal desk to weigh; it never overwrites quotes.status, and neither should any integration you build on it. A model gate that could relax a policy gate wouldn't be a gate. (And note the latency: a kimi-k2.6 call runs ~12s–170s — the route reads the quote, releases, calls, and returns; it never holds a pg connection across the model call.)

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

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 — exposing current_user and require_permission for the one job the app keeps: role-based gating.

After the audit, this component kept exactly one gate: POST /quotes/{quote_id}/adjudicate demands the quotes:approve permission — the discount judgement is a deal-desk power, and the route spends a model call. Everything else (catalog/book/quote CRUD, even the pricer — deterministic and idempotent, its status stamped by the policy row, not by whoever calls it) rides on the gateway's authentication alone. The suite's other surviving gates: orgs:qualify (lead-to-opportunity), leads:score (qualification-scoring), forecast:override (pipeline-forecast) — all checked against the pool's sales / analyst / manager role catalog (manager carries quotes:approve).

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=quotes:approve), 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-quote-cpq/v1.tar. This post is a walkthrough of exactly those files; the tarball is source only (no Dockerfile — the crm-cpq-pricer engine and its deploy live in Step 3 / Ship it):

models.py          # SQLAlchemy — products, price_books, price_book_entries, quotes, quote_lines, discount_policy
routes.py          # FastAPI    — an APIRouter the suite mounts + a standalone app; catalog/book/quote
                   #              CRUD + price (the pricer) + adjudicate (Models gate, quotes:approve-gated)
db.py              # lazy engine (openapi() builds with no creds) + the ON CONFLICT upsert helper
auth.py            # gateway header-trust: current_user + require_permission — no verifier
sa_token.py        # deployed: mints + refreshes the service-account token for the pg-wire password
.env.example       # BUCKET + PG_HOST/PG_PORT + DODIL_TOKEN (or the SA pair) + DEV_ALLOW_ANON
requirements.txt   # sqlalchemy, psycopg[binary], pgvector, fastapi, uvicorn, pydantic, httpx

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

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
 
cp .env.example .env      # then set DODIL_TOKEN (your `dodil auth login` token)
# BUCKET defaults to "crm"; create it once (Step 1), then create the tables from the models:
python -c "import db, models; models.Base.metadata.create_all(db.engine)"
 
uvicorn routes:app --reload
# POST /products · /price_books · /price_book_entries · /quotes · /quote_lines
# POST /quotes/{quote_id}/price      (the pricer)
# POST /quotes/{quote_id}/adjudicate (the kimi-k2.6 discount gate)

models.Base.metadata.create_all is the ORM tab's every class at once — the same six tables Steps 1–2 built by CLI, created from the natural-key models (money as Numeric(18,2) DECIMAL) 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 /quote-cpq) 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 3 — Price the quote: the crm-cpq-pricer engine

The pricer is the deterministic heart. Given a quote_id it: resolves each line's unit_price from the quote's price book (price_book_entries, fallback products.list_price), computes line_total = quantity × unit_price × (1 − discount_pct), rolls subtotal / discount_total / total onto the header, and stamps status from the effective discount vs the policy. The pricing logic is one SQL statement — you can run it inline to see exactly what the engine will write:

You

In $BUCKET, resolve each quote_line's unit_price from the quote's price book (fallback products.list_price) and compute line_total = quantity × unit_price × (1 − discount_pct) for quote q-approved.

DODIL MCP tools called
data_sql
Agent

qa-1 → unit 900, line_total 8550; qa-2 → unit 200, line_total 950. Sum 9500.

In production that logic runs in an Ignite app — a separate workload with its own service account (client-credentials → bearer token), so re-pricing on every line change never touches your login. It's an image-mode app: a small HTTP server (GET /healthz for the probe, POST /price for the work) built from the Dockerfile below and deployed with --dockerfile-path. It reaches DataK3 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. It does not call Models, so it needs only k3.editor (+ ignite.app-developer as the deploy identity). The policy thresholds are hoisted knobs (injected as env at deploy) that mirror the discount_policy row — one source of truth.

# engine/server.py — crm-cpq-pricer. IMAGE-mode Ignite app: an HTTP server on $PORT.
#   GET  /healthz -> 200 {"status":"ready"}      (probe path; no auth)
#   POST /price   -> body {"quote_id": "..."}     (reprices the quote; BUCKET comes from env)
# It mints its OWN SA token and writes over the DROP-IN POSTGRES WIRE (no K3 HTTP API).
# Only psycopg is third-party; everything else is stdlib. This pricer never calls Models —
# the kimi-k2.6 discount gate (Step 4) runs out-of-handler.
 
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
from psycopg import errors as pg_errors
 
# --- hoisted policy knobs: mirror the discount_policy row, injected as env at deploy ---
APPROVAL_THRESHOLD_PCT = float(os.environ.get("APPROVAL_THRESHOLD_PCT", "0.10"))  # at/under -> auto-approve
MAX_DISCOUNT_PCT       = float(os.environ.get("MAX_DISCOUNT_PCT", "0.25"))        # above    -> auto-reject
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"
# an explicit User-Agent is REQUIRED — stdlib urllib's default "Python-urllib/x" is
# banned by Cloudflare at id.dodil.io (HTTP 403 "error code: 1010").
UA = "crm-cpq-pricer/1.0"
 
LINE_COLS  = ["quote_line_id", "quote_id", "product_id", "quantity",
              "unit_price", "discount_pct", "line_total"]
QUOTE_COLS = ["quote_id", "opportunity_id", "account_domain", "price_book_id",
              "status", "currency", "subtotal", "discount_total", "total",
              "valid_until", "created_at"]
 
# Resolve unit_price from the quote's price book (fallback products.list_price); compute line_total.
PRICE_SQL = """
    SELECT ql.quote_line_id, ql.quote_id, ql.product_id, ql.quantity, ql.discount_pct,
           COALESCE(pbe.unit_price, p.list_price) AS unit_price,
           CAST(ql.quantity * COALESCE(pbe.unit_price, p.list_price) * (1 - ql.discount_pct) AS DOUBLE) AS line_total
    FROM quote_lines ql
    JOIN quotes q ON q.quote_id = ql.quote_id
    LEFT JOIN price_book_entries pbe ON pbe.price_book_id = q.price_book_id AND pbe.product_id = ql.product_id
    LEFT JOIN products p ON p.product_id = ql.product_id
    WHERE ql.quote_id = %s"""
 
 
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 _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 _upsert(cur, table, cols, row, pk=None):
    # Re-pricing re-writes the SAME PK (quote_line_id / quote_id) each run, so 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); the managed data_table_upsert is the equivalent.
    pk   = pk or cols[0]
    setc = ", ".join(f"{c} = EXCLUDED.{c}" for c in cols if c != pk)
    cur.execute(f"INSERT INTO {table} ({', '.join(cols)}) "
                f"VALUES ({', '.join(['%s'] * len(cols))}) "
                f"ON CONFLICT ({pk}) DO UPDATE SET {setc}",
                [row.get(c) for c in cols])
 
 
def price_quote(quote_id):
    token = _token()
 
    def _run():
        with _pg(token) as conn, conn.cursor() as cur:
            # 1. Resolve unit_price + line_total for every line (the pricing SQL).
            cur.execute(PRICE_SQL, (quote_id,))
            cols = [d.name for d in cur.description]
            lines = [dict(zip(cols, r)) for r in cur.fetchall()]
            if not lines:
                raise ValueError(f"no lines for quote {quote_id}")
 
            # 2. Full-row upsert each priced line back into quote_lines.
            for ln in lines:
                _upsert(cur, "quote_lines", LINE_COLS, ln)
 
            # 3. Roll the header from the priced lines.
            subtotal       = sum(ln["quantity"] * ln["unit_price"] for ln in lines)
            discount_total = sum(ln["quantity"] * ln["unit_price"] * ln["discount_pct"] for ln in lines)
            total          = sum(ln["line_total"] for ln in lines)
            eff            = (discount_total / subtotal) if subtotal else 0.0
 
            # 4. Policy status: <= threshold auto-approve; <= max needs_approval; else reject.
            status = ("approved"        if eff <= APPROVAL_THRESHOLD_PCT
                      else "needs_approval" if eff <= MAX_DISCOUNT_PCT
                      else "rejected")
 
            # 5. Full-row upsert the quote header (read the existing row, restamp the totals + status).
            cur.execute(f"SELECT {', '.join(QUOTE_COLS)} FROM quotes WHERE quote_id = %s", (quote_id,))
            head = dict(zip([d.name for d in cur.description], cur.fetchone()))
            head.update(subtotal=subtotal, discount_total=discount_total, total=total, status=status)
            _upsert(cur, "quotes", QUOTE_COLS, head)
            conn.commit()
            return {"quote_id": quote_id, "subtotal": subtotal,
                    "discount_total": discount_total, "total": total,
                    "effective_discount": eff, "status": status}
 
    return _retry(_run)
 
 
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 != "/price":
            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"{}")
            if not req.get("quote_id"):
                return self._send(400, {"error": "missing quote_id"})
            return self._send(200, price_quote(req["quote_id"]))
        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-cpq-pricer serving on 0.0.0.0:{port} bucket={BUCKET}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

Two files ride alongside it — the Dockerfile the platform builds on deploy (Lane B / Kaniko build-on-deploy) and the one third-party dep, the Postgres driver:

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

Give the engine its own least-privilege identity, then deploy it (image mode) and reprice the quote:

You

Create a service account for crm-cpq-pricer, grant it k3.editor + ignite.app-developer, deploy my ./engine app to Ignite (image mode, Dockerfile) as crm-cpq-pricer with the SA creds + policy knobs as runtime env, then POST quote_id q-approved to its /price endpoint to reprice it.

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

Created crm-cpq-pricer-sa, granted k3.editor + ignite.app-developer, deployed crm-cpq-pricer (image:build, scale-to-zero). POST /price q-approved: subtotal 10000, discount_total 500, total 9500, effective_discount 0.05, status approved.

Because every table is merge-keyed, re-pricing is idempotent — invoke the pricer twice on the same quote_id and you get the same totals and one row, never a duplicate. Retries and shard replays are safe; say so to anyone wiring this to a "reprice on line change" trigger.

NOTE

Deploy: image mode (Lane B), validated live 2026-09-02. crm-cpq-pricer 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-cpq-pricer 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 4 — The discount-approval gate (kimi-k2.6)

The deterministic threshold is the floor — literally: the gate's verdict is advisory and can never relax the pricer's status (the 2026-09-06 validation caught the model auto-approving a 10.48% discount the policy routes for review — see ## Routes). The gate is for the cases a number can't settle: 22% is over the 10% auto line but under the 25% cap — deterministically that's needs_approval, but should deal desk even look at it, or is it an obvious route-to-review because it's a strategic first logo? Feed the model the quote's discount, the policy, the account tier, and the deal size; it returns a structured verdict with a reason you can log on the quote. The system prompt renders approval_threshold_pct and max_discount_pct from the same params — one source.

NOTE

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

You

On kimi-k2.6, adjudicate a discount: account acme.com (tier strategic), opportunity $36,000, first order, total discount requested 22% (subtotal 20000, discount 4400). Policy: auto-approve at/under 10%, route review up to 25%, reject above 25%. Return ONLY JSON: {decision (auto_approve|route_review|reject), reason (<=20 words)}.

DODIL MCP tools called
ignite_models_chat
Agent

{"decision": "route_review", "reason": "Strategic tier; 22% discount falls in mandatory review range."}

How the pillars map

One bucket, one bill, one auth context — CPQ is mostly the SQL pillar, with Ignite for the stateful reprice and Models for judgement. No second pricing database, no nightly catalog sync.

ConcernWhere it livesReplaces
Catalog, price books, quotes, lines, policyDataK3 SQL tables (merge-keyed, one copy)Postgres + a CPQ product's own store
Resolve price / roll totals / stamp statusIgnite crm-cpq-pricer (its own SA, k3.editor)a cron job + app server
Discount edge-case judgementModels kimi-k2.6 gate → {decision, reason}a deal-desk email thread
Drop-in access for a billing/ERP clientdata connect $BUCKET → pg / bolt / grpc endpointsbespoke API integration

The policy is data: discount_policy is one row, so "what self-approves" is a value you edit, not a deploy. And because products is owns_master here, crm/account-360 can read this same catalog for its whitespace grid — set link_whitespace: true to also project products as product nodes into crm/core's crm_node/crm_edge for the graph skill (off by default; keeps CPQ self-contained).

Customize — the decisions this skill asks you

Q1 · currency / multi_currency — one book or many?

"Do you sell in one currency, or several regions/currencies?"

  • multi_currency: false (default) → one price_books row in currency (default USD); every quote prices against it.
  • multi_currency: true → one price_books row per currency/region (e.g. pb-usd, pb-eur), each with its own price_book_entries. A quote's price_book_id selects the book, so the pricer resolves that quote's lines in its currency with no code change.

Q2 · approval_threshold_pct / max_discount_pct — the discount policy

"Below what effective discount should a quote self-approve, and above what should it be auto-rejected?" → Written to the discount_policy row and the crm-cpq-pricer constants (APPROVAL_THRESHOLD_PCT, MAX_DISCOUNT_PCT) and the gate's system prompt — one source, three projections. Defaults 0.10 / 0.25: at/under 10% → approved, over 10% up to 25% → needs_approval, over 25% → rejected. This is the money knob — it decides what fraction of quotes never wait on a human. Raise the threshold to self-approve more (faster, looser); lower it to route more to deal desk (slower, tighter margin control).

Q3 · link_whitespace — feed the account graph?

"Should products show up in the account-360 whitespace graph?"

  • false (default) → CPQ is self-contained; no graph writes.
  • true → also project products as product nodes (+ owns_product edges) into crm/core's crm_node/crm_edge, so crm/account-360's "who owns what across the family / where's the gap" query works. Only turn this on in the suite (it needs crm/core's node/edge tables).

Industry variants (saas / manufacturing / finserv / real-estate) compose this skill with small additive diffs — e.g. manufacturing adds a quote-to-order leg. See the per-industry CRM pages.

Test

Run this against live DataK3 after building (the default, single-currency branch). Every number below was produced live on a validation bucket — the pricer bands, the idempotent re-price (3 quotes / 6 lines / SUM(total) 35400.00 unchanged), and the kimi-k2.6 gate — and the package was re-validated 2026-09-06 on the persistent crm bucket over the suite's funnel quote: q-grey-1 priced 31500.00 − 3300.00 = 28200.00 at an effective 10.48% → needs_approval, exact DECIMAL(18,2) over the pg wire, and the adjudicate gate's more-lenient auto_approve verdict stayed advisory (the pricer's status is the floor).

# 1. Six CPQ tables exist.
dodil data table list -b "$BUCKET" -o json | python3 -c "import sys,json; t={r['table'] for r in json.load(sys.stdin)['rows']}; assert {'products','price_books','price_book_entries','quotes','quote_lines','discount_policy'} <= t; print('ok: 6 CPQ tables')"
 
# 2. A 5% quote → approved, total = 9500 (10×900×0.95 + 5×200×0.95 = 8550 + 950).
dodil data sql -b "$BUCKET" "SELECT status, total FROM quotes WHERE quote_id='q-approved'"
#   -> status approved, total 9500
 
# 3. A 15% quote (> threshold, <= max) → needs_approval, total 11900.
#    (lines: 10× prod-seat @900, 1× prod-support @5000, both at 0.15)
dodil data sql -b "$BUCKET" "SELECT status, total FROM quotes WHERE quote_id='q-needs'"
#   -> status needs_approval, total 11900   (subtotal 14000, discount_total 2100)
 
# 4. A 30% quote (> max) → rejected, total 14000.
#    (lines: 20× prod-seat @900, 10× prod-storage @200, both at 0.30)
dodil data sql -b "$BUCKET" "SELECT status, total FROM quotes WHERE quote_id='q-rejected'"
#   -> status rejected, total 14000   (subtotal 20000, discount_total 6000)
 
# 5. The discount gate returns valid JSON on the 22% strategic case (Step 4).
#   -> {"decision":"route_review","reason":"Strategic tier; 22% discount falls in mandatory review range."}
 
# 6. Idempotent re-price: reprice all quotes again → same totals, same row count (no duplicates).
dodil data sql -b "$BUCKET" "SELECT count(*) AS rows, SUM(total) AS sum_total FROM quotes"
#   -> rows 3, sum_total 35400  (unchanged across re-runs)  [4 incl. the EUR quote if multi_currency]

Multi-currency branch (multi_currency: true): add pb-eur (EUR, entries seat 850 / storage 190) and a quote q-eur on it at 5% (10× seat, 5× storage). The pricer resolves EUR book prices (850/190, not the USD 900/200) → total 8977.5, status approved — proving a quote prices in its own book's currency.

One-shot

Scaffold crm/quote-cpq on DataK3 bucket "crm" (currency USD, approval_threshold_pct 0.10, max_discount_pct
0.25). Create merge-keyed tables products(product_id), price_books(price_book_id),
price_book_entries(entry_id), quotes(quote_id), quote_lines(quote_line_id), discount_policy(policy_id).
Seed 3 products (seat/storage/support), a USD price book with entries, and one discount_policy row
(auto-approve at/under 0.10, max 0.25). Draft three quotes at 5% / 15% / 30% effective discount, then
deploy the crm-cpq-pricer Ignite app (image mode, Dockerfile — an HTTP server exposing POST /price that,
over the drop-in Postgres wire, resolves unit_price from price_book_entries with a products.list_price
fallback, computes line_total = quantity × unit_price × (1 − discount_pct), rolls
subtotal/discount_total/total, and stamps status: approved / needs_approval / rejected) under a service
account with k3.editor + ignite.app-developer. POST each quote_id to /price and
assert 5%→approved(9500), 15%→needs_approval(11900), 30%→rejected(14000). Finally run one kimi-k2.6
discount-approval gate on a 22% strategic case and confirm it returns {decision, reason} JSON.

Ship it — crm-cpq-pricer as a repricing endpoint

The image-mode pricer already is that endpoint: POST /price with a quote_id returns the priced verdict, so an ERP/billing service calls it straight over HTTP. It runs request-invoked (scale-to-zero) — reprice on demand, or wire it to a "reprice on line change" trigger. Redeploy it and read back its public FQDN + health:

You

Deploy the crm-cpq-pricer app (image mode, Dockerfile) and give me its public URL and invoke contract.

DODIL MCP tools called
ignite_app_deploy→ignite_app_get
Agent

Deployed crm-cpq-pricer (image:build); reach it at https://crm-cpq-pricer-$DODIL_ORG-8080.ignite.dodil.cloud — POST /price {quote_id} for the priced verdict, GET /healthz for the probe.

Full lifecycle (DODIL git → CI checks → a scanned image in the registry → versioning and rollback): 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) — the catalog, price books, quotes, and lines are all standard Postgres tables.

Full, live-validated walkthrough: Connect your tools.

Conclusion

You have a one-bucket CPQ: a product catalog, multi-currency price books, quotes whose lines reprice idempotently from the book, a discount policy that lives as an editable row, and a kimi-k2.6 gate for the judgement calls. No second pricing database, no nightly sync, no deal-desk spreadsheet — the verdict is a column, and "what self-approves" is a value you edit. Wire it up next to the CRM core (opportunities/accounts it quotes against) and account-360 (flip link_whitespace: true and your catalog feeds the whitespace graph).