mcp: full FastMCP server — brand-scoped tools, Google Ads, ClickHouse, recommendations

docker/mcp/main.py

@@ -5,7 +5,7 @@ All tools validate brand_membership before execution.
 """
 
 import os
-import asyncio
+import json
 import logging
 from contextlib import asynccontextmanager
 from typing import Optional
@@ -13,68 +13,70 @@ from typing import Optional
 import asyncpg
 import clickhouse_connect
 from fastmcp import FastMCP
-from fastmcp.server import Context
 from dotenv import load_dotenv
+from health import start_health_thread
 
 load_dotenv()
 
-logging.basicConfig(level=logging.INFO)
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
 log = logging.getLogger("manar-mcp")
 
-# ── Config ────────────────────────────────────────────────
 CORE_DB_URL    = os.environ["CORE_DB_URL"]
 TENANTS_DB_URL = os.environ["TENANTS_DB_URL"]
 CH_HOST        = os.environ.get("CLICKHOUSE_HOST", "clickhouse")
 CH_PORT        = int(os.environ.get("CLICKHOUSE_PORT", 8123))
 CH_DB          = os.environ.get("CLICKHOUSE_DB", "manar")
-MCP_PORT       = int(os.environ.get("MCP_PORT", 8000))
+MCP_PORT       = int(os.environ.get("MCP_PORT", 8090))
+
 
-# ── App State ─────────────────────────────────────────────
 class AppState:
     core_pool:    Optional[asyncpg.Pool] = None
     tenants_pool: Optional[asyncpg.Pool] = None
-    ch_client:    Optional[any]          = None
+    ch_client:    Optional[object]       = None
+
 
 state = AppState()
 
-# ── Lifespan ──────────────────────────────────────────────
+
 @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(host=CH_HOST, port=CH_PORT, database=CH_DB)
+    state.ch_client = clickhouse_connect.get_client(
-    log.info("Manar MCP ready.")
+        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.")
     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:
-    """
-    Validates that actor_email has membership in brand_slug.
-    Super admins bypass membership check.
-    Returns brand row on success, raises ValueError on failure.
-    """
     async with state.core_pool.acquire() as conn:
-        # Check super admin
         user = await conn.fetchrow(
-            "SELECT id, is_super_admin FROM users WHERE email = $1 AND is_active = true",
+            "SELECT id, role FROM users WHERE email = $1",
             actor_email
         )
         if not user:
-            raise ValueError(f"User {actor_email} not found or inactive")
+            raise ValueError(f"User {actor_email} not found")
 
         brand = await conn.fetchrow(
-            "SELECT id, slug, name, trust_level, is_active FROM brands WHERE slug = $1",
+            "SELECT id, slug, name, trust_level, status FROM brands WHERE slug = $1",
             brand_slug
         )
-        if not brand or not brand["is_active"]:
+        if not brand or brand["status"] != "active":
             raise ValueError(f"Brand '{brand_slug}' not found or inactive")
 
-        if not user["is_super_admin"]:
+        if user["role"] != "super_admin":
             membership = await conn.fetchrow(
                 "SELECT role FROM brand_memberships WHERE brand_id = $1 AND user_id = $2",
                 brand["id"], user["id"]
@@ -84,29 +86,22 @@ async def assert_brand_access(brand_slug: str, actor_email: str) -> dict:
 
         return dict(brand)
 
-# ── Health ────────────────────────────────────────────────
-@mcp.resource("health://status")
-async def health_status() -> str:
-    return "ok"
 
-# ── Tool: list_brands ─────────────────────────────────────
+# ── Tools ─────────────────────────────────────────────────────────────────────
+
 @mcp.tool()
 async def list_brands(actor_email: str) -> list[dict]:
-    """
+    """List all active brands accessible to actor_email."""
-    List all active brands accessible to actor_email.
-    Super admins see all brands. Regular users see their memberships only.
-    """
     async with state.core_pool.acquire() as conn:
         user = await conn.fetchrow(
-            "SELECT id, is_super_admin FROM users WHERE email = $1 AND is_active = true",
+            "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":
-        if user["is_super_admin"]:
             rows = await conn.fetch(
-                "SELECT slug, name, trust_level, timezone FROM brands WHERE is_active = true ORDER BY name"
+                "SELECT slug, name, trust_level, timezone FROM brands WHERE status = 'active' ORDER BY name"
             )
         else:
             rows = await conn.fetch(
@@ -114,145 +109,128 @@ async def list_brands(actor_email: str) -> list[dict]:
                    FROM brands b
                    JOIN brand_memberships m ON m.brand_id = b.id
                    JOIN users u ON u.id = m.user_id
-                   WHERE u.email = $1 AND b.is_active = true
+                   WHERE u.email = $1 AND b.status = 'active' ORDER BY b.name""",
-                   ORDER BY b.name""",
                 actor_email
             )
         return [dict(r) for r in rows]
 
-# ── Tool: get_ad_accounts ─────────────────────────────────
+
 @mcp.tool()
 async def get_ad_accounts(brand_slug: str, actor_email: str) -> list[dict]:
-    """
+    """Get all ad accounts for a brand."""
-    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, platform_slug, external_id, name, currency, is_active, last_synced_at "
+            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]
 
