MCP Crypto API Servers

""" Blockchain Explorer MCP Server This server provides tools for interacting with various blockchain explorers (Etherscan, BSCScan, etc.) to fetch token and wallet information. """ from __future__ import annotations from typing import Any, Dict, List, Optional from mcp.server.fastmcp import FastMCP from common import fetch_json, config, API_KEYS, SUPPORTED_CHAINS # Initialize FastMCP mcp = FastMCP(name="blockchain-explorer-server") @mcp.tool() async def get_wallet_balances( wallet_address: str, chain_ids: Optional[List[str]] = None, api_keys: Optional[Dict[str, str]] = None, timeout: Optional[float] = None, max_retries: Optional[int] = None ) -> str: """Get native token (ETH/MATIC/BNB) balance for a wallet across multiple chains. Args: wallet_address: The wallet address to check balances for chain_ids: List of chain IDs to check (e.g., ['ethereum', 'arbitrum', 'optimism']) api_keys: Optional dictionary of API keys per chain timeout: Optional request timeout override max_retries: Optional retry count override Returns: Formatted string with balances for each chain """ chain_ids = chain_ids or config['default_chains'] api_keys = api_keys or config['api_keys'] results = [] for chain_id in chain_ids: if chain_id not in SUPPORTED_CHAINS: continue chain = SUPPORTED_CHAINS[chain_id] api_key = api_keys.get(chain_id) if not api_key: results.append(f"⚠️ {chain.name}: API key tidak tersedia") continue params = { 'module': 'account', 'action': 'balance', 'address': wallet_address, 'apikey': api_key } url = chain.explorer_url data = await fetch_json(url, params=params, timeout=timeout) if not data or data.get('status') != '1': results.append(f"❌ {chain.name}: Gagal mendapatkan saldo") continue balance = int(data['result']) / 1e18 # Convert from wei/satoshi results.append(f"✅ {chain.name}: {balance:.4f} {chain.symbol}") if not results: return "Tidak dapat mendapatkan saldo dari blockchain manapun." return "# Saldo Wallet\n\n" + "\n".join(results) @mcp.tool() async def get_token_holders( token_address: str, chain_id: str = "ethereum", top_n: int = 10 ) -> str: """Mendapatkan informasi kepemilikan token pada alamat smart contract tertentu. Args: token_address: Alamat smart contract token chain_id: ID blockchain (ethereum, bsc, polygon, arbitrum) top_n: Jumlah holder teratas yang ingin ditampilkan Returns: String berisi informasi kepemilikan token """ if chain_id not in SUPPORTED_CHAINS: return f"Chain {chain_id} tidak didukung." chain = SUPPORTED_CHAINS[chain_id] api_key = API_KEYS.get(chain_id) if not api_key: return f"API key untuk {chain.name} tidak tersedia." try: # Verifikasi kontrak token params = { 'module': 'contract', 'action': 'getabi', 'address': token_address, 'apikey': api_key } url = chain.explorer_url contract_data = await fetch_json(url, params=params) if not contract_data or contract_data.get('status') != '1': return f"Kontrak {token_address} tidak terverifikasi di {chain.name}." # Dapatkan holder params = { 'module': 'token', 'action': 'tokenholderlist', 'contractaddress': token_address, 'page': 1, 'offset': top_n, 'apikey': api_key } data = await fetch_json(url, params=params) if not data or data.get('status') != '1': return f"Gagal mendapatkan data holder untuk token {token_address}" holders = data.get('result', []) if not holders: return "Tidak ditemukan data holder." # Format hasil result = f"# Top {len(holders)} Token Holders di {chain.name}\n\n" result += "| # | Address | Balance |\n" result += "|---|---------|----------|\n" total_supply = sum(float(h.get('TokenHolderQuantity', 0)) for h in holders) for i, holder in enumerate(holders, 1): address = holder.get('TokenHolderAddress', 'N/A') balance = float(holder.get('TokenHolderQuantity', 0)) percentage = (balance / total_supply * 100) if total_supply > 0 else 0 result += f"| {i} | `{address}` | {balance:,.0f} ({percentage:.2f}%) |\n" # Analisis konsentrasi if holders and total_supply > 0: top_holder = float(holders[0]['TokenHolderQuantity']) top_holder_pct = (top_holder / total_supply * 100) top5_pct = sum(float(h['TokenHolderQuantity']) for h in holders[:5]) / total_supply * 100 result += "\n## Analisis Konsentrasi\n" result += f"- Holder terbesar memegang **{top_holder_pct:.2f}%** dari total supply\n" result += f"- 5 holder teratas memegang **{top5_pct:.2f}%** dari total supply\n" # Evaluasi risiko risk_level = "🟢 Rendah" if top_holder_pct > 50: risk_level = "🔴 Tinggi" elif top_holder_pct > 20: risk_level = "🟡 Sedang" result += f"- **Tingkat Risiko Konsentrasi**: {risk_level}\n" return result except Exception as e: return f"Error saat mengambil data holder: {str(e)}" @mcp.tool() async def analyze_token_distribution(token_address: str, chain_id: str = "ethereum") -> str: """Menganalisis distribusi kepemilikan token secara mendalam. Args: token_address: Alamat smart contract token chain_id: ID blockchain (ethereum, bsc, polygon, arbitrum) Returns: String berisi analisis distribusi token """ if chain_id not in SUPPORTED_CHAINS: return f"Chain {chain_id} tidak didukung." chain = SUPPORTED_CHAINS[chain_id] api_key = API_KEYS.get(chain_id) if not api_key: return f"API key untuk {chain.name} tidak tersedia." try: # Dapatkan data holder holders_data = await get_token_holders(token_address, chain_id, top_n=100) # Analisis kontrak params = { 'module': 'contract', 'action': 'getsourcecode', 'address': token_address, 'apikey': api_key } url = chain.explorer_url contract_data = await fetch_json(url, params=params) result = "# Analisis Distribusi Token\n\n" # Tambahkan data holder if holders_data: result += "## Distribusi Kepemilikan\n" result += holders_data.split("\n", 1)[1] if "\n" in holders_data else holders_data result += "\n" # Analisis kontrak if contract_data and contract_data.get('status') == '1': contract_info = contract_data['result'][0] result += "\n## Analisis Kontrak\n" # Verifikasi kontrak if contract_info.get('ABI') == "Contract source code not verified": result += "⚠️ **PERINGATAN**: Kontrak tidak terverifikasi!\n" else: result += "✅ Kontrak terverifikasi\n" # Cek fitur kontrak yang berisiko source_code = contract_info.get('SourceCode', '').lower() risk_features = [] if "selfdestruct" in source_code or "suicide" in source_code: risk_features.append("🔴 Memiliki fungsi self-destruct") if "mint" in source_code and "onlyowner" in source_code: risk_features.append("🟡 Owner dapat melakukan minting") if "pause" in source_code: risk_features.append("🟡 Kontrak dapat di-pause") if "blacklist" in source_code: risk_features.append("ℹ️ Memiliki fungsi blacklist") if risk_features: result += "\n### Fitur Berisiko yang Ditemukan:\n" for feature in risk_features: result += f"- {feature}\n" else: result += "\n✅ Tidak ditemukan fitur berisiko dalam kontrak\n" return result except Exception as e: return f"Error saat menganalisis distribusi token: {str(e)}" if __name__ == "__main__": mcp.serve()