Odos MCP Server

""" Retrieve current gas price estimates across blockchain networks. """ from datetime import datetime from typing import Any, Dict, List, Optional from pydantic import BaseModel from ...helpers import make_zerion_request from ...mcp import mcp class GetGasPricesArgs(BaseModel): chain_ids: Optional[List[str]] = None gas_types: Optional[List[str]] = None def _build_gas_params(args: GetGasPricesArgs) -> Dict[str, str]: """Build query parameters for gas prices request""" params = {} if args.chain_ids: params["filter[chain_ids]"] = ",".join(args.chain_ids) if args.gas_types: params["filter[gas_types]"] = ",".join(args.gas_types) return params def _group_by_chain(gas_data: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]: """Group gas data by chain for better readability""" chains: Dict[str, List[Dict[str, Any]]] = {} for item in gas_data: chain_info = item.get("relationships", {}).get("chain", {}).get("data", {}) chain_id = chain_info.get("id", "unknown") if chain_id not in chains: chains[chain_id] = [] chains[chain_id].append(item) return chains def _format_time_estimation(estimation_seconds: int) -> str: """Format time estimation in minutes""" return f"~{estimation_seconds // 60}min" def _format_eip1559_gas(info: Dict[str, Any]) -> str: """Format EIP-1559 gas information""" output = "" base_fee = info.get("base_fee", 0) / 1e9 # Convert to gwei output += f" • Base Fee: {base_fee:.1f} gwei\n" slow = info.get("slow", {}) if slow: slow_max = slow.get("max_fee", 0) / 1e9 slow_time = slow.get("estimation_seconds", 0) output += ( f" • Slow: {slow_max:.1f} gwei ({_format_time_estimation(slow_time)})\n" ) standard = info.get("standard", {}) if standard: std_max = standard.get("max_fee", 0) / 1e9 std_time = standard.get("estimation_seconds", 0) output += ( f" • Standard: {std_max:.1f} gwei ({_format_time_estimation(std_time)})\n" ) fast = info.get("fast", {}) if fast: fast_max = fast.get("max_fee", 0) / 1e9 fast_time = fast.get("estimation_seconds", 0) output += ( f" • Fast: {fast_max:.1f} gwei ({_format_time_estimation(fast_time)})\n" ) return output def _format_classic_gas(info: Dict[str, Any]) -> str: """Format classic gas information""" slow = info.get("slow", 0) / 1e9 standard = info.get("standard", 0) / 1e9 fast = info.get("fast", 0) / 1e9 output = f" • Slow: {slow:.1f} gwei\n" output += f" • Standard: {standard:.1f} gwei\n" output += f" • Fast: {fast:.1f} gwei\n" return output def _format_optimistic_gas(info: Dict[str, Any]) -> str: """Format Optimism L2 gas information""" l1_cost = info.get("l1", 0) / 1e9 l2_cost = info.get("l2", 0) / 1e9 output = f" • L1 Cost: {l1_cost:.1f} gwei\n" output += f" • L2 Cost: {l2_cost:.1f} gwei\n" return output def _format_timestamp(updated_at: str) -> str: """Format timestamp for display""" if not updated_at: return "" try: dt = datetime.fromisoformat(updated_at.replace("Z", "+00:00")) time_str = dt.strftime("%H:%M UTC") return f" • Updated: {time_str}\n" except (ValueError, TypeError): return "" def _format_gas_item(item: Dict[str, Any]) -> str: """Format a single gas price item""" attrs = item.get("attributes", {}) gas_type = attrs.get("gas_type", "unknown") info = attrs.get("info", {}) updated_at = attrs.get("updated_at", "") output = "" if gas_type == "eip1559": output += _format_eip1559_gas(info) elif gas_type == "classic": output += _format_classic_gas(info) elif gas_type == "optimistic": output += _format_optimistic_gas(info) output += _format_timestamp(updated_at) return output def _format_chain_gas_data(chain_id: str, chain_data: List[Dict[str, Any]]) -> str: """Format gas data for a single chain""" chain_name = chain_id.replace("-", " ").title() output = f"🌐 {chain_name.upper()}\n" for item in chain_data: output += _format_gas_item(item) output += "\n" return output @mcp.tool( name="get_gas_prices", description="Get current gas prices across blockchain networks using Zerion API", ) async def get_gas_prices(args: GetGasPricesArgs): try: # Build params and fetch data params = _build_gas_params(args) data = await make_zerion_request(query="/gas-prices/", params=params) # Parse and format response gas_data = data.get("data", []) if not gas_data: return "❌ No gas price data available" output = "⛽ GAS PRICES\n" output += "--------------------------------------------------\n\n" # Group by chain and format chains = _group_by_chain(gas_data) for chain_id, chain_data in chains.items(): output += _format_chain_gas_data(chain_id, chain_data) return output.strip() except ConnectionError as e: return f"❌ API Error fetching gas prices: {str(e)}" except Exception as e: # pylint: disable=broad-except return f"❌ Unexpected Error fetching gas prices: {str(e)}"