-# ── Tool: fetch_google_ads_metrics ────────────────────────
+
 @mcp.tool()
 async def fetch_google_ads_metrics(
-    brand_slug:   str,
+    brand_slug: str, actor_email: str,
-    actor_email:  str,
+    date_from: str, date_to: str,
-    date_from:    str,
+    breakdown: str = "campaign"
-    date_to:      str,
-    breakdown:    str = "campaign",  # campaign | adgroup | day
 ) -> list[dict]:
     """
-    Fetch performance metrics from ClickHouse for a brand.
+    Fetch performance metrics from ClickHouse.
-    Returns aggregated impressions, clicks, cost, conversions, ROAS.
     breakdown: campaign | adgroup | day
+    date_from / date_to: YYYY-MM-DD
     """
     await assert_brand_access(brand_slug, actor_email)
 
     group_fields = {
         "campaign": "campaign_id, campaign_name",
-        "adgroup":  "campaign_id, campaign_name, adgroup_id, adgroup_name",
+        "adgroup":  "campaign_id, campaign_name, ad_group_id, ad_group_name",
-        "day":      "date",
+        "day":      "toDate(hour) AS date",
     }.get(breakdown, "campaign_id, campaign_name")
 
     query = f"""
-        SELECT
+        SELECT {group_fields},
-            {group_fields},
             sum(impressions)  AS impressions,
             sum(clicks)       AS clicks,
-            sum(cost_micros)  AS cost_micros,
+            round(sum(cost_micros) / 1e6, 2) AS cost,
             sum(conversions)  AS conversions,
-            sum(conv_value)   AS conv_value,
+            sum(conversion_value) AS conv_value,
-            if(sum(impressions) > 0, sum(clicks) / sum(impressions), 0) AS ctr,
+            round(if(sum(impressions) > 0, sum(clicks) / sum(impressions), 0), 4) AS ctr,
-            if(sum(clicks) > 0, sum(cost_micros) / sum(clicks), 0) AS cpc_micros,
+            round(if(sum(clicks) > 0, sum(cost_micros) / 1e6 / sum(clicks), 0), 2) AS cpc,
-            if(sum(cost_micros) > 0,
+            round(if(sum(cost_micros) > 0, sum(conversion_value) / (sum(cost_micros) / 1e6), 0), 2) AS roas
-               sum(conv_value) / (sum(cost_micros) / 1000000), 0) AS roas
         FROM manar.ad_metrics_hourly
         WHERE brand_id = '{brand_slug}'
-          AND platform = 'google_ads'
+          AND platform_slug = 'google_ads'
-          AND date >= '{date_from}'
+          AND toDate(hour) >= '{date_from}'
-          AND date <= '{date_to}'
+          AND toDate(hour) <= '{date_to}'
         GROUP BY {group_fields}
-        ORDER BY cost_micros DESC
+        ORDER BY cost DESC
         LIMIT 500
     """
     result = state.ch_client.query(query)
     return [dict(zip(result.column_names, row)) for row in result.result_rows]
 
