Odos MCP Server

""" Helper functions for Odos API client """ import base64 import os from typing import Any, Dict, Optional, Union, cast import requests from dotenv import load_dotenv from .constants import CHAIN_ALIASES, CHAIN_INFO, COMMON_TOKENS, ODOS_API_BASE load_dotenv() async def make_odos_request( path: str, method: str = "GET", payload: Optional[Dict] = None, params: Optional[Dict] = None, ) -> Dict[str, Any]: """Make a request to the Odos API""" url = f"{ODOS_API_BASE}{path}" headers = {"Content-Type": "application/json"} try: if method.upper() == "POST": response = requests.post( url, json=payload, headers=headers, params=params, timeout=30 ) elif method.upper() == "GET": response = requests.get(url, headers=headers, params=params, timeout=30) else: raise ValueError(f"Unsupported HTTP method: {method}") response.raise_for_status() return cast(Dict[str, Any], response.json()) except requests.exceptions.HTTPError as e: error_text = e.response.text if e.response else "No response text" raise ConnectionError( f"API Error({e.response.status_code if e.response else 'Unknown status'}):{error_text}" ) from e except requests.exceptions.RequestException as e: raise ConnectionError(f"Odos API Request Error: {e}") from e def resolve_chain_id(chain_identifier: Union[str, int]) -> int: """Resolve a chain identifier to a chain ID""" if isinstance(chain_identifier, int): if chain_identifier in CHAIN_INFO: return chain_identifier raise ValueError(f"Unsupported chain ID: {chain_identifier}") lowercased_identifier = chain_identifier.lower() chain_id = CHAIN_ALIASES.get(lowercased_identifier) if chain_id: return chain_id # Try parsing as number try: numeric_id = int(chain_identifier) if numeric_id in CHAIN_INFO: return numeric_id except ValueError: pass # Iterate directly over dictionaries supported_chains = list(CHAIN_ALIASES) + [str(k) for k in CHAIN_INFO] raise ValueError( f"Unknown chain: {chain_identifier}. Supported chains: {', '.join(supported_chains)}" ) async def resolve_token_address(chain_id: int, token_identifier: str) -> str: """Resolve a token identifier to a token address""" # Check if it's already an address if token_identifier.startswith("0x") and len(token_identifier) == 42: return token_identifier # First try common tokens (avoids API call) common_token = COMMON_TOKENS.get(chain_id, {}).get(token_identifier.upper()) if common_token: return common_token try: tokens_response: Dict[str, Any] = await make_odos_request( f"/info/tokens/{chain_id}" ) token_map_data = tokens_response.get("tokenMap") if not isinstance(token_map_data, dict): print(f"Warning: 'tokenMap'or is missing in response for chain {chain_id}.") return token_identifier token_map: Dict[str, Dict[str, Any]] = cast( Dict[str, Dict[str, Any]], token_map_data ) identifier_lower = token_identifier.lower() for address, token_info in token_map.items(): if not isinstance(token_info, dict): continue if ( token_info.get("symbol", "").lower() == identifier_lower or token_info.get("name", "").lower() == identifier_lower or token_info.get("assetId", "").lower() == identifier_lower ): return cast(str, address) except ConnectionError as e: # pylint: disable=broad-except print( f" Token API failed for {token_identifier} on chain {chain_id} due to: {e}" ) except (TypeError, KeyError, AttributeError) as e: # pylint: disable=broad-except print( f"Token API failed for {token_identifier} on chain {chain_id} due to: {e}" ) except Exception as e: # pylint: disable=broad-except print( f"Token API lookup failed for {token_identifier} on chain {chain_id} due to: {e}" ) return token_identifier async def make_zerion_request(query: str, params: Dict[str, str]) -> Dict[str, Any]: """ Fetch wallet portfolio data from Zerion API """ url = f"https://api.zerion.io/v1{query}" api_key_str = os.getenv("ZERION_API_KEY") if api_key_str is None: raise ValueError("ZERION_API_KEY environment variable not set.") api_key_bytes = api_key_str.encode("utf-8") headers = { "accept": "application/json", "authorization": f"Basic {base64.b64encode(api_key_bytes).decode('utf-8')}", } try: response = requests.get(url, headers=headers, params=params, timeout=30) response.raise_for_status() return cast(Dict[str, Any], response.json()) except requests.exceptions.RequestException as e: raise ConnectionError(f"Zerion API Error: {e}") from e