#!/usr/bin/env python3 """ Base Blockchain CLI Tool for Hermes Agent ------------------------------------------ Queries the Base (Ethereum L2) JSON-RPC API and CoinGecko for enriched on-chain data. Uses only Python standard library — no external packages required. Usage: python3 base_client.py stats python3 base_client.py wallet
[--limit N] [--all] [--no-prices] python3 base_client.py tx python3 base_client.py token python3 base_client.py gas python3 base_client.py contract
python3 base_client.py whales [--min-eth N] python3 base_client.py price Environment: BASE_RPC_URL Override the default RPC endpoint (default: https://mainnet.base.org) """ import argparse import json import os import sys import time import urllib.request import urllib.error from typing import Any, Dict, List, Optional, Tuple RPC_URL = os.environ.get( "BASE_RPC_URL", "https://mainnet.base.org", ) WEI_PER_ETH = 10**18 GWEI = 10**9 # ERC-20 function selectors (first 4 bytes of keccak256 hash) SEL_BALANCE_OF = "70a08231" SEL_NAME = "06fdde03" SEL_SYMBOL = "95d89b41" SEL_DECIMALS = "313ce567" SEL_TOTAL_SUPPLY = "18160ddd" # ERC-165 supportsInterface(bytes4) selector SEL_SUPPORTS_INTERFACE = "01ffc9a7" # Interface IDs for ERC-165 detection IFACE_ERC721 = "80ac58cd" IFACE_ERC1155 = "d9b67a26" # Transfer(address,address,uint256) event topic TRANSFER_TOPIC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" # Well-known Base tokens — maps lowercase address -> (symbol, name, decimals). KNOWN_TOKENS: Dict[str, Tuple[str, str, int]] = { "0x4200000000000000000000000000000000000006": ("WETH", "Wrapped Ether", 18), "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913": ("USDC", "USD Coin", 6), "0x2ae3f1ec7f1f5012cfeab0185bfc7aa3cf0dec22": ("cbETH", "Coinbase Wrapped Staked ETH", 18), "0x940181a94a35a4569e4529a3cdfb74e38fd98631": ("AERO", "Aerodrome Finance", 18), "0x4ed4e862860bed51a9570b96d89af5e1b0efefed": ("DEGEN", "Degen", 18), "0xac1bd2486aaf3b5c0fc3fd868558b082a531b2b4": ("TOSHI", "Toshi", 18), "0x532f27101965dd16442e59d40670faf5ebb142e4": ("BRETT", "Brett", 18), "0xa88594d404727625a9437c3f886c7643872296ae": ("WELL", "Moonwell", 18), "0xc1cba3fcea344f92d9239c08c0568f6f2f0ee452": ("wstETH", "Wrapped Lido Staked ETH", 18), "0xb6fe221fe9eef5aba221c348ba20a1bf5e73624c": ("rETH", "Rocket Pool ETH", 18), "0xcbb7c0000ab88b473b1f5afd9ef808440eed33bf": ("cbBTC", "Coinbase Wrapped BTC", 8), } # Reverse lookup: symbol -> contract address (for the `price` command). _SYMBOL_TO_ADDRESS = {v[0].upper(): k for k, v in KNOWN_TOKENS.items()} _SYMBOL_TO_ADDRESS["ETH"] = "ETH" # --------------------------------------------------------------------------- # HTTP / RPC helpers # --------------------------------------------------------------------------- def _http_get_json(url: str, timeout: int = 10, retries: int = 2) -> Any: """GET JSON from a URL with retry on 429 rate-limit. Returns parsed JSON or None.""" for attempt in range(retries + 1): req = urllib.request.Request( url, headers={"Accept": "application/json", "User-Agent": "HermesAgent/1.0"}, ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return json.load(resp) except urllib.error.HTTPError as exc: if exc.code == 429 and attempt < retries: time.sleep(2.0 * (attempt + 1)) continue return None except Exception: return None return None def _rpc_call(method: str, params: list = None, retries: int = 2) -> Any: """Send a JSON-RPC request with retry on 429 rate-limit.""" payload = json.dumps({ "jsonrpc": "2.0", "id": 1, "method": method, "params": params or [], }).encode() _headers = {"Content-Type": "application/json", "User-Agent": "HermesAgent/1.0"} for attempt in range(retries + 1): req = urllib.request.Request( RPC_URL, data=payload, headers=_headers, method="POST", ) try: with urllib.request.urlopen(req, timeout=20) as resp: body = json.load(resp) if "error" in body: err = body["error"] if isinstance(err, dict) and err.get("code") == 429: if attempt < retries: time.sleep(1.5 * (attempt + 1)) continue sys.exit(f"RPC error: {err}") return body.get("result") except urllib.error.HTTPError as exc: if exc.code == 429 and attempt < retries: time.sleep(1.5 * (attempt + 1)) continue sys.exit(f"RPC HTTP error: {exc}") except urllib.error.URLError as exc: sys.exit(f"RPC connection error: {exc}") return None # Keep backward compat alias. rpc = _rpc_call _BATCH_LIMIT = 10 # Base public RPC limits to 10 calls per batch def _rpc_batch_chunk(items: list) -> list: """Send a single batch of JSON-RPC requests (max _BATCH_LIMIT).""" payload = json.dumps(items).encode() _headers = {"Content-Type": "application/json", "User-Agent": "HermesAgent/1.0"} for attempt in range(3): req = urllib.request.Request( RPC_URL, data=payload, headers=_headers, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as resp: data = json.load(resp) # If the RPC returns an error dict instead of a list, treat as failure if isinstance(data, dict) and "error" in data: sys.exit(f"RPC batch error: {data['error']}") return data if isinstance(data, list) else [] except urllib.error.HTTPError as exc: if exc.code == 429 and attempt < 2: time.sleep(1.5 * (attempt + 1)) continue sys.exit(f"RPC batch HTTP error: {exc}") except urllib.error.URLError as exc: sys.exit(f"RPC batch error: {exc}") return [] def rpc_batch(calls: list) -> list: """Send a batch of JSON-RPC requests, auto-chunking to respect limits.""" items = [ {"jsonrpc": "2.0", "id": i, "method": c["method"], "params": c.get("params", [])} for i, c in enumerate(calls) ] if len(items) <= _BATCH_LIMIT: return _rpc_batch_chunk(items) # Split into chunks of _BATCH_LIMIT all_results = [] for start in range(0, len(items), _BATCH_LIMIT): chunk = items[start:start + _BATCH_LIMIT] all_results.extend(_rpc_batch_chunk(chunk)) return all_results def wei_to_eth(wei: int) -> float: return wei / WEI_PER_ETH def wei_to_gwei(wei: int) -> float: return wei / GWEI def hex_to_int(hex_str: Optional[str]) -> int: """Convert hex string (0x...) to int. Returns 0 for None/empty.""" if not hex_str or hex_str == "0x": return 0 return int(hex_str, 16) def print_json(obj: Any) -> None: print(json.dumps(obj, indent=2)) def _short_addr(addr: str) -> str: """Abbreviate an address for display: first 6 + last 4.""" if len(addr) <= 14: return addr return f"{addr[:6]}...{addr[-4:]}" # --------------------------------------------------------------------------- # ABI encoding / decoding helpers # --------------------------------------------------------------------------- def _encode_address(addr: str) -> str: """ABI-encode an address as a 32-byte hex string (no 0x prefix).""" clean = addr.lower().replace("0x", "") return clean.zfill(64) def _decode_uint(hex_data: Optional[str]) -> int: """Decode a hex-encoded uint256 return value.""" if not hex_data or hex_data == "0x": return 0 return int(hex_data.replace("0x", ""), 16) def _decode_string(hex_data: Optional[str]) -> str: """Decode an ABI-encoded string return value.""" if not hex_data or hex_data == "0x" or len(hex_data) < 130: return "" data = hex_data[2:] if hex_data.startswith("0x") else hex_data try: length = int(data[64:128], 16) if length == 0 or length > 256: return "" str_hex = data[128:128 + length * 2] return bytes.fromhex(str_hex).decode("utf-8").strip("\x00") except (ValueError, UnicodeDecodeError): return "" def _eth_call(to: str, selector: str, args: str = "", block: str = "latest") -> Optional[str]: """Execute eth_call with a function selector. Returns None on revert/error.""" data = "0x" + selector + args try: payload = json.dumps({ "jsonrpc": "2.0", "id": 1, "method": "eth_call", "params": [{"to": to, "data": data}, block], }).encode() req = urllib.request.Request( RPC_URL, data=payload, headers={"Content-Type": "application/json", "User-Agent": "HermesAgent/1.0"}, method="POST", ) with urllib.request.urlopen(req, timeout=20) as resp: body = json.load(resp) if "error" in body: return None return body.get("result") except Exception: return None # --------------------------------------------------------------------------- # Price & token name helpers (CoinGecko — free, no API key) # --------------------------------------------------------------------------- def fetch_prices(addresses: List[str], max_lookups: int = 20) -> Dict[str, float]: """Fetch USD prices for Base token addresses via CoinGecko (one per request). CoinGecko free tier doesn't support batch Base token lookups, so we do individual calls — capped at *max_lookups* to stay within rate limits. Returns {lowercase_address: usd_price}. """ prices: Dict[str, float] = {} for i, addr in enumerate(addresses[:max_lookups]): url = ( f"https://api.coingecko.com/api/v3/simple/token_price/base" f"?contract_addresses={addr}&vs_currencies=usd" ) data = _http_get_json(url, timeout=10) if data and isinstance(data, dict): for key, info in data.items(): if isinstance(info, dict) and "usd" in info: prices[addr.lower()] = info["usd"] break # Pause between calls to respect CoinGecko free-tier rate-limits if i < len(addresses[:max_lookups]) - 1: time.sleep(1.0) return prices def fetch_eth_price() -> Optional[float]: """Fetch current ETH price in USD via CoinGecko.""" data = _http_get_json( "https://api.coingecko.com/api/v3/simple/price?ids=ethereum&vs_currencies=usd" ) if data and "ethereum" in data: return data["ethereum"].get("usd") return None def resolve_token_name(addr: str) -> Optional[Dict[str, str]]: """Look up token name and symbol. Checks known tokens first, then on-chain. Returns {"name": ..., "symbol": ...} or None. """ addr_lower = addr.lower() if addr_lower in KNOWN_TOKENS: sym, name, _ = KNOWN_TOKENS[addr_lower] return {"symbol": sym, "name": name} # Try reading name() and symbol() from the contract name_hex = _eth_call(addr, SEL_NAME) symbol_hex = _eth_call(addr, SEL_SYMBOL) name = _decode_string(name_hex) if name_hex else "" symbol = _decode_string(symbol_hex) if symbol_hex else "" if symbol: return {"symbol": symbol.upper(), "name": name} return None def _token_label(addr: str) -> str: """Return a human-readable label: symbol if known, else abbreviated address.""" addr_lower = addr.lower() if addr_lower in KNOWN_TOKENS: return KNOWN_TOKENS[addr_lower][0] return _short_addr(addr) # --------------------------------------------------------------------------- # 1. Network Stats # --------------------------------------------------------------------------- def cmd_stats(_args): """Base network health: block, gas, chain ID, ETH price.""" results = rpc_batch([ {"method": "eth_blockNumber"}, {"method": "eth_gasPrice"}, {"method": "eth_chainId"}, {"method": "eth_getBlockByNumber", "params": ["latest", False]}, ]) by_id = {r["id"]: r.get("result") for r in results} block_num = hex_to_int(by_id.get(0)) gas_price = hex_to_int(by_id.get(1)) chain_id = hex_to_int(by_id.get(2)) block = by_id.get(3) or {} base_fee = hex_to_int(block.get("baseFeePerGas")) if block.get("baseFeePerGas") else None timestamp = hex_to_int(block.get("timestamp")) if block.get("timestamp") else None gas_used = hex_to_int(block.get("gasUsed")) if block.get("gasUsed") else None gas_limit = hex_to_int(block.get("gasLimit")) if block.get("gasLimit") else None tx_count = len(block.get("transactions", [])) eth_price = fetch_eth_price() out = { "chain": "Base" if chain_id == 8453 else f"Chain {chain_id}", "chain_id": chain_id, "latest_block": block_num, "gas_price_gwei": round(wei_to_gwei(gas_price), 4), } if base_fee is not None: out["base_fee_gwei"] = round(wei_to_gwei(base_fee), 4) if timestamp: out["block_timestamp"] = timestamp if gas_used is not None and gas_limit: out["block_gas_used"] = gas_used out["block_gas_limit"] = gas_limit out["block_utilization_pct"] = round(gas_used / gas_limit * 100, 2) out["block_tx_count"] = tx_count if eth_price is not None: out["eth_price_usd"] = eth_price print_json(out) # --------------------------------------------------------------------------- # 2. Wallet Info (ETH + ERC-20 balances with prices) # --------------------------------------------------------------------------- def cmd_wallet(args): """ETH balance + ERC-20 token holdings with USD values.""" address = args.address.lower() show_all = getattr(args, "all", False) limit = getattr(args, "limit", 20) or 20 skip_prices = getattr(args, "no_prices", False) # Batch: ETH balance + balanceOf for all known tokens calls = [{"method": "eth_getBalance", "params": [address, "latest"]}] token_addrs = list(KNOWN_TOKENS.keys()) for token_addr in token_addrs: calls.append({ "method": "eth_call", "params": [ {"to": token_addr, "data": "0x" + SEL_BALANCE_OF + _encode_address(address)}, "latest", ], }) results = rpc_batch(calls) by_id = {r["id"]: r.get("result") for r in results} eth_balance = wei_to_eth(hex_to_int(by_id.get(0))) # Parse token balances tokens = [] for i, token_addr in enumerate(token_addrs): raw = hex_to_int(by_id.get(i + 1)) if raw == 0: continue sym, name, decimals = KNOWN_TOKENS[token_addr] amount = raw / (10 ** decimals) tokens.append({ "address": token_addr, "symbol": sym, "name": name, "amount": amount, "decimals": decimals, }) # Fetch prices eth_price = None prices: Dict[str, float] = {} if not skip_prices: eth_price = fetch_eth_price() if tokens: mints_to_price = [t["address"] for t in tokens] prices = fetch_prices(mints_to_price, max_lookups=20) # Enrich with USD values, filter dust, sort enriched = [] dust_count = 0 dust_value = 0.0 for t in tokens: usd_price = prices.get(t["address"]) usd_value = round(usd_price * t["amount"], 2) if usd_price else None if not show_all and usd_value is not None and usd_value < 0.01: dust_count += 1 dust_value += usd_value continue entry = {"token": t["symbol"], "address": t["address"], "amount": t["amount"]} if usd_price is not None: entry["price_usd"] = usd_price entry["value_usd"] = usd_value enriched.append(entry) # Sort: tokens with known USD value first (highest->lowest), then unknowns enriched.sort( key=lambda x: (x.get("value_usd") is not None, x.get("value_usd") or 0), reverse=True, ) # Apply limit unless --all total_tokens = len(enriched) if not show_all and len(enriched) > limit: enriched = enriched[:limit] hidden_tokens = total_tokens - len(enriched) # Compute portfolio total total_usd = sum(t.get("value_usd", 0) for t in enriched) eth_value_usd = round(eth_price * eth_balance, 2) if eth_price else None if eth_value_usd: total_usd += eth_value_usd total_usd += dust_value output = { "address": args.address, "eth_balance": round(eth_balance, 18), } if eth_price: output["eth_price_usd"] = eth_price output["eth_value_usd"] = eth_value_usd output["tokens_shown"] = len(enriched) if hidden_tokens > 0: output["tokens_hidden"] = hidden_tokens output["erc20_tokens"] = enriched if dust_count > 0: output["dust_filtered"] = {"count": dust_count, "total_value_usd": round(dust_value, 4)} if total_usd > 0: output["portfolio_total_usd"] = round(total_usd, 2) if hidden_tokens > 0 and not show_all: output["warning"] = ( "portfolio_total_usd may be partial because hidden tokens are not " "included when --limit is applied." ) output["note"] = f"Checked {len(KNOWN_TOKENS)} known Base tokens. Unknown ERC-20s not shown." print_json(output) # --------------------------------------------------------------------------- # 3. Transaction Details # --------------------------------------------------------------------------- def cmd_tx(args): """Full transaction details by hash.""" tx_hash = args.hash results = rpc_batch([ {"method": "eth_getTransactionByHash", "params": [tx_hash]}, {"method": "eth_getTransactionReceipt", "params": [tx_hash]}, ]) by_id = {r["id"]: r.get("result") for r in results} tx = by_id.get(0) receipt = by_id.get(1) if tx is None: sys.exit("Transaction not found.") value_wei = hex_to_int(tx.get("value")) tx_gas_price = hex_to_int(tx.get("gasPrice")) gas_used = hex_to_int(receipt.get("gasUsed")) if receipt else None effective_gas_price = ( hex_to_int(receipt.get("effectiveGasPrice")) if receipt and receipt.get("effectiveGasPrice") else tx_gas_price ) l2_fee_wei = effective_gas_price * gas_used if gas_used is not None else None l1_fee_wei = hex_to_int(receipt.get("l1Fee")) if receipt and receipt.get("l1Fee") else 0 fee_wei = (l2_fee_wei + l1_fee_wei) if l2_fee_wei is not None else None eth_price = fetch_eth_price() out = { "hash": tx_hash, "block": hex_to_int(tx.get("blockNumber")), "from": tx.get("from"), "to": tx.get("to"), "value_ETH": round(wei_to_eth(value_wei), 18) if value_wei else 0, "gas_price_gwei": round(wei_to_gwei(effective_gas_price), 4), } if gas_used is not None: out["gas_used"] = gas_used if l2_fee_wei is not None: out["l2_fee_ETH"] = round(wei_to_eth(l2_fee_wei), 12) if l1_fee_wei: out["l1_fee_ETH"] = round(wei_to_eth(l1_fee_wei), 12) if fee_wei is not None: out["fee_ETH"] = round(wei_to_eth(fee_wei), 12) if receipt: out["status"] = "success" if receipt.get("status") == "0x1" else "failed" out["contract_created"] = receipt.get("contractAddress") out["log_count"] = len(receipt.get("logs", [])) # Decode ERC-20 transfers from logs transfers = [] if receipt: for log in receipt.get("logs", []): topics = log.get("topics", []) if len(topics) >= 3 and topics[0] == TRANSFER_TOPIC: from_addr = "0x" + topics[1][-40:] to_addr = "0x" + topics[2][-40:] token_contract = log.get("address", "") label = _token_label(token_contract) entry = { "token": label, "contract": token_contract, "from": from_addr, "to": to_addr, } # ERC-20: 3 topics, amount in data if len(topics) == 3: amount_hex = log.get("data", "0x") if amount_hex and amount_hex != "0x": raw_amount = hex_to_int(amount_hex) addr_lower = token_contract.lower() if addr_lower in KNOWN_TOKENS: decimals = KNOWN_TOKENS[addr_lower][2] entry["amount"] = raw_amount / (10 ** decimals) else: entry["raw_amount"] = raw_amount # ERC-721: 4 topics, tokenId in topics[3] elif len(topics) == 4: entry["token_id"] = hex_to_int(topics[3]) entry["type"] = "ERC-721" transfers.append(entry) if transfers: out["token_transfers"] = transfers if eth_price is not None: if value_wei: out["value_USD"] = round(wei_to_eth(value_wei) * eth_price, 2) if l2_fee_wei is not None: out["l2_fee_USD"] = round(wei_to_eth(l2_fee_wei) * eth_price, 4) if l1_fee_wei: out["l1_fee_USD"] = round(wei_to_eth(l1_fee_wei) * eth_price, 4) if fee_wei is not None: out["fee_USD"] = round(wei_to_eth(fee_wei) * eth_price, 4) print_json(out) # --------------------------------------------------------------------------- # 4. Token Info # --------------------------------------------------------------------------- def cmd_token(args): """ERC-20 token metadata, supply, price, market cap.""" addr = args.address.lower() # Batch: name, symbol, decimals, totalSupply, code check calls = [ {"method": "eth_call", "params": [{"to": addr, "data": "0x" + SEL_NAME}, "latest"]}, {"method": "eth_call", "params": [{"to": addr, "data": "0x" + SEL_SYMBOL}, "latest"]}, {"method": "eth_call", "params": [{"to": addr, "data": "0x" + SEL_DECIMALS}, "latest"]}, {"method": "eth_call", "params": [{"to": addr, "data": "0x" + SEL_TOTAL_SUPPLY}, "latest"]}, {"method": "eth_getCode", "params": [addr, "latest"]}, ] results = rpc_batch(calls) by_id = {r["id"]: r.get("result") for r in results} code = by_id.get(4) if not code or code == "0x": sys.exit("Address is not a contract.") name = _decode_string(by_id.get(0)) symbol = _decode_string(by_id.get(1)) decimals_raw = by_id.get(2) decimals = _decode_uint(decimals_raw) total_supply_raw = _decode_uint(by_id.get(3)) # Fall back to known tokens if on-chain read failed if not symbol and addr in KNOWN_TOKENS: symbol = KNOWN_TOKENS[addr][0] name = KNOWN_TOKENS[addr][1] decimals = KNOWN_TOKENS[addr][2] is_known_token = addr in KNOWN_TOKENS is_erc20 = bool((symbol or is_known_token) and decimals_raw and decimals_raw != "0x") if not is_erc20: sys.exit("Contract does not appear to be an ERC-20 token.") total_supply = total_supply_raw / (10 ** decimals) if decimals else total_supply_raw # Fetch price price_data = fetch_prices([addr]) out = {"address": args.address} if name: out["name"] = name if symbol: out["symbol"] = symbol out["decimals"] = decimals out["total_supply"] = round(total_supply, min(decimals, 6)) out["code_size_bytes"] = (len(code) - 2) // 2 if addr in price_data: out["price_usd"] = price_data[addr] out["market_cap_usd"] = round(price_data[addr] * total_supply, 0) print_json(out) # --------------------------------------------------------------------------- # 5. Gas Analysis (Base-specific: L2 execution + L1 data costs) # --------------------------------------------------------------------------- def cmd_gas(_args): """Detailed gas analysis with L1 data fee context and cost estimates.""" latest_hex = _rpc_call("eth_blockNumber") latest = hex_to_int(latest_hex) # Get last 10 blocks for trend analysis + current gas price block_calls = [] for i in range(10): block_calls.append({ "method": "eth_getBlockByNumber", "params": [hex(latest - i), False], }) block_calls.append({"method": "eth_gasPrice"}) results = rpc_batch(block_calls) by_id = {r["id"]: r.get("result") for r in results} current_gas_price = hex_to_int(by_id.get(10)) base_fees = [] gas_utilizations = [] tx_counts = [] latest_block_info = None for i in range(10): b = by_id.get(i) if not b: continue bf = hex_to_int(b.get("baseFeePerGas", "0x0")) gu = hex_to_int(b.get("gasUsed", "0x0")) gl = hex_to_int(b.get("gasLimit", "0x0")) txc = len(b.get("transactions", [])) base_fees.append(bf) if gl > 0: gas_utilizations.append(gu / gl * 100) tx_counts.append(txc) if i == 0: latest_block_info = { "block": hex_to_int(b.get("number")), "base_fee_gwei": round(wei_to_gwei(bf), 6), "gas_used": gu, "gas_limit": gl, "utilization_pct": round(gu / gl * 100, 2) if gl > 0 else 0, "tx_count": txc, } avg_base_fee = sum(base_fees) / len(base_fees) if base_fees else 0 avg_utilization = sum(gas_utilizations) / len(gas_utilizations) if gas_utilizations else 0 avg_tx_count = sum(tx_counts) / len(tx_counts) if tx_counts else 0 # Estimate costs for common operations eth_price = fetch_eth_price() simple_transfer_gas = 21_000 erc20_transfer_gas = 65_000 swap_gas = 200_000 def _estimate_cost(gas: int) -> Dict[str, Any]: cost_wei = gas * current_gas_price cost_eth = wei_to_eth(cost_wei) entry: Dict[str, Any] = {"gas_units": gas, "cost_ETH": round(cost_eth, 10)} if eth_price: entry["cost_USD"] = round(cost_eth * eth_price, 6) return entry out: Dict[str, Any] = { "current_gas_price_gwei": round(wei_to_gwei(current_gas_price), 6), "latest_block": latest_block_info, "trend_10_blocks": { "avg_base_fee_gwei": round(wei_to_gwei(avg_base_fee), 6), "avg_utilization_pct": round(avg_utilization, 2), "avg_tx_count": round(avg_tx_count, 1), "min_base_fee_gwei": round(wei_to_gwei(min(base_fees)), 6) if base_fees else None, "max_base_fee_gwei": round(wei_to_gwei(max(base_fees)), 6) if base_fees else None, }, "cost_estimates": { "eth_transfer": _estimate_cost(simple_transfer_gas), "erc20_transfer": _estimate_cost(erc20_transfer_gas), "swap": _estimate_cost(swap_gas), }, "note": "Base is an L2. Total tx cost = L2 execution fee + L1 data posting fee. " "L1 data fee depends on calldata size and L1 gas prices (not shown here). " "Actual costs may be slightly higher than estimates.", } if eth_price: out["eth_price_usd"] = eth_price print_json(out) # --------------------------------------------------------------------------- # 6. Contract Inspection # --------------------------------------------------------------------------- def cmd_contract(args): """Inspect an address: EOA vs contract, ERC type detection, proxy resolution.""" addr = args.address.lower() # Batch: getCode, getBalance, name, symbol, decimals, totalSupply, ERC-721, ERC-1155 calls = [ {"method": "eth_getCode", "params": [addr, "latest"]}, {"method": "eth_getBalance", "params": [addr, "latest"]}, {"method": "eth_call", "params": [{"to": addr, "data": "0x" + SEL_NAME}, "latest"]}, {"method": "eth_call", "params": [{"to": addr, "data": "0x" + SEL_SYMBOL}, "latest"]}, {"method": "eth_call", "params": [{"to": addr, "data": "0x" + SEL_DECIMALS}, "latest"]}, {"method": "eth_call", "params": [{"to": addr, "data": "0x" + SEL_TOTAL_SUPPLY}, "latest"]}, {"method": "eth_call", "params": [ {"to": addr, "data": "0x" + SEL_SUPPORTS_INTERFACE + IFACE_ERC721.zfill(64)}, "latest", ]}, {"method": "eth_call", "params": [ {"to": addr, "data": "0x" + SEL_SUPPORTS_INTERFACE + IFACE_ERC1155.zfill(64)}, "latest", ]}, ] results = rpc_batch(calls) # Handle per-item errors gracefully by_id: Dict[int, Any] = {} for r in results: if "error" not in r: by_id[r["id"]] = r.get("result") else: by_id[r["id"]] = None code = by_id.get(0, "0x") eth_balance = hex_to_int(by_id.get(1)) if not code or code == "0x": out = { "address": args.address, "is_contract": False, "eth_balance": round(wei_to_eth(eth_balance), 18), "note": "This is an externally owned account (EOA), not a contract.", } print_json(out) return code_size = (len(code) - 2) // 2 # Check ERC-20 name = _decode_string(by_id.get(2)) symbol = _decode_string(by_id.get(3)) decimals_raw = by_id.get(4) supply_raw = by_id.get(5) is_erc20 = bool(symbol and decimals_raw and decimals_raw != "0x") # Check ERC-721 / ERC-1155 via ERC-165 erc721_result = by_id.get(6) erc1155_result = by_id.get(7) is_erc721 = erc721_result is not None and _decode_uint(erc721_result) == 1 is_erc1155 = erc1155_result is not None and _decode_uint(erc1155_result) == 1 # Detect proxy pattern (EIP-1967 implementation slot) impl_slot = "0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc" impl_result = _rpc_call("eth_getStorageAt", [addr, impl_slot, "latest"]) is_proxy = False impl_address = None if impl_result and impl_result != "0x" + "0" * 64: impl_address = "0x" + impl_result[-40:] if impl_address != "0x" + "0" * 40: is_proxy = True out: Dict[str, Any] = { "address": args.address, "is_contract": True, "code_size_bytes": code_size, "eth_balance": round(wei_to_eth(eth_balance), 18), } interfaces = [] if is_erc20: interfaces.append("ERC-20") if is_erc721: interfaces.append("ERC-721") if is_erc1155: interfaces.append("ERC-1155") if interfaces: out["detected_interfaces"] = interfaces if is_erc20: decimals = _decode_uint(decimals_raw) supply = _decode_uint(supply_raw) out["erc20"] = { "name": name, "symbol": symbol, "decimals": decimals, "total_supply": supply / (10 ** decimals) if decimals else supply, } if is_proxy: out["proxy"] = { "is_proxy": True, "implementation": impl_address, "standard": "EIP-1967", } # Check known tokens if addr in KNOWN_TOKENS: sym, tname, _ = KNOWN_TOKENS[addr] out["known_token"] = {"symbol": sym, "name": tname} print_json(out) # --------------------------------------------------------------------------- # 7. Whale Detector # --------------------------------------------------------------------------- def cmd_whales(args): """Scan the latest block for large ETH transfers with USD values.""" min_wei = int(args.min_eth * WEI_PER_ETH) block = rpc("eth_getBlockByNumber", ["latest", True]) if block is None: sys.exit("Could not retrieve latest block.") eth_price = fetch_eth_price() whales = [] for tx in (block.get("transactions") or []): value = hex_to_int(tx.get("value")) if value >= min_wei: entry: Dict[str, Any] = { "hash": tx.get("hash"), "from": tx.get("from"), "to": tx.get("to"), "value_ETH": round(wei_to_eth(value), 6), } if eth_price: entry["value_USD"] = round(wei_to_eth(value) * eth_price, 2) whales.append(entry) # Sort by value descending whales.sort(key=lambda x: x["value_ETH"], reverse=True) out: Dict[str, Any] = { "block": hex_to_int(block.get("number")), "block_time": hex_to_int(block.get("timestamp")), "min_threshold_ETH": args.min_eth, "large_transfers": whales, "note": "Scans latest block only — point-in-time snapshot.", } if eth_price: out["eth_price_usd"] = eth_price print_json(out) # --------------------------------------------------------------------------- # 8. Price Lookup # --------------------------------------------------------------------------- def cmd_price(args): """Quick price lookup for a token by contract address or known symbol.""" query = args.token # Check if it's a known symbol addr = _SYMBOL_TO_ADDRESS.get(query.upper(), query).lower() # Special case: ETH itself if addr == "eth": eth_price = fetch_eth_price() out: Dict[str, Any] = {"query": query, "token": "ETH", "name": "Ethereum"} if eth_price: out["price_usd"] = eth_price else: out["price_usd"] = None out["note"] = "Price not available." print_json(out) return # Resolve name token_meta = resolve_token_name(addr) # Fetch price prices = fetch_prices([addr]) out = {"query": query, "address": addr} if token_meta: out["name"] = token_meta["name"] out["symbol"] = token_meta["symbol"] if addr in prices: out["price_usd"] = prices[addr] else: out["price_usd"] = None out["note"] = "Price not available — token may not be listed on CoinGecko." print_json(out) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser( prog="base_client.py", description="Base blockchain query tool for Hermes Agent", ) sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("stats", help="Network stats: block, gas, chain ID, ETH price") p_wallet = sub.add_parser("wallet", help="ETH balance + ERC-20 tokens with USD values") p_wallet.add_argument("address") p_wallet.add_argument("--limit", type=int, default=20, help="Max tokens to display (default: 20)") p_wallet.add_argument("--all", action="store_true", help="Show all tokens (no limit, no dust filter)") p_wallet.add_argument("--no-prices", action="store_true", help="Skip price lookups (faster, RPC-only)") p_tx = sub.add_parser("tx", help="Transaction details by hash") p_tx.add_argument("hash") p_token = sub.add_parser("token", help="ERC-20 token metadata, price, and market cap") p_token.add_argument("address") sub.add_parser("gas", help="Gas analysis with cost estimates and L1 data fee context") p_contract = sub.add_parser("contract", help="Contract inspection: type detection, proxy check") p_contract.add_argument("address") p_whales = sub.add_parser("whales", help="Large ETH transfers in the latest block") p_whales.add_argument("--min-eth", type=float, default=1.0, help="Minimum ETH transfer size (default: 1.0)") p_price = sub.add_parser("price", help="Quick price lookup by address or symbol") p_price.add_argument("token", help="Contract address or known symbol (ETH, USDC, AERO, ...)") args = parser.parse_args() dispatch = { "stats": cmd_stats, "wallet": cmd_wallet, "tx": cmd_tx, "token": cmd_token, "gas": cmd_gas, "contract": cmd_contract, "whales": cmd_whales, "price": cmd_price, } dispatch[args.command](args) if __name__ == "__main__": main()