mcp: live query architecture — all metrics hit platform API directly, no ClickHouse batch

docker/mcp/main.py

@@ -1,13 +1,15 @@
 """
-Manar MCP Server — Google Ads Connector
+Manar MCP Server — Live query edition
-FastMCP server exposing brand-scoped Google Ads tools.
+- FastMCP tools on :8090/mcp  (brand-scoped, live platform queries)
-All tools validate brand_membership before execution.
+- REST API on :8092            (same live queries, called by frontend)
+- Health on :8091
 """
 
 import os
 import json
 import logging
 from contextlib import asynccontextmanager
+from datetime import date, timedelta
 from typing import Optional
 
 import asyncpg
@@ -15,6 +17,8 @@ import clickhouse_connect
 from fastmcp import FastMCP
 from dotenv import load_dotenv
 from health import start_health_thread
+from api import app as rest_app, start_api_thread, build_gads_client, get_brand_credentials
+from oauth import register_oauth_routes
 
 load_dotenv()
 
@@ -41,29 +45,27 @@ state = AppState()
 @asynccontextmanager
 async def lifespan(app):
     log.info("Manar MCP starting...")
-    state.core_pool = await asyncpg.create_pool(CORE_DB_URL, min_size=2, max_size=10)
+    state.core_pool    = await asyncpg.create_pool(CORE_DB_URL,    min_size=2, max_size=10)
     state.tenants_pool = await asyncpg.create_pool(TENANTS_DB_URL, min_size=2, max_size=10)
