app/mcp/server.py

"""T33 MCP Server — exposes 8 tools to AI agents at mcp.rugmunch.io.

Per v4.0 §T33. JSON-RPC over SSE (the protocol Claude/Cursor speak).

Tools (per v4.0):
  1. get_token_risk        — Real-time risk score (FREE 5/day or $0.01)
  2. get_wallet_analysis   — Wallet activity + reputation
  3. get_deployer_reputation — Deployer reputation (0-100)
  4. get_news_sentiment    — Latest news + sentiment
  5. generate_report       — Full AI research report ($5)
  6. query_catalog         — Natural language catalog query
  7. find_similar_tokens   — Vector-similar tokens
  8. resolve_entity        — Cross-chain entity resolution

Backend implementations: app/catalog/* + app/domain/reports/generator.py
"""
from __future__ import annotations

import logging
from typing import Any

log = logging.getLogger(__name__)

# Tool catalog — inputSchema follows JSON Schema 2020-12
TOOL_CATALOG: list[dict[str, Any]] = [
    {
        "name": "get_token_risk",
        "description": "Real-time risk score for any token across 13+ chains. Returns score (0-100), tier (low/medium/high/critical), and risk factors. Free tier: 5 calls/day, $0.01 thereafter.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "chain": {"type": "string", "enum": ["solana", "ethereum", "base", "arbitrum", "optimism", "polygon", "bsc", "tron", "bitcoin", "avalanche", "fantom", "gnosis"]},
                "address": {"type": "string", "description": "Token contract address"},
            },
            "required": ["chain", "address"],
        },
    },
    {
        "name": "get_wallet_analysis",
        "description": "Wallet activity, balance, transaction history, and reputation. Returns wallet profile + risk flags.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "chain": {"type": "string"},
                "address": {"type": "string"},
            },
            "required": ["chain", "address"],
        },
    },
    {
        "name": "get_deployer_reputation",
        "description": "Deployer reputation score 0-100 (100=clean, 0=serial rugger). Deterministic from on-chain history + news + RAG findings. Cached 1h.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "chain": {"type": "string"},
                "address": {"type": "string"},
            },
            "required": ["chain", "address"],
        },
    },
    {
        "name": "get_news_sentiment",
        "description": "Latest news for a token or wallet with sentiment classification. Returns articles + composite sentiment score.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "subject_id": {"type": "string", "description": "chain:address, or 'all' for general news"},
                "since_hours": {"type": "integer", "default": 24, "minimum": 1, "maximum": 720},
                "limit": {"type": "integer", "default": 10, "minimum": 1, "maximum": 50},
            },
        },
    },
    {
        "name": "generate_report",
        "description": "Full AI research report on a token or wallet. 7 sections composed in parallel via LLM. $5/report. Returns full Markdown.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "subject_type": {"type": "string", "enum": ["token", "wallet"]},
                "subject_id": {"type": "string", "description": "chain:address"},
            },
            "required": ["subject_type", "subject_id"],
        },
    },
    {
        "name": "query_catalog",
        "description": "Natural language catalog query. Returns matching tokens, wallets, deployers, news, RAG findings. $0.05/query.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Natural language question"},
            },
            "required": ["query"],
        },
    },
    {
        "name": "find_similar_tokens",
        "description": "Vector-similar tokens to a given token. Returns tokens with cosine similarity >= 0.85. $0.03/query.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "chain": {"type": "string"},
                "address": {"type": "string"},
                "limit": {"type": "integer", "default": 10, "maximum": 50},
            },
            "required": ["chain", "address"],
        },
    },
    {
        "name": "resolve_entity",
        "description": "Cross-chain entity resolution. Given a wallet, find all linked wallets across chains via SAME_AS / FUNDED_BY_SAME / CLONE_OF / BEHAVIORAL_MATCH edges. $0.10/query.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "wallet_id": {"type": "string", "description": "chain:address"},
            },
            "required": ["wallet_id"],
        },
    },
    # ── TIER 1 moat tools (June 23 2026) ─────────────────────────
    {
        "name": "analytics_query",
        "description": "Run a read-only SQL query against the embedded DuckDB analytics engine. Use for sub-1GB analytical queries (counts, aggregations, joins). For larger queries, use ClickHouse directly. Max 10K rows returned.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "sql": {"type": "string", "description": "SQL query (SELECT only, no writes)"},
                "params": {"type": "array", "items": {}, "default": []},
                "max_rows": {"type": "integer", "default": 1000, "maximum": 10000},
            },
            "required": ["sql"],
        },
    },
    {
        "name": "mcp_discover",
        "description": "Discover all available MCP tools with versioning, deprecation status, and schema introspection. Returns the full tool catalog with auth requirements.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "include_deprecated": {"type": "boolean", "default": False},
                "category": {"type": "string", "description": "filter by category (free, pro, enterprise, moat)"},
            },
        },
    },
    {
        "name": "status_check",
        "description": "Health check across all RMI subsystems (backend, postgres, clickhouse, qdrant, redis, minio, reth, mcp server, x402, certstream, glitchtip). Returns pass/fail/degraded for each with latency.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "include_metrics": {"type": "boolean", "default": True},
            },
        },
    },
]


