| """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 json |
| import logging |
| from typing import Any |
|
|
| log = logging.getLogger(__name__) |
|
|
| |
| 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"], |
| }, |
| }, |
| ] |
|
|
|
|
| |
| 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} |
| |
| from app.catalog.reputation import compute_deployer_reputation |
| from app.catalog.models import Deployer |
| 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, |
| }} |
| except Exception as e: |
| return {"error": f"report_fail: {e}"} |
|
|
| if name == "query_catalog": |
| |
| 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']}"} |
| |
| 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": []} |
| |
| 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} |
|
|
| 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" |
|
|
