Odos MCP Server

""" Retrieve and analyze historical wallet transaction data. """ from typing import Any, Dict, List, Optional from pydantic import BaseModel from ...helpers import make_zerion_request from ...mcp import mcp class TransactionArgs(BaseModel): wallet_address: str days_back: int = 30 page_size: int = 50 contract_filter: Optional[str] = None input_tokens: Optional[List[str]] = None output_tokens: Optional[List[str]] = None min_trade_value: Optional[float] = None max_trade_value: Optional[float] = None @mcp.tool( name="get_past_transactions", description=( "Fetch transactions with comprehensive information. " "For Odos, use get_router_address and then pass that in as contract filter." ), ) async def get_past_transactions(args: TransactionArgs) -> str: """Get transactions with comprehensive information retention""" try: params = {"page[size]": str(args.page_size)} if args.contract_filter: params["filter[search_query]"] = args.contract_filter data = await make_zerion_request( f"/wallets/{args.wallet_address}/transactions", params ) if not data or "data" not in data: return "No transaction data found" trades = data["data"] if not trades: return f"No transactions found in the last {args.days_back} days" filtered_trades = apply_filters(trades, args) if not filtered_trades: return f"No transactions match filters. Found {len(trades)} total trades." return format_transactions( args.wallet_address, filtered_trades, args.days_back, args.contract_filter ) except ConnectionError as e: return f"❌ API Error in get_past_transactions: {str(e)}" except Exception as e: # pylint: disable=broad-except return f"❌ Unexpected Error in get_past_transactions: {str(e)}" def apply_filters(trades: List[Dict], args: TransactionArgs) -> List[Dict]: """Apply filters to trades""" if not any( [ args.input_tokens, args.output_tokens, args.min_trade_value, args.max_trade_value, ] ): return trades filtered = [] for trade in trades: parsed = parse_transaction(trade) if args.input_tokens: input_symbols = [ t["token"]["symbol"].upper() for t in parsed["transfers"] if t["direction"] == "out" ] if not any(token.upper() in input_symbols for token in args.input_tokens): continue if args.output_tokens: output_symbols = [ t["token"]["symbol"].upper() for t in parsed["transfers"] if t["direction"] == "in" ] if not any(token.upper() in output_symbols for token in args.output_tokens): continue trade_value = sum( t["value_usd"] for t in parsed["transfers"] if t["direction"] == "out" ) if args.min_trade_value and trade_value < args.min_trade_value: continue if args.max_trade_value and trade_value > args.max_trade_value: continue filtered.append(trade) return filtered def _parse_fee(attrs: Dict) -> Dict[str, Any]: """Parse fee information from transaction attributes""" fee = attrs.get("fee", {}) or {} if not fee: return {} fee_data = { "value_usd": fee.get("value", 0), "price_per_unit": fee.get("price", 0), } fee_quantity = fee.get("quantity", {}) or {} if fee_quantity: fee_data["quantity"] = { "int": fee_quantity.get("int", "0"), "decimals": fee_quantity.get("decimals", 18), "float": fee_quantity.get("float", 0), "numeric": fee_quantity.get("numeric", "0"), } fee_fungible = fee.get("fungible_info", {}) or {} if fee_fungible: fee_data["token"] = { "name": fee_fungible.get("name", ""), "symbol": fee_fungible.get("symbol", ""), "icon_url": ( fee_fungible.get("icon", {}).get("url", "") if fee_fungible.get("icon") else "" ), "flags": fee_fungible.get("flags", {}), "implementations": fee_fungible.get("implementations", []), } return fee_data def _parse_application_metadata(attrs: Dict) -> Dict[str, Any]: """Parse application metadata from transaction attributes""" app_meta = attrs.get("application_metadata", {}) or {} if not app_meta: return {} app_data = { "name": app_meta.get("name", ""), "contract_address": app_meta.get("contract_address", ""), "icon_url": ( app_meta.get("icon", {}).get("url", "") if app_meta.get("icon") else "" ), } method_info = app_meta.get("method", {}) or {} if method_info: app_data["method"] = { "id": method_info.get("id", ""), "name": method_info.get("name", ""), } return app_data def _parse_transfers(attrs: Dict): """Parse transfer information from transaction attributes""" transfers = [] input_total_usd = 0 output_total_usd = 0 for transfer in attrs.get("transfers", []): fungible = transfer.get("fungible_info", {}) or {} token_info = { "name": fungible.get("name", ""), "symbol": fungible.get("symbol", ""), "icon_url": ( fungible.get("icon", {}).get("url", "") if fungible.get("icon") else "" ), "flags": fungible.get("flags", {}), "implementations": fungible.get("implementations", []), } quantity = transfer.get("quantity", {}) or {} quantity_info = { "int": quantity.get("int", "0"), "decimals": quantity.get("decimals", 18), "float": quantity.get("float", 0), "numeric": quantity.get("numeric", "0"), } direction = transfer.get("direction", "") value_usd = transfer.get("value", 0) or 0 price = transfer.get("price", 0) or 0 transfer_data = { "token": token_info, "quantity": quantity_info, "direction": direction, "value_usd": value_usd, "price_per_token": price, "sender": transfer.get("sender", ""), "recipient": transfer.get("recipient", ""), } transfers.append(transfer_data) if direction == "out": input_total_usd += value_usd elif direction == "in": output_total_usd += value_usd return transfers, input_total_usd, output_total_usd def _parse_approvals(attrs: Dict) -> list: """Parse approval information from transaction attributes""" approvals = attrs.get("approvals", []) or [] result = [] for approval in approvals: approval_token = approval.get("fungible_info", {}) or {} approval_quantity = approval.get("quantity", {}) or {} approval_data = { "token": { "name": approval_token.get("name", ""), "symbol": approval_token.get("symbol", ""), "icon_url": ( approval_token.get("icon", {}).get("url", "") if approval_token.get("icon") else "" ), "flags": approval_token.get("flags", {}), "implementations": approval_token.get("implementations", []), }, "quantity": { "int": approval_quantity.get("int", "0"), "decimals": approval_quantity.get("decimals", 18), "float": approval_quantity.get("float", 0), "numeric": approval_quantity.get("numeric", "0"), }, "spender": approval.get("spender", ""), "value_usd": approval.get("value", 0), } result.append(approval_data) return result def _parse_relationships(tx: Dict) -> Dict[str, Any]: """Parse relationship information from transaction""" relationships = tx.get("relationships", {}) or {} result = {} chain_rel = relationships.get("chain", {}) or {} chain_data = chain_rel.get("data", {}) or {} if chain_data: result["chain"] = { "type": chain_data.get("type", ""), "id": chain_data.get("id", ""), "links": chain_rel.get("links", {}), } dapp_rel = relationships.get("dapp", {}) or {} dapp_data = dapp_rel.get("data", {}) or {} if dapp_data: result["dapp"] = { "type": dapp_data.get("type", ""), "id": dapp_data.get("id", ""), "links": dapp_rel.get("links", {}), } return result def _parse_gas_metrics(fee: Dict, input_total_usd: float) -> Dict[str, Any]: """Parse gas efficiency metrics""" fee_value = fee.get("value_usd", 0) or 0 if fee_value > 0 and input_total_usd > 0: gas_ratio = (fee_value / input_total_usd) * 100 return { "fee_to_trade_ratio_percent": gas_ratio, "fee_usd": fee_value, "trade_value_usd": input_total_usd, } return {} def format_transactions( wallet_address: str, trades: List[Dict], days_back: int, contract_filter: Optional[str] = None, ) -> str: """Format transactions with maximum information retention""" output = _format_header(wallet_address, days_back, contract_filter) if not trades: return output + "No transactions to display." for i, trade_data in enumerate(trades): parsed = parse_transaction(trade_data) output += _format_transaction(parsed, i) if len(trades) > 15: output += f"Additional {len(trades) - 15} transactions not displayed\n" return output def _format_header( wallet_address: str, days_back: int, contract_filter: Optional[str] ) -> str: """Format transaction analysis header""" output = "TRANSACTION ANALYSIS\n" output += f"Wallet: {wallet_address}\n" output += f"Analysis period: {days_back} days\n" if contract_filter: output += f"Filtered by contract: {contract_filter}\n" output += "--------------------------------------------------\n\n" return output def _format_basic_info(parsed: Dict, i: int) -> str: """Format basic transaction information""" output = f"TRANSACTION {i}\n" output += f"Transaction ID: {parsed.get('transaction_id', '')}\n" output += f"Transaction Type: {parsed.get('transaction_type', '')}\n" output += f"Hash: {parsed.get('hash', '')}\n" output += f"Timestamp: {parsed.get('timestamp', '')}\n" output += f"Status: {parsed.get('status', '')}\n" output += f"Operation Type: {parsed.get('operation_type', '')}\n" output += f"Block Number: {parsed.get('mined_at_block', 0)}\n" output += f"Nonce: {parsed.get('nonce', 0)}\n" output += f"Sent From: {parsed.get('sent_from', '')}\n" output += f"Sent To: {parsed.get('sent_to', '')}\n" return output def _format_application_info(app_info: Dict) -> str: """Format application information""" if app_info: app_name = app_info.get("name", "N/A") contract_addr = app_info.get("contract_address", "N/A") method_name = app_info.get("method", {}).get("name", "N/A") return f"Application: {app_name} ({method_name}) @ {contract_addr}\n" return "Application: Not available\n" def _format_fee_info(fee_info: Dict) -> str: """Format fee information""" if not fee_info: return "" output = f"Fee Value USD: {fee_info.get('value_usd', 0)}\n" output += f"Fee Price Per Unit: {fee_info.get('price_per_unit', 0)}\n" fee_quantity = fee_info.get("quantity", {}) if fee_quantity: output += f"Fee Amount Int: {fee_quantity.get('int', '0')}\n" output += f"Fee Amount Decimals: {fee_quantity.get('decimals', 18)}\n" output += f"Fee Amount Float: {fee_quantity.get('float', 0)}\n" output += f"Fee Amount Numeric: {fee_quantity.get('numeric', '0')}\n" fee_token = fee_info.get("token", {}) if fee_token: output += f"Fee Token Name: {fee_token.get('name', '')}\n" output += f"Fee Token Symbol: {fee_token.get('symbol', '')}\n" output += f"Fee Token Icon: {fee_token.get('icon_url', '')}\n" output += f"Fee Token Flags: {fee_token.get('flags', {})}\n" implementations = fee_token.get("implementations", []) if implementations: output += f"Fee Token Implementations Count: {len(implementations)}\n" for j, impl in enumerate(implementations[:5], 1): output += ( f" Implementation {j}: Chain {impl.get('chain_id', '')}, " f"Address{impl.get('address', '')},Decimals{impl.get('decimals', 18)}\n" ) return output def _format_trade_summary_info(trade_summary: Dict) -> str: """Format trade summary information""" if not trade_summary: return "" output = "Trade Summary:\n" output += f" Input Count: {trade_summary.get('input_count', 0)}\n" output += f" Output Count: {trade_summary.get('output_count', 0)}\n" output += f" Input Total USD: {trade_summary.get('input_total_usd', 0)}\n" output += f" Output Total USD: {trade_summary.get('output_total_usd', 0)}\n" output += f" Net Change USD: {trade_summary.get('net_change_usd', 0)}\n" output += f" Net Change Percent: {trade_summary.get('net_change_percent', 0)}\n" # Input tokens input_tokens = trade_summary.get("input_tokens", []) if input_tokens: output += "Input Tokens:\n" for j, token in enumerate(input_tokens, 1): output += f" Input Token {j}:\n" output += f" Symbol: {token.get('symbol', '')}\n" output += f" Name: {token.get('name', '')}\n" output += f" Amount Int: {token.get('amount_int', '0')}\n" output += f" Amount Float: {token.get('amount_float', 0)}\n" output += f" Amount Numeric: {token.get('amount_numeric', '0')}\n" output += f" Decimals: {token.get('decimals', 18)}\n" output += f" Value USD: {token.get('value_usd', 0)}\n" output += f" Price Per Token: {token.get('price_per_token', 0)}\n" output += f" Sender: {token.get('sender', '')}\n" output += f" Recipient: {token.get('recipient', '')}\n" # Output tokens output_tokens = trade_summary.get("output_tokens", []) if output_tokens: output += "Output Tokens:\n" for j, token in enumerate(output_tokens, 1): output += f" Output Token {j}:\n" output += f" Symbol: {token.get('symbol', '')}\n" output += f" Name: {token.get('name', '')}\n" output += f" Amount Int: {token.get('amount_int', '0')}\n" output += f" Amount Float: {token.get('amount_float', 0)}\n" output += f" Amount Numeric: {token.get('amount_numeric', '0')}\n" output += f" Decimals: {token.get('decimals', 18)}\n" output += f" Value USD: {token.get('value_usd', 0)}\n" output += f" Price Per Token: {token.get('price_per_token', 0)}\n" output += f" Sender: {token.get('sender', '')}\n" output += f" Recipient: {token.get('recipient', '')}\n" return output def _format_transfer_details(transfers: list, transfer_count: int) -> str: """Format complete transfer details""" output = f"Transfer Count: {transfer_count}\n" for j, transfer in enumerate(transfers, 1): output += f"Transfer {j}:\n" output += f" Direction: {transfer.get('direction', '')}\n" output += f" Sender: {transfer.get('sender', '')}\n" output += f" Recipient: {transfer.get('recipient', '')}\n" output += f" Value USD: {transfer.get('value_usd', 0)}\n" output += f" Price Per Token: {transfer.get('price_per_token', 0)}\n" token = transfer.get("token", {}) output += f" Token Name: {token.get('name', '')}\n" output += f" Token Symbol: {token.get('symbol', '')}\n" output += f" Token Icon: {token.get('icon_url', '')}\n" output += f" Token Flags: {token.get('flags', {})}\n" quantity = transfer.get("quantity", {}) output += f" Quantity Int: {quantity.get('int', '0')}\n" output += f" Quantity Decimals: {quantity.get('decimals', 18)}\n" output += f" Quantity Float: {quantity.get('float', 0)}\n" output += f" Quantity Numeric: {quantity.get('numeric', '0')}\n" implementations = token.get("implementations", []) if implementations: output += f" Token Implementations Count: {len(implementations)}\n" for k, impl in enumerate(implementations[:3], 1): output += ( f" Implementation {k}: Chain {impl.get('chain_id', '')}, " f"Address {impl.get('address', '')}, Decimals {impl.get('decimals', 18)}\n" ) return output def _format_chain_dapp_info(parsed: Dict) -> str: """Format chain and dapp information""" output = "" # Chain information chain_info = parsed.get("chain", {}) if chain_info: output += f"Chain Type: {chain_info.get('type', '')}\n" output += f"Chain ID: {chain_info.get('id', '')}\n" output += f"Chain Links: {chain_info.get('links', {})}\n" # DApp information dapp_info = parsed.get("dapp", {}) if dapp_info: output += f"DApp Type: {dapp_info.get('type', '')}\n" output += f"DApp ID: {dapp_info.get('id', '')}\n" output += f"DApp Links: {dapp_info.get('links', {})}\n" return output def _format_gas_approval_info(parsed: Dict) -> str: """Format gas metrics and approval information""" output = "" # Gas metrics gas_metrics = parsed.get("gas_metrics", {}) if gas_metrics: output += ( f"Fee to Trade Ratio Percent:" f"{gas_metrics.get('fee_to_trade_ratio_percent', 0)}\n" ) output += f"Fee USD: {gas_metrics.get('fee_usd', 0)}\n" output += f"Trade Value USD: {gas_metrics.get('trade_value_usd', 0)}\n" # Approvals approvals = parsed.get("approvals", []) if approvals: output += f"Approvals Count: {len(approvals)}\n" for j, approval in enumerate(approvals, 1): output += f"Approval {j}:\n" output += f" Spender: {approval.get('spender', '')}\n" output += f" Value USD: {approval.get('value_usd', 0)}\n" approval_token = approval.get("token", {}) output += f" Token Name: {approval_token.get('name', '')}\n" output += f" Token Symbol: {approval_token.get('symbol', '')}\n" output += f" Token Flags: {approval_token.get('flags', {})}\n" approval_quantity = approval.get("quantity", {}) output += f" Quantity Int: {approval_quantity.get('int', '0')}\n" output += f" Quantity Float: {approval_quantity.get('float', 0)}\n" output += f" Quantity Numeric: {approval_quantity.get('numeric', '0')}\n" output += f" Quantity Decimals: {approval_quantity.get('decimals', 18)}\n" return output def _format_flags_protocol_info(parsed: Dict) -> str: """Format flags and protocol information""" output = "" # Flags flags = parsed.get("flags", {}) if flags: output += f"Flags: {flags}\n" # Protocol Information (if available, e.g., Odos specific) protocol_info = parsed.get("protocol_info", {}) if protocol_info: output += "Protocol Info:\n" for key, value in protocol_info.items(): output += f" {key.replace('_', ' ').title()}: {value}\n" return output def _format_transaction(parsed: Dict, i: int) -> str: """Format a single transaction with all details""" output = _format_basic_info(parsed, i) output += _format_application_info(parsed.get("application", {})) output += _format_fee_info(parsed.get("fee", {})) output += _format_trade_summary_info(parsed.get("trade_summary", {})) output += _format_transfer_details( parsed.get("transfers", []), parsed.get("transfer_count", 0) ) output += _format_chain_dapp_info(parsed) output += _format_gas_approval_info(parsed) output += _format_flags_protocol_info(parsed) output += "--------------------------------------------------\n" return output def parse_transaction(tx: Dict) -> Dict[str, Any]: """Parse transaction with maximum information retention""" attrs = tx.get("attributes", {}) or {} # Core transaction data parsed = { "transaction_id": tx.get("id", ""), "transaction_type": tx.get("type", ""), "hash": attrs.get("hash", ""), "timestamp": attrs.get("mined_at", ""), "status": attrs.get("status", ""), "operation_type": attrs.get("operation_type", ""), "mined_at_block": attrs.get("mined_at_block", 0), "nonce": attrs.get("nonce", 0), "sent_from": attrs.get("sent_from", ""), "sent_to": attrs.get("sent_to", ""), } # Use helper functions to parse sections fee = _parse_fee(attrs) if fee: parsed["fee"] = fee app_meta = _parse_application_metadata(attrs) if app_meta: parsed["application"] = app_meta transfers, input_total_usd, output_total_usd = _parse_transfers(attrs) parsed["transfers"] = transfers parsed["transfer_count"] = len(transfers) # Build trade summary input_transfers = [t for t in transfers if t["direction"] == "out"] output_transfers = [t for t in transfers if t["direction"] == "in"] parsed["trade_summary"] = { "input_count": len(input_transfers), "output_count": len(output_transfers), "input_total_usd": input_total_usd, "output_total_usd": output_total_usd, "net_change_usd": output_total_usd - input_total_usd, "net_change_percent": ( ((output_total_usd - input_total_usd) / input_total_usd * 100) if input_total_usd > 0 else 0 ), } # Add detailed token information to trade summary if input_transfers: parsed["trade_summary"]["input_tokens"] = [ { "symbol": t["token"]["symbol"], "name": t["token"]["name"], "amount_int": t["quantity"]["int"], "amount_float": t["quantity"]["float"], "amount_numeric": t["quantity"]["numeric"], "decimals": t["quantity"]["decimals"], "value_usd": t["value_usd"], "price_per_token": t["price_per_token"], "sender": t["sender"], "recipient": t["recipient"], } for t in input_transfers ] if output_transfers: parsed["trade_summary"]["output_tokens"] = [ { "symbol": t["token"]["symbol"], "name": t["token"]["name"], "amount_int": t["quantity"]["int"], "amount_float": t["quantity"]["float"], "amount_numeric": t["quantity"]["numeric"], "decimals": t["quantity"]["decimals"], "value_usd": t["value_usd"], "price_per_token": t["price_per_token"], "sender": t["sender"], "recipient": t["recipient"], } for t in output_transfers ] # Use helper functions for remaining sections approvals = _parse_approvals(attrs) if approvals: parsed["approvals"] = approvals parsed["flags"] = attrs.get("flags", {}) relationships = _parse_relationships(tx) parsed.update(relationships) gas_metrics = _parse_gas_metrics(fee, input_total_usd) if gas_metrics: parsed["gas_metrics"] = gas_metrics return parsed