Build a Data Analytics Agent Using Custom Connectors

import argparse
import json
import os
from pathlib import Path

from perplexity import Perplexity

LAST_RESPONSE = Path("last_response.json")
RUNS = Path("runs")


def ask(client, profile_id, question, version="latest", model=None, previous_id=None):
    """Call Agent API; the saved profile supplies the analyst's configuration."""
    request = {
        # Reuse the configured model, instructions, skills, connector, and tools.
        "profile": {"type": "custom", "id": profile_id, "version": version},
        "input": question,
    }
    if model:
        # Compare a provider without rebuilding the profile.
        request["model"] = model
        request["max_output_tokens"] = 16384
    if previous_id:
        # Continue the investigation through Agent API conversation state.
        request["previous_response_id"] = previous_id
    return client.responses.create(**request)


def show_work(response):
    folder = RUNS / response.id
    folder.mkdir(parents=True, exist_ok=True)
    receipt = folder / "response.json"
    receipt.write_text(response.model_dump_json(indent=2))
    print(f"evidence: {receipt}")
    print(f"status: {response.status}")
    if response.error:
        print(f"response error: {response.error}")
    calls = 0
    for item in response.output:
        if item.type == "skill_loaded":
            print(f"[skill] load attempt: {item.name}")
        elif item.type == "mcp_list_tools":
            print(f"[{item.server_label}] {len(item.tools or [])} tools available")
            if item.error:
                print(f"connector error: {item.error}")
        elif item.type == "mcp_call":
            calls += 1
            arguments = json.loads(item.arguments or "{}")
            detail = arguments.get("sql") or arguments.get("query") or json.dumps(arguments)
            suffix = f" -> error: {item.error}" if item.error else ""
            print(f"[{item.server_label}] {item.name}: {detail}{suffix}")
            try:
                result = json.loads(item.output or "{}")
                if isinstance(result, dict) and result.get("error"):
                    print(f"SQL/tool error: {result['error']}")
            except (TypeError, json.JSONDecodeError):
                pass
        elif item.type == "tool_search_output":
            print(f"[tool search] {item.arguments}")
        elif item.type == "sandbox_results":
            print(f"[sandbox] ran {item.language} code, status {item.status}")
        elif item.type == "share_file":
            print(f"[file] {item.filename}")
    print()
    print(response.output_text)
    print()
    print(f"mcp_call items: {calls}")
    cost = getattr(getattr(response.usage, "cost", None), "total_cost", None)
    print(f"cost: ${cost:.4f}" if cost is not None else "cost: not reported")
    print(f"response id: {response.id}")
    if response.status == "completed":
        LAST_RESPONSE.write_text(json.dumps({"id": response.id}))
        return True
    return False


def download_files(client, response):
    if not any(item.type == "share_file" for item in response.output):
        return
    folder = RUNS / response.id / "files"
    folder.mkdir(parents=True, exist_ok=True)
    for file in client.responses.files.list(response.id).data:
        name = file.filename
        if not name or name in {".", ".."} or "/" in name or "\\" in name:
            raise ValueError(f"Unsafe filename: {name!r}")
        target = folder / name
        if target.exists():
            raise FileExistsError(f"Refusing to overwrite {target}")
        content = client.responses.files.content(file_id=file.id, response_id=response.id)
        content.write_to_file(target)
        print(f"downloaded {target} ({file.bytes} bytes)")


def main():
    parser = argparse.ArgumentParser(description="Ask the Brightside analyst a question.")
    parser.add_argument("--ask", help="start a new conversation with this question")
    parser.add_argument("--follow-up", help="continue the last conversation with this question")
    parser.add_argument("--model", help="run the profile with a different model for this request")
    parser.add_argument("--version", default="latest", help="profile version to run (default: latest)")
    parser.add_argument("--profile", default=os.environ.get("BRIGHTSIDE_PROFILE_ID"))
    args = parser.parse_args()
    if not args.profile:
        parser.error("pass --profile or set BRIGHTSIDE_PROFILE_ID")
    if bool(args.ask) == bool(args.follow_up):
        parser.error("pass exactly one of --ask or --follow-up")
    previous_id = None
    if args.follow_up:
        try:
            previous_id = json.loads(LAST_RESPONSE.read_text())["id"]
        except (OSError, ValueError, KeyError):
            parser.error("no valid conversation checkpoint; run --ask first")
    question = args.ask or args.follow_up
    client = Perplexity()
    response = ask(client, args.profile, question, args.version, args.model, previous_id)
    if not show_work(response):
        raise SystemExit("Run did not complete. Read the saved response; checkpoint unchanged.")
    download_files(client, response)


