The Data Layer

One Interface

The CMO agent doesn't open dashboards. It runs cmo <module> <command>. Every data source — GA4, GSC, Stripe, Supabase, Instantly — is a module behind the same CLI wrapper. The agent reads JSON, not screenshots.

# Same interface, 12 different APIs underneath:
cmo ga4 overview --site=kaicalls --days=7
cmo gsc opportunities --site=kaicalls
cmo stripe_report mrr
cmo kaicalls leads --days=7
cmo bwk dashboard

# All pipe through cmo_wrapper.py with telemetry tracking:
#   → validates module + command
#   → invokes scripts/{module}.py
#   → logs execution duration + outcome to JSONL
#   → returns JSON to the agent

cmo_wrapper.py — what the agent actually calls

def run_module(module_name: str, command: str, args: dict):
    """Run a CMO module command and track telemetry."""
    script_path = f"/opt/cmo-analytics/scripts/{module_name}.py"

    # Build CLI args from dict — passed directly to module
    cmd_args = [command]
    for k, v in args.items():
        cmd_args.append(f"--{k}={v}")

    start_time = time.time()
    # Run in subprocess — isolated, no import side effects
    result = subprocess.run(
        [sys.executable, script_path] + cmd_args,
        capture_output=True, text=True
    )
    duration_ms = int((time.time() - start_time) * 1000)

    # Log to telemetry JSONL regardless of outcome
    log_execution(
        module=module_name, command=command, params=args,
        duration_ms=duration_ms,
        outcome="success" if result.returncode == 0 else "error",
        error=result.stderr[:200] if result.returncode != 0 else None
    )
    return json.loads(result.stdout)

12 Data Sources

Each source is a Python class in analytics/. All use lazy client init — the API client is created on first use, not import. Credentials come from .env.

Lazy client init — why it matters

# All five clients follow the same pattern.
# Importing the module is instant — no API calls, no auth handshake.
# The client only connects when a query actually fires.

class GoogleAnalytics:
    def __init__(self, property_id: str, credentials_path: str):
        self.property_id = property_id
        self.credentials_path = credentials_path
        self._client = None          ← not initialized here

    def _get_client(self):
        if self._client is None:   ← first call only
            from google.analytics.data_v1beta import BetaAnalyticsDataClient
            from google.oauth2 import service_account
            credentials = service_account.Credentials.from_service_account_file(
                self.credentials_path,
                scopes=["https://www.googleapis.com/auth/analytics.readonly"]
            )
            self._client = BetaAnalyticsDataClient(credentials=credentials)
        return self._client

# Same pattern in SupabaseAnalytics, SearchConsole, StripeAnalytics.
# Means the daily_report.py can import all four without triggering
# any network calls — only the ones actually invoked connect.

Supabase — Data Isolation by Design

The KaiCalls Supabase instance contains data for all paying clients — not just Connor's businesses. Every query is hard-scoped at the module level. The agent can't accidentally pull a client's lead data.

# Owned business IDs — the only ones ever queried
KAI_CALLS_BUSINESS_ID = "15e7eca8-9e34-4ec7-9a3c-24a2dd69df79"
ABP_BUSINESS_ID       = "25d75618-109c-4fb2-ac26-8bede260d26f"
OWNED_BUSINESS_IDS    = [KAI_CALLS_BUSINESS_ID, ABP_BUSINESS_ID]

def get_leads(self, limit=100, business_id=None, ...):
    query = client.table("leads").select("*, businesses(name)")

    if business_id:
        query = query.eq("business_id", business_id)
    else:
        query = query.in_("business_id", OWNED_BUSINESS_IDS)
    # Never reaches this point without a business_id filter.
    # Client data from other businesses is structurally unreachable.

Two businesses, two contexts — never mixed

# KaiCalls (15e7eca8) = SALES pipeline
# → outbound prospecting, converting to $499/mo subscribers
# → metrics: leads, calls made, conversion rate
# → command: cmo kaicalls leads --days=7

# ABP/Starrs (25d75618) = OPERATIONAL health
# → inbound calls landing correctly, transcripts capturing
# → metrics: call volume, transcript success rate
# → command: cmo kaicalls abp-health

# kaicalls.py enforces this split at the command level —
# abp-health routes to a separate query, never mixed with leads:
def cmd_abp_health(args):
    """ABP operational health — NOT the KaiCalls sales pipeline."""
    return db.get_call_outcomes_analysis(
        business_id=ABP_BUSINESS_ID,  ← explicit, not the default
        days=int(args.get("days", 7))
    )

Stripe — Zero External Dependencies

The Stripe client uses no SDK. Raw urllib with basic auth. The reason: the Stripe Python SDK is 4MB and pulls in 6 transitive dependencies. For a background analytics process that runs 3x per day, that's unnecessary. The implementation covers everything needed: MRR, subscriptions, at-risk detection.

BASE_URL = "https://api.stripe.com/v1"

def _request(self, endpoint: str, params: dict = None) → dict:
    """Raw HTTP — no SDK, no transitive deps."""
    url = f"{BASE_URL}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)

    req = Request(url)
    # Basic auth: api_key as username, empty password
    credentials = base64.b64encode(f"{self.api_key}:".encode()).decode()
    req.add_header("Authorization", f"Basic {credentials}")

    with urlopen(req) as response:
        return json.loads(response.read().decode())

def get_mrr(self) → dict:
    """MRR from active subscriptions. Shared across BWK + KaiCalls + VocalScribe."""
    subs = self._paginate("subscriptions", {"status": "active", "limit": 100})
    total_mrr = sum(s["plan"]["amount"] for s in subs) / 100
    return {"mrr_usd": total_mrr, "active_subs": len(subs)}

