Odos MCP Server

""" Analyze wallet holdings and portfolio composition via Zerion API. """ from pydantic import BaseModel from ...helpers import make_zerion_request from ...mcp import mcp class GetWalletPortfolioArgs(BaseModel): wallet_address: str def _extract_portfolio_totals(portfolio: dict) -> tuple[float, float, float]: """Extract total portfolio values and changes""" total_value = portfolio["total"]["positions"] change_24h_usd = portfolio["changes"]["absolute_1d"] change_24h_percent = portfolio["changes"]["percent_1d"] return total_value, change_24h_usd, change_24h_percent def _extract_position_breakdown(portfolio: dict) -> dict: """Extract position breakdown by type""" pos_by_type = portfolio["positions_distribution_by_type"] return { "wallet": pos_by_type.get("wallet", 0), "deposited": pos_by_type.get("deposited", 0), "borrowed": pos_by_type.get("borrowed", 0), "locked": pos_by_type.get("locked", 0), "staked": pos_by_type.get("staked", 0), } def _get_top_chains(portfolio: dict) -> list: """Get top 5 chains by value""" pos_by_chain = portfolio["positions_distribution_by_chain"] return sorted(pos_by_chain.items(), key=lambda x: x[1], reverse=True)[:5] def _format_portfolio_header( wallet_address: str, total_value: float, change_24h_usd: float, change_24h_percent: float, ) -> str: """Format the portfolio header section""" change_indicator = "+" if change_24h_usd >= 0 else "" return ( "WALLET PORTFOLIO ANALYSIS\n" f"Address: {wallet_address[:6]}...{wallet_address[-4:]}\n" f"Total Value: ${total_value:,.2f}\n" f"24h Change: {change_indicator}${change_24h_usd:,.2f}" f" ({change_indicator}{change_24h_percent:.2f}%)\n\n" "POSITION BREAKDOWN:\n" ) def _format_position_breakdown(positions: dict, total_value: float) -> str: """Format the position breakdown section""" wallet_val = positions["wallet"] output = f"Wallet:${wallet_val:,.2f}({wallet_val/total_value*100:.1f}%)\n" if positions["deposited"] > 0: dep_val = positions["deposited"] output += f"Deposited:${dep_val:,.2f}({dep_val/total_value*100:.1f}%)\n" if positions["staked"] > 0: stake_val = positions["staked"] output += f"Staked: ${stake_val:,.2f} ({stake_val/total_value*100:.1f}%)\n" if positions["locked"] > 0: lock_val = positions["locked"] output += f"Locked: ${lock_val:,.2f} ({lock_val/total_value*100:.1f}%)\n" if positions["borrowed"] > 0: borrow_val = positions["borrowed"] output += f"Borrowed: ${borrow_val:,.2f} ({borrow_val/total_value*100:.1f}%)\n" return output def _format_top_chains(top_chains: list, total_value: float) -> str: """Format the top chains section""" output = "\nTOP CHAINS BY VALUE:\n" for chain, value in top_chains: if value > 1: # Only show chains with > $1 chain_name = chain.replace("-", " ").title() percentage = value / total_value * 100 output += f"{chain_name}: ${value:,.2f} ({percentage:.1f}%)\n" return output @mcp.tool( name="get_wallet_portfolio", description="Get complete wallet portfolio using Zerion API", ) async def get_wallet_portfolio(args: GetWalletPortfolioArgs): try: # Use helper function to fetch data data = await make_zerion_request( query=f"/wallets/{args.wallet_address}/portfolio", params={"filter[positions]": "no_filter", "currency": "usd"}, ) # Parse the response portfolio = data["data"]["attributes"] # Extract data using helper functions total_value, change_24h_usd, change_24h_percent = _extract_portfolio_totals( portfolio ) positions = _extract_position_breakdown(portfolio) top_chains = _get_top_chains(portfolio) # Format output using helper functions header = _format_portfolio_header( args.wallet_address, total_value, change_24h_usd, change_24h_percent ) output = header output += _format_position_breakdown(positions, total_value) output += _format_top_chains(top_chains, total_value) return output except ConnectionError as e: return f"❌ API Error fetching portfolio: {str(e)}" except Exception as e: # pylint: disable=broad-except return f"❌ Unexpected Error fetching portfolio: {str(e)}"