if __name__ == "__main__":
    main()

---
name: metric-investigation
description: Load when asked why a business metric changed, dropped, or spiked, or for a follow-up that narrows or charts an earlier investigation. Do not load for simple lookups or single-number questions.
compatibility: Requires a data connector that exposes describe_tables, search_context, and run_sql. Charts need the Sandbox tool.
---

# Metric investigation

You are investigating a change in a business metric. Work in this order.

1. Call `describe_tables` first. Learn which system writes each table. If a tool you need is missing from one tool search, search again by its exact name before you say it is unavailable.
2. Call `search_context` for the metric name before you write your own definition. This demo matches every search word; if nothing matches, retry with fewer words, such as `paid conversion` or `mobile`. Use the team's definition when one exists and say which one you used.
3. Run SQL for every number you report, including differences and percentages. Never estimate or guess a count. Exclude test accounts from every signup, conversion, and billing diagnostic unless explicitly comparing all, non-test, and test accounts. Filter the rows being counted: `LEFT JOIN accounts ... AND a.is_test = 0` does not filter `COUNT(s.account_id)`. Use an inner join, a prefiltered CTE, or a conditional count. Read the daily example in `references/week-bounds.md` when retaining zero-count days.
4. Compare the period in question with at least three earlier Monday to Sunday weeks, using the same query with explicit date bounds. Read `references/week-bounds.md` before writing the first weekly query. Both signup and subscription start must fall inside that same week's bounds. Keep these bounds in every device, segment, and plan breakdown; only the grouping or cohort filter changes. Neither same-day matching nor a rolling seven days after signup implements this metric. Check that mutually exclusive breakdown counts sum to the weekly totals.
5. Before you conclude that the business changed, check whether the measurement changed. Break the metric down by device, segment, and plan. Look for a dimension that goes to zero or near zero on specific days, and check whether the tables that feed the metric moved together or separately.
6. Missing signup rows damage both the numerator and denominator of this metric. Label the measured rate as observed; it does not establish whether the true rate improved, stayed stable, or deteriorated. Billing accounts without an observed signup do not reveal their missing signup dates. Do not reconstruct exact true cohort counts or rates from them. Keep observed counts, billing totals, and estimates separate, and state what remains unknown.
7. Check every tool result before using it. A failed or truncated diagnostic is not evidence. If a required check fails, name the check and error, give only the supported findings, and say the investigation is incomplete. Do not treat a completed API response as proof that its analysis is correct.

When the user asks for a chart, send Sandbox a prompt containing the complete successful, untruncated SQL result, preserving `columns`, `rows`, and `truncated`. Columns must include `week_start`, `signups`, and `paid_conversions`. Ask Sandbox to save it as `result.json` and use the bundled `scripts/plot_weekly.py` helper from this skill's directory with a title matching the requested cohort. The helper accepts `result.json`, `chart.png`, and the title as its three arguments. Ask Sandbox to share both the PNG and input JSON. If the helper is unavailable, ask Sandbox to generate an equivalent grouped bar chart with a zero baseline, direct labels, explicit week dates, and the cohort title. Note that the data is synthetic and September 9 to 12 mobile signup records are missing. Sandbox determines what code runs; do not claim the helper executed without evidence. Compare the returned chart and input data with the SQL result before presenting them.

Answer in this shape:

- **Finding.** One or two sentences on what moved and whether the evidence points to the business or the measurement.
- **Evidence.** The numbers, with the periods they cover.
- **SQL.** Every query you ran, in the order you ran it.
- **Assumptions.** Definitions you chose, filters you applied, and anything you could not check.

Do not change source data or publish messages. Creating the requested local chart and its input file is allowed.

# Week bounds in SQLite

Weeks run Monday to Sunday. Write both bounds out. Do not derive the week from the day inside the query.

Good:

