get_daily_active_addresses
Track daily active wallet counts for Solana memecoin launchpads to monitor platform engagement and user activity trends.
Instructions
Fetch the daily count of active wallet addresses interacting with Solana memecoin launchpads.
This tool retrieves data from a Dune Analytics query and pivots it to show the number of unique
active wallets per day by each platform.
Args:
limit (int, optional): Maximum number of rows to fetch from the Dune query. Defaults to 1000.
Returns:
str: A markdown-formatted table of daily active wallets by platform, or an error message if
the query fails.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No |
Implementation Reference
- main.py:125-148 (handler)The primary handler function for the 'get_daily_active_addresses' tool. Decorated with @mcp.tool() for automatic registration in FastMCP. Fetches data from Dune query 5002622, pivots daily active wallets by platform using pandas, sorts descending by date, and returns markdown table or error.
@mcp.tool() def get_daily_active_addresses(limit: int = 1000) -> str: """ Fetch the daily count of active wallet addresses interacting with Solana memecoin launchpads. This tool retrieves data from a Dune Analytics query and pivots it to show the number of unique active wallets per day by each platform. Args: limit (int, optional): Maximum number of rows to fetch from the Dune query. Defaults to 1000. Returns: str: A markdown-formatted table of daily active wallets by platform, or an error message if the query fails. """ try: data = get_latest_result(5002622, limit=limit) df = pd.DataFrame(data) df['date_time'] = pd.to_datetime(df['date_time']).dt.date pivot_df = df.pivot(index="date_time", columns="platform", values="daily_active_wallets") 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)Helper utility function called by the handler (and other tools) to retrieve the latest results from a Dune Analytics query using the configured API key and base URL.
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