get_trending_tokens_by_kol_trading_volume
Identify trending Solana memecoins by analyzing trading activity from key opinion leaders (KOLs). Retrieve a formatted table with token names, mint addresses, KOL buys, total buys, and trading volume for informed decision-making.
Instructions
Retrieve tokens with the highest trading volume by memecoin KOLs.
Args:
limit (int): Maximum number of tokens to return. Defaults to 100.
Returns:
str: A formatted table of trending tokens by KOL trading volume including
token name, mint address, unique KOL buys, total buys, and total volume,
or an error message if the query fails.
Raises:
httpx.HTTPStatusError: If the Dune API request fails.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No |
Implementation Reference
- main.py:222-252 (handler)The handler function for the 'get_trending_tokens_by_kol_trading_volume' tool, decorated with @mcp.tool() for registration. It fetches latest results from Dune query ID 4838351, processes the rows into a formatted table using tabulate, and returns the markdown table or error.
@mcp.tool() def get_trending_tokens_by_kol_trading_volume(limit: int = 100) -> str: """Retrieve tokens with the highest trading volume by memecoin KOLs. Args: limit (int): Maximum number of tokens to return. Defaults to 100. Returns: str: A formatted table of trending tokens by KOL trading volume including token name, mint address, unique KOL buys, total buys, and total volume, or an error message if the query fails. Raises: httpx.HTTPStatusError: If the Dune API request fails. """ try: data = get_latest_result(4838351, limit=limit) rows = [ [ row["token"], row["contract_address"], row["unique_kols"], row["total_buys"], f'${row["total_volume"]:.2f}' ] for row in data ] headers = ["Token", "Mint Address", "Unique KOL Buys", "Total Buys", "Total Volume"] return f"# Top {limit} Trending Tokens by KOL Trading Volume\n\n" + tabulate(rows, headers=headers) except Exception as e: return str(e) - main.py:23-46 (helper)Helper function used by the tool to fetch the latest execution results from a specified Dune Analytics query using the Dune API.
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