-    state.ch_client = clickhouse_connect.get_client(
+    state.ch_client    = clickhouse_connect.get_client(
         host=CH_HOST, port=CH_PORT, database=CH_DB,
         username=os.environ.get("CLICKHOUSE_USER", "default"),
         password=os.environ.get("CLICKHOUSE_PASSWORD", "")
     )
-    log.info("Manar MCP ready — all connections established.")
+    log.info("Manar MCP ready.")
     yield
     await state.core_pool.close()
     await state.tenants_pool.close()
-    log.info("Manar MCP shut down.")
 
 
 mcp = FastMCP("manar-mcp", lifespan=lifespan)
 
 
-# ── Access Guard ──────────────────────────────────────────────────────────────
+# ── Access guard ──────────────────────────────────────────────────────────────
 
 async def assert_brand_access(brand_slug: str, actor_email: str) -> dict:
     async with state.core_pool.acquire() as conn:
-        user = await conn.fetchrow(
+        user = await conn.fetchrow("SELECT id, role FROM users WHERE email = $1", actor_email)
-            "SELECT id, role FROM users WHERE email = $1", actor_email)
         if not user:
             raise ValueError(f"User {actor_email} not found")
         brand = await conn.fetchrow(
@@ -72,130 +74,214 @@ async def assert_brand_access(brand_slug: str, actor_email: str) -> dict:
             raise ValueError(f"Brand '{brand_slug}' not found or inactive")
         if user["role"] != "super_admin":
             membership = await conn.fetchrow(
-                "SELECT role FROM brand_memberships WHERE brand_id = $1 AND user_id = $2",
+                "SELECT role FROM brand_memberships WHERE brand_id=$1 AND user_id=$2",
                 brand["id"], user["id"])
             if not membership:
-                raise ValueError(f"User {actor_email} has no access to brand '{brand_slug}'")
+                raise ValueError(f"No access to '{brand_slug}'")
         return dict(brand)
 
 
-# ── Tools ─────────────────────────────────────────────────────────────────────
+# ── Register REST routes ──────────────────────────────────────────────────────
+
+register_oauth_routes(rest_app, lambda: state.tenants_pool)
+
+@rest_app.get("/brands/{brand_slug}/recommendations")
+async def rest_recommendations(brand_slug: str, status: str = "pending"):
+    schema = f"brand_{brand_slug}"
+    try:
+        async with state.tenants_pool.acquire() as conn:
+            rows = await conn.fetch(
+                f"""SELECT id::text, type, title, description, priority, status, created_at::text
+                    FROM {schema}.recommendations WHERE status=$1
+                    ORDER BY created_at DESC LIMIT 50""",
+                status)
+        return [dict(r) for r in rows]
+    except Exception as e:
+        log.error(f"recommendations error: {e}")
+        return []
+
+@rest_app.get("/brands/{brand_slug}/alerts")
+async def rest_alerts(brand_slug: str):
+    try:
+        ch = clickhouse_connect.get_client(
+            host=CH_HOST, port=CH_PORT, database=CH_DB,
+            username=os.environ.get("CLICKHOUSE_USER", "default"),
+            password=os.environ.get("CLICKHOUSE_PASSWORD", "")
+        )
+        result = ch.query(f"""
+            SELECT campaign_id, metric, expected_value, actual_value,
+                   deviation_pct, severity, toString(detected_at) AS detected_at
+            FROM manar.anomaly_events
+            WHERE brand_id='{brand_slug}' AND resolved_at IS NULL
+            ORDER BY detected_at DESC LIMIT 50
+        """)
+        return [dict(zip(result.column_names, r)) for r in result.result_rows]
+    except Exception as e:
+        log.error(f"alerts error: {e}")
+        return []
+
+
+# ── MCP Tools ─────────────────────────────────────────────────────────────────
 
 @mcp.tool()
 async def list_brands(actor_email: str) -> list[dict]:
-    """List all active brands accessible to actor_email."""
+    """List all active brands the actor can access."""
     async with state.core_pool.acquire() as conn:
-        user = await conn.fetchrow("SELECT id, role FROM users WHERE email = $1", actor_email)
+        user = await conn.fetchrow("SELECT id, role FROM users WHERE email=$1", actor_email)
         if not user:
             raise ValueError(f"User {actor_email} not found")
         if user["role"] == "super_admin":
             rows = await conn.fetch(
-                "SELECT slug, name, trust_level, timezone FROM brands WHERE status = 'active' ORDER BY name")
+                "SELECT slug, name, trust_level, timezone FROM brands WHERE status='active' ORDER BY name")
         else:
             rows = await conn.fetch(
-                """SELECT b.slug, b.name, b.trust_level, b.timezone, m.role
+                """SELECT b.slug, b.name, b.trust_level, b.timezone
-                   FROM brands b JOIN brand_memberships m ON m.brand_id = b.id
+                   FROM brands b
-                   JOIN users u ON u.id = m.user_id
+                   JOIN brand_memberships m ON m.brand_id=b.id
-                   WHERE u.email = $1 AND b.status = 'active' ORDER BY b.name""",
+                   JOIN users u ON u.id=m.user_id
+                   WHERE u.email=$1 AND b.status='active' ORDER BY b.name""",
                 actor_email)
         return [dict(r) for r in rows]
 
 
 @mcp.tool()
-async def get_ad_accounts(brand_slug: str, actor_email: str) -> list[dict]:
+async def get_campaigns(
-    """Get all ad accounts for a brand."""
-    await assert_brand_access(brand_slug, actor_email)
-    schema = f"brand_{brand_slug}"
-    async with state.tenants_pool.acquire() as conn:
-        rows = await conn.fetch(
-            f"SELECT id::text, platform_slug, external_id, name, currency, status, created_at::text "
-            f"FROM {schema}.ad_accounts ORDER BY platform_slug, name")
-    return [dict(r) for r in rows]
-
-
-@mcp.tool()
-async def fetch_google_ads_metrics(
     brand_slug: str, actor_email: str,
-    date_from: str, date_to: str,
+    days: int = 30, platform: str = "google_ads"
-    breakdown: str = "campaign"
 ) -> list[dict]:
-    """
+    """Live — campaign performance: spend, clicks, conversions, ROAS."""
-    Fetch performance metrics from ClickHouse.
-    breakdown: campaign | adgroup | day
-    date_from / date_to: YYYY-MM-DD
-    """
     await assert_brand_access(brand_slug, actor_email)
+    creds = await get_brand_credentials(brand_slug, platform)
+    if not creds.get("refresh_token"):
+        return [{"error": "not_connected", "brand": brand_slug, "platform": platform}]
 
-    group_fields = {
+    client      = build_gads_client(creds["refresh_token"], creds.get("developer_token"))
-        "campaign": "campaign_id, campaign_name",
+    ga          = client.get_service("GoogleAdsService")
-        "adgroup":  "campaign_id, campaign_name, adgroup_id, adgroup_name",
+    cid         = creds["account_id"]
-        "day":      "date",
+    today       = date.today()
-    }.get(breakdown, "campaign_id, campaign_name")
+    date_from   = today - timedelta(days=days)
 
-    query = f"""
+    q = f"""SELECT campaign.id, campaign.name, campaign.status,
-        SELECT {group_fields},
+                   metrics.impressions, metrics.clicks, metrics.cost_micros,
-            sum(impressions)  AS impressions,
+                   metrics.conversions, metrics.conversions_value
-            sum(clicks)       AS clicks,
+            FROM campaign
-            round(sum(cost_micros) / 1e6, 2) AS cost,
+            WHERE segments.date BETWEEN '{date_from}' AND '{today}'
-            sum(conversions)  AS conversions,
+              AND campaign.status != 'REMOVED'"""
-            sum(conv_value)   AS conv_value,
+
-            round(if(sum(impressions) > 0, sum(clicks) / sum(impressions), 0), 4) AS ctr,
+    cmap = {}
-            round(if(sum(clicks) > 0, sum(cost_micros) / 1e6 / sum(clicks), 0), 2) AS cpc,
+    for batch in ga.search_stream(customer_id=cid, query=q):
-            round(if(sum(cost_micros) > 0, sum(conv_value) / (sum(cost_micros) / 1e6), 0), 2) AS roas
+        for row in batch.results:
-        FROM manar.ad_metrics_hourly
+            k = str(row.campaign.id)
-        WHERE brand_id = '{brand_slug}'
+            if k not in cmap:
-          AND platform = 'google_ads'
+                cmap[k] = {"campaign_id": k, "campaign_name": row.campaign.name,
-          AND date >= '{date_from}'
+                           "status": row.campaign.status.name, "impressions": 0,
-          AND date <= '{date_to}'
+                           "clicks": 0, "cost": 0.0, "conversions": 0.0, "conv_value": 0.0}
-        GROUP BY {group_fields}
+            cmap[k]["impressions"] += row.metrics.impressions
-        ORDER BY cost DESC
+            cmap[k]["clicks"]      += row.metrics.clicks
-        LIMIT 500
+            cmap[k]["cost"]        += row.metrics.cost_micros / 1_000_000
-    """
+            cmap[k]["conversions"] += row.metrics.conversions
-    result = state.ch_client.query(query)
+            cmap[k]["conv_value"]  += row.metrics.conversions_value
-    return [dict(zip(result.column_names, row)) for row in result.result_rows]
+
+    result = []
+    for c in cmap.values():
+        c["cost"] = round(c["cost"], 2)
+        c["ctr"]  = round(c["clicks"] / c["impressions"], 4) if c["impressions"] else 0
+        c["cpc"]  = round(c["cost"] / c["clicks"], 2)        if c["clicks"]      else 0
+        c["roas"] = round(c["conv_value"] / c["cost"], 2)    if c["cost"]        else 0
+        result.append(c)
+    return sorted(result, key=lambda x: x["cost"], reverse=True)
 
 
 @mcp.tool()
 async def get_search_terms(
     brand_slug: str, actor_email: str,
-    date_from: str, date_to: str,
+    days: int = 30, platform: str = "google_ads"
-    min_cost: float = 0,
-    limit: int = 200
 ) -> list[dict]:
-    """Get search term performance — for keyword expansion / negative keyword decisions."""
+    """Live — search terms. Use to find negatives or expansion opportunities."""
     await assert_brand_access(brand_slug, actor_email)
-    query = f"""
+    creds = await get_brand_credentials(brand_slug, platform)
-        SELECT search_term,
+    if not creds.get("refresh_token"):
-            sum(impressions) AS impressions,
+        return [{"error": "not_connected", "brand": brand_slug}]
-            sum(clicks)      AS clicks,
+
-            round(sum(cost_micros) / 1e6, 2) AS cost,
+    client      = build_gads_client(creds["refresh_token"], creds.get("developer_token"))
-            sum(conversions) AS conversions,
+    ga          = client.get_service("GoogleAdsService")
-            round(if(sum(clicks) > 0, sum(cost_micros) / 1e6 / sum(clicks), 0), 2) AS cpc
+    cid         = creds["account_id"]
-        FROM manar.search_term_performance
+    today       = date.today()
-        WHERE brand_id = '{brand_slug}'
+    date_from   = today - timedelta(days=days)
-          AND date >= '{date_from}' AND date <= '{date_to}'
+
-        GROUP BY search_term
+    q = f"""SELECT search_term_view.search_term,
-        HAVING sum(cost_micros) >= {int(min_cost * 1_000_000)}
+                   metrics.impressions, metrics.clicks, metrics.cost_micros, metrics.conversions
-        ORDER BY cost DESC
+            FROM search_term_view
-        LIMIT {limit}
+            WHERE segments.date BETWEEN '{date_from}' AND '{today}'"""
-    """
+
-    result = state.ch_client.query(query)
+    terms = {}
-    return [dict(zip(result.column_names, row)) for row in result.result_rows]
+    for batch in ga.search_stream(customer_id=cid, query=q):
+        for row in batch.results:
+            t = row.search_term_view.search_term
+            if t not in terms:
+                terms[t] = {"search_term": t, "impressions": 0, "clicks": 0, "cost": 0.0, "conversions": 0.0}
+            terms[t]["impressions"] += row.metrics.impressions
+            terms[t]["clicks"]      += row.metrics.clicks
+            terms[t]["cost"]        += row.metrics.cost_micros / 1_000_000
+            terms[t]["conversions"] += row.metrics.conversions
+
+    result = sorted(terms.values(), key=lambda x: x["cost"], reverse=True)[:100]
+    for r in result:
+        r["cost"] = round(r["cost"], 2)
+        r["cpc"]  = round(r["cost"] / r["clicks"], 2) if r["clicks"] else 0
+    return result
 
 
 @mcp.tool()
-async def get_anomalies(brand_slug: str, actor_email: str, resolved: bool = False) -> list[dict]:
+async def get_account_summary(
-    """Get open anomaly events for a brand."""
+    brand_slug: str, actor_email: str,
+    days: int = 30, platform: str = "google_ads"
+) -> dict:
+    """Live — account summary with period-over-period comparison."""
     await assert_brand_access(brand_slug, actor_email)
-    resolved_filter = "IS NOT NULL" if resolved else "IS NULL"
+    creds = await get_brand_credentials(brand_slug, platform)
-    result = state.ch_client.query(f"""
+    if not creds.get("refresh_token"):
-        SELECT brand_id, account_id, campaign_id, metric,
+        return {"error": "not_connected", "brand": brand_slug}
-               detected_at, expected_value, actual_value, deviation_pct, severity
+
-        FROM manar.anomaly_events
+    client      = build_gads_client(creds["refresh_token"], creds.get("developer_token"))
-        WHERE brand_id = '{brand_slug}' AND resolved_at {resolved_filter}
+    ga          = client.get_service("GoogleAdsService")
-        ORDER BY detected_at DESC LIMIT 100
+    cid         = creds["account_id"]
-    """)
+    today       = date.today()
-    return [dict(zip(result.column_names, row)) for row in result.result_rows]
+
+    def fetch(d_from, d_to):
+        q = f"""SELECT metrics.impressions, metrics.clicks,
+                       metrics.cost_micros, metrics.conversions, metrics.conversions_value
+                FROM customer WHERE segments.date BETWEEN '{d_from}' AND '{d_to}'"""
+        t = {"impressions": 0, "clicks": 0, "cost": 0.0, "conversions": 0.0, "conv_value": 0.0}
+        for batch in ga.search_stream(customer_id=cid, query=q):
+            for row in batch.results:
+                t["impressions"] += row.metrics.impressions
+                t["clicks"]      += row.metrics.clicks
+                t["cost"]        += row.metrics.cost_micros / 1_000_000
+                t["conversions"] += row.metrics.conversions
+                t["conv_value"]  += row.metrics.conversions_value
+        t["cost"]    = round(t["cost"], 2)
+        t["ctr_pct"] = round(t["clicks"] / t["impressions"] * 100, 2) if t["impressions"] else 0
+        t["cpc"]     = round(t["cost"] / t["clicks"], 2)               if t["clicks"]      else 0
+        t["roas"]    = round(t["conv_value"] / t["cost"], 2)           if t["cost"]        else 0
+        return t
+
+    cur_from = today - timedelta(days=days)
+    pri_to   = cur_from - timedelta(days=1)
+    pri_from = pri_to   - timedelta(days=days)
+    current  = fetch(cur_from, today)
+    prior    = fetch(pri_from, pri_to)
+
+    def pct(a, b):
+        try: return round((a - b) / b * 100, 1) if b else None
+        except: return None
+
+    return {
+        "brand": brand_slug, "days": days, "platform": platform,
+        "current": current, "prior": prior,
+        "changes": {k: pct(current.get(k, 0), prior.get(k, 0))
+                    for k in ["impressions", "clicks", "cost", "conversions", "roas", "cpc"]}
+    }
 
 
 @mcp.tool()
@@ -203,21 +289,16 @@ async def create_recommendation(
     brand_slug: str, actor_email: str,
     rec_type: str, title: str, description: str,
     priority: str = "medium",
-    payload: dict = None,
+    payload: dict = None, estimated_impact: dict = None
-    estimated_impact: dict = None
 ) -> dict:
-    """
+    """Store a recommendation for human approval."""
-    Create a recommendation in the brand queue.
-    rec_type: budget_increase | keyword_add | negative_add | bid_adjust | pause_campaign
-    priority: low | medium | high | critical
-    """
     await assert_brand_access(brand_slug, actor_email)
     schema = f"brand_{brand_slug}"
     async with state.tenants_pool.acquire() as conn:
         row = await conn.fetchrow(
             f"""INSERT INTO {schema}.recommendations
                 (type, title, description, priority, payload, estimated_impact)
-                VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb)
+                VALUES ($1,$2,$3,$4,$5::jsonb,$6::jsonb)
                 RETURNING id::text, type, title, status, created_at::text""",
             rec_type, title, description, priority,
             json.dumps(payload or {}), json.dumps(estimated_impact or {}))
@@ -228,15 +309,13 @@ async def create_recommendation(
 async def list_recommendations(
     brand_slug: str, actor_email: str, status: str = "pending"
 ) -> list[dict]:
-    """List recommendations. status: pending | approved | rejected | executed | expired"""
+    """List recommendations. status: pending | approved | rejected | executed"""
     await assert_brand_access(brand_slug, actor_email)
     schema = f"brand_{brand_slug}"
     async with state.tenants_pool.acquire() as conn:
         rows = await conn.fetch(
-            f"""SELECT id::text, type, title, description, priority, status,
+            f"""SELECT id::text, type, title, description, priority, status, created_at::text
-                       estimated_impact::text, created_at::text
+                FROM {schema}.recommendations WHERE status=$1 ORDER BY created_at DESC""",
-                FROM {schema}.recommendations
-                WHERE status = $1 ORDER BY created_at DESC""",
             status)
     return [dict(r) for r in rows]
 
@@ -251,96 +330,18 @@ async def approve_recommendation(
     schema = f"brand_{brand_slug}"
     async with state.tenants_pool.acquire() as conn:
         await conn.execute(
-            f"UPDATE {schema}.recommendations SET status = $1, actioned_at = NOW() WHERE id = $2::uuid",
+            f"UPDATE {schema}.recommendations SET status=$1, actioned_at=NOW() WHERE id=$2::uuid",
             decision, recommendation_id)
         row = await conn.fetchrow(
             f"""INSERT INTO {schema}.approvals (recommendation_id, decision, notes)
-                VALUES ($1::uuid, $2, $3)
+                VALUES ($1::uuid,$2,$3) RETURNING id::text, decision, created_at::text""",
-                RETURNING id::text, decision, created_at::text""",
             recommendation_id, decision, notes)
     return dict(row)
 
 
-@mcp.tool()
-async def get_geo_performance(
-    brand_slug: str, actor_email: str,
-    date_from: str, date_to: str,
-    campaign_id: str = ""
-) -> list[dict]:
-    """Get geographic performance breakdown."""
-    await assert_brand_access(brand_slug, actor_email)
-    where_campaign = f"AND campaign_id = '{campaign_id}'" if campaign_id else ""
-    result = state.ch_client.query(f"""
-        SELECT country_code, region, city,
-            sum(impressions) AS impressions, sum(clicks) AS clicks,
-            round(sum(cost_micros) / 1e6, 2) AS cost, sum(conversions) AS conversions
-        FROM manar.geo_performance
-        WHERE brand_id = '{brand_slug}'
-          AND date >= '{date_from}' AND date <= '{date_to}'
-          {where_campaign}
-        GROUP BY country_code, region, city
-        ORDER BY cost DESC LIMIT 200
-    """)
-    return [dict(zip(result.column_names, row)) for row in result.result_rows]
-
-
-@mcp.tool()
-async def get_summary(brand_slug: str, actor_email: str, days: int = 30) -> dict:
-    """
-    Get high-level KPI summary for a brand over the last N days.
-    Returns totals + comparison to prior period.
-    """
-    await assert_brand_access(brand_slug, actor_email)
-
-    result = state.ch_client.query(f"""
-        SELECT
-            period,
-            sum(impressions)                                       AS impressions,
-            sum(clicks)                                            AS clicks,
-            round(sum(cost_micros) / 1e6, 2)                      AS cost,
-            round(sum(conversions), 1)                             AS conversions,
-            round(sum(conv_value), 2)                              AS conv_value,
-            round(avg(ctr) * 100, 2)                               AS ctr_pct,
-            round(if(sum(clicks)>0, sum(cost_micros)/1e6/sum(clicks), 0), 2) AS cpc,
-            round(if(sum(cost_micros)>0, sum(conv_value)/(sum(cost_micros)/1e6), 0), 2) AS roas
-        FROM (
-            SELECT
-                if(date >= today() - {days}, 'current', 'prior') AS period,
-                impressions, clicks, cost_micros, conversions, conv_value, ctr
-            FROM manar.ad_metrics_hourly
-            WHERE brand_id = '{brand_slug}'
-              AND date >= today() - {days * 2}
-        )
-        GROUP BY period
-    """)
-
-    data = {{}}
-    for row in result.result_rows:
-        d = dict(zip(result.column_names, row))
-        period = d.pop('period')
-        data[period] = d
-
-    current = data.get('current', {{}})
-    prior   = data.get('prior', {{}})
-
-    def pct_change(curr, prev):
-        if not prev or prev == 0:
-            return None
-        return round((curr - prev) / prev * 100, 1)
-
-    return {{
-        "brand": brand_slug,
-        "days": days,
-        "current": current,
-        "prior": prior,
-        "changes": {{
-            k: pct_change(current.get(k, 0), prior.get(k, 0))
-            for k in ['impressions', 'clicks', 'cost', 'conversions', 'roas', 'cpc']
-        }}
-    }}
-
-
 # ── Entry ─────────────────────────────────────────────────────────────────────
+
 if __name__ == "__main__":
     start_health_thread()
+    start_api_thread()
     mcp.run(transport="http", host="0.0.0.0", port=MCP_PORT)