mcp: core server — brand-scoped tools for Google Ads + recommendations

docker/mcp/main.py

@@ -0,0 +1,296 @@
+"""
+Manar MCP Server — Google Ads Connector
+FastMCP server exposing brand-scoped Google Ads tools.
+All tools validate brand_membership before execution.
+"""
+
+import os
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+from typing import Optional
+
+import asyncpg
+import clickhouse_connect
+from fastmcp import FastMCP
+from fastmcp.server import Context
+from dotenv import load_dotenv
+
+load_dotenv()
+
+logging.basicConfig(level=logging.INFO)
+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))
+
+# ── App State ─────────────────────────────────────────────
+class AppState:
+    core_pool:    Optional[asyncpg.Pool] = None
+    tenants_pool: Optional[asyncpg.Pool] = None
+    ch_client:    Optional[any]          = 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.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)
+    log.info("Manar MCP ready.")
+    yield
+    await state.core_pool.close()
+    await state.tenants_pool.close()
+
+mcp = FastMCP("manar-mcp", lifespan=lifespan)
+
+# ── 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",
+            actor_email
+        )
+        if not user:
+            raise ValueError(f"User {actor_email} not found or inactive")
+
+        brand = await conn.fetchrow(
+            "SELECT id, slug, name, trust_level, is_active FROM brands WHERE slug = $1",
+            brand_slug
+        )
+        if not brand or not brand["is_active"]:
+            raise ValueError(f"Brand '{brand_slug}' not found or inactive")
+
+        if not user["is_super_admin"]:
+            membership = await conn.fetchrow(
+                "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}'")
+
+        return dict(brand)
+
+# ── Health ────────────────────────────────────────────────
+@mcp.resource("health://status")
+async def health_status() -> str:
+    return "ok"
+
+# ── Tool: list_brands ─────────────────────────────────────
+@mcp.tool()
+async def list_brands(actor_email: str) -> list[dict]:
+    """
+    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",
+            actor_email
+        )
+        if not user:
+            raise ValueError(f"User {actor_email} not found")
+
+        if user["is_super_admin"]:
+            rows = await conn.fetch(
+                "SELECT slug, name, trust_level, timezone FROM brands WHERE is_active = true ORDER BY name"
+            )
+        else:
+            rows = await conn.fetch(
+                """SELECT b.slug, b.name, b.trust_level, b.timezone, m.role
+                   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
+                   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.
+    """
+    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"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,
+    actor_email:  str,
+    date_from:    str,
+    date_to:      str,
+    breakdown:    str = "campaign",  # campaign | adgroup | day
+) -> list[dict]:
+    """
+    Fetch performance metrics from ClickHouse for a brand.
+    Returns aggregated impressions, clicks, cost, conversions, ROAS.
+    breakdown: campaign | adgroup | day
+    """
+    await assert_brand_access(brand_slug, actor_email)
+
+    group_fields = {
+        "campaign": "campaign_id, campaign_name",
+        "adgroup":  "campaign_id, campaign_name, adgroup_id, adgroup_name",
+        "day":      "date",
+    }.get(breakdown, "campaign_id, campaign_name")
+
+    query = f"""
+        SELECT
+            {group_fields},
+            sum(impressions)  AS impressions,
+            sum(clicks)       AS clicks,
+            sum(cost_micros)  AS cost_micros,
+            sum(conversions)  AS conversions,
+            sum(conv_value)   AS conv_value,
+            if(sum(impressions) > 0, sum(clicks) / sum(impressions), 0) AS ctr,
+            if(sum(clicks) > 0, sum(cost_micros) / sum(clicks), 0) AS cpc_micros,
+            if(sum(cost_micros) > 0,
+               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 date >= '{date_from}'
+          AND date <= '{date_to}'
+        GROUP BY {group_fields}
+        ORDER BY cost_micros 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,
+    actor_email: str,
+    date_from:   str,
+    date_to:     str,
+    min_cost:    float = 0,
+    limit:       int   = 200,
+) -> list[dict]:
+    """
+    Get search term performance. Used for keyword expansion / negative keyword decisions.
+    """
+    await assert_brand_access(brand_slug, actor_email)
+    query = f"""
+        SELECT
+            search_term,
+            sum(impressions) AS impressions,
+            sum(clicks) AS clicks,
+            sum(cost_micros) AS cost_micros,
+            sum(conversions) AS conversions,
+            if(sum(clicks) > 0, sum(cost_micros) / sum(clicks), 0) AS cpc_micros
+        FROM manar.search_term_performance
+        WHERE brand_id = '{brand_slug}'
+          AND date >= '{date_from}'
+          AND date <= '{date_to}'
+        GROUP BY search_term
+        HAVING sum(cost_micros) >= {int(min_cost * 1_000_000)}
+        ORDER BY cost_micros 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,
+    actor_email: str,
+    resolved:    bool = False,
+) -> list[dict]:
+    """
+    Get open anomaly events for a brand.
+    """
+    await assert_brand_access(brand_slug, actor_email)
+    query = f"""
+        SELECT brand_id, platform, entity_type, entity_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}
+        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,
+    actor_email:      str,
+    rec_type:         str,
+    title:            str,
+    description:      str,
+    priority:         int  = 2,
+    payload:          dict = None,
+    estimated_impact: dict = None,
+) -> dict:
+    """
+    Create a recommendation in the brand's tenants DB.
+    rec_type: budget_increase | keyword_add | negative_add | bid_adjust | pause_campaign | etc.
+    """
+    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)
+                RETURNING id, type, title, status, created_at""",
+            rec_type, title, description, priority,
+            str(payload or {}), str(estimated_impact or {})
+        )
+    return dict(row)
+
+# ── Tool: list_recommendations ────────────────────────────
+@mcp.tool()
+async def list_recommendations(
+    brand_slug:  str,
+    actor_email: str,
+    status:      str = "pending",
+) -> list[dict]:
+    """
+    List recommendations for a brand, filtered by status.
+    status: pending | approved | rejected | executed | vetoed
+    """
+    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
+                FROM {schema}.recommendations
+                WHERE status = $1
+                ORDER BY priority DESC, created_at DESC""",
+            status
+        )
+    return [dict(r) for r in rows]
+
+# ── Entry Point ───────────────────────────────────────────
+if __name__ == "__main__":
+    import uvicorn
+    mcp.run(transport="http", host="0.0.0.0", port=MCP_PORT)