from __future__ import annotations
from typing import Any, Dict, List, Optional
import json
from mcp.server.fastmcp import FastMCP
from common import fetch_json, API_KEYS, SUPPORTED_CHAINS
# Initialize FastMCP
mcp = FastMCP(name="defillama")
DEFILLAMA_API_BASE = "https://api.llama.fi"
@mcp.tool()
async def get_all_protocols() -> str:
"""Mendapatkan daftar semua protokol yang didukung oleh DefiLlama.
Returns:
String berisi daftar protokol dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/protocols"
data = await fetch_json(url)
if not data or not isinstance(data, list):
return "Tidak ada data protokol yang tersedia"
table = "| Name | Category | TVL (USD) | Chain |\
"
table += "|------|----------|-----------|-------|\
"
for protocol in data:
name = protocol.get('name', '???')
category = protocol.get('category', '???')
tvl = protocol.get('tvl', 0)
tvl_str = f"${tvl:,.0f}" if isinstance(tvl, (int, float)) else '???'
chains = ", ".join(protocol.get('chains', []))
table += f"| {name} | {category} | {tvl_str} | {chains} |\
"
return table.rstrip()
@mcp.tool()
async def get_protocol_tvl(protocol_name: str) -> str:
"""Mendapatkan TVL saat ini untuk protokol tertentu.
Args:
protocol_name: Nama protokol (slug).
Returns:
String berisi TVL protokol dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/protocol/{protocol_name}"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return f"Tidak ada data TVL untuk protokol {protocol_name} yang tersedia"
tvl = data.get('tvl', 0)
tvl_str = f"${tvl:,.0f}" if isinstance(tvl, (int, float)) else '???'
return f"TVL saat ini untuk {protocol_name}: {tvl_str}"
@mcp.tool()
async def get_historical_tvl_protocol(protocol_name: str) -> str:
"""Mendapatkan TVL historis untuk protokol tertentu.
Args:
protocol_name: Nama protokol (slug).
Returns:
String berisi TVL historis protokol dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/protocol/{protocol_name}"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return f"Tidak ada data TVL untuk protokol {protocol_name} yang tersedia"
tvl_data = data.get('tvl', [])
if not tvl_data:
return f"Tidak ada data TVL historis untuk protokol {protocol_name} yang tersedia"
table = "| Date | TVL (USD) |\
"
table += "|------|-----------|\
"
for entry in tvl_data:
date = entry.get('date', '???')
tvl = entry.get('totalLiquidityUSD', 0)
tvl_str = f"${tvl:,.0f}" if isinstance(tvl, (int, float)) else '???'
table += f"| {date} | {tvl_str} |\
"
return table.rstrip()
@mcp.tool()
async def get_historical_tvl_chain(chain_name: str) -> str:
"""Mendapatkan TVL historis untuk chain tertentu.
Args:
chain_name: Nama chain (slug).
Returns:
String berisi TVL historis chain dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/charts/{chain_name}"
data = await fetch_json(url)
if not data or not isinstance(data, list):
return f"Tidak ada data TVL untuk chain {chain_name} yang tersedia"
table = "| Date | TVL (USD) |\
"
table += "|------|-----------|\
"
for entry in data:
date = entry.get('date', '???')
tvl = entry.get('totalLiquidityUSD', 0)
tvl_str = f"${tvl:,.0f}" if isinstance(tvl, (int, float)) else '???'
table += f"| {date} | {tvl_str} |\
"
return table.rstrip()
@mcp.tool()
async def get_historical_coin_prices(coin_id: str, start_timestamp: int, end_timestamp: int) -> str:
"""Mendapatkan harga historis untuk koin tertentu dalam rentang waktu.
Args:
coin_id: ID koin (misalnya, 'coingecko:ethereum').
start_timestamp: Waktu mulai dalam format Unix timestamp.
end_timestamp: Waktu akhir dalam format Unix timestamp.
Returns:
String berisi harga historis koin dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/charts/{coin_id}?start={start_timestamp}&end={end_timestamp}"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return f"Tidak ada data harga untuk koin {coin_id} yang tersedia"
prices = data.get('prices', [])
if not prices:
return f"Tidak ada data harga historis untuk koin {coin_id} yang tersedia"
table = "| Date | Price (USD) |\
"
table += "|------|-------------|\
"
for entry in prices:
date = entry.get('timestamp', '???')
price = entry.get('price', 0)
price_str = f"${price:.4f}" if isinstance(price, (int, float)) else '???'
table += f"| {date} | {price_str} |\
"
return table.rstrip()
@mcp.tool()
async def get_current_coin_price(coin_id: str) -> str:
"""Mendapatkan harga koin saat ini.
Args:
coin_id: ID koin (misalnya, 'coingecko:ethereum').
Returns:
String berisi harga koin saat ini dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/prices/current/{coin_id}"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return f"Tidak ada data harga untuk koin {coin_id} yang tersedia"
coins_data = data.get('coins', {})
if not coins_data:
return f"Tidak ada data harga untuk koin {coin_id} yang tersedia"
price_data = coins_data.get(coin_id, {})
price = price_data.get('price', 0)
timestamp = price_data.get('timestamp', 0)
return f"Harga saat ini untuk {coin_id}: ${price:.4f} (Timestamp: {timestamp})"
@mcp.tool()
async def get_historical_coin_price_at_timestamp(timestamp: int, coins: str) -> str:
"""Mendapatkan harga historis token pada timestamp tertentu.
Args:
timestamp: Waktu dalam format Unix timestamp.
coins: Set koin yang dipisahkan koma (misalnya, 'ethereum:0xdF574c24545E5FfEcb9a659c229253D4111d87e1,coingecko:ethereum').
Returns:
String berisi harga historis koin dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/prices/historical/{timestamp}/{coins}"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return f"Tidak ada data harga historis untuk koin {coins} pada timestamp {timestamp} yang tersedia"
coins_data = data.get('coins', {})
if not coins_data:
return f"Tidak ada data harga historis untuk koin {coins} pada timestamp {timestamp} yang tersedia"
table = "| Coin ID | Price (USD) | Symbol | Timestamp |\
"
table += "|---------|-------------|--------|-----------|\
"
for coin_id, price_info in coins_data.items():
price = price_info.get('price', 0)
symbol = price_info.get('symbol', '???')
timestamp = price_info.get('timestamp', 0)
price_str = f"${price:.4f}" if isinstance(price, (int, float)) else '???'
table += f"| {coin_id} | {price_str} | {symbol} | {timestamp} |\
"
return table.rstrip()
@mcp.tool()
async def get_batch_historical_prices(coins_json: str) -> str:
"""Mendapatkan harga historis untuk beberapa token pada beberapa timestamp yang berbeda.
Args:
coins_json: Objek JSON string di mana key adalah koin dalam format {chain}:{address}, dan value adalah array timestamp yang diminta.
Contoh: '{"avax:0xb97ef9ef8734c71904d8002f8b6bc66dd9c48a6e": [1666876743, 1666862343], "coingecko:ethereum": [1666869543, 1666862343]}'
Returns:
String berisi harga historis koin dalam format markdown.
"""
try:
coins_data_dict = json.loads(coins_json)
except json.JSONDecodeError:
return "Error: Format JSON coins_json tidak valid."
url = f"{DEFILLAMA_API_BASE}/batchHistorical"
params = {"coins": coins_json}
data = await fetch_json(url, params=params)
if not data or not isinstance(data, dict):
return f"Tidak ada data harga historis batch yang tersedia untuk {coins_json}"
result_table = "| Coin ID | Timestamp | Price (USD) | Symbol | Confidence |\
"
result_table += "|---------|-----------|-------------|--------|------------|\
"
for coin_id, coin_info in data.get('coins', {}).items():
symbol = coin_info.get('symbol', '???')
prices = coin_info.get('prices', [])
for price_entry in prices:
timestamp = price_entry.get('timestamp', '???')
price = price_entry.get('price', 0)
confidence = price_entry.get('confidence', '???')
price_str = f"${price:.4f}" if isinstance(price, (int, float)) else '???'
result_table += f"| {coin_id} | {timestamp} | {price_str} | {symbol} | {confidence} |\
"
return result_table.rstrip()
@mcp.tool()
async def get_stablecoins() -> str:
"""Mendapatkan daftar semua stablecoin yang didukung oleh DefiLlama.
Returns:
String berisi daftar stablecoin dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/stablecoins"
data = await fetch_json(url)
if not data or not isinstance(data, list):
return "Tidak ada data stablecoin yang tersedia"
table = "| Name | Symbol | Total Supply (USD) | Peg Type | Chains |\
"
table += "|------|--------|--------------------|----------|--------|\
"
for sc in data:
name = sc.get('name', '???')
symbol = sc.get('symbol', '???')
total_supply = sc.get('totalSupply', 0)
total_supply_str = f"${total_supply:,.0f}" if isinstance(total_supply, (int, float)) else '???'
peg_type = sc.get('pegType', '???')
chains = ", ".join(sc.get('chains', []))
table += f"| {name} | {symbol} | {total_supply_str} | {peg_type} | {chains} |\
"
return table.rstrip()
@mcp.tool()
async def get_stablecoins_by_peg_type(peg_type: str) -> str:
"""Mendapatkan daftar stablecoin yang di-pegged ke aset tertentu.
Args:
peg_type: Tipe peg (misalnya, 'USD', 'EUR', 'GOLD').
Returns:
String berisi daftar stablecoin dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/stablecoins/{peg_type}"
data = await fetch_json(url)
if not data or not isinstance(data, list):
return f"Tidak ada data stablecoin dengan peg type {peg_type} yang tersedia"
table = "| Name | Symbol | Total Supply (USD) | Peg Type | Chains |\
"
table += "|------|--------|--------------------|----------|--------|\
"
for sc in data:
name = sc.get('name', '???')
symbol = sc.get('symbol', '???')
total_supply = sc.get('totalSupply', 0)
total_supply_str = f"${total_supply:,.0f}" if isinstance(total_supply, (int, float)) else '???'
peg_type_found = sc.get('pegType', '???')
chains = ", ".join(sc.get('chains', []))
table += f"| {name} | {symbol} | {total_supply_str} | {peg_type_found} | {chains} |\
"
return table.rstrip()
@mcp.tool()
async def get_historical_tvl_all_chains() -> str:
"""Mendapatkan TVL historis untuk semua chain.
Returns:
String berisi TVL historis semua chain dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/v2/historicalChainTvl"
data = await fetch_json(url)
if not data or not isinstance(data, list):
return "Tidak ada data TVL historis untuk semua chain yang tersedia"
table = "| Date | TVL (USD) |\
"
table += "|------|-----------|\
"
for entry in data:
date = entry.get('date', '???')
tvl = entry.get('totalLiquidityUSD', 0)
tvl_str = f"${tvl:,.0f}" if isinstance(tvl, (int, float)) else '???'
table += f"| {date} | {tvl_str} |\
"
return table.rstrip()
@mcp.tool()
async def get_current_tvl_all_chains() -> str:
"""Mendapatkan TVL saat ini untuk semua chain.
Returns:
String berisi TVL semua chain dalam format markdown
"""
url = f"{DEFILLAMA_API_BASE}/v2/chains"
data = await fetch_json(url)
if not data or not isinstance(data, list):
return "Tidak ada data TVL yang tersedia"
table = "| Chain | TVL (USD) |\
"
table += "|-------|-----------|\
"
for chain_data in data:
name = chain_data.get('name', '???')
tvl = chain_data.get('tvl', 0)
tvl_str = f"${tvl:,.0f}" if isinstance(tvl, (int, float)) else '???'
table += f"| {name} | {tvl_str} |\
"
return table.rstrip()
@mcp.tool()
async def get_tvl_protocol(protocol_name: str) -> str:
"""Mendapatkan TVL saat ini untuk protokol tertentu (endpoint sederhana).
Args:
protocol_name: Nama protokol (slug).
Returns:
String berisi TVL protokol dalam format markdown.
"""
url = f"{DEFILLAMA_API_BASE}/tvl/{protocol_name}"
data = await fetch_json(url)
if data is None:
return f"Tidak ada data TVL untuk protokol {protocol_name} yang tersedia"
# Data langsung berupa nilai TVL
return f"TVL saat ini untuk {protocol_name}: ${data:,.0f}"
@mcp.tool()
async def get_stablecoin_charts_by_chain(chain: str) -> str:
"""Mendapatkan historical mcap sum dari semua stablecoin di chain tertentu.
Args:
chain: Nama chain (slug).
Returns:
String berisi historical mcap stablecoin dalam format markdown.
"""
url = f"https://stablecoins.llama.fi/stablecoincharts/{chain}"
data = await fetch_json(url)
if not data or not isinstance(data, list):
return f"Tidak ada data historical mcap stablecoin untuk chain {chain} yang tersedia"
table = "| Date | Market Cap (USD) |\n"
table += "|------|-----------------|\n"
for entry in data:
date = entry.get('date', '???')
mcap = entry.get('totalCirculating', 0)
mcap_str = f"${mcap:,.0f}" if isinstance(mcap, (int, float)) else '???'
table += f"| {date} | {mcap_str} |\n"
return table.rstrip()
@mcp.tool()
async def get_stablecoin_by_asset(asset_id: int) -> str:
"""Mendapatkan historical mcap dan distribusi chain dari stablecoin.
Args:
asset_id: ID stablecoin.
Returns:
String berisi informasi stablecoin dalam format markdown.
"""
url = f"https://stablecoins.llama.fi/stablecoin/{asset_id}"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return f"Tidak ada data untuk stablecoin dengan ID {asset_id} yang tersedia"
# Informasi dasar
name = data.get('name', '???')
symbol = data.get('symbol', '???')
mcap = data.get('circulating', 0)
mcap_str = f"${mcap:,.0f}" if isinstance(mcap, (int, float)) else '???'
peg_type = data.get('pegType', '???')
price = data.get('price', 0)
price_str = f"${price:.4f}" if isinstance(price, (int, float)) else '???'
result = f"## {name} ({symbol})\n\n"
result += f"- **Market Cap**: {mcap_str}\n"
result += f"- **Peg Type**: {peg_type}\n"
result += f"- **Current Price**: {price_str}\n\n"
# Distribusi chain
chains_data = data.get('chains', {})
if chains_data:
result += "### Chain Distribution\n\n"
result += "| Chain | Amount | Percentage |\n"
result += "|-------|--------|------------|\n"
for chain, amount in chains_data.items():
percentage = (amount / mcap) * 100 if mcap > 0 else 0
amount_str = f"${amount:,.0f}" if isinstance(amount, (int, float)) else '???'
result += f"| {chain} | {amount_str} | {percentage:.2f}% |\n"
# Historical data
history = data.get('history', [])
if history:
result += "\n### Historical Market Cap\n\n"
result += "| Date | Market Cap (USD) |\n"
result += "|------|-----------------|\n"
for entry in history[:10]: # Limit to 10 entries to avoid too long output
date = entry.get('date', '???')
hist_mcap = entry.get('circulating', 0)
hist_mcap_str = f"${hist_mcap:,.0f}" if isinstance(hist_mcap, (int, float)) else '???'
result += f"| {date} | {hist_mcap_str} |\n"
if len(history) > 10:
result += "*Note: Showing only 10 most recent entries*\n"
return result.rstrip()
@mcp.tool()
async def get_stablecoin_chains() -> str:
"""Mendapatkan mcap saat ini dari semua stablecoin di setiap chain.
Returns:
String berisi mcap stablecoin per chain dalam format markdown.
"""
url = "https://stablecoins.llama.fi/stablecoinchains"
data = await fetch_json(url)
if not data or not isinstance(data, list):
return "Tidak ada data mcap stablecoin per chain yang tersedia"
table = "| Chain | Market Cap (USD) | Percentage |\n"
table += "|-------|-----------------|------------|\n"
total_mcap = sum(chain.get('totalCirculating', 0) for chain in data)
for chain_data in data:
chain = chain_data.get('name', '???')
mcap = chain_data.get('totalCirculating', 0)
mcap_str = f"${mcap:,.0f}" if isinstance(mcap, (int, float)) else '???'
percentage = (mcap / total_mcap) * 100 if total_mcap > 0 else 0
table += f"| {chain} | {mcap_str} | {percentage:.2f}% |\n"
return table.rstrip()
@mcp.tool()
async def get_stablecoin_prices() -> str:
"""Mendapatkan harga saat ini dari semua stablecoin.
Returns:
String berisi harga stablecoin dalam format markdown.
"""
url = "https://stablecoins.llama.fi/stablecoinprices"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return "Tidak ada data harga stablecoin yang tersedia"
table = "| Symbol | Name | Price (USD) | Market Cap (USD) |\n"
table += "|--------|------|-------------|-----------------|\n"
for coin_id, coin_data in data.items():
symbol = coin_data.get('symbol', '???')
name = coin_data.get('name', '???')
price = coin_data.get('price', 0)
price_str = f"${price:.4f}" if isinstance(price, (int, float)) else '???'
mcap = coin_data.get('circulating', 0)
mcap_str = f"${mcap:,.0f}" if isinstance(mcap, (int, float)) else '???'
table += f"| {symbol} | {name} | {price_str} | {mcap_str} |\n"
return table.rstrip()
@mcp.tool()
async def get_percentage_change(coins: str) -> str:
"""Mendapatkan persentase perubahan harga koin dalam berbagai periode waktu.
Args:
coins: Set koin yang dipisahkan koma (misalnya, 'ethereum:0xdF574c24545E5FfEcb9a659c229253D4111d87e1,coingecko:ethereum').
Returns:
String berisi persentase perubahan harga dalam format markdown.
"""
url = f"https://coins.llama.fi/percentage/{coins}"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return f"Tidak ada data persentase perubahan untuk koin {coins} yang tersedia"
coins_data = data.get('coins', {})
if not coins_data:
return f"Tidak ada data persentase perubahan untuk koin {coins} yang tersedia"
table = "| Coin ID | Symbol | Price (USD) | 1h | 24h | 7d | 30d |\n"
table += "|---------|--------|-------------|-----|-----|-----|-----|\n"
for coin_id, coin_info in coins_data.items():
symbol = coin_info.get('symbol', '???')
price = coin_info.get('price', 0)
price_str = f"${price:.4f}" if isinstance(price, (int, float)) else '???'
# Persentase perubahan
h1 = coin_info.get('1h', 0)
h24 = coin_info.get('24h', 0)
d7 = coin_info.get('7d', 0)
d30 = coin_info.get('30d', 0)
h1_str = f"{h1:.2f}%" if isinstance(h1, (int, float)) else '???'
h24_str = f"{h24:.2f}%" if isinstance(h24, (int, float)) else '???'
d7_str = f"{d7:.2f}%" if isinstance(d7, (int, float)) else '???'
d30_str = f"{d30:.2f}%" if isinstance(d30, (int, float)) else '???'
table += f"| {coin_id} | {symbol} | {price_str} | {h1_str} | {h24_str} | {d7_str} | {d30_str} |\n"
return table.rstrip()
@mcp.tool()
async def get_block_by_timestamp(chain: str, timestamp: int) -> str:
"""Mendapatkan nomor blok terdekat dengan timestamp tertentu.
Args:
chain: Nama chain (slug).
timestamp: Waktu dalam format Unix timestamp.
Returns:
String berisi informasi blok dalam format markdown.
"""
url = f"https://coins.llama.fi/block/{chain}/{timestamp}"
data = await fetch_json(url)
if not data or not isinstance(data, dict):
return f"Tidak ada data blok untuk chain {chain} pada timestamp {timestamp} yang tersedia"
height = data.get('height', '???')
timestamp_actual = data.get('timestamp', '???')
return f"Blok terdekat untuk {chain} pada timestamp {timestamp}:\n\n- **Block Height**: {height}\n- **Actual Timestamp**: {timestamp_actual}"
@mcp.tool()
async def get_yields_pools() -> str:
"""Mendapatkan daftar semua pool yield yang tersedia.
Returns:
String berisi daftar pool yield dalam format markdown.
"""
url = "https://yields.llama.fi/pools"
data = await fetch_json(url)
if not data or not isinstance(data, dict) or 'data' not in data:
return "Tidak ada data pool yield yang tersedia"
pools = data.get('data', [])
if not pools:
return "Tidak ada data pool yield yang tersedia"
table = "| Pool | Project | Chain | APY | TVL (USD) |\n"
table += "|------|---------|-------|-----|-----------|\n"
# Limit to 20 pools to avoid too long output
for pool in pools[:20]:
pool_name = pool.get('symbol', '???')
project = pool.get('project', '???')
chain = pool.get('chain', '???')
apy = pool.get('apy', 0)
apy_str = f"{apy:.2f}%" if isinstance(apy, (int, float)) else '???'
tvl = pool.get('tvlUsd', 0)
tvl_str = f"${tvl:,.0f}" if isinstance(tvl, (int, float)) else '???'
table += f"| {pool_name} | {project} | {chain} | {apy_str} | {tvl_str} |\n"
if len(pools) > 20:
table += "\n*Note: Showing only 20 pools out of " + str(len(pools)) + "*"
return table.rstrip()
@mcp.tool()
async def get_yields_pool(pool_id: str) -> str:
"""Mendapatkan informasi detail tentang pool yield tertentu.
Args:
pool_id: ID pool.
Returns:
String berisi informasi pool yield dalam format markdown.
"""
url = f"https://yields.llama.fi/pool/{pool_id}"
data = await fetch_json(url)
if not data or not isinstance(data, dict) or 'data' not in data:
return f"Tidak ada data untuk pool yield dengan ID {pool_id} yang tersedia"
pool = data.get('data', {})
if not pool:
return f"Tidak ada data untuk pool yield dengan ID {pool_id} yang tersedia"
pool_name = pool.get('symbol', '???')
project = pool.get('project', '???')
chain = pool.get('chain', '???')
apy = pool.get('apy', 0)
apy_str = f"{apy:.2f}%" if isinstance(apy, (int, float)) else '???'
tvl = pool.get('tvlUsd', 0)
tvl_str = f"${tvl:,.0f}" if isinstance(tvl, (int, float)) else '???'
result = f"## {pool_name} ({project} on {chain})\n\n"
result += f"- **APY**: {apy_str}\n"
result += f"- **TVL**: {tvl_str}\n"
# Additional details if available
if 'apyBase' in pool:
apy_base = pool.get('apyBase', 0)
apy_base_str = f"{apy_base:.2f}%" if isinstance(apy_base, (int, float)) else '???'
result += f"- **Base APY**: {apy_base_str}\n"
if 'apyReward' in pool:
apy_reward = pool.get('apyReward', 0)
apy_reward_str = f"{apy_reward:.2f}%" if isinstance(apy_reward, (int, float)) else '???'
result += f"- **Reward APY**: {apy_reward_str}\n"
if 'rewardTokens' in pool:
reward_tokens = pool.get('rewardTokens', [])
if reward_tokens:
result += f"- **Reward Tokens**: {', '.join(reward_tokens)}\n"
if 'underlyingTokens' in pool:
underlying_tokens = pool.get('underlyingTokens', [])
if underlying_tokens:
result += f"- **Underlying Tokens**: {', '.join(underlying_tokens)}\n"
# Historical APY data if available
apy_history = pool.get('apyHistory', {})
if apy_history:
result += "\n### Historical APY\n\n"
result += "| Date | APY |\n"
result += "|------|-----|\n"
for date, apy_value in list(apy_history.items())[:10]: # Limit to 10 entries
apy_value_str = f"{apy_value:.2f}%" if isinstance(apy_value, (int, float)) else '???'
result += f"| {date} | {apy_value_str} |\n"
if len(apy_history) > 10:
result += "\n*Note: Showing only 10 most recent entries*"
return result.rstrip()
@mcp.tool()
async def get_volumes_overview_dexs(chain: str = None) -> str:
"""Mendapatkan overview volume DEX.
Args:
chain: Nama chain (slug). Jika None, akan menampilkan data untuk semua chain.
Returns:
String berisi overview volume DEX dalam format markdown.
"""
url = "https://api.llama.fi/overview/dexs"
if chain:
url = f"https://api.llama.fi/overview/dexs/{chain}"
data = await fetch_json(url)
if not data or not isinstance(data, dict) or 'totalDataChart' not in data:
return "Tidak ada data volume DEX yang tersedia"
# Total volume data
total_data = data.get('totalDataChart', [])
total_volume_24h = data.get('total24h', 0)
total_volume_24h_str = f"${total_volume_24h:,.0f}" if isinstance(total_volume_24h, (int, float)) else '???'
result = f"## DEX Volume Overview{' for ' + chain if chain else ''}\n\n"
result += f"**Total 24h Volume**: {total_volume_24h_str}\n\n"
# DEX breakdown
dexs_data = data.get('dexs', [])
if dexs_data:
result += "### DEX Breakdown\n\n"
result += "| DEX | 24h Volume | Market Share |\n"
result += "|-----|------------|--------------|\n"
for dex in dexs_data[:15]: # Limit to 15 entries
name = dex.get('name', '???')
volume_24h = dex.get('volume', 0)
volume_24h_str = f"${volume_24h:,.0f}" if isinstance(volume_24h, (int, float)) else '???'
market_share = dex.get('volumeShare', 0) * 100 # Convert to percentage
market_share_str = f"{market_share:.2f}%" if isinstance(market_share, (int, float)) else '???'
result += f"| {name} | {volume_24h_str} | {market_share_str} |\n"
if len(dexs_data) > 15:
result += "\n*Note: Showing only top 15 DEXs*"
return result.rstrip()
@mcp.tool()
async def get_fees_overview(chain: str = None) -> str:
"""Mendapatkan overview fees dan revenue.
Args:
chain: Nama chain (slug). Jika None, akan menampilkan data untuk semua chain.
Returns:
String berisi overview fees dan revenue dalam format markdown.
"""
url = "https://api.llama.fi/overview/fees"
if chain:
url = f"https://api.llama.fi/overview/fees/{chain}"
data = await fetch_json(url)
if not data or not isinstance(data, dict) or 'totalDataChart' not in data:
return "Tidak ada data fees dan revenue yang tersedia"
# Total fees data
total_fees_24h = data.get('total24h', 0)
total_fees_24h_str = f"${total_fees_24h:,.0f}" if isinstance(total_fees_24h, (int, float)) else '???'
result = f"## Fees & Revenue Overview{' for ' + chain if chain else ''}\n\n"
result += f"**Total 24h Fees**: {total_fees_24h_str}\n\n"
# Protocol breakdown
protocols_data = data.get('protocols', [])
if protocols_data:
result += "### Protocol Breakdown\n\n"
result += "| Protocol | 24h Fees | 24h Revenue | Market Share |\n"
result += "|----------|----------|-------------|--------------|\n"
for protocol in protocols_data[:15]: # Limit to 15 entries
name = protocol.get('name', '???')
fees_24h = protocol.get('total24h', 0)
fees_24h_str = f"${fees_24h:,.0f}" if isinstance(fees_24h, (int, float)) else '???'
revenue_24h = protocol.get('revenue24h', 0)
revenue_24h_str = f"${revenue_24h:,.0f}" if isinstance(revenue_24h, (int, float)) else '???'
market_share = protocol.get('marketShare', 0) * 100 # Convert to percentage
market_share_str = f"{market_share:.2f}%" if isinstance(market_share, (int, float)) else '???'
result += f"| {name} | {fees_24h_str} | {revenue_24h_str} | {market_share_str} |\n"
if len(protocols_data) > 15:
result += "\n*Note: Showing only top 15 protocols*"
return result.rstrip()
if __name__ == "__main__":
mcp.run()