```sql
WHERE s.day BETWEEN '2026-09-07' AND '2026-09-13'
```

Bad, because `date(day, 'weekday 1', '-7 days')` moves every Monday into the week before it:

```sql
GROUP BY date(s.day, 'weekday 1', '-7 days')
```

Also avoid `strftime('%W', day)`: it numbers weeks from the first Monday of the year and does not match the team's calendar at year boundaries.

Template for same-week paid conversions, one row per week:

```sql
WITH weeks(week_start) AS (
  VALUES ('2026-08-17'), ('2026-08-24'), ('2026-08-31'), ('2026-09-07')
)
SELECT
  w.week_start,
  COUNT(s.account_id) AS signups,
  SUM(EXISTS (
    SELECT 1 FROM subscriptions b
    WHERE b.account_id = s.account_id
      AND b.started_day BETWEEN w.week_start AND date(w.week_start, '+6 days')
  )) AS paid_conversions
FROM weeks w
JOIN signups s ON s.day BETWEEN w.week_start AND date(w.week_start, '+6 days')
JOIN accounts a ON a.id = s.account_id AND a.is_test = 0
GROUP BY w.week_start
ORDER BY w.week_start;
```

Add `AND a.segment = 'enterprise'` to the accounts join for the enterprise version.

## Keep the metric unchanged in breakdowns

For device, segment, or plan breakdowns, add that dimension to `SELECT` and `GROUP BY` in the weekly template. Keep both date predicates anchored to `w.week_start`. Sum the breakdown counts back to the headline counts for each week.

Do not replace the subscription predicate with `b.started_day = s.day` or `b.started_day BETWEEN s.day AND date(s.day, '+6 days')`. Those count same-day and rolling-seven-day conversions, respectively, not same-calendar-week conversions.

## Filter the counted cohort before a left join

This template keeps days with zero observed signups while excluding test accounts before counting:

```sql
WITH days(day) AS (
  VALUES ('2026-09-09'), ('2026-09-10'), ('2026-09-11'), ('2026-09-12')
),
devices(device) AS (VALUES ('web'), ('mobile')),
eligible_signups AS (
  SELECT s.account_id, s.day, s.device
  FROM signups s
  JOIN accounts a ON a.id = s.account_id
  WHERE a.is_test = 0
)
SELECT d.day, v.device, COUNT(s.account_id) AS signups
FROM days d
CROSS JOIN devices v
LEFT JOIN eligible_signups s ON s.day = d.day AND s.device = v.device
GROUP BY d.day, v.device
ORDER BY d.day, v.device;
```

By contrast, `LEFT JOIN accounts a ON a.id = s.account_id AND a.is_test = 0` followed by `COUNT(s.account_id)` still counts test signups. It only makes the joined account fields null. Apply the cohort filter to billing diagnostics too.

## Separate billing activity from observed conversions

The following diagnostic counts non-test billing starts and those without a matching observed signup in the same week:

```sql
SELECT
  COUNT(*) AS billing_starts,
  SUM(NOT EXISTS (
    SELECT 1 FROM signups s
    WHERE s.account_id = b.account_id
      AND s.day BETWEEN '2026-09-07' AND '2026-09-13'
  )) AS without_observed_same_week_signup
FROM subscriptions b
JOIN accounts a ON a.id = b.account_id AND a.is_test = 0
WHERE b.started_day BETWEEN '2026-09-07' AND '2026-09-13';
```

An unmatched billing row may belong to an earlier signup or a missing signup. Without its signup date, you cannot assign it to the true weekly signup cohort. Report the observed rate separately and leave the true cohort size, conversions, and rate unknown.

import json
import subprocess
import sys


def load_rows(path):
    with open(path) as source:
        result = json.load(source)
    if result.get("truncated") or result.get("error"):
        raise ValueError("Use a successful, untruncated SQL result.")
    rows = [dict(zip(result["columns"], row, strict=True)) for row in result["rows"]]
    required = {"week_start", "signups", "paid_conversions"}
    if not rows or not required <= set(result["columns"]):
        raise ValueError("Expected weekly signups and paid_conversions.")
    return rows


