Odos MCP Server

""" Get swap quotes and transaction data from Odos API. """ import json from typing import Any, Dict, 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 from ...mcp_types import OdosQuoteResponse class getQuoteArgs(BaseModel): """Arguments for fetching swap details""" chain: Union[str, int] inputToken: str outputToken: str inputAmount: str slippagePercent: float userAddress: str receiverAddress: Optional[str] = None pathViz: Optional[bool] = True class Config: extra = "allow" @mcp.tool( name="get_quote_swap", description=( "Fetches Odos swap quote and transaction assembly data based on input parameters." "Use Symbols for the tokens " ), ) async def get_quote_swap(args: getQuoteArgs) -> list[types.TextContent]: """ Fetches Odos swap quote and transaction assembly/simulation data. This tool combines getting a quote and simulating the transaction. The arguments are defined by the FetchOdosSwapDetailsArgs Pydantic model. """ try: api_details = await fetch_odos_swap_details_from_api(args) quote_response: OdosQuoteResponse = api_details["quote_response"] output = { "message": "Successfully fetched Odos swap details (FastMCP).", "chainId": api_details["chain_id"], "resolvedInputAddress": api_details["resolved_input_address"], "resolvedOutputAddress": api_details["resolved_output_address"], "effectiveReceiverAddress": api_details["effective_receiver_address"], "quote": { "pathId": quote_response.pathId, "outAmount": ( quote_response.outAmounts[0] if quote_response.outAmounts else "N/A" ), "gasEstimate": quote_response.gasEstimate, "pathViz": json.dumps(quote_response.pathViz, indent=2), }, } 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_swap: {str(e)}" ) ] except (ValueError, KeyError, TypeError) as e: return [ types.TextContent( type="text", text=f"❌ Unexpected Error in get_quote_swap: {str(e)}" ) ] async def fetch_odos_swap_details_from_api(args: getQuoteArgs) -> Dict[str, Any]: """Fetch both quote and assembly data from Odos API""" chain_id = resolve_chain_id(args.chain) resolved_input_address = await resolve_token_address(chain_id, args.inputToken) resolved_output_address = await resolve_token_address(chain_id, args.outputToken) effective_receiver_address = args.receiverAddress or args.userAddress # Get quote quote_payload = { "chainId": chain_id, "inputTokens": [ {"tokenAddress": resolved_input_address, "amount": args.inputAmount} ], "outputTokens": [{"tokenAddress": resolved_output_address, "proportion": 1}], "slippageLimitPercent": args.slippagePercent, "userAddr": args.userAddress, "referralCode": 0, "disableRFQs": True, "pathViz": args.pathViz, } quote_data = await make_odos_request( "/sor/quote/v2", method="POST", payload=quote_payload ) quote_response = OdosQuoteResponse(**quote_data) if not quote_response.pathId: raise ValueError("Could not retrieve pathId from Odos quote") return { "quote_response": quote_response, "chain_id": chain_id, "resolved_input_address": resolved_input_address, "resolved_output_address": resolved_output_address, "effective_receiver_address": effective_receiver_address, }