-# ── Tool: get_search_terms ────────────────────────────────
+
 @mcp.tool()
 async def get_search_terms(
-    brand_slug:  str,
+    brand_slug: str, actor_email: str,
-    actor_email: str,
+    date_from: str, date_to: str,
-    date_from:   str,
+    min_cost: float = 0,
-    date_to:     str,
+    limit: int = 200
-    min_cost:    float = 0,
-    limit:       int   = 200,
 ) -> list[dict]:
-    """
+    """Get search term performance for keyword expansion / negative keyword decisions."""
-    Get search term performance. Used for keyword expansion / negative keyword decisions.
-    """
     await assert_brand_access(brand_slug, actor_email)
     query = f"""
-        SELECT
+        SELECT search_term,
-            search_term,
             sum(impressions) AS impressions,
-            sum(clicks) AS clicks,
+            sum(clicks)      AS clicks,
-            sum(cost_micros) AS cost_micros,
+            round(sum(cost_micros) / 1e6, 2) AS cost,
             sum(conversions) AS conversions,
-            if(sum(clicks) > 0, sum(cost_micros) / sum(clicks), 0) AS cpc_micros
+            round(if(sum(clicks) > 0, sum(cost_micros) / 1e6 / sum(clicks), 0), 2) AS cpc
         FROM manar.search_term_performance
         WHERE brand_id = '{brand_slug}'
-          AND date >= '{date_from}'
+          AND date >= '{date_from}' AND date <= '{date_to}'
-          AND date <= '{date_to}'
         GROUP BY search_term
         HAVING sum(cost_micros) >= {int(min_cost * 1_000_000)}
-        ORDER BY cost_micros DESC
+        ORDER BY cost DESC
         LIMIT {limit}
     """
     result = state.ch_client.query(query)
     return [dict(zip(result.column_names, row)) for row in result.result_rows]
 
-# ── Tool: get_anomalies ───────────────────────────────────
+
 @mcp.tool()
 async def get_anomalies(
-    brand_slug:  str,
+    brand_slug: str, actor_email: str,
-    actor_email: str,
+    resolved: bool = False
-    resolved:    bool = False,
 ) -> list[dict]:
-    """
+    """Get open anomaly events for a brand."""
-    Get open anomaly events for a brand.
-    """
     await assert_brand_access(brand_slug, actor_email)
+    resolved_filter = "IS NOT NULL" if resolved else "IS NULL"
     query = f"""
-        SELECT brand_id, platform, entity_type, entity_id, metric,
+        SELECT brand_id, ad_account_id, campaign_id, metric,
                detected_at, expected_value, actual_value, deviation_pct, severity
         FROM manar.anomaly_events
         WHERE brand_id = '{brand_slug}'
-          AND resolved = {1 if resolved else 0}
+          AND resolved_at {resolved_filter}
         ORDER BY detected_at DESC
         LIMIT 100
     """
     result = state.ch_client.query(query)
     return [dict(zip(result.column_names, row)) for row in result.result_rows]
 
-# ── Tool: create_recommendation ──────────────────────────
+
 @mcp.tool()
 async def create_recommendation(
-    brand_slug:       str,
+    brand_slug: str, actor_email: str,
-    actor_email:      str,
+    rec_type: str, title: str, description: str,
-    rec_type:         str,
+    priority: str = "medium",
-    title:            str,
+    payload: dict = None,
-    description:      str,
+    estimated_impact: dict = None
-    priority:         int  = 2,
-    payload:          dict = None,
-    estimated_impact: dict = None,
 ) -> dict:
     """
-    Create a recommendation in the brand's tenants DB.
+    Create a recommendation in the brand queue.
-    rec_type: budget_increase | keyword_add | negative_add | bid_adjust | pause_campaign | etc.
+    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}"
@@ -261,36 +239,121 @@ async def create_recommendation(
             f"""INSERT INTO {schema}.recommendations
                 (type, title, description, priority, payload, estimated_impact)
                 VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb)
