dex-metrics-mcp

Instructions

Implementation Reference

• main.py:45-63 (handler)

  Handler function decorated with @mcp.tool() that implements the tool logic. Fetches data from Dune query 4319 using the helper get_latest_result, processes into a pandas DataFrame indexed by Rank, sorted ascending, and returns as markdown table or error string.
  @mcp.tool() def get_latest_trading_volume_by_dex(limit: int = 1000) -> str: """ Retrieve the latest 24-hour and 7-day trading volume by decentralized exchange (DEX). Args: limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000. Returns: str: A markdown-formatted table of trading volume data, or an error message if the query fails. """ try: data = get_latest_result(4319, limit=limit) df = pd.DataFrame(data) df = df.set_index("Rank") df = df.sort_index(ascending=True) return df.to_markdown() except Exception as e: return str(e)
• main.py:21-43 (helper)

  Supporting utility function used by the tool (and others) to retrieve latest execution results from a specified Dune Analytics query ID via API, returning list of rows.
  def get_latest_result(query_id: int, limit: int = 1000) -> list: """ Fetch the latest results from a Dune Analytics query. Args: query_id (int): The ID of the Dune query to fetch results from. limit (int, optional): Maximum number of rows to return. Defaults to 1000. Returns: list: A list of dictionaries containing the query results, or an empty list if the request fails. Raises: httpx.HTTPStatusError: If the API request fails due to a client or server error. """ url = f"{BASE_URL}/query/{query_id}/results" params = {"limit": limit} with httpx.Client() as client: response = client.get(url, params=params, headers=HEADERS, timeout=300) response.raise_for_status() data = response.json() result_data = data.get("result", {}).get("rows", []) return result_data