colorlink/manar
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

mcp: fix ClickHouse column names, add get_summary tool with period comparison

Browse Source
This commit is contained in:
colorlink
2026-05-06 18:18:44 +00:00
parent 759ce61e3a
commit b58b69bc3b
1 changed files with 94 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker/mcp/main.py
+94 -107
View File
@@ -63,27 +63,19 @@ mcp = FastMCP("manar-mcp", lifespan=lifespan)
async def assert_brand_access(brand_slug: str, actor_email: str) -> dict:
async with state.core_pool.acquire() as conn:
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(
"SELECT id, slug, name, trust_level, status FROM brands WHERE slug = $1",
brand_slug
)
"SELECT id, slug, name, trust_level, status FROM brands WHERE slug = $1", brand_slug)
if not brand or brand["status"] != "active":
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",
brand["id"], user["id"]
)
brand["id"], user["id"])
if not membership:
raise ValueError(f"User {actor_email} has no access to brand '{brand_slug}'")
return dict(brand)
@@ -93,25 +85,19 @@ async def assert_brand_access(brand_slug: str, actor_email: str) -> dict:
async def list_brands(actor_email: str) -> list[dict]:
"""List all active brands accessible to actor_email."""
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
FROM brands b
JOIN brand_memberships m ON m.brand_id = b.id
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.status = 'active' ORDER BY b.name""",
actor_email
)
actor_email)
return [dict(r) for r in rows]
@@ -123,8 +109,7 @@ async def get_ad_accounts(brand_slug: str, actor_email: str) -> list[dict]:
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"
)
f"FROM {schema}.ad_accounts ORDER BY platform_slug, name")
return [dict(r) for r in rows]
@@ -143,8 +128,8 @@ async def fetch_google_ads_metrics(
group_fields = {
"campaign": "campaign_id, campaign_name",
"adgroup": "campaign_id, campaign_name, ad_group_id, ad_group_name",
"day": "toDate(hour) AS date",
"adgroup": "campaign_id, campaign_name, adgroup_id, adgroup_name",
"day": "date",
}.get(breakdown, "campaign_id, campaign_name")
query = f"""
@@ -153,15 +138,15 @@ async def fetch_google_ads_metrics(
sum(clicks) AS clicks,
round(sum(cost_micros) / 1e6, 2) AS cost,
sum(conversions) AS conversions,
sum(conversion_value) AS conv_value,
sum(conv_value) AS conv_value,
round(if(sum(impressions) > 0, sum(clicks) / sum(impressions), 0), 4) AS ctr,
round(if(sum(clicks) > 0, sum(cost_micros) / 1e6 / sum(clicks), 0), 2) AS cpc,
round(if(sum(cost_micros) > 0, sum(conversion_value) / (sum(cost_micros) / 1e6), 0), 2) AS roas
round(if(sum(cost_micros) > 0, sum(conv_value) / (sum(cost_micros) / 1e6), 0), 2) AS roas
FROM manar.ad_metrics_hourly
WHERE brand_id = '{brand_slug}'
AND platform_slug = 'google_ads'
AND toDate(hour) >= '{date_from}'
AND toDate(hour) <= '{date_to}'
AND platform = 'google_ads'
AND date >= '{date_from}'
AND date <= '{date_to}'
GROUP BY {group_fields}
ORDER BY cost DESC
LIMIT 500
@@ -177,7 +162,7 @@ async def get_search_terms(
min_cost: float = 0,
limit: int = 200
) -> list[dict]:
"""Get search term performance for keyword expansion / negative keyword decisions."""
"""Get search term performance — for keyword expansion / negative keyword decisions."""
await assert_brand_access(brand_slug, actor_email)
query = f"""
SELECT search_term,
@@ -199,23 +184,17 @@ async def get_search_terms(
@mcp.tool()
async def get_anomalies(
brand_slug: str, actor_email: str,
resolved: bool = False
) -> list[dict]:
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)
resolved_filter = "IS NOT NULL" if resolved else "IS NULL"
query = f"""
SELECT brand_id, ad_account_id, campaign_id, metric,
result = state.ch_client.query(f"""
SELECT brand_id, 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_at {resolved_filter}
ORDER BY detected_at DESC
LIMIT 100
"""
result = state.ch_client.query(query)
WHERE brand_id = '{brand_slug}' AND resolved_at {resolved_filter}
ORDER BY detected_at DESC LIMIT 100
""")
return [dict(zip(result.column_names, row)) for row in result.result_rows]
@@ -241,21 +220,15 @@ async def create_recommendation(
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 {})
)
json.dumps(payload or {}), json.dumps(estimated_impact or {}))
return dict(row)
@mcp.tool()
async def list_recommendations(
brand_slug: str, actor_email: str,
status: str = "pending"
brand_slug: str, actor_email: str, status: str = "pending"
) -> list[dict]:
"""
List recommendations for a brand.
status: pending | approved | rejected | executed | expired
"""
"""List recommendations. 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:
@@ -263,69 +236,31 @@ async def list_recommendations(
f"""SELECT id::text, type, title, description, priority, status,
estimated_impact::text, created_at::text
FROM {schema}.recommendations
WHERE status = $1
ORDER BY created_at DESC""",
status
)
WHERE status = $1 ORDER BY created_at DESC""",
status)
return [dict(r) for r in rows]
@mcp.tool()
async def approve_recommendation(
brand_slug: str, actor_email: str,
recommendation_id: str,
decision: str,
notes: str = ""
recommendation_id: str, decision: str, notes: str = ""
) -> dict:
"""
Approve or reject a recommendation.
decision: approved | rejected | vetoed
"""
"""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
)
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)
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
)
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,
@@ -335,24 +270,76 @@ async def get_geo_performance(
"""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"""
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
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)
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()