fleetanalytics_mcp/docs/ANALYTICS_MCP.md

Read-only Analytics MCP Server — Implementation Guide

Audience: engineer deploying/maintaining the server. Status: built — pending deploy. Repo: fleetanalytics_mcp (standalone; repo.rahamafresh.com/kianiadee/fleetanalytics_mcp.git). Hosted on the same Coolify host as tracksolid_db. Date: 2026-06-16.

1. Purpose & context

The decision & analytics team needs to pull fleet reporting data (fuel, utilisation, driver behaviour, INC tickets, raw telemetry) from tracksolid_db to make decisions — read-only, never edit/delete. The only programmatic surface today is the dashboard_api FastAPI bridge with a fixed set of /analytics/* / /webhook/* endpoints — too rigid for ad-hoc analysis.

This adds a hosted, read-only MCP server that lets analysts query the database directly from Claude: a guarded general SELECT tool plus schema-introspection tools, pointed at the existing PostgreSQL 16 + TimescaleDB + PostGIS database through a new least-privilege analytics_ro role.

The DB is internal-only (DATABASE_URL → timescale_db:5432 on the Docker network, not reachable from a laptop), so the server is hosted on the same Coolify host as the DB. It ships as its own repo with its own Dockerfile (Coolify-buildable), and joins the network that can reach timescale_db. A deploy.sh is included as a manual host-deploy fallback that mirrors the proven dashboard_api bridge pattern.

Read-only is enforced at four layers: the analytics_ro GRANTs (no INSERT/UPDATE/DELETE), a session default_transaction_read_only = on, a transaction that is rolled back (never committed), and a single-statement / keyword SQL guard in the query tool.

Where this sits

analyst's Claude  ──HTTPS (Bearer)──►  fleetmcp.fivetitude.com (Traefik)
                                              │
                                       analytics_mcp container  (uvicorn :8892, coolify net)
                                              │  psycopg2, role = analytics_ro, READ ONLY
                                              ▼
                                       timescale_db:5432  (tracksolid_db)
                                          reporting.*  ·  tracksolid.*

Ports in use: 8890 prod dashboard_api · 8891 staging dashboard_api · 8892 analytics_mcp.

---

2. Repo contents

File	What
analytics_mcp.py	the MCP server (FastMCP streamable-HTTP; uvicorn target analytics_mcp:app)
Dockerfile	Coolify-buildable image (port 8892)
pyproject.toml	dependencies (mcp[cli], psycopg2-binary, uvicorn)
deploy.sh	manual host deploy (standalone Traefik bridge) — fallback to Coolify
scripts/analytics_ro_role.sql	read-only role DDL (modelled on the backend's dashboard_ro_role.sql + hardening)
scripts/bootstrap_analytics_ro.sh	host bootstrap: generate pw → apply role SQL
docs/ANALYTICS_MCP.md / .html	this guide

The backend repo (tracksolid_timescale_grafana_prod) keeps only a pointer note in its CLAUDE.md recording that analytics_ro exists and is owned by this repo.

---

3. Step 1 — the analytics_ro role

scripts/analytics_ro_role.sql

Modelled on scripts/dashboard_ro_role.sql. Run as the postgres superuser (it does CREATE ROLE), supplied a password as psql var :'ro_pw' — no secret in the repo.

-- analytics_ro_role.sql — dedicated read-only LOGIN role for the analytics MCP server.
-- Run as postgres SUPERUSER via scripts/bootstrap_analytics_ro.sh (NOT run_migrations.py).
-- Grants exactly the read surface: SELECT on reporting.* + tracksolid.*, the v_trips
-- matview, and EXECUTE on reporting.fn_*. No INSERT/UPDATE/DELETE, not the matview owner,
-- so analytics_ro can never write or REFRESH. Idempotent — safe to re-apply (rotates pw).
\set ON_ERROR_STOP on

DO $role$
BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'analytics_ro') THEN
    CREATE ROLE analytics_ro LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE;
  END IF;
END $role$;

ALTER ROLE analytics_ro WITH LOGIN PASSWORD :'ro_pw';

GRANT CONNECT ON DATABASE tracksolid_db TO analytics_ro;
GRANT USAGE   ON SCHEMA reporting, tracksolid TO analytics_ro;

GRANT SELECT  ON ALL TABLES    IN SCHEMA reporting  TO analytics_ro;  -- tables + views
GRANT SELECT  ON ALL TABLES    IN SCHEMA tracksolid TO analytics_ro;  -- tables + views
GRANT SELECT  ON reporting.v_trips TO analytics_ro;                   -- MATERIALIZED VIEW
GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA reporting  TO analytics_ro;

-- Future objects created by the migration role auto-grant (matviews still need explicit GRANT).
ALTER DEFAULT PRIVILEGES FOR ROLE tracksolid_owner IN SCHEMA reporting  GRANT SELECT  ON TABLES    TO analytics_ro;
ALTER DEFAULT PRIVILEGES FOR ROLE tracksolid_owner IN SCHEMA tracksolid GRANT SELECT  ON TABLES    TO analytics_ro;
ALTER DEFAULT PRIVILEGES FOR ROLE tracksolid_owner IN SCHEMA reporting  GRANT EXECUTE ON FUNCTIONS TO analytics_ro;

-- Extra hardening over dashboard_ro: this role serves ad-hoc HUMAN queries.
ALTER ROLE analytics_ro SET default_transaction_read_only = on;
ALTER ROLE analytics_ro SET statement_timeout = '30s';
ALTER ROLE analytics_ro SET idle_in_transaction_session_timeout = '60s';

scripts/bootstrap_analytics_ro.sh

Clone of scripts/bootstrap_dashboard_ro.sh — generates ~/.analytics_ro.pw (0600) on first run, applies the SQL via docker exec ... psql.

#!/usr/bin/env bash
# bootstrap_analytics_ro.sh — create/refresh the analytics_ro read-only role.
# Run ON THE HOST. Generates a strong pw into ~/.analytics_ro.pw (0600) on first
# run, then applies scripts/analytics_ro_role.sql as the postgres superuser.
# Deploy:
#   scp scripts/analytics_ro_role.sql scripts/bootstrap_analytics_ro.sh kianiadee@twala.rahamafresh.com:~/
#   ssh kianiadee@twala.rahamafresh.com 'bash ~/bootstrap_analytics_ro.sh'
set -euo pipefail

PW_FILE="${ANALYTICS_RO_PW_FILE:-$HOME/.analytics_ro.pw}"
SQL_FILE="${1:-$HOME/analytics_ro_role.sql}"
test -f "$SQL_FILE" || { echo "ERROR: role SQL not found at $SQL_FILE"; exit 1; }

if [ ! -s "$PW_FILE" ]; then
  ( umask 077; openssl rand -hex 24 > "$PW_FILE" ); chmod 600 "$PW_FILE"
  echo "Generated new analytics_ro password -> $PW_FILE (0600)"
else
  echo "Reusing existing analytics_ro password from $PW_FILE"
fi
PW=$(cat "$PW_FILE")

DB=$(docker ps --filter name=timescale_db --format "{{.Names}}" | head -1)
[ -n "$DB" ] || { echo "ERROR: timescale_db container not found"; exit 1; }

echo "Applying analytics_ro role DDL to $DB as postgres ..."
docker exec -i "$DB" psql -U postgres -d tracksolid_db -v ON_ERROR_STOP=1 -v ro_pw="$PW" < "$SQL_FILE"
echo "analytics_ro ready (password not printed). Now (re)run deploy_analytics_mcp.sh."

---

4. The MCP server (analytics_mcp.py)

FastMCP streamable-HTTP server, served by uvicorn (target analytics_mcp:app). It uses its own read-only psycopg2 pool and a small local logger — it deliberately does not import the backend's ts_shared_rev (that module eagerly requires the Tracksolid ingestion secrets, which this read-only server has no business holding). The canonical source is ../analytics_mcp.py; the abridged version below shows the guard, the tools, and the auth:

"""
analytics_mcp.py — Fireside Communications · Read-only Analytics MCP Server
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Hosted MCP server for the decision & analytics team. Exposes the fleet reporting
data (reporting.* + tracksolid.*) to Claude as READ-ONLY query + introspection
tools. Connects as the analytics_ro role; every query runs in a read-only txn
that is rolled back. Served over streamable HTTP behind Traefik with Bearer auth.
"""
from __future__ import annotations

import os
import re
import time
from contextlib import contextmanager

import psycopg2
import psycopg2.extras
import psycopg2.pool
from mcp.server.fastmcp import FastMCP
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

# NOTE: a small local logger is defined here (see analytics_mcp.py); we do NOT import
# ts_shared_rev, so this read-only server carries none of the ingestion secrets.
log = _get_logger("server")

DATABASE_URL = os.environ["DATABASE_URL"]          # set to the analytics_ro DSN by deploy
MAX_ROWS_CEIL = int(os.getenv("MCP_MAX_ROWS", "10000"))
READABLE_SCHEMAS = ("reporting", "tracksolid")

# ── read-only connection pool ────────────────────────────────────────────────
# Force read-only + a statement timeout at the connection level (belt + braces;
# the analytics_ro role already sets these, but a self-contained server is safer).
_pool = psycopg2.pool.ThreadedConnectionPool(
    1, int(os.getenv("MCP_POOL_MAX", "8")), DATABASE_URL,
    options="-c default_transaction_read_only=on -c statement_timeout=30000 -c client_encoding=UTF8",
)

@contextmanager
def _ro_conn():
    """Read-only connection; the transaction is ALWAYS rolled back (never commits)."""
    conn = _pool.getconn()
    try:
        conn.set_session(readonly=True, autocommit=False)
        yield conn
    finally:
        conn.rollback()
        _pool.putconn(conn)

def _rows(cur):
    cols = [d[0] for d in cur.description]
    return [dict(zip(cols, r)) for r in cur.fetchall()]

# ── SQL guard for the general query tool ─────────────────────────────────────
_FORBIDDEN = re.compile(
    r"\b(insert|update|delete|drop|alter|create|grant|revoke|truncate|copy|call|do|merge|"
    r"vacuum|reindex|refresh|comment|lock|set|reset)\b", re.IGNORECASE)

def _strip_comments(sql: str) -> str:
    sql = re.sub(r"/\*.*?\*/", " ", sql, flags=re.DOTALL)
    sql = re.sub(r"--[^\n]*", " ", sql)
    return sql.strip()

