Odos MCP Server

""" Get zap quotes for liquidity provision from Odos API. """ import json from typing import Any, Dict, List, Optional, Union from mcp import types from pydantic import BaseModel from ...helpers import make_odos_request, resolve_chain_id, resolve_token_address from ...mcp import mcp class InputToken(BaseModel): tokenAddress: str amount: str class OutputToken(BaseModel): tokenAddress: str proportion: float class PathVizImageConfig(BaseModel): linkColors: Optional[List[str]] = None nodeColor: Optional[str] = None nodeTextColor: Optional[str] = None legendTextColor: Optional[str] = None width: Optional[int] = None height: Optional[int] = None class GetQuoteZapArgs(BaseModel): """Arguments for fetching zap quote""" chainId: Union[str, int] inputTokens: List[InputToken] outputTokens: List[OutputToken] gasPrice: Optional[float] = None userAddr: Optional[str] = None slippageLimitPercent: Optional[float] = 0.3 sourceBlacklist: Optional[List[str]] = [] sourceWhitelist: Optional[List[str]] = [] poolBlacklist: Optional[List[str]] = [] pathVizImage: Optional[bool] = False pathVizImageConfig: Optional[PathVizImageConfig] = None disableRFQs: Optional[bool] = True referralCode: Optional[int] = 0 compact: Optional[bool] = True likeAsset: Optional[bool] = False simple: Optional[bool] = False @mcp.tool( name="get_quote_zap", description=( "Quote a path for zapping into or out of liquidity positions atomically." "Input tokens can be both regular ERC-20 tokens and liquidity pool tokens." ), ) async def get_quote_zap(args: GetQuoteZapArgs) -> list[types.TextContent]: """ Fetches Odos zap quote for liquidity positions. """ try: chain_id = resolve_chain_id(args.chainId) # Resolve token addresses resolved_input_tokens: List[Dict[str, Any]] = [] for input_token_item in args.inputTokens: resolved_address = await resolve_token_address( chain_id, input_token_item.tokenAddress ) resolved_input_tokens.append( {"tokenAddress": resolved_address, "amount": input_token_item.amount} ) resolved_output_tokens: List[Dict[str, Any]] = [] for output_token_item in args.outputTokens: resolved_address = await resolve_token_address( chain_id, output_token_item.tokenAddress ) resolved_output_tokens.append( { "tokenAddress": resolved_address, "proportion": output_token_item.proportion, } ) # Prepare payload payload = { "chainId": chain_id, "inputTokens": resolved_input_tokens, "outputTokens": resolved_output_tokens, "slippageLimitPercent": args.slippageLimitPercent, "referralCode": args.referralCode, "disableRFQs": args.disableRFQs, "compact": args.compact, "likeAsset": args.likeAsset, "simple": args.simple, "sourceBlacklist": args.sourceBlacklist or [], "sourceWhitelist": args.sourceWhitelist or [], "poolBlacklist": args.poolBlacklist or [], "pathVizImage": args.pathVizImage, } # Add optional fields if args.gasPrice is not None: payload["gasPrice"] = args.gasPrice if args.userAddr: payload["userAddr"] = args.userAddr if args.pathVizImageConfig: payload["pathVizImageConfig"] = args.pathVizImageConfig.dict( exclude_none=True ) response_data = await make_odos_request( "/sor/quote/v2/zap", method="POST", payload=payload ) output = { "message": "Successfully fetched Odos zap quote.", "chainId": chain_id, "quote": response_data, } return [types.TextContent(type="text", text=json.dumps(output, indent=2))] except ConnectionError as e: return [ types.TextContent( type="text", text=f"❌ API Error in get_quote_zap: {str(e)}" ) ] except (ValueError, KeyError, TypeError) as e: return [ types.TextContent( type="text", text=f"❌ Unexpected Error in get_quote_zap: {str(e)}" ) ]