get_trending_tokens_on_raydium
Identify trending tokens on Raydium based on trading volume within 5h, 12h, or 24h. Returns a formatted table with token details for strategic decision-making in Solana memecoin markets.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | ||
| time_span | No | 5h |
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