def _guard(sql: str) -> str:
    stripped = _strip_comments(sql)
    if not stripped:
        raise ValueError("Empty query.")
    # exactly one statement (allow a single trailing ;)
    parts = [p for p in stripped.split(";") if p.strip()]
    if len(parts) != 1:
        raise ValueError("Only a single statement is allowed.")
    stmt = parts[0].strip()
    if not re.match(r"^(select|with)\b", stmt, re.IGNORECASE):
        raise ValueError("Only SELECT / WITH queries are allowed.")
    if _FORBIDDEN.search(stmt):
        raise ValueError("Query contains a forbidden (write/DDL) keyword.")
    return stmt

mcp = FastMCP("fireside-analytics", stateless_http=True)

@mcp.tool()
def query(sql: str, max_rows: int = 1000) -> dict:
    """Run a read-only SELECT/WITH query against the fleet database.
    Only the reporting.* and tracksolid.* schemas are readable. Returns up to
    `max_rows` rows (default 1000, hard cap 10000). Auto-applies LIMIT if absent."""
    stmt = _guard(sql)
    cap = max(1, min(int(max_rows), MAX_ROWS_CEIL))
    if not re.search(r"\blimit\b", stmt, re.IGNORECASE):
        stmt = f"{stmt}\nLIMIT {cap + 1}"     # +1 to detect truncation
    t0 = time.monotonic()
    with _ro_conn() as conn, conn.cursor() as cur:
        cur.execute(stmt)
        rows = _rows(cur)
    truncated = len(rows) > cap
    rows = rows[:cap]
    dur_ms = int((time.monotonic() - t0) * 1000)
    log.info("query rows=%d trunc=%s %dms :: %s", len(rows), truncated, dur_ms, sql[:200])
    return {"row_count": len(rows), "truncated": truncated, "rows": rows}

