dexscreener_server.py•13.7 kB
"""DEX Screener MCP Server
This server provides tools for interacting with the DexScreener API to get information
about trading pairs, tokens, and market data from various decentralized exchanges.
"""
from __future__ import annotations
from typing import Any, List, Optional
import urllib.parse
import aiohttp
import json
import sys
import traceback
from mcp.server.fastmcp import FastMCP
# ---------------------------------------------------------------------------
# Constants & configuration
# ---------------------------------------------------------------------------
# Initialize FastMCP
mcp = FastMCP(name="dexscreener-server")
# DexScreener API configuration
BASE_URL = "https://api.dexscreener.com"
HEADERS = {
"User-Agent": "dexscreener-mcp/1.0 (+https://dexscreener.com)",
"Accept": "application/json"
}
import aiohttp
import json
import sys
async def _fetch_dex_json(endpoint: str, params: Optional[dict] = None) -> dict[str, Any] | list[Any] | None:
"""Fetch from DexScreener API and return parsed JSON or None.
Args:
endpoint: API endpoint to fetch (without base URL)
params: Optional query parameters
"""
url = f"{BASE_URL}{endpoint}"
try:
print(f"DEBUG - Sending request to: {url}", file=sys.stderr)
if params:
print(f"DEBUG - With params: {params}", file=sys.stderr)
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=HEADERS, params=params) as response:
if response.status != 200:
print(f"ERROR: API returned status {response.status} for URL {url}", file=sys.stderr)
return None
text = await response.text()
print(f"DEBUG - Response text (first 100 chars): {text[:100]}...", file=sys.stderr)
try:
return json.loads(text)
except json.JSONDecodeError as e:
print(f"ERROR: Failed to parse JSON: {str(e)}", file=sys.stderr)
print(f"Response text (first 200 chars): {text[:200]}...", file=sys.stderr)
return None
except Exception as e:
print(f"ERROR in _fetch_dex_json: {str(e)}", file=sys.stderr)
import traceback
print(traceback.format_exc(), file=sys.stderr)
return None
# ---------------------------------------------------------------------------
# Helper functions for formatting output
# ---------------------------------------------------------------------------
def _format_pair_summary(pairs: List[dict]) -> str:
"""Format pair data as markdown table."""
if not pairs:
return "No pairs found"
summary = "| Chain | Pair | Price | Volume 24h | Liquidity |\n"
summary += "|-------|------|-------|------------|------------|\n"
for pair in pairs:
dex = pair.get('dexId', 'Unknown DEX')
chain = pair.get('chainId', 'Unknown')
base_symbol = pair.get('baseToken', {}).get('symbol', '???')
quote_symbol = pair.get('quoteToken', {}).get('symbol', '???')
pair_name = f"{base_symbol}/{quote_symbol} ({dex})"
price_usd = pair.get('priceUsd', '???')
volume_usd = pair.get('volume', {}).get('h24', '???')
liquidity_usd = pair.get('liquidity', {}).get('usd', '???')
if isinstance(price_usd, (int, float)):
price_usd = f"${price_usd:.4f}"
if isinstance(volume_usd, (int, float)):
volume_usd = f"${volume_usd:,.0f}"
if isinstance(liquidity_usd, (int, float)):
liquidity_usd = f"${liquidity_usd:,.0f}"
summary += f"| {chain} | {pair_name} | {price_usd} | {volume_usd} | {liquidity_usd} |\n"
return summary.rstrip()
# Fungsi _format_pool_summary dihapus karena tidak diperlukan lagi
# Fungsi _format_token_summary dan _format_order_summary dihapus karena tidak diperlukan lagi
# ---------------------------------------------------------------------------
# MCP Tools
# ---------------------------------------------------------------------------
@mcp.tool()
async def get_pair_by_chain_and_address(chain_id: str, pair_id: str) -> str:
"""Get specific trading pair information by chain and pair address.
Args:
chain_id: Chain identifier (e.g., ethereum, bsc, polygon)
pair_id: Pair contract address
"""
try:
# Endpoint sesuai dengan spesifikasi API terbaru: /latest/dex/pairs/{chainId}/{pairId}
endpoint = f"/latest/dex/pairs/{chain_id}/{pair_id}"
data = await _fetch_dex_json(endpoint)
if not data or 'pair' not in data:
return "Pair not found or no data available"
# API mengembalikan objek 'pair' bukan array 'pairs'
pair = data['pair']
return _format_pair_summary([pair])
except Exception as e:
return f"Error fetching pair data: {str(e)}"
@mcp.tool()
async def get_token_pairs(chain_id: str, token_address: str) -> str:
"""Get all trading pairs for a specific token.
Args:
chain_id: Chain identifier (e.g., ethereum, bsc, polygon)
token_address: Token contract address
"""
try:
# Endpoint sesuai dengan spesifikasi API terbaru: /token-pairs/v1/{chainId}/{tokenAddress}
endpoint = f"/token-pairs/v1/{chain_id}/{token_address}"
data = await _fetch_dex_json(endpoint)
if not data or 'pairs' not in data or not data['pairs']:
return "No pairs found for the given token"
return _format_pair_summary(data['pairs'])
except Exception as e:
return f"Error fetching token pairs data: {str(e)}"
@mcp.tool()
async def search_pairs(query: str) -> str:
"""Search for trading pairs by token symbol or address.
Args:
query: Search query (token symbol, name, or address)
"""
try:
# Endpoint sesuai dengan spesifikasi API terbaru: /latest/dex/search
endpoint = "/latest/dex/search"
params = {"q": query}
data = await _fetch_dex_json(endpoint, params)
if not data or 'pairs' not in data or not data['pairs']:
return "No pairs found matching the search query"
return _format_pair_summary(data['pairs'][:5])
except Exception as e:
return f"Error searching pairs: {str(e)}"
@mcp.tool()
async def get_token_profiles_latest(token_addresses: str = None, chain_id: str = None) -> str:
"""Get latest token profiles with metadata and social information.
Args:
token_addresses: Comma-separated list of token addresses
chain_id: Chain identifier
"""
try:
endpoint = "/token-profiles/latest/v1"
params = {}
if token_addresses:
params["tokenAddresses"] = token_addresses
if chain_id:
params["chainId"] = chain_id
data = await _fetch_dex_json(endpoint, params)
if not data or 'profiles' not in data or not data['profiles']:
return "No token profiles found"
profiles = data['profiles']
summary = "| Chain | Token | Description | Links |\n"
summary += "|-------|-------|-------------|-------|\n"
for profile in profiles:
chain = profile.get('chainId', 'Unknown')
token_address = profile.get('tokenAddress', 'Unknown')
description = profile.get('description', 'No description')
if len(description) > 50:
description = description[:47] + "..."
links = profile.get('links', [])
link_text = ", ".join([f"[{link.get('label', 'Link')}]({link.get('url', '#')})" for link in links[:3]])
if not link_text:
link_text = "No links"
summary += f"| {chain} | {token_address} | {description} | {link_text} |\n"
return summary.rstrip()
except Exception as e:
return f"Error fetching token profiles: {str(e)}"
@mcp.tool()
async def get_token_boosts_latest(token_addresses: str = None) -> str:
"""Get latest token boosts information.
Args:
token_addresses: Comma-separated list of token addresses
"""
try:
print(f"DEBUG - Fetching latest token boosts for addresses: {token_addresses}", file=sys.stderr)
endpoint = "/token-boosts/latest/v1"
params = {}
if token_addresses:
params["tokenAddresses"] = token_addresses
data = await _fetch_dex_json(endpoint, params)
if data:
print(f"DEBUG - Response received: {str(data)[:200]}...", file=sys.stderr)
if not data:
return "No data received from API"
if 'boosts' not in data:
return f"API response does not contain 'boosts' field. Response: {str(data)[:200]}..."
if not data['boosts']:
return "No token boosts found"
boosts = data['boosts']
summary = "| Chain | Token | Amount | Total Amount | Description |\n"
summary += "|-------|-------|--------|--------------|-------------|\n"
for boost in boosts:
chain = boost.get('chainId', 'Unknown')
token_address = boost.get('tokenAddress', 'Unknown')
amount = boost.get('amount', 0)
total_amount = boost.get('totalAmount', 0)
description = boost.get('description', 'No description')
if len(description) > 30:
description = description[:27] + "..."
summary += f"| {chain} | {token_address} | {amount} | {total_amount} | {description} |\n"
return summary.rstrip()
except Exception as e:
print(f"ERROR in get_token_boosts_latest: {str(e)}", file=sys.stderr)
print(traceback.format_exc(), file=sys.stderr)
return f"Error fetching token boosts: {str(e)}"
@mcp.tool()
async def get_token_boosts_top(chain_id: str = None, limit: int = None) -> str:
"""Get top token boosts ranked by amount.
Args:
chain_id: Chain identifier
limit: Number of results to return
"""
try:
print(f"DEBUG - Sending request to get top token boosts for chain: {chain_id}, limit: {limit}", file=sys.stderr)
endpoint = "/token-boosts/top/v1"
params = {}
if chain_id:
params["chainId"] = chain_id
if limit:
params["limit"] = limit
print(f"DEBUG - Headers: {HEADERS}", file=sys.stderr)
print(f"DEBUG - Params: {params}", file=sys.stderr)
print(f"DEBUG - Full URL: {BASE_URL}{endpoint}?{'&'.join([f'{k}={v}' for k, v in params.items()])}", file=sys.stderr)
data = await _fetch_dex_json(endpoint, params)
if data:
print(f"DEBUG - Response received: {str(data)[:200]}...", file=sys.stderr)
if not data:
return "No data received from API"
if 'boosts' not in data:
return f"API response does not contain 'boosts' field. Response: {str(data)[:200]}..."
if not data['boosts']:
return "No top token boosts found"
boosts = data['boosts']
summary = "| Chain | Token | Amount | Total Amount | Description |\n"
summary += "|-------|-------|--------|--------------|-------------|\n"
for boost in boosts:
chain = boost.get('chainId', 'Unknown')
token_address = boost.get('tokenAddress', 'Unknown')
amount = boost.get('amount', 0)
total_amount = boost.get('totalAmount', 0)
description = boost.get('description', 'No description')
if len(description) > 30:
description = description[:27] + "..."
summary += f"| {chain} | {token_address} | {amount} | {total_amount} | {description} |\n"
return summary.rstrip()
except Exception as e:
print(f"ERROR in get_token_boosts_top: {str(e)}", file=sys.stderr)
print(traceback.format_exc(), file=sys.stderr)
return f"Error fetching top token boosts: {str(e)}"
@mcp.tool()
async def get_orders(chain_id: str, token_address: str) -> str:
"""Get order book data for a specific token.
Args:
chain_id: Chain identifier (e.g., ethereum, bsc, polygon)
token_address: Token contract address
"""
try:
endpoint = f"/orders/v1/{chain_id}/{token_address}"
data = await _fetch_dex_json(endpoint)
if not data or 'orders' not in data:
return "No order book data found for the token"
orders = data['orders']
bids = orders.get('bids', [])
asks = orders.get('asks', [])
summary = "## Bids (Buy Orders)\n\n"
summary += "| Price | Amount | Total |\n"
summary += "|-------|--------|-------|\n"
for bid in bids[:10]: # Limit to top 10 bids
price = bid.get('price', '0')
amount = bid.get('amount', '0')
total = bid.get('total', '0')
summary += f"| {price} | {amount} | {total} |\n"
summary += "\n## Asks (Sell Orders)\n\n"
summary += "| Price | Amount | Total |\n"
summary += "|-------|--------|-------|\n"
for ask in asks[:10]: # Limit to top 10 asks
price = ask.get('price', '0')
amount = ask.get('amount', '0')
total = ask.get('total', '0')
summary += f"| {price} | {amount} | {total} |\n"
return summary.rstrip()
except Exception as e:
return f"Error fetching order book data: {str(e)}"
if __name__ == "__main__":
mcp.run()