main.py•4.25 kB
import httpx
import json
import re
from typing import Dict, Any, List
from mcp.server.fastmcp import FastMCP
from tabulate import tabulate
# Initialize the MCP server
mcp = FastMCP("Chainlist MCP", dependencies=["httpx", "tabulate"])
# Cache for chain data
chain_data = None
async def fetch_chain_data() -> list[Dict[str, Any]]:
"""Fetch and cache chain data from Chainlist API."""
global chain_data
if chain_data is None:
async with httpx.AsyncClient() as client:
response = await client.get("https://chainlist.org/rpcs.json")
response.raise_for_status()
chain_data = response.json()
return chain_data
def format_chain_as_markdown(chain: Dict[str, Any]) -> str:
"""
Format a single chain's details as Markdown, extracting specified fields with RPC and Explorers as tables.
"""
# Extract required fields
name = chain.get('name', 'N/A')
chain_id = chain.get('chainId', 'N/A')
native_currency = chain.get('nativeCurrency', {})
tvl = chain.get('tvl', 'N/A')
rpc_list = chain.get('rpc', [])
explorers_list = chain.get('explorers', [])
# Format native currency
currency_info = f"{native_currency.get('name', 'N/A')} ({native_currency.get('symbol', 'N/A')}, {native_currency.get('decimals', 'N/A')} decimals)" if native_currency else 'N/A'
# Format RPC list as a table
rpc_data = [[rpc.get('url', 'N/A'), rpc.get('tracking', 'N/A')] for rpc in rpc_list]
rpc_output = "**RPC Endpoints**:\n"
if rpc_data:
rpc_output += tabulate(rpc_data, headers=["URL", "Tracking"], tablefmt="pipe")
else:
rpc_output += "None"
# Format explorers list as a table
explorers_data = [[explorer.get('name', 'N/A'), explorer.get('url', 'N/A'), explorer.get('standard', 'N/A')] for explorer in explorers_list]
explorers_output = "**Explorers**:\n"
if explorers_data:
explorers_output += tabulate(explorers_data, headers=["Name", "URL", "Standard"], tablefmt="pipe")
else:
explorers_output += "None"
# Combine all fields
return f"""**Chain Details**
- **Name**: {name}
- **Chain ID**: {chain_id}
- **Native Currency**: {currency_info}
- **TVL**: {tvl}
{rpc_output}
{explorers_output}
"""
@mcp.tool()
async def getChainById(chain_id: int) -> str:
"""
Retrieve information about a blockchain by its chain ID, returned as Markdown.
**Parameters**:
- `chain_id` (integer): The unique identifier of the blockchain (e.g., 1 for Ethereum Mainnet).
**Returns**:
- A Markdown-formatted string containing the chain's details (Name, Chain ID, Native Currency, TVL, RPC Endpoints, Explorers) or an error message if no chain is found.
"""
chains = await fetch_chain_data()
for chain in chains:
if chain.get("chainId") == chain_id:
return format_chain_as_markdown(chain)
return f"**Error**: No chain found with ID {chain_id}"
@mcp.tool()
async def getChainsByKeyword(keyword: str, limit: int = 5) -> str:
"""
Retrieve information about blockchains matching a keyword (case-insensitive partial match), returned as Markdown.
**Parameters**:
- `keyword` (string): The keyword or partial name of the blockchain to search for (e.g., 'eth' for Ethereum).
- `limit` (integer, optional): Maximum number of matching chains to return (default: 5).
**Returns**:
- A Markdown-formatted string listing up to `limit` matching chains with their details (Name, Chain ID, Native Currency, TVL, RPC Endpoints, Explorers) or an error message if no chains are found.
"""
chains = await fetch_chain_data()
pattern = re.escape(keyword.strip())
matches: List[Dict[str, Any]] = []
for chain in chains:
chain_name = chain.get("name", "")
if re.search(pattern, chain_name, re.IGNORECASE):
matches.append(chain)
if matches:
output = "**Matching Chains**\n\n"
# Apply limit to matches
for i, chain in enumerate(matches[:limit], 1):
output += f"### Chain {i}\n{format_chain_as_markdown(chain)}\n"
return output
return f"**Error**: No chains found matching keyword '{keyword}'"
if __name__ == "__main__":
mcp.run()