def plot(rows, target, title):
    subprocess.run([sys.executable, "-m", "pip", "install", "--quiet", "matplotlib"], check=True)
    import matplotlib

    matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    weeks = [row["week_start"] for row in rows]
    signups = [row["signups"] for row in rows]
    paid = [row["paid_conversions"] for row in rows]
    positions = range(len(weeks))
    figure, axis = plt.subplots(figsize=(8, 4.5))
    axis.bar([p - 0.2 for p in positions], signups, width=0.4, label="Signups", color="#1f77b4")
    axis.bar([p + 0.2 for p in positions], paid, width=0.4, label="Paid conversions", color="#ff7f0e")
    for p, value in zip(positions, signups):
        axis.text(p - 0.2, value, str(value), ha="center", va="bottom", fontsize=9)
    for p, value in zip(positions, paid):
        axis.text(p + 0.2, value, str(value), ha="center", va="bottom", fontsize=9)
    axis.set_xticks(list(positions))
    axis.set_xticklabels([f"Week of {week}" for week in weeks])
    axis.set_ylabel("Accounts")
    axis.set_title(title)
    axis.legend()
    figure.text(0.5, 0.02, "Synthetic data; Sep 9 to 12 mobile signup records are missing.",
                ha="center", fontsize=8)
    figure.tight_layout(rect=(0, 0.06, 1, 1))
    figure.savefig(target, dpi=150)
    plt.close(figure)
    print(f"wrote {target}")


if __name__ == "__main__":
    plot(load_rows(sys.argv[1]), sys.argv[2], sys.argv[3])

import atexit
import hmac
import json
import os
import random
import re
import sqlite3
from datetime import date, timedelta

import uvicorn
from mcp.server.mcpserver import MCPServer
from mcp.server.transport_security import TransportSecuritySettings
from starlette.responses import JSONResponse

LAST_DAY = date(2026, 9, 13)
FIRST_DAY = LAST_DAY - timedelta(days=41)
GAP_DAYS = {date(2026, 9, 9) + timedelta(days=i) for i in range(4)}
PLANS = [("starter", 29), ("pro", 79), ("enterprise", 299)]
SEGMENTS = ["smb", "smb", "mid_market", "enterprise"]


def build_rows():
    rng = random.Random(7)
    accounts, signups, subscriptions = [], [], []
    account_id = 1000
    day = FIRST_DAY
    while day <= LAST_DAY:
        for _ in range(rng.randint(70, 90)):
            account_id += 1
            plan, price = rng.choice(PLANS)
            device = "mobile" if rng.random() < 0.55 else "web"
            is_test = 1 if rng.random() < 0.03 else 0
            provisioned_by = "sales" if rng.random() < 0.1 else "self_serve"
            accounts.append((account_id, plan, rng.choice(SEGMENTS), is_test, provisioned_by))
            if not (device == "mobile" and day in GAP_DAYS):
                signups.append((account_id, day.isoformat(), device))
            if rng.random() < 0.3:
                started = day + timedelta(days=rng.randint(0, 3))
                if started <= LAST_DAY:
                    subscriptions.append((account_id, started.isoformat(), price))
        day += timedelta(days=1)
    return accounts, signups, subscriptions


def create_db():
    connection = sqlite3.connect(":memory:", check_same_thread=False)
    connection.executescript(
        "CREATE TABLE accounts (id INTEGER PRIMARY KEY, plan TEXT, segment TEXT,"
        " is_test INTEGER, provisioned_by TEXT);"
        "CREATE TABLE signups (account_id INTEGER, day TEXT, device TEXT);"
        "CREATE TABLE subscriptions (account_id INTEGER, started_day TEXT, amount_usd REAL);"
    )
    accounts, signups, subscriptions = build_rows()
    connection.executemany("INSERT INTO accounts VALUES (?, ?, ?, ?, ?)", accounts)
    connection.executemany("INSERT INTO signups VALUES (?, ?, ?)", signups)
    connection.executemany("INSERT INTO subscriptions VALUES (?, ?, ?)", subscriptions)
    connection.commit()
    connection.execute("PRAGMA query_only = ON")
    return connection


