common.py•4.39 kB
"""
Common utilities and configuration shared across MCP servers.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional
import os
import sys
import httpx
from dataclasses import dataclass
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
@dataclass
class ChainInfo:
id: int
name: str
explorer_url: str
symbol: str = "ETH"
# Load API keys from environment
API_KEYS = {
'ethereum': os.getenv('ETHERSCAN_API_KEY', ''),
'arbitrum': os.getenv('ARBISCAN_API_KEY', ''),
'moralis': os.getenv('MORALIS_API_KEY', ''),
'optimism': os.getenv('OPTIMISTIC_ETHERSCAN_API_KEY', ''),
'base': os.getenv('BASESCAN_API_KEY', ''),
'bsc': os.getenv('BSCSCAN_API_KEY', ''),
'polygon': os.getenv('POLYGONSCAN_API_KEY', '')
}
# Default chains to check
DEFAULT_CHAINS = os.getenv('DEFAULT_CHAINS', 'ethereum,arbitrum,optimism,base').split(',')
# Supported chains with their IDs and explorer URLs
SUPPORTED_CHAINS = {
'ethereum': ChainInfo(1, 'Ethereum', 'https://api.etherscan.io/api'),
'arbitrum': ChainInfo(42161, 'Arbitrum', 'https://api.arbiscan.io/api'),
'optimism': ChainInfo(10, 'Optimism', 'https://api-optimistic.etherscan.io/api'),
'base': ChainInfo(8453, 'Base', 'https://api.basescan.org/api'),
'bsc': ChainInfo(56, 'BSC', 'https://api.bscscan.com/api', 'BNB'),
'polygon': ChainInfo(137, 'Polygon', 'https://api.polygonscan.com/api', 'MATIC'),
'solana': ChainInfo(1399811149, 'Solana', 'https://api.solscan.io', 'SOL')
}
# Configuration dictionary
config = {
'api_keys': {k: v for k, v in API_KEYS.items() if v}, # Only include non-empty API keys
'default_chains': DEFAULT_CHAINS,
'timeout': float(os.getenv('REQUEST_TIMEOUT', '20.0')),
'max_retries': int(os.getenv('MAX_RETRIES', '3'))
}
async def fetch_json(
url: str,
headers: Optional[Dict[str, str]] = None,
params: Optional[Dict[str, Any]] = None,
timeout: Optional[float] = None,
max_retries: int = 3
) -> dict[str, Any] | list[Any] | None:
"""Fetch URL and return parsed JSON or None.
Args:
url: Full URL to fetch
headers: Optional request headers
params: Optional query parameters
timeout: Request timeout in seconds
max_retries: Maximum number of retry attempts
"""
timeout = timeout or config['timeout']
default_headers = {
"Accept": "application/json",
}
if headers:
default_headers.update(headers)
response = None
async with httpx.AsyncClient(timeout=timeout, headers=default_headers) as client:
try:
print(f"DEBUG - Sending request to: {url}", file=sys.stderr)
print(f"DEBUG - Headers: {default_headers}", file=sys.stderr)
print(f"DEBUG - Params: {params}", file=sys.stderr)
response = await client.get(url, params=params)
print(f"DEBUG - Response status: {response.status_code}", file=sys.stderr)
# Only print a limited preview of the response text to avoid overwhelming logs
response_preview = response.text[:200] + '...' if len(response.text) > 200 else response.text
print(f"DEBUG - Response preview: {response_preview}", file=sys.stderr)
response.raise_for_status()
# Parse JSON with explicit error handling
try:
return response.json()
except httpx.HTTPError as json_err:
print(f"DEBUG - JSON parsing error: {str(json_err)}", file=sys.stderr)
return None
except Exception as e:
print(f"DEBUG - Error in fetch_json: {str(e)}", file=sys.stderr)
if response:
print(f"DEBUG - Response status: {getattr(response, 'status_code', None)}", file=sys.stderr)
# Only print a limited preview of the response text
response_text = getattr(response, 'text', '')
if response_text:
preview = response_text[:200] + '...' if len(response_text) > 200 else response_text
print(f"DEBUG - Response preview: {preview}", file=sys.stderr)
# Silently swallow all errors - the caller will return a readable fallback message
return None