diff --git a/optional-skills/blockchain/base/SKILL.md b/optional-skills/blockchain/base/SKILL.md new file mode 100644 index 000000000..a1d197147 --- /dev/null +++ b/optional-skills/blockchain/base/SKILL.md @@ -0,0 +1,231 @@ +--- +name: base +description: Query Base (Ethereum L2) blockchain data with USD pricing — wallet balances, token info, transaction details, gas analysis, contract inspection, whale detection, and live network stats. Uses Base RPC + CoinGecko. No API key required. +version: 0.1.0 +author: youssefea +license: MIT +metadata: + hermes: + tags: [Base, Blockchain, Crypto, Web3, RPC, DeFi, EVM, L2, Ethereum] + related_skills: [] +--- + +# Base Blockchain Skill + +Query Base (Ethereum L2) on-chain data enriched with USD pricing via CoinGecko. +8 commands: wallet portfolio, token info, transactions, gas analysis, +contract inspection, whale detection, network stats, and price lookup. + +No API key needed. Uses only Python standard library (urllib, json, argparse). + +--- + +## When to Use + +- User asks for a Base wallet balance, token holdings, or portfolio value +- User wants to inspect a specific transaction by hash +- User wants ERC-20 token metadata, price, supply, or market cap +- User wants to understand Base gas costs and L1 data fees +- User wants to inspect a contract (ERC type detection, proxy resolution) +- User wants to find large ETH transfers (whale detection) +- User wants Base network health, gas price, or ETH price +- User asks "what's the price of USDC/AERO/DEGEN/ETH?" + +--- + +## Prerequisites + +The helper script uses only Python standard library (urllib, json, argparse). +No external packages required. + +Pricing data comes from CoinGecko's free API (no key needed, rate-limited +to ~10-30 requests/minute). For faster lookups, use `--no-prices` flag. + +--- + +## Quick Reference + +RPC endpoint (default): https://mainnet.base.org +Override: export BASE_RPC_URL=https://your-private-rpc.com + +Helper script path: ~/.hermes/skills/blockchain/base/scripts/base_client.py + +``` +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 stats +python3 base_client.py price +``` + +--- + +## Procedure + +### 0. Setup Check + +```bash +python3 --version + +# Optional: set a private RPC for better rate limits +export BASE_RPC_URL="https://mainnet.base.org" + +# Confirm connectivity +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py stats +``` + +### 1. Wallet Portfolio + +Get ETH balance and ERC-20 token holdings with USD values. +Checks ~15 well-known Base tokens (USDC, WETH, AERO, DEGEN, etc.) +via on-chain `balanceOf` calls. Tokens sorted by value, dust filtered. + +```bash +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py \ + wallet 0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045 +``` + +Flags: +- `--limit N` — show top N tokens (default: 20) +- `--all` — show all tokens, no dust filter, no limit +- `--no-prices` — skip CoinGecko price lookups (faster, RPC-only) + +Output includes: ETH balance + USD value, token list with prices sorted +by value, dust count, total portfolio value in USD. + +Note: Only checks known tokens. Unknown ERC-20s are not discovered. +Use the `token` command with a specific contract address for any token. + +### 2. Transaction Details + +Inspect a full transaction by its hash. Shows ETH value transferred, +gas used, fee in ETH/USD, status, and decoded ERC-20/ERC-721 transfers. + +```bash +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py \ + tx 0xabc123...your_tx_hash_here +``` + +Output: hash, block, from, to, value (ETH + USD), gas price, gas used, +fee, status, contract creation address (if any), token transfers. + +### 3. Token Info + +Get ERC-20 token metadata: name, symbol, decimals, total supply, price, +market cap, and contract code size. + +```bash +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py \ + token 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913 +``` + +Output: name, symbol, decimals, total supply, price, market cap. +Reads name/symbol/decimals directly from the contract via eth_call. + +### 4. Gas Analysis + +Detailed gas analysis with cost estimates for common operations. +Shows current gas price, base fee trends over 10 blocks, block +utilization, and estimated costs for ETH transfers, ERC-20 transfers, +and swaps. + +```bash +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py gas +``` + +Output: current gas price, base fee, block utilization, 10-block trend, +cost estimates in ETH and USD. + +Note: Base is an L2 — actual transaction costs include an L1 data +posting fee that depends on calldata size and L1 gas prices. The +estimates shown are for L2 execution only. + +### 5. Contract Inspection + +Inspect an address: determine if it's an EOA or contract, detect +ERC-20/ERC-721/ERC-1155 interfaces, resolve EIP-1967 proxy +implementation addresses. + +```bash +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py \ + contract 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913 +``` + +Output: is_contract, code size, ETH balance, detected interfaces +(ERC-20, ERC-721, ERC-1155), ERC-20 metadata, proxy implementation +address. + +### 6. Whale Detector + +Scan the most recent block for large ETH transfers with USD values. + +```bash +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py \ + whales --min-eth 1.0 +``` + +Note: scans the latest block only — point-in-time snapshot, not historical. +Default threshold is 1.0 ETH (lower than Solana's default since ETH +values are higher). + +### 7. Network Stats + +Live Base network health: latest block, chain ID, gas price, base fee, +block utilization, transaction count, and ETH price. + +```bash +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py stats +``` + +### 8. Price Lookup + +Quick price check for any token by contract address or known symbol. + +```bash +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py price ETH +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py price USDC +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py price AERO +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py price DEGEN +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py price 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913 +``` + +Known symbols: ETH, WETH, USDC, cbETH, AERO, DEGEN, TOSHI, BRETT, +WELL, wstETH, rETH, cbBTC. + +--- + +## Pitfalls + +- **CoinGecko rate-limits** — free tier allows ~10-30 requests/minute. + Price lookups use 1 request per token. Use `--no-prices` for speed. +- **Public RPC rate-limits** — Base's public RPC limits requests. + For production use, set BASE_RPC_URL to a private endpoint + (Alchemy, QuickNode, Infura). +- **Wallet shows known tokens only** — unlike Solana, EVM chains have no + built-in "get all tokens" RPC. The wallet command checks ~15 popular + Base tokens via `balanceOf`. Unknown ERC-20s won't appear. Use the + `token` command for any specific contract. +- **Token names read from contract** — if a contract doesn't implement + `name()` or `symbol()`, these fields may be empty. Known tokens have + hardcoded labels as fallback. +- **Gas estimates are L2 only** — Base transaction costs include an L1 + data posting fee (depends on calldata size and L1 gas prices). The gas + command estimates L2 execution cost only. +- **Whale detector scans latest block only** — not historical. Results + vary by the moment you query. Default threshold is 1.0 ETH. +- **Proxy detection** — only EIP-1967 proxies are detected. Other proxy + patterns (EIP-1167 minimal proxy, custom storage slots) are not checked. +- **Retry on 429** — both RPC and CoinGecko calls retry up to 2 times + with exponential backoff on rate-limit errors. + +--- + +## Verification + +```bash +# Should print Base chain ID (8453), latest block, gas price, and ETH price +python3 ~/.hermes/skills/blockchain/base/scripts/base_client.py stats +``` diff --git a/optional-skills/blockchain/base/scripts/base_client.py b/optional-skills/blockchain/base/scripts/base_client.py new file mode 100644 index 000000000..cafffb49f --- /dev/null +++ b/optional-skills/blockchain/base/scripts/base_client.py @@ -0,0 +1,1008 @@ +#!/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()