@mcp.tool()
def list_schemas() -> list[dict]:
    """List the readable schemas and their table/view counts."""
    with _ro_conn() as conn, conn.cursor() as cur:
        cur.execute(
            "SELECT table_schema AS schema, count(*) AS objects "
            "FROM information_schema.tables WHERE table_schema = ANY(%s) "
            "GROUP BY 1 ORDER BY 1", (list(READABLE_SCHEMAS),))
        return _rows(cur)

@mcp.tool()
def list_tables(schema: str) -> list[dict]:
    """List tables + views in a schema (reporting or tracksolid)."""
    if schema not in READABLE_SCHEMAS:
        raise ValueError(f"schema must be one of {READABLE_SCHEMAS}")
    with _ro_conn() as conn, conn.cursor() as cur:
        cur.execute(
            "SELECT table_name AS name, table_type AS kind "
            "FROM information_schema.tables WHERE table_schema = %s "
            "ORDER BY 1", (schema,))
        return _rows(cur)

@mcp.tool()
def describe_table(schema: str, table: str) -> list[dict]:
    """Describe a table/view: columns, types, nullability, defaults."""
    if schema not in READABLE_SCHEMAS:
        raise ValueError(f"schema must be one of {READABLE_SCHEMAS}")
    with _ro_conn() as conn, conn.cursor() as cur:
        cur.execute(
            "SELECT column_name AS column, data_type AS type, is_nullable AS nullable, "
            "column_default AS default FROM information_schema.columns "
            "WHERE table_schema = %s AND table_name = %s ORDER BY ordinal_position",
            (schema, table))
        return _rows(cur)