# ── Tool versioning + deprecation registry ───────────────────────
TOOL_VERSIONS: dict[str, str] = {
    "get_token_risk":         "1.2.0",
    "get_wallet_analysis":    "1.1.0",
    "get_deployer_reputation": "2.0.0",  # M3 — Bayesian posterior
    "get_news_sentiment":     "1.0.0",
    "generate_report":        "2.0.0",  # M3 — RAG-grounded
    "query_catalog":          "1.0.0",
    "find_similar_tokens":    "1.0.0",
    "resolve_entity":         "1.0.0",
    "analytics_query":        "1.0.0",  # M3 moat TIER 1
    "mcp_discover":           "1.0.0",  # M3 moat TIER 1
    "status_check":           "1.0.0",  # M3 moat TIER 1
}

TOOL_DEPRECATED: set[str] = set()  # empty — no deprecated tools yet
TOOL_SUCCESSORS: dict[str, str] = {}  # empty — no successor mappings yet

# Server version (single source of truth for /mcp/info)
MCP_SERVER_VERSION = "5.0.0"
MCP_PROTOCOL_VERSION = "2024-11-05"


# ── Tool implementations ──────────────────────────────────────────
async def call_tool(name: str, arguments: dict) -> dict:
    """Dispatch a tool call to the appropriate backend."""
    from app.catalog.service import get_catalog

    catalog = get_catalog()
    await catalog._init_stores()

    if name == "get_token_risk":
        from app.catalog.models import Chain
        try:
            c = Chain(arguments["chain"])
        except ValueError:
            return {"error": f"unknown chain: {arguments['chain']}"}
        result = await catalog.get_token_risk(c, arguments["address"])
        return {"result": result, "tier": "free_or_pro"}

    if name == "get_wallet_analysis":
        from app.catalog.models import Chain
        try:
            c = Chain(arguments["chain"])
        except ValueError:
            return {"error": f"unknown chain: {arguments['chain']}"}
        w = await catalog.get_wallet(c, arguments["address"])
        if not w:
            return {"error": "wallet not found in catalog"}
        return {"result": w.model_dump(mode="json")}

    if name == "get_deployer_reputation":
        from app.catalog.models import Chain
        try:
            c = Chain(arguments["chain"])
        except ValueError:
            return {"error": f"unknown chain: {arguments['chain']}"}
        w = await catalog.get_wallet(c, arguments["address"])
        if not w:
            return {"error": "deployer wallet not found", "reputation_score": 50}
        # Compute reputation deterministically
        from app.catalog.models import Deployer
        from app.catalog.reputation import compute_deployer_reputation
        deployer = Deployer(
            wallet_id=w.wallet_id, chain=w.chain, address=w.address,
            first_seen=w.first_seen, last_seen=w.last_seen,
            tx_count=w.tx_count, total_volume_usd=w.total_volume_usd,
            is_deployer=True, reputation_score=w.reputation_score,
            deployments=getattr(w, "deployments", []),
            rug_count=getattr(w, "rug_count", 0),
        )
        score = await compute_deployer_reputation(deployer, catalog)
        return {"result": {"reputation_score": score, "tier": _tier_from_score(score)}}

    if name == "get_news_sentiment":
        subject = arguments.get("subject_id", "all")
        since = int(arguments.get("since_hours", 24))
        limit = int(arguments.get("limit", 10))
        if not catalog._health.postgres:
            return {"error": "postgres unavailable", "articles": []}
        try:
            async with catalog._pg_pool.acquire() as conn:
                rows = await conn.fetch(
                    """SELECT news_id, title, summary, source, published_at, sentiment_score
                       FROM news_items
                       WHERE published_at > NOW() - ($1 || ' hours')::interval
                       ORDER BY published_at DESC LIMIT $2""",
                    str(since), limit,
                )
            articles = [
                {
                    "news_id": r["news_id"],
                    "title": r["title"],
                    "summary": (r["summary"] or "")[:200],
                    "source": r["source"],
                    "published_at": r["published_at"].isoformat(),
                    "sentiment_score": r["sentiment_score"],
                }
                for r in rows
            ]
            avg_sent = sum(a["sentiment_score"] or 0 for a in articles) / max(1, len(articles))
            return {"result": {
                "subject": subject,
                "article_count": len(articles),
                "avg_sentiment": round(avg_sent, 3),
                "articles": articles,
            }}
        except Exception as e:
            return {"error": f"news_query_fail: {e}"}

    if name == "generate_report":
        from app.domain.reports.generator import generate_token_report, generate_wallet_report
        chain, address = arguments["subject_id"].split(":", 1)
        try:
            if arguments["subject_type"] == "token":
                report = await generate_token_report(catalog, chain, address)
            else:
                report = await generate_wallet_report(catalog, chain, address)
            from app.domain.reports.generator import save_report
            await save_report(catalog, report)
            return {"result": {
                "report_id": report.report_id,
                "risk_score": report.risk_score,
                "risk_tier": report.risk_tier.value,
                "markdown": report.to_markdown(),
                "paid_via_x402": None,  # MCP doesn't enforce payment in v1
            }}
        except Exception as e:
            return {"error": f"report_fail: {e}"}

    if name == "query_catalog":
        # NL query -> RAG search
        q = arguments.get("query", "")
        hits = await catalog.rag_search(query=q, top_k=5)
        return {"result": {"query": q, "hits": hits, "count": len(hits)}}

    if name == "find_similar_tokens":
        from app.catalog.models import Chain
        try:
            c = Chain(arguments["chain"])
        except ValueError:
            return {"error": f"unknown chain: {arguments['chain']}"}
        # Use token's rag_embedding_id to find similar via Qdrant
        token = await catalog.get_token(c, arguments["address"])
        if not token or not token.rag_embedding_id:
            return {"error": "token not in catalog or no RAG embedding", "similar": []}
        # Use RAG to search for similar by querying with the token's content
        rag_hits = await catalog.rag_search(query=token.symbol or "token", top_k=int(arguments.get("limit", 10)))
        return {"result": {"subject": arguments["address"], "similar": rag_hits[:10]}}

    if name == "resolve_entity":
        result = await catalog.resolve_entity(arguments["wallet_id"])
        return {"result": result}

    # ── TIER 1 moat tools (June 23 2026) ─────────────────────────
    if name == "analytics_query":
        # T13: Run read-only SQL via embedded DuckDB
        # TODO: M3 moat TIER 2 — add API key check before opening to public
        from app.core.duckdb_analytics import DuckDBAnalytics
        sql = arguments.get("sql", "").strip()
        if not sql:
            return {"error": "sql parameter required"}
        # Safety: only allow SELECT / WITH statements, no writes
        sql_upper = sql.upper().lstrip()
        if not (sql_upper.startswith("SELECT") or sql_upper.startswith("WITH") or sql_upper.startswith("SHOW") or sql_upper.startswith("DESCRIBE")):
            return {"error": "only SELECT/WITH/SHOW/DESCRIBE queries are allowed"}
        max_rows = min(int(arguments.get("max_rows", 1000)), 10000)
        params = arguments.get("params", [])
        try:
            d = DuckDBAnalytics()
            rows = d.query(sql, params=params, max_rows=max_rows)
            return {"result": {"rows": rows, "count": len(rows), "truncated": len(rows) >= max_rows}, "engine": "duckdb"}
        except Exception as exc:
            return {"error": f"duckdb query failed: {exc}"}

    if name == "mcp_discover":
        # MCP catalog discovery with versioning + deprecation
        include_deprecated = arguments.get("include_deprecated", False)
        category = arguments.get("category")
        tools = []
        for tool in TOOL_CATALOG:
            name = tool["name"]
            if not include_deprecated and name in TOOL_DEPRECATED:
                continue
            if category and category not in tool.get("description", "").lower():
                # category is a hint, not a strict filter
                pass
            entry = {
                "name": name,
                "version": TOOL_VERSIONS.get(name, "1.0.0"),
                "description": tool.get("description", ""),
                "input_schema": tool.get("inputSchema", {}),
                "deprecated": name in TOOL_DEPRECATED,
                "successor": TOOL_SUCCESSORS.get(name),
            }
            tools.append(entry)
        return {
            "result": {
                "server": "rugmunch-intelligence",
                "server_version": MCP_SERVER_VERSION,
                "protocol_version": MCP_PROTOCOL_VERSION,
                "tool_count": len(tools),
                "tools": tools,
            }
        }

    if name == "status_check":
        # M3 moat TIER 1 — unified health check across all subsystems
        # Use the async path directly (we're in an event loop already)
        from app.core.health import run_health_checks
        health = await run_health_checks()
        return {"result": health}

    return {"error": f"unknown tool: {name}"}


def _tier_from_score(score: int) -> str:
    if score < 25:
        return "low"
    if score < 50:
        return "medium"
    if score < 75:
        return "high"
    return "critical"