get_trending_tokens_by_source
Retrieve top traded tokens from specified platforms like Telegram, Web, or Mobile within the last 12 hours. Returns a formatted table with rank, token name, mint address, trading volume, and total trades. Ideal for tracking trending memecoins and Solana-based tokens.
Instructions
Retrieve top traded tokens on specified source platform in the last 12 hours.
Args:
source (str): The platform to query tokens from. Must be one of: 'Telegram', 'Web', 'Mobile'.
Defaults to 'Telegram'.
limit (int): Maximum number of tokens to return. Defaults to 100.
Returns:
str: A formatted table of trending tokens including rank, token name, mint address,
trading volume, and total trades, or an error message if the query fails.
Raises:
ValueError: If an invalid source value is provided.
httpx.HTTPStatusError: If the Dune API request fails.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | ||
| source | No | Telegram |
Implementation Reference
- main.py:56-95 (handler)The handler function decorated with @mcp.tool() that implements the get_trending_tokens_by_source tool. It fetches data from Dune API queries based on the source (Telegram, Web, Mobile), processes the rows into a table format using helpers, and returns a markdown table or error message.
def get_trending_tokens_by_source(source: str = "Telegram", limit: int = 100) -> str: """Retrieve top traded tokens on specified source platform in the last 12 hours. Args: source (str): The platform to query tokens from. Must be one of: 'Telegram', 'Web', 'Mobile'. Defaults to 'Telegram'. limit (int): Maximum number of tokens to return. Defaults to 100. Returns: str: A formatted table of trending tokens including rank, token name, mint address, trading volume, and total trades, or an error message if the query fails. Raises: ValueError: If an invalid source value is provided. httpx.HTTPStatusError: If the Dune API request fails. """ query_ids = { "Telegram": 4830187, "Web": 4830192, "Mobile": 4930328, } try: query_id = query_ids.get(source) if query_id is None: raise ValueError("Invalid source value. Allowed: Telegram | Web | Mobile") data = get_latest_result(query_id, limit=limit) rows = [ [ row["rank"], strip_a_tag(row["token_link"]), row["token_mint_address"], f'${row["total_volume_usd"]:.2f}', row["total_trades"] ] for row in data ] headers = ["Rank", "Token", "Mint Address", "Volume(12h)", "Total Trades"] return f"# Top {limit} Trending Tokens on {source} - Last 12 Hours\n\n" + tabulate(rows, headers=headers) except Exception as e: return str(e) - main.py:23-46 (helper)Helper function to retrieve the latest execution results from a specified Dune Analytics query ID, used by the tool to fetch raw data.
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-50 (helper)Helper utility to extract text content from HTML anchor tags, used to clean token names in the tool's output.
def strip_a_tag(html): match = re.search(r'>(.*?)</a>', html) return match.group(1) if match else html