-                RETURNING id, type, title, status, created_at""",
+                RETURNING id::text, type, title, status, created_at::text""",
             rec_type, title, description, priority,
-            str(payload or {}), str(estimated_impact or {})
+            json.dumps(payload or {}),
+            json.dumps(estimated_impact or {})
         )
     return dict(row)
 
-# ── Tool: list_recommendations ────────────────────────────
+
 @mcp.tool()
 async def list_recommendations(
-    brand_slug:  str,
+    brand_slug: str, actor_email: str,
-    actor_email: str,
+    status: str = "pending"
-    status:      str = "pending",
 ) -> list[dict]:
     """
-    List recommendations for a brand, filtered by status.
+    List recommendations for a brand.
-    status: pending | approved | rejected | executed | vetoed
+    status: pending | approved | rejected | executed | expired
     """
     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, type, title, description, priority, status, estimated_impact, created_at
+            f"""SELECT id::text, type, title, description, priority, status,
+                       estimated_impact::text, created_at::text
                 FROM {schema}.recommendations
                 WHERE status = $1
-                ORDER BY priority DESC, created_at DESC""",
+                ORDER BY created_at DESC""",
             status
         )
     return [dict(r) for r in rows]
 
-# ── Entry Point ───────────────────────────────────────────
+
+@mcp.tool()
+async def approve_recommendation(
+    brand_slug: str, actor_email: str,
+    recommendation_id: str,
+    decision: str,
+    notes: str = ""
+) -> dict:
+    """
+    Approve or reject a recommendation.
+    decision: approved | rejected | vetoed
+    """
+    await assert_brand_access(brand_slug, actor_email)
+    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""",
+            decision, recommendation_id
+        )
+        row = await conn.fetchrow(
+            f"""INSERT INTO {schema}.approvals
+                (recommendation_id, decision, notes)
+                VALUES ($1::uuid, $2, $3)
+                RETURNING id::text, decision, created_at::text""",
+            recommendation_id, decision, notes
+        )
+    return dict(row)
+
+
+@mcp.tool()
+async def get_day_part_performance(
+    brand_slug: str, actor_email: str,
+    campaign_id: str,
+    date_from: str, date_to: str
+) -> list[dict]:
+    """Get hourly/day-of-week performance breakdown for a campaign."""
+    await assert_brand_access(brand_slug, actor_email)
+    query = f"""
+        SELECT hour_of_day, day_of_week,
+            sum(impressions) AS impressions,
+            sum(clicks)      AS clicks,
+            round(sum(cost_micros) / 1e6, 2) AS cost,
+            sum(conversions) AS conversions,
+            round(if(sum(impressions) > 0, sum(clicks) / sum(impressions), 0), 4) AS ctr
+        FROM manar.day_part_performance
+        WHERE brand_id = '{brand_slug}'
+          AND campaign_id = '{campaign_id}'
+          AND date >= '{date_from}' AND date <= '{date_to}'
+        GROUP BY hour_of_day, day_of_week
+        ORDER BY day_of_week, hour_of_day
+    """
+    result = state.ch_client.query(query)
+    return [dict(zip(result.column_names, row)) for row in result.result_rows]
+
+
+@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 ""
+    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
+    """
+    result = state.ch_client.query(query)
+    return [dict(zip(result.column_names, row)) for row in result.result_rows]
+
+
+# ── Entry ─────────────────────────────────────────────────────────────────────
 if __name__ == "__main__":
-    import uvicorn
+    start_health_thread()
     mcp.run(transport="http", host="0.0.0.0", port=MCP_PORT)