import httpx
import logging
import sys
import asyncio
from mcp_app import mcp
logging.basicConfig(
stream=sys.stderr, level=logging.INFO, format="[LOG] %(message)s", force=True
)
# RPC List
CHAINS = {
"Ethereum": {"url": "https://eth.llamarpc.com", "symbol": "ETH"},
"Arbitrum": {"url": "https://arb1.arbitrum.io/rpc", "symbol": "ETH"},
"Optimism": {"url": "https://mainnet.optimism.io", "symbol": "ETH"},
"Polygon": {"url": "https://polygon-rpc.com", "symbol": "MATIC"},
"Base": {"url": "https://mainnet.base.org", "symbol": "ETH"},
}
async def fetch_balance(client, name, url, symbol, address):
"""Helper function to fetch one single chain"""
try:
payload = {
"jsonrpc": "2.0",
"method": "eth_getBalance",
"params": [address, "latest"],
"id": 1,
}
response = await client.post(url, json=payload, timeout=2.0)
data = response.json()
if "result" in data:
wei = int(data["result"], 16)
eth = wei / 10**18
return name, f"{eth:.4f} {symbol}"
return name, "Error"
except Exception:
return name, "Offline"
@mcp.tool()
async def wallet_check(address: str) -> dict:
"""
Check the native token balance of a wallet across multiple blockchains.
Args:
address: The wallet address (must start with 0x)
"""
if not address or not address.startswith("0x"):
return {"Error": "Invalid address format"}
results = {}
# Use a single client for all requests
async with httpx.AsyncClient() as client:
tasks = []
for name, data in CHAINS.items():
chain_url = data["url"]
chain_symbol = data["symbol"]
tasks.append(fetch_balance(client, name, chain_url, chain_symbol, address))
responses = await asyncio.gather(*tasks)
for name, balance in responses:
results[name] = balance
return results
