Kommo + Hex: Sales Analytics in Collaborative Data Notebooks Without a BI License

Hex is a modern data workspace: SQL + Python notebooks with the ability to publish as interactive dashboards. Data teams use it as an alternative to Jupyter + Tableau - one place to explore data, build models, and share results with the business. For teams using Kommo, Hex lets you export deal data, build cohort analysis, visualize the conversion funnel, and publish it as a live dashboard for leadership - without BI tools and without rewriting SQL queries.

Architecture: a Python ETL exports data from the Kommo API into PostgreSQL/Snowflake, Hex connects to the database and runs SQL + Python cells. Optionally, you can trigger a Hex Project via API on a schedule.

Hex API uses Bearer token authentication. Key operations: POST /api/v1/projects/{hex_project_id}/runs - trigger a project refresh (re-run all SQL cells with fresh data). GET /api/v1/projects/{id}/runs/{run_id} - check run status.

Hex Project - a notebook with SQL and Python cells + a Published App (dashboard) for non-technical audiences. A project can be run on a schedule (Hex Schedules) or triggered via API.

ETL: Kommo -> PostgreSQL for Hex

import requests, os, psycopg2
from datetime import datetime, timezone

KOMMO_SUBDOMAIN = os.environ["KOMMO_SUBDOMAIN"]
KOMMO_TOKEN     = os.environ["KOMMO_ACCESS_TOKEN"]
PG_DSN          = os.environ["PG_DSN"]

KOMMO_BASE = f"https://{KOMMO_SUBDOMAIN}.kommo.com/api/v4"
KOMMO_HDR  = {"Authorization": f"Bearer {KOMMO_TOKEN}"}

def fetch_leads_paginated(params: dict) -> list[dict]:
    all_leads = []
    page = 1
    while True:
        params["page"] = page
        r = requests.get(f"{KOMMO_BASE}/leads", headers=KOMMO_HDR, params=params)
        if r.status_code == 204:
            break
        items = r.json().get("_embedded", {}).get("leads", []) or []
        if not items:
            break
        all_leads.extend(items)
        page += 1
    return all_leads

def upsert_leads(leads: list[dict], conn):
    cur = conn.cursor()
    cur.execute(
        "CREATE TABLE IF NOT EXISTS kommo_leads ("
        "  id BIGINT PRIMARY KEY, name TEXT, status_id INT, pipeline_id INT,"
        "  responsible_id INT, price NUMERIC,"
        "  created_at TIMESTAMPTZ, updated_at TIMESTAMPTZ, closed_at TIMESTAMPTZ,"
        "  loss_reason_id INT"
        ")"
    )
    for lead in leads:
        ts = lambda k: (
            datetime.fromtimestamp(lead[k], tz=timezone.utc) if lead.get(k) else None
        )
        cur.execute(
            "INSERT INTO kommo_leads"
            " (id,name,status_id,pipeline_id,responsible_id,price,created_at,updated_at,closed_at,loss_reason_id)"
            " VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
            " ON CONFLICT (id) DO UPDATE SET"
            " name=EXCLUDED.name, status_id=EXCLUDED.status_id, price=EXCLUDED.price,"
            " updated_at=EXCLUDED.updated_at, closed_at=EXCLUDED.closed_at",
            (lead["id"], lead.get("name"), lead.get("status_id"), lead.get("pipeline_id"),
             lead.get("responsible_user_id"), lead.get("price"),
             ts("created_at"), ts("updated_at"), ts("closed_at"), lead.get("loss_reason_id")),
        )
    conn.commit()
    cur.close()

def run_etl():
    conn   = psycopg2.connect(PG_DSN)
    params = {"limit": 250, "with": "custom_fields_values"}
    leads  = fetch_leads_paginated(params)
    upsert_leads(leads, conn)
    conn.close()
    print(f"ETL: {len(leads)} leads upserted")

if __name__ == "__main__":
    run_etl()

Triggering a Hex Project via API

HEX_TOKEN      = os.environ["HEX_API_TOKEN"]
HEX_PROJECT_ID = os.environ["HEX_PROJECT_ID"]
HEX_BASE       = "https://app.hex.tech/api/v1"
HEX_HDR        = {"Authorization": f"Bearer {HEX_TOKEN}", "Content-Type": "application/json"}