def get_subscriptions(self, status="active") → list[StripeSubscription]:
    raw = self._paginate("subscriptions", {"status": status, "expand[]": "data.customer"})
    return [StripeSubscription(
        id=s["id"],
        customer_email=s["customer"]["email"],
        plan_name=s["items"]["data"][0]["price"]["nickname"] or "unknown",
        mrr=s["plan"]["amount"] / 100,
        cancel_at_period_end=s["cancel_at_period_end"],
    ) for s in raw]

GA4 — 10 Properties, One Client

Nine different businesses. Nine GA4 properties. One GoogleAnalytics class, instantiated per site key. The site registry lives in ga4.py — adding a new site is one dict entry and one env var.

SITES = {
    "kaicalls":    {"property_id": os.getenv("GA_KAICALLS_PROPERTY_ID")},
    "vocalscribe": {"property_id": os.getenv("GA_VOCALSCRIBE_PROPERTY_ID")},
    "buildwithkai":{"property_id": os.getenv("GA_BUILDWITHKAI_PROPERTY_ID")},
    "kaithescribe":{"property_id": os.getenv("GA_KAITHESCRIBE_PROPERTY_ID")},
    "meetkai":     {"property_id": os.getenv("GA_MEETKAI_PROPERTY_ID")},
    "connorgallic":{"property_id": os.getenv("GA_CONNORGALLIC_PROPERTY_ID")},

    "indexify":    {"property_id": os.getenv("GA_INDEXIFY_PROPERTY_ID")},
    "abp":         {"property_id": os.getenv("GA_AWESOMEBACKYARDPARTIES_PROPERTY_ID")},
    "starrsparty": {"property_id": os.getenv("GA_STARRSPARTY_PROPERTY_ID")},
}

def get_ga(site_key: str) → GoogleAnalytics:
    site = SITES.get(site_key)
    if not site or not site["property_id"]:
        raise ValueError(f"Unknown site: {site_key}")
    return GoogleAnalytics(property_id=site["property_id"], credentials_path=CREDS)

GA4 Data API request structure

def get_overview(self, start_date="30daysAgo", end_date="today") → dict:
    """Sessions · users · bounce rate · avg duration."""
    from google.analytics.data_v1beta.types import (
        RunReportRequest, DateRange, Metric, Dimension
    )
    request = RunReportRequest(
        property=f"properties/{self.property_id}",
        date_ranges=[DateRange(start_date=start_date, end_date=end_date)],
        metrics=[
            Metric(name="sessions"),
            Metric(name="totalUsers"),
            Metric(name="bounceRate"),
            Metric(name="averageSessionDuration"),
        ],
    )
    response = client.run_report(request)
    # Returns normalized dict — agent reads JSON, not GA4 proto objects
    return self._normalize(response)

Cross-site sweep — one command

# cmo ga4 all --days=7
# Runs get_overview() against all 9 properties, aggregates results.
def cmd_all(args):
    days = int(args.get("days", 7))
    start = f"{days}daysAgo"
    results = {}
    for key, site in SITES.items():
        if not site["property_id"]:
            continue           ← skip unconfigured properties silently
        try:
            ga = get_ga(key)
            results[key] = ga.get_overview(start_date=start)
        except Exception as e:
            results[key] = {"error": str(e)[:100]}
    return results

Daily Report — Data Composition

The 8am daily report doesn't call a single API — it composes results from all active sources. Each section uses safe() wrapping so a single source failure doesn't abort the whole report.

def safe(fn, fallback="error"):
    """Call fn(). If it throws, return {fallback: error_message}."""
    try:
        return fn()
    except Exception as e:
        return {fallback: str(e)[:200]}

def cmd_daily(args):
    # All clients — lazy init, no upfront connections
    kc     = SupabaseAnalytics(url=..., key=...)
    stripe = StripeAnalytics(api_key=...)
    ga_kc  = GoogleAnalytics(property_id=..., credentials_path=...)
    ga_bwk = GoogleAnalytics(property_id=..., credentials_path=...)
    ga_abp = GoogleAnalytics(property_id=..., credentials_path=...)

    return {
        "generated_at": datetime.now().isoformat(),
        "kaicalls": {
            "leads_today": safe(lambda: kc.get_leads_summary(days=1)),
            "leads_7d":    safe(lambda: kc.get_leads_summary(days=7)),
            "calls_today": safe(lambda: kc.get_call_outcomes_analysis(days=1)),
        },
        "stripe": {
            "mrr":        safe(lambda: stripe.get_mrr()),
            "revenue_7d": safe(lambda: stripe.get_revenue_summary(days=7)),
        },
        "traffic": {
            "kaicalls":    safe(lambda: ga_kc.get_overview("1daysAgo", "today")),
            "buildwithkai":safe(lambda: ga_bwk.get_overview("1daysAgo", "today")),
            "abp":         safe(lambda: ga_abp.get_overview("1daysAgo", "today")),
        },
    }
    # One source fails → its key has {"error": "..."}, rest of report intact

Telemetry — Every Command Logged

Every cmo command is logged to a daily JSONL file. Duration, outcome, error message if any. No external service — local append-only files. The agent can query its own usage patterns.

# One line per command execution:
{
  "module":      "kaicalls",
  "command":     "leads",
  "params":      {"days": "7"},
  "duration_ms": 1243,
  "outcome":     "success",
  "api_calls":   1,
  "timestamp":   "2026-03-15T08:00:12Z"
}

VALID_MODULES = {
  "kaicalls",
  "bwk",
  "abp",
  "ga4",
  "gsc",
  "stripe_report",
  "daily_report",
  "telemetry",
}
# Modules outside this set → rejected before execution

Full Command Surface