memecoin-radar-mcp

Instructions

Args:
    time_span (str): Time period for the query. Must be one of: '5h', '12h', '24h'.
        Defaults to '5h'.
    limit (int): Maximum number of tokens to return. Defaults to 100.

Returns:
    str: A formatted table of trending tokens on Raydium including token name,
        mint address, and trading volume, or an error message if the query fails.

Raises:
    ValueError: If an invalid time_span value is provided.
    httpx.HTTPStatusError: If the Dune API request fails.

Implementation Reference

• main.py:254-292 (handler)

  The handler function for the get_trending_tokens_on_raydium tool. It is registered via the @mcp.tool() decorator. Fetches trending token data from Dune Analytics queries based on the specified time_span, processes the results using helper functions like get_latest_result and strip_a_tag, formats them into a table using tabulate, and returns the markdown table or an error message.

  @mcp.tool()
  def get_trending_tokens_on_raydium(time_span: str = "5h", limit: int = 100) -> str:
      """Retrieve tokens with the highest trading volume on Raydium within a specified time span.
  
      Args:
          time_span (str): Time period for the query. Must be one of: '5h', '12h', '24h'.
              Defaults to '5h'.
          limit (int): Maximum number of tokens to return. Defaults to 100.
  
      Returns:
          str: A formatted table of trending tokens on Raydium including token name,
              mint address, and trading volume, or an error message if the query fails.
  
      Raises:
          ValueError: If an invalid time_span value is provided.
          httpx.HTTPStatusError: If the Dune API request fails.
      """
      query_ids = {
          "5h": 4840714,
          "12h": 4840651, 
          "24h": 4840709,
      }
      try:
          query_id = query_ids.get(time_span)
          if query_id is None:
              raise ValueError("Invalid time_span value. Allowed: 5h | 12h | 24h")
          data = get_latest_result(query_id, limit=limit)
          rows = [
              [
                  strip_a_tag(row["asset_with_chart"]),
                  row["token_address"],
                  f'${row.get(f"total_volume_{time_span}", 0):.2f}'
              ]
              for row in data
          ]
          headers = ["Token", "Mint Address", "Volume"]
          return f"# Top {limit} Trending Tokens on Raydium - Last {time_span}\n\n" + tabulate(rows, headers=headers) 
      except Exception as e:
          return str(e)

• main.py:23-46 (helper)

  Shared helper function used by the tool to retrieve the latest execution results from a specified Dune Analytics query ID.

  def get_latest_result(query_id: int, limit: int = 1000):
      """
      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

• main.py:47-49 (helper)

  Helper function to extract text content from HTML anchor tags, used to clean token names in the results.

  def strip_a_tag(html):
      match = re.search(r'>(.*?)</a>', html)
      return match.group(1) if match else html