def trigger_hex_run(input_params: dict | None = None) -> str:
    payload = {}
    if input_params:
        payload["inputParams"] = input_params
    r = requests.post(
        f"{HEX_BASE}/projects/{HEX_PROJECT_ID}/runs",
        headers=HEX_HDR,
        json=payload,
    )
    r.raise_for_status()
    return r.json().get("runId", "")

def wait_hex_run(run_id: str, timeout_sec: int = 300) -> str:
    import time
    deadline = time.time() + timeout_sec
    while time.time() < deadline:
        r = requests.get(
            f"{HEX_BASE}/projects/{HEX_PROJECT_ID}/runs/{run_id}",
            headers=HEX_HDR,
        )
        status = r.json().get("status", "")
        if status in ("COMPLETED", "ERRORED", "KILLED"):
            return status
        time.sleep(10)
    return "TIMEOUT"

# Usage: run via cron after ETL completes
run_id = trigger_hex_run()
status = wait_hex_run(run_id)
print(f"Hex run {run_id}: {status}")

SQL Cells in Hex for Kommo Analysis

-- Monthly conversion funnel
SELECT
    DATE_TRUNC('month', created_at) AS month,
    COUNT(*)                         AS total_leads,
    COUNT(*) FILTER (WHERE status_id = {{ won_stage_id }})  AS won,
    COUNT(*) FILTER (WHERE status_id = {{ lost_stage_id }}) AS lost,
    ROUND(
        100.0 * COUNT(*) FILTER (WHERE status_id = {{ won_stage_id }}) / COUNT(*),
        1
    ) AS win_rate_pct
FROM kommo_leads
WHERE created_at >= NOW() - INTERVAL '12 months'
GROUP BY 1
ORDER BY 1;

-- Average deal cycle (days to close) by sales rep
SELECT
    responsible_id,
    ROUND(AVG(EXTRACT(DAY FROM (closed_at - created_at))), 1) AS avg_days_to_close,
    COUNT(*) FILTER (WHERE status_id = {{ won_stage_id }})    AS won_count,
    SUM(price) FILTER (WHERE status_id = {{ won_stage_id }})  AS total_revenue
FROM kommo_leads
WHERE closed_at IS NOT NULL
  AND status_id = {{ won_stage_id }}
GROUP BY 1
ORDER BY 3 DESC;

In Hex, {{ variable }} is an Input Cell (parameter). Create Input Cells for won_stage_id and lost_stage_id - dashboard users can change them without touching any code.

Hex Scheduled Runs

Hex App -> [Project] -> Schedules -> Add Schedule. For example: every day at 7:00 UTC - reruns SQL cells with fresh data. The Published App updates automatically.

For the ETL + Hex Run chain: set up a cron job (0 6 * * *) that runs the ETL script and then triggers Hex via API. Hex starts after ETL finishes.

Who This Is For

Data teams at B2B SaaS companies using Kommo: an analyst or data scientist who needs to regularly prepare a funnel report for the sales director. Hex lets you write SQL + Python once and publish it as a live dashboard without configuring a separate BI tool.

A similar approach with a dbt layer: Kommo + Lightdash.

Frequently Asked Questions

Does Hex require a paid plan for API runs?

Yes. API access for triggering project runs is available on the Hex Team plan ($24/user/mo) and above. On the free plan, runs are manual only. Scheduled Runs are available on paid plans only.

How does Hex connect to PostgreSQL?

Hex Connections -> Add Connection -> PostgreSQL -> enter host, port, database, user, password. Hex supports SSH tunnels and SSL. Once connected, all SQL cells in the project can use that connection. Data is not cached in Hex - each run makes a fresh query to the database.

Can data be exported from Hex to CSV automatically?

Hex allows exporting cell results to CSV via the UI. For automated export via API: after a project run, use GET /api/v1/projects/{id}/runs/{run_id}/results/{cell_id} to retrieve data from a specific cell as JSON. CSV export via API is in development.

Summary

Kommo + Hex - sales analytics in data notebooks:

• Python ETL: fetch_leads_paginated -> upsert into PostgreSQL with ON CONFLICT DO UPDATE
• Hex API: POST /api/v1/projects/{id}/runs -> trigger after ETL
• SQL cells: conversion funnel, days_to_close, revenue by sales rep
• Input Cells {{ won_stage_id }} - parameters for non-technical audiences
• Scheduled Runs: automatic dashboard refresh without manual intervention

If you need help setting up analytics on Kommo data in Hex or another data warehouse - describe your requirements to the Exceltic.dev team.