Odos MCP Server

""" Get available liquidity sources and DEX integrations for trading. """ import json from typing import Union from mcp import types from pydantic import BaseModel from ...helpers import make_odos_request, resolve_chain_id from ...mcp import mcp class GetLiquiditySourcesArgs(BaseModel): """Arguments for getting liquidity sources""" chain_id: Union[str, int] @mcp.tool( name="get_liquidity_sources", description=( "Get list of all liquidity sources supported by Odos." "This list can be used to filter out unwanted or problematic liquidity pools" ), ) async def get_liquidity_sources( args: GetLiquiditySourcesArgs, ) -> list[types.TextContent]: """ Gets all supported liquidity sources for a specific chain. """ try: chain_id = resolve_chain_id(args.chain_id) # Build URL with path parameters url = f"/info/liquidity-sources/{chain_id}" response_data = await make_odos_request(url, method="GET") sources = response_data.get("sources", []) output = { "message": f"Successfully fetched liquidity sources for chain {chain_id}.", "chainId": chain_id, "sources": sources, "sourceCount": len(sources), } 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_liquidity_sources: {str(e)}" ) ] except (ValueError, KeyError, TypeError) as e: return [ types.TextContent( type="text", text=f"❌ Unexpected Error in get_liquidity_sources: {str(e)}", ) ]