dex-metrics-mcp

Instructions

Retrieve year-over-year monthly trading volume by aggregator.

This tool fetches year-over-year monthly trading volume data for aggregators from a Dune Analytics query and returns it in a markdown-formatted pivot table, with years as the index and months as columns.

Args:
    limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.

Returns:
    str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.

Implementation Reference

• main.py:175-195 (handler)

  The main handler function for the tool 'get_yoy_monthly_trading_volume_by_aggragator'. It is decorated with @mcp.tool() which serves as both the implementation and registration. Fetches data from Dune query 4424 using the helper get_latest_result, pivots the data by year and month, and returns a markdown table.

  @mcp.tool()
  def get_yoy_monthly_trading_volume_by_aggragator(limit: int = 1000) -> str:
      """
      Retrieve year-over-year monthly trading volume by aggregator.
  
      This tool fetches year-over-year monthly trading volume data for aggregators from a Dune Analytics query and returns it in a markdown-formatted pivot table, with years as the index and months as columns.
  
      Args:
          limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
  
      Returns:
          str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.
      """
      try:
          data = get_latest_result(4424, limit=limit)
          df = pd.DataFrame(data)
          pivot_df = df.pivot(index="year", columns="month", values="usd_volume")
          pivot_df = pivot_df.sort_index(ascending=False)
          return pivot_df.to_markdown()
      except Exception as e:
          return str(e)  

• main.py:21-43 (helper)

  Shared helper function used by multiple tools, including this one, to fetch latest results from a specified Dune Analytics query ID.

  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