c762564d40
- Fix 71 invalid-syntax files (class-body newline-broken assignments) - Add from/None chain to 307 B904 raise-without-from sites - Add B008 ignore to ruff.toml (already in pyproject.toml) - Noqa F401 on __init__.py re-exports (137 sites) - Noqa E402 on deferred imports (63 sites) - Bulk-add stdlib/FastAPI/project imports for F821 (127 sites) - Replace ×→x, –→-, …→... in docstrings (4093 chars) - Manual refactor of 5 SIM103/SIM116 patterns Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py) Co-authored-by: opencode <opencode@rugmunch.io>
325 lines
14 KiB
Python
325 lines
14 KiB
Python
"""
|
|
RMI PREMIUM MCP SERVERS - Bot Attractors → x402 Revenue
|
|
=========================================================
|
|
8 new MCP servers designed for maximum bot adoption.
|
|
Each: free tier → rate limit → x402 micropayment upsell.
|
|
Target users: trading bots, MEV searchers, degens, researchers.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
|
|
import httpx as req
|
|
|
|
|
|
def gredis():
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv("/app/.env", override=True)
|
|
import redis
|
|
|
|
return redis.Redis(
|
|
host="rmi-redis", port=6379, password=os.getenv("REDIS_PASSWORD"), decode_responses=True
|
|
)
|
|
|
|
|
|
def trial(fingerprint: str, tool: str, limit: int = 10) -> dict:
|
|
r, k = gredis(), f"mcp:trial:{tool}:{fingerprint}"
|
|
c = int(r.get(k) or 0)
|
|
if c < limit:
|
|
r.incr(k)
|
|
r.expire(k, 86400)
|
|
return {"tier": "free", "remaining": limit - c - 1, "calls_used": c + 1}
|
|
return {
|
|
"tier": "free_exhausted",
|
|
"remaining": 0,
|
|
"upgrade": "$0.01-0.05/call via x402 on Base/Solana/13 chains",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════
|
|
# MCP #1: WHALE ALERT - Real-time large transfers
|
|
# ═══════════════════════════════════════════════════
|
|
def whale_alert(
|
|
chain: str = "ethereum", min_value: float = 1000000, fingerprint: str = "anon"
|
|
) -> dict:
|
|
"""Real-time whale transaction alerts. 15 free/day. $0.03/call premium."""
|
|
auth = trial(fingerprint, "whale", 15)
|
|
if auth["tier"] == "free_exhausted":
|
|
return {"error": "Free exhausted", "upgrade": auth["upgrade"]}
|
|
result = {"chain": chain, "min_value_usd": min_value, "alerts": []}
|
|
if chain == "ethereum":
|
|
try:
|
|
r = req.get(
|
|
"https://api.etherscan.io/api?module=account&action=txlist&address=0xBE0eB53F46cd790Cd13851d5EFf43D12404d33E8&sort=desc&page=1&offset=5",
|
|
timeout=10,
|
|
)
|
|
if r.status_code == 200:
|
|
for tx in r.json().get("result", [])[:5]:
|
|
val = int(tx.get("value", 0)) / 1e18
|
|
if val > 100:
|
|
result["alerts"].append(
|
|
{
|
|
"hash": tx["hash"],
|
|
"from": tx["from"],
|
|
"to": tx["to"],
|
|
"value_eth": val,
|
|
"estimated_usd": val * 3000,
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
result["auth"] = auth
|
|
result["mcp"] = "rmi-whale-alert"
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════
|
|
# MCP #2: TOKEN LAUNCH SCANNER
|
|
# ═══════════════════════════════════════════════════
|
|
def token_launch_scanner(
|
|
chain: str = "solana", max_age_hours: int = 1, fingerprint: str = "anon"
|
|
) -> dict:
|
|
"""New token detection. 10 free/day. $0.05/call premium."""
|
|
auth = trial(fingerprint, "launch", 10)
|
|
if auth["tier"] == "free_exhausted":
|
|
return {"error": "Free exhausted", "upgrade": auth["upgrade"]}
|
|
result = {"chain": chain, "max_age_hours": max_age_hours, "new_tokens": []}
|
|
if chain == "solana":
|
|
try:
|
|
r = req.get("https://api.dexscreener.com/token-profiles/latest/v1", timeout=10)
|
|
if r.status_code == 200:
|
|
for t in r.json()[:10]:
|
|
result["new_tokens"].append(
|
|
{
|
|
"address": t.get("tokenAddress"),
|
|
"symbol": t.get("symbol"),
|
|
"name": t.get("name"),
|
|
"age": "new",
|
|
"chain": "solana",
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
result["auth"] = auth
|
|
result["mcp"] = "rmi-launch-scanner"
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════
|
|
# MCP #3: SMART MONEY TRACKER
|
|
# ═══════════════════════════════════════════════════
|
|
def smart_money_tracker(chain: str = "ethereum", fingerprint: str = "anon") -> dict:
|
|
"""Track profitable wallet activity. 10 free/day. $0.03/call premium."""
|
|
auth = trial(fingerprint, "smartmoney", 10)
|
|
if auth["tier"] == "free_exhausted":
|
|
return {"error": "Free exhausted", "upgrade": auth["upgrade"]}
|
|
r = gredis()
|
|
wallets = []
|
|
for addr in r.keys("rmi:label:ethereum:*")[:20]:
|
|
label = json.loads(r.get(addr) or "{}")
|
|
if (
|
|
"smart" in str(label).lower()
|
|
or "fund" in str(label).lower()
|
|
or "capital" in str(label).lower()
|
|
):
|
|
wallets.append(
|
|
{
|
|
"address": addr.split(":")[-1],
|
|
"label": label.get("label", ""),
|
|
"entity": label.get("name_tag", ""),
|
|
}
|
|
)
|
|
return {
|
|
"chain": chain,
|
|
"smart_wallets": wallets,
|
|
"total_tracked": "82K labeled addresses",
|
|
"auth": auth,
|
|
"mcp": "rmi-smart-money",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════
|
|
# MCP #4: MEV/SANDWICH DETECTION
|
|
# ═══════════════════════════════════════════════════
|
|
def mev_detector(address: str = "", chain: str = "ethereum", fingerprint: str = "anon") -> dict:
|
|
"""Detect MEV sandwich attacks on any address. 10 free/day. $0.05/call."""
|
|
auth = trial(fingerprint, "mev", 10)
|
|
if auth["tier"] == "free_exhausted":
|
|
return {"error": "Free exhausted", "upgrade": auth["upgrade"]}
|
|
result = {"address": address, "chain": chain, "sandwich_risk": "low", "mev_exposure": 0}
|
|
try:
|
|
r = req.get(
|
|
f"https://api.etherscan.io/api?module=account&action=txlist&address={address}&sort=desc&page=1&offset=20",
|
|
timeout=10,
|
|
)
|
|
if r.status_code == 200:
|
|
txs = r.json().get("result", [])
|
|
if len(txs) > 5:
|
|
# Simple heuristic: >5 txs in same block = possible MEV
|
|
blocks = {}
|
|
for tx in txs:
|
|
b = tx.get("blockNumber", "")
|
|
blocks[b] = blocks.get(b, 0) + 1
|
|
result["sandwich_risk"] = "medium" if any(v > 2 for v in blocks.values()) else "low"
|
|
result["mev_exposure"] = min(
|
|
1.0, sum(1 for v in blocks.values() if v > 2) / max(len(blocks), 1)
|
|
)
|
|
except Exception:
|
|
pass
|
|
result["auth"] = auth
|
|
result["mcp"] = "rmi-mev-detector"
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════
|
|
# MCP #5: NARRATIVE DETECTION
|
|
# ═══════════════════════════════════════════════════
|
|
def narrative_detector(fingerprint: str = "anon") -> dict:
|
|
"""AI-identified trending narratives from 500+ sources. 10 free/day."""
|
|
auth = trial(fingerprint, "narrative", 10)
|
|
if auth["tier"] == "free_exhausted":
|
|
return {"error": "Free exhausted", "upgrade": auth["upgrade"]}
|
|
r = gredis()
|
|
titles = []
|
|
for idx in ["rmi:news:500feeds", "rmi:news:index"]:
|
|
for aid in r.zrevrange(idx, 0, min(50, r.zcard(idx))):
|
|
a = json.loads(r.get(f"rmi:news:article:{aid}") or "{}")
|
|
titles.append(a.get("title", ""))
|
|
# Simple keyword frequency
|
|
words = {}
|
|
for t in titles:
|
|
for w in t.lower().split():
|
|
if len(w) > 3:
|
|
words[w] = words.get(w, 0) + 1
|
|
trending = sorted(words.items(), key=lambda x: x[1], reverse=True)[:20]
|
|
return {
|
|
"articles_analyzed": len(titles),
|
|
"trending_terms": [
|
|
{"term": w, "frequency": c}
|
|
for w, c in trending
|
|
if w not in ["that", "this", "with", "from", "have", "will", "your", "than"]
|
|
],
|
|
"sources": 500,
|
|
"auth": auth,
|
|
"mcp": "rmi-narrative",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════
|
|
# MCP #6: CROSS-CHAIN ARBITRAGE SCANNER
|
|
# ═══════════════════════════════════════════════════
|
|
def arbitrage_scanner(token: str = "ETH", fingerprint: str = "anon") -> dict:
|
|
"""Cross-chain/DEX price differences. 10 free/day. $0.05/call."""
|
|
auth = trial(fingerprint, "arb", 10)
|
|
if auth["tier"] == "free_exhausted":
|
|
return {"error": "Free exhausted", "upgrade": auth["upgrade"]}
|
|
prices = {}
|
|
try:
|
|
r = req.get(f"https://api.dexscreener.com/latest/dex/search?q={token}", timeout=10)
|
|
if r.status_code == 200:
|
|
for pair in r.json().get("pairs", [])[:10]:
|
|
dex = pair.get("dexId", "unknown")
|
|
price = pair.get("priceUsd", "0")
|
|
chain = pair.get("chainId", "unknown")
|
|
prices[f"{dex}_{chain}"] = float(price)
|
|
except Exception:
|
|
pass
|
|
if prices:
|
|
pvals = [v for v in prices.values() if v > 0]
|
|
spread = max(pvals) - min(pvals) if pvals else 0
|
|
return {
|
|
"token": token,
|
|
"exchanges": len(prices),
|
|
"prices": prices,
|
|
"max_spread_usd": round(spread, 4) if prices else 0,
|
|
"auth": auth,
|
|
"mcp": "rmi-arbitrage",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════
|
|
# MCP #7: CONTRACT AUDIT QUICK-SCAN
|
|
# ═══════════════════════════════════════════════════
|
|
def quick_audit(address: str, chain: str = "ethereum", fingerprint: str = "anon") -> dict:
|
|
"""Pre-trade security check. 15 free/day. $0.02/call."""
|
|
auth = trial(fingerprint, "audit", 15)
|
|
if auth["tier"] == "free_exhausted":
|
|
return {"error": "Free exhausted", "upgrade": auth["upgrade"]}
|
|
result = {"address": address, "chain": chain}
|
|
try:
|
|
r = req.get(
|
|
f"https://api.gopluslabs.io/api/v1/token_security/{chain}?contract_addresses={address}",
|
|
timeout=10,
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json().get("result", {}).get(address.lower(), {})
|
|
result["honeypot"] = data.get("is_honeypot") == "1"
|
|
result["buy_tax"] = data.get("buy_tax", "0")
|
|
result["sell_tax"] = data.get("sell_tax", "0")
|
|
result["is_open_source"] = data.get("is_open_source") == "1"
|
|
result["owner_renounced"] = data.get("is_owner_renounced", "0") == "1"
|
|
result["liquidity"] = "locked" if data.get("lp_holders") else "unknown"
|
|
red_flags = sum(
|
|
[
|
|
result.get("honeypot", False),
|
|
float(result.get("buy_tax", "0")) > 5,
|
|
float(result.get("sell_tax", "0")) > 5,
|
|
]
|
|
)
|
|
result["verdict"] = (
|
|
"SAFE" if red_flags == 0 else ("CAUTION" if red_flags == 1 else "DANGER")
|
|
)
|
|
except Exception:
|
|
pass
|
|
result["auth"] = auth
|
|
result["mcp"] = "rmi-quick-audit"
|
|
return result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════
|
|
# MCP #8: PORTFOLIO RISK SCORER
|
|
# ═══════════════════════════════════════════════════
|
|
def portfolio_risk(address: str, fingerprint: str = "anon") -> dict:
|
|
"""Cross-chain risk scoring. 10 free/day. $0.03/call."""
|
|
auth = trial(fingerprint, "risk", 10)
|
|
if auth["tier"] == "free_exhausted":
|
|
return {"error": "Free exhausted", "upgrade": auth["upgrade"]}
|
|
r = gredis()
|
|
risk = {"address": address, "risk_score": 0, "risk_factors": []}
|
|
# Check sanctions
|
|
if r.get(f"rmi:label:ethereum:{address.lower()}"):
|
|
label = json.loads(r.get(f"rmi:label:ethereum:{address.lower()}") or "{}")
|
|
if "scam" in str(label).lower():
|
|
risk["risk_factors"].append("known_scam")
|
|
risk["risk_score"] += 30
|
|
if "hack" in str(label).lower():
|
|
risk["risk_factors"].append("hack_related")
|
|
risk["risk_score"] += 40
|
|
# Check Chainabuse
|
|
try:
|
|
resp = req.get(f"https://api.chainabuse.com/v0/reports?address={address}", timeout=5)
|
|
if resp.status_code == 200 and len(resp.json().get("reports", [])) > 0:
|
|
risk["risk_factors"].append("community_reported")
|
|
risk["risk_score"] += 20
|
|
except Exception:
|
|
pass
|
|
risk["risk_level"] = (
|
|
"LOW" if risk["risk_score"] < 20 else ("MEDIUM" if risk["risk_score"] < 50 else "HIGH")
|
|
)
|
|
risk["auth"] = auth
|
|
risk["mcp"] = "rmi-portfolio-risk"
|
|
return risk
|
|
|
|
|
|
# Registry
|
|
PREMIUM_MCP = {
|
|
"rmi-whale-alert": whale_alert,
|
|
"rmi-launch-scanner": token_launch_scanner,
|
|
"rmi-smart-money": smart_money_tracker,
|
|
"rmi-mev-detector": mev_detector,
|
|
"rmi-narrative": narrative_detector,
|
|
"rmi-arbitrage": arbitrage_scanner,
|
|
"rmi-quick-audit": quick_audit,
|
|
"rmi-portfolio-risk": portfolio_risk,
|
|
}
|