CONTEXT = {
    "date_range": f"{FIRST_DAY} to {LAST_DAY}; weeks run Monday to Sunday",
    "tables": {
        "accounts": "One row per customer account. is_test = 1 marks internal test accounts.",
        "signups": "One row per account signup, written by the checkout tracking pipeline. device is 'web' or 'mobile'.",
        "subscriptions": "One row per paid subscription, written by billing when the first payment clears.",
    },
    "definitions": [
        {
            "name": "paid_conversion_rate",
            "definition": "Subscriptions started in a week from accounts that signed up in that same week, divided by signups that week. Exclude accounts where is_test = 1.",
        },
        {
            "name": "enterprise_conversion_rate",
            "definition": "paid_conversion_rate restricted to accounts where segment = 'enterprise'.",
        },
    ],
    "documents": [
        {
            "title": "Mobile checkout tracking change",
            "text": f"Mobile checkout tracking changed on {min(GAP_DAYS)}. Signups from the mobile app may be missing until the fix ships.",
        }
    ],
    "queries": [
        {
            "name": "weekly_signups_by_device",
            "reviewed": True,
            "sql": "WITH weeks(start) AS (VALUES ('2026-08-17'), ('2026-08-24'), ('2026-08-31'), ('2026-09-07')) SELECT w.start, s.device, COUNT(*) AS signups FROM weeks w JOIN signups s ON s.day BETWEEN w.start AND date(w.start, '+6 days') JOIN accounts a ON a.id = s.account_id WHERE a.is_test = 0 GROUP BY w.start, s.device ORDER BY w.start, s.device",
        }
    ],
}

db = create_db()
atexit.register(db.close)
mcp = MCPServer("brightside-data")
WORDS = lambda text: [word.rstrip("s") for word in re.findall(r"[a-z0-9]+", text.lower())]


@mcp.tool()
def describe_tables() -> dict:
    """List every table with its columns and a note on what writes it. Call this first."""
    tables = {}
    for (name,) in db.execute("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name"):
        columns = [row[1] for row in db.execute(f"PRAGMA table_info({name})")]
        tables[name] = {"columns": columns, "note": CONTEXT["tables"].get(name, "")}
    return {"date_range": CONTEXT["date_range"], "tables": tables}


@mcp.tool()
def search_context(query: str) -> dict:
    """Find metric definitions, team notes, and reviewed SQL that mention the query text. Search here before defining a metric yourself."""
    words = set(WORDS(query))
    results = []
    for kind in ("definitions", "documents", "queries"):
        for item in CONTEXT[kind]:
            if words and words <= set(WORDS(json.dumps(item))):
                results.append({"kind": kind, **item})
    return {"query": query, "results": results}


@mcp.tool()
def run_sql(sql: str) -> dict:
    """Run one read-only SQLite SELECT or WITH statement and return up to 100 rows."""
    statement = sql.strip().rstrip(";").strip()
    if ";" in statement or not re.match(r"(?is)(select|with)\b", statement):
        return {"error": "Only one SELECT or WITH statement is allowed."}
    try:
        cursor = db.execute(statement)
        rows = cursor.fetchmany(101)
    except sqlite3.Error as error:
        return {"error": str(error)}
    columns = [column[0] for column in cursor.description or []]
    return {"columns": columns, "rows": [list(row) for row in rows[:100]], "truncated": len(rows) > 100}


KEY = os.environ.get("DATA_SERVER_KEY", "")


class RequireKey:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            headers = {name.decode().lower(): value.decode() for name, value in scope["headers"]}
            sent = headers.get("authorization", "")
            if not hmac.compare_digest(sent, f"Bearer {KEY}"):
                print(f"rejected {scope['method']} {scope['path']}: no valid key")
                response = JSONResponse({"error": "missing or wrong API key"}, status_code=401)
                await response(scope, receive, send)
                return
        await self.app(scope, receive, send)


if __name__ == "__main__":
    if not KEY:
        raise SystemExit("Set DATA_SERVER_KEY before starting the server.")
    print("MCP endpoint: http://127.0.0.1:8000/mcp")
    print("Next, in another terminal: cloudflared tunnel --url http://localhost:8000")
    security = TransportSecuritySettings(enable_dns_rebinding_protection=False)
    app = RequireKey(mcp.streamable_http_app(transport_security=security, json_response=True))
    uvicorn.run(app, host="127.0.0.1", port=8000)