main.py•10.1 kB
from mcp.server.fastmcp import FastMCP
import httpx
import os
from dotenv import load_dotenv
import pandas as pd
# Load environment variables
load_dotenv()
# Initialize MCP server
mcp = FastMCP(
name="DEX Metrics",
dependencies=["httpx", "python-dotenv", "pandas"]
)
# Configuration
DUNE_API_KEY = os.getenv("DUNE_API_KEY")
BASE_URL = "https://api.dune.com/api/v1"
HEADERS = {"X-Dune-API-Key": DUNE_API_KEY}
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
@mcp.tool()
def get_latest_trading_volume_by_dex(limit: int = 1000) -> str:
"""
Retrieve the latest 24-hour and 7-day trading volume by decentralized exchange (DEX).
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(4319, limit=limit)
df = pd.DataFrame(data)
df = df.set_index("Rank")
df = df.sort_index(ascending=True)
return df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_latest_trading_volume_by_aggregator(limit: int = 1000) -> str:
"""
Retrieve the latest 24-hour and 7-day trading volume by aggregator.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(21689, limit=limit)
df = pd.DataFrame(data)
df = df.set_index("Rank")
df = df.sort_index(ascending=True)
return df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_weekly_trading_volume_by_chain(limit: int = 1000) -> str:
"""
Retrieve weekly trading volume by blockchain.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(2180075, limit=limit)
df = pd.DataFrame(data)
df["date"] = pd.to_datetime(df["_col1"]).dt.date
pivot_df = df.pivot(index="date", columns="blockchain", values="usd_volume")
pivot_df = pivot_df.sort_index(ascending=False)
return pivot_df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_daily_trading_volume_by_dex(limit: int = 1000) -> str:
"""
Retrieve daily trading volume by decentralized exchange (DEX).
This tool fetches daily trading volume data for various DEXs from a Dune Analytics query and returns it in a markdown-formatted pivot table, with dates as the index and DEX projects as columns.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(4388, limit=limit)
df = pd.DataFrame(data)
df["date"] = pd.to_datetime(df["_col1"]).dt.date
pivot_df = df.pivot(index="date", columns="project", values="usd_volume")
pivot_df = pivot_df.sort_index(ascending=False)
return pivot_df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_weekly_trading_volume_by_dex(limit: int = 1000) -> str:
"""
Retrieve weekly trading volume by decentralized exchange (DEX).
This tool fetches weekly trading volume data for various DEXs from a Dune Analytics query and returns it in a markdown-formatted pivot table, with dates as the index and DEX projects as columns.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(4323, limit=limit)
df = pd.DataFrame(data)
df["date"] = pd.to_datetime(df["_col1"]).dt.date
pivot_df = df.pivot(index="date", columns="project", values="usd_volume")
pivot_df = pivot_df.sort_index(ascending=False)
return pivot_df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_monthly_trading_volume_by_dex(limit: int = 1000) -> str:
"""
Retrieve monthly trading volume by decentralized exchange (DEX).
This tool fetches monthly trading volume data for various DEXs from a Dune Analytics query and returns it in a markdown-formatted pivot table, with dates as the index and DEX projects as columns.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(4323, limit=limit)
df = pd.DataFrame(data)
df["date"] = pd.to_datetime(df["_col1"]).dt.date
pivot_df = df.pivot(index="date", columns="project", values="usd_volume")
pivot_df = pivot_df.sort_index(ascending=False)
return pivot_df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_yoy_monthly_trading_volume_by_aggragator(limit: int = 1000) -> str:
"""
Retrieve year-over-year monthly trading volume by aggregator.
This tool fetches year-over-year monthly trading volume data for aggregators from a Dune Analytics query and returns it in a markdown-formatted pivot table, with years as the index and months as columns.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(4424, limit=limit)
df = pd.DataFrame(data)
pivot_df = df.pivot(index="year", columns="month", values="usd_volume")
pivot_df = pivot_df.sort_index(ascending=False)
return pivot_df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_weekly_solana_trading_volume_by_dex(limit: int = 1000) -> str:
"""
Retrieve weekly trading volume by decentralized exchange (DEX) on the Solana blockchain.
This tool fetches weekly trading volume data for DEXs on Solana from a Dune Analytics query and returns it in a markdown-formatted pivot table, with dates as the index and DEX projects as columns.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(3084516, limit=limit)
df = pd.DataFrame(data)
df["date"] = pd.to_datetime(df["time"]).dt.date
pivot_df = df.pivot(index="date", columns="project", values="amount_usd")
pivot_df = pivot_df.sort_index(ascending=False)
return pivot_df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_weekly_trading_volume_by_frontend(limit: int = 1000) -> str:
"""
Retrieve 7-day trading volume by frontend.
This tool fetches 7-day trading volume data for different frontends from a Dune Analytics query and returns it in a markdown-formatted table, sorted by volume in descending order.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(3364122, limit=limit)
df = pd.DataFrame(data)
df = df.sort_values(by="volume", ascending=False)
return df.to_markdown()
except Exception as e:
return str(e)
@mcp.tool()
def get_daily_trading_volume_by_telegram_bot(limit: int = 1000) -> str:
"""
Retrieve daily trading volume by Telegram bot.
This tool fetches daily trading volume data for Telegram bots from a Dune Analytics query and returns it in a markdown-formatted pivot table, with dates as the index and bot names as columns.
Args:
limit (int, optional): Maximum number of rows to retrieve from the query. Defaults to 1000.
Returns:
str: A markdown-formatted pivot table of trading volume data, or an error message if the query fails.
"""
try:
data = get_latest_result(2687239, limit=limit)
df = pd.DataFrame(data)
df["date"] = pd.to_datetime(df["block_date"]).dt.date
pivot_df = df.pivot(index="date", columns="bot", values="volumeUSD")
pivot_df = pivot_df.sort_index(ascending=False)
return pivot_df.to_markdown()
except Exception as e:
return str(e)
# Run the server
if __name__ == "__main__":
mcp.run()