@mcp.tool()
def list_functions(schema: str = "reporting") -> list[dict]:
    """List callable functions (e.g. reporting.fn_*) with their signatures."""
    if schema not in READABLE_SCHEMAS:
        raise ValueError(f"schema must be one of {READABLE_SCHEMAS}")
    with _ro_conn() as conn, conn.cursor() as cur:
        cur.execute(
            "SELECT p.proname AS name, pg_get_function_arguments(p.oid) AS args "
            "FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace "
            "WHERE n.nspname = %s ORDER BY 1", (schema,))
        return _rows(cur)

@mcp.tool()
def sample_table(schema: str, table: str, n: int = 20) -> dict:
    """Return the first `n` rows of a table/view (convenience over query)."""
    if schema not in READABLE_SCHEMAS:
        raise ValueError(f"schema must be one of {READABLE_SCHEMAS}")
    # quote_ident via format() on validated identifiers
    return query(f'SELECT * FROM "{schema}"."{table}"', max_rows=n)

# ── Bearer-token auth middleware ──────────────────────────────────────────────
# MCP_AUTH_TOKENS = "alice:tok1,bob:tok2"  (per-analyst → revocable + attributable)
_TOKENS = {
    t.split(":", 1)[1]: t.split(":", 1)[0]
    for t in os.getenv("MCP_AUTH_TOKENS", "").split(",")
    if ":" in t
}

class BearerAuth(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path == "/healthz":
            return await call_next(request)
        auth = request.headers.get("authorization", "")
        token = auth[7:] if auth.lower().startswith("bearer ") else ""
        if token not in _TOKENS:
            return JSONResponse({"error": "unauthorized"}, status_code=401)
        request.state.caller = _TOKENS[token]
        return await call_next(request)

async def healthz(_request):
    return JSONResponse({"ok": True, "tokens": len(_TOKENS)})

app = mcp.streamable_http_app()
app.add_middleware(BearerAuth)
app.add_route("/healthz", healthz, methods=["GET"])   # Starlette: add_route, not @app.route

Notes. stateless_http=True suits a behind-proxy / multi-worker deploy. The token map is token → name so each query is attributable in the logs. The _FORBIDDEN guard also blocks SET/RESET so a query can't relax default_transaction_read_only. Validate the exact FastMCP app-factory method name against the installed mcp version (streamable_http_app() vs http_app()); the deploy command's app target must match.

---

5. Packaging — Dockerfile + pyproject.toml

This repo is self-contained: its pyproject.toml declares the deps (mcp[cli], psycopg2-binary, uvicorn[standard]) and the Dockerfile builds a slim image that runs uvicorn analytics_mcp:app on port 8892. The project is a flat single module, so [tool.uv] package = false and the Dockerfile installs dependencies only (uv sync --no-dev --no-install-project) — it never tries to build the module as a package. No dependency on the backend image.

---

6. Deploy

The DB is internal-only, so the server runs on the same Coolify host as timescale_db.

Recommended — Coolify-managed app. Create a Coolify application from this repo (repo.rahamafresh.com/kianiadee/fleetanalytics_mcp.git), Dockerfile build, app port 8892, domain fleetmcp.rahamafresh.com (prod) / fleetmcp.fivetitude.com (staging). Set as secrets DATABASE_URL=postgresql://analytics_ro:<pw>@timescale_db:5432/tracksolid_db and MCP_AUTH_TOKENS=alice:<tok>,bob:<tok>, then connect the app to the network that can reach timescale_db (the tracksolid stack's network) so the hostname resolves. Coolify manages the Traefik labels + TLS from the domain. Auto-deploys on push via the Forgejo webhook.

Fallback — manual host deploy (deploy.sh). If not using the Coolify UI, check the repo out on the host and run deploy.sh — it builds the image, resolves the DB network + DSN from the running stack, swaps in the analytics_ro credentials, and runs a standalone Traefik-labelled bridge (the proven dashboard_api pattern). See the script header.

cd ~/fleetanalytics_mcp && git pull
MCP_AUTH_TOKENS="alice:$(openssl rand -hex 16)" bash deploy.sh

---

7. Deploy runbook (ordered)

1. Role (once): scp scripts/analytics_ro_role.sql scripts/bootstrap_analytics_ro.sh kianiadee@twala.rahamafresh.com:~/ then ssh ... 'bash ~/bootstrap_analytics_ro.sh' (writes ~/.analytics_ro.pw).
2. App: either point Coolify at this repo (§6 recommended) or run deploy.sh on the host. Record each analyst's token securely (it is shown once when generated).
3. Network: ensure the MCP container shares a Docker network with timescale_db so the DSN host resolves (Coolify network setting, or deploy.sh reuses the stack network automatically).
4. DNS/Traefik: ensure fleetmcp.* resolves to the host; Coolify/Traefik issues the cert.

---

8. Add to Claude (for analysts)

Claude Code

claude mcp add --transport http fireside-analytics https://fleetmcp.fivetitude.com \
  --header "Authorization: Bearer <your-token>"
claude mcp list      # should show "fireside-analytics: connected"

Claude Desktop / claude.ai — add a custom connector with the same URL and an Authorization: Bearer <your-token> header.

Example session prompts: "list the schemas", "describe reporting.v_daily_summary", "top 10 cost centres by distance in the last 30 days" (the model writes the SELECT and calls query).

---

9. Verification checklist

• psql -U analytics_ro -d tracksolid_db -c "SELECT count(*) FROM reporting.v_daily_summary" succeeds.
• psql -U analytics_ro ... -c "CREATE TABLE x(i int)" fails (permission denied) — proves read-only.
• the image builds (docker build . or Coolify build succeeds); analytics_mcp container is Up.
• DATABASE_URL shows analytics_ro (pw masked); curl localhost:8892/healthz returns {"ok":true,...}.
• the container can resolve/reach timescale_db (shares its network).
• claude mcp list shows the server connected; list_schemas / describe_table / a real query return data.
• query("UPDATE reporting.refresh_log SET notes='x'") is rejected by the guard.
• A request with a missing/bad bearer token returns 401.
• docker logs analytics_mcp shows one audit line per query (caller name, SQL, rows, ms).

---

10. Security notes

• Four read-only layers: role GRANTs · default_transaction_read_only=on (role + connection) · rolled-back txn · SQL keyword guard.
• Least privilege: analytics_ro only has USAGE+SELECT on reporting/tracksolid and EXECUTE on reporting functions — no other schema, no write of any kind.
• Per-analyst tokens make access revocable and queries attributable; rotate by editing MCP_AUTH_TOKENS and re-running the deploy (a recreate).
• Resource guards: statement_timeout=30s, idle-txn timeout, row cap (default 1000 / ceiling 10000) protect the DB from runaway analyst queries.
• Future: swap static Bearer tokens for OAuth (MCP supports it) if/when the team scales; consider a column-level deny-list if any PII lives in tracksolid.*.