Coin MCP Server
by longmans
- src
- coin_api_mcp
from typing import Any
import asyncio
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
import urllib.parse
import json
API_BASE = "pro-api.coinmarketcap.com"
API_KEY = None
server = Server("coin-api")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
List available tools.
Each tool specifies its arguments using JSON Schema validation.
"""
return [
types.Tool(
name="listing-coins",
description="Returns a paginated list of all active cryptocurrencies with latest market data",
inputSchema={
"type": "object",
"properties": {
"start": {
"type": "integer",
"description": "Optionally offset the start (1-based index) of the paginated list of items to return.",
"minimum": 1,
},
"limit": {
"type": "integer",
"description": "Optionally specify the number of results to return.",
"minimum": 1,
"maximum": 5000,
},
"price_min": {
"type": "number",
"description": "Optionally specify a threshold of minimum USD price to filter results by.",
"minimum": 0,
},
"price_max": {
"type": "number",
"description": "Optionally specify a threshold of maximum USD price to filter results by.",
"minimum": 0,
},
"market_cap_min": {
"type": "number",
"description": "Optionally specify a threshold of minimum market cap to filter results by.",
"minimum": 0,
},
"market_cap_max": {
"type": "number",
"description": "Optionally specify a threshold of maximum market cap to filter results by.",
"minimum": 0,
},
"volume_24h_min": {
"type": "number",
"description": "Optionally specify a threshold of minimum 24 hour USD volume to filter results by.",
"minimum": 0,
},
"volume_24h_max": {
"type": "number",
"description": "Optionally specify a threshold of maximum 24 hour USD volume to filter results by.",
"minimum": 0,
},
"circulating_supply_min": {
"type": "number",
"description": "Optionally specify a threshold of minimum circulating supply to filter results by.",
"minimum": 0,
},
"circulating_supply_max": {
"type": "number",
"description": "Optionally specify a threshold of maximum circulating supply to filter results by.",
"minimum": 0,
},
"percent_change_24h_min": {
"type": "number",
"description": "Optionally specify a threshold of minimum 24 hour percent change to filter results by.",
"minimum": -100,
},
"percent_change_24h_max": {
"type": "number",
"description": "Optionally specify a threshold of maximum 24 hour percent change to filter results by.",
"minimum": -100,
},
"convert": {
"type": "string",
"description": "Optionally calculate market quotes in up to 120 currencies at once by passing a comma-separated list of cryptocurrency or fiat currency symbols.",
},
"convert_id": {
"type": "string",
"description": "Optionally calculate market quotes by CoinMarketCap ID instead of symbol.",
},
"sort": {
"type": "string",
"description": "What field to sort the list of cryptocurrencies by.",
"enum": [
"market_cap",
"name",
"symbol",
"date_added",
"market_cap_strict",
"price",
"circulating_supply",
"total_supply",
"max_supply",
"num_market_pairs",
"volume_24h",
"percent_change_1h",
"percent_change_24h",
"percent_change_7d",
"market_cap_by_total_supply_strict",
"volume_7d",
"volume_30d",
],
},
"sort_dir": {
"type": "string",
"description": "The direction in which to order cryptocurrencies against the specified sort.",
"enum": ["asc", "desc"],
},
"cryptocurrency_type": {
"type": "string",
"description": "The type of cryptocurrency to include.",
"enum": ["all", "coins", "tokens"],
},
"tag": {
"type": "string",
"description": "The tag of cryptocurrency to include.",
"enum": ["all", "defi", "filesharing"],
},
"aux": {
"type": "string",
"description": "Optionally specify a comma-separated list of supplemental data fields to return.",
},
},
"required": [],
},
),
types.Tool(
name="get-coin-info",
description="Get coins' information includes details like logo, description, official website URL, social links, and links to a cryptocurrency's technical documentation.",
inputSchema={
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "One or more comma-separated CoinMarketCap cryptocurrency IDs. Example: \"1,2\"",
},
"slug": {
"type": "string",
"description": "Alternatively pass a comma-separated list of cryptocurrency slugs. Example: \"bitcoin,ethereum\"",
},
"symbol": {
"type": "string",
"description": "Alternatively pass one or more comma-separated cryptocurrency symbols. Example: \"BTC,ETH\"",
},
"address": {
"type": "string",
"description": "Alternatively pass in a contract address. Example: \"0xc40af1e4fecfa05ce6bab79dcd8b373d2e436c4e\"",
},
"skip_invalid": {
"type": "boolean",
"description": "Pass true to relax request validation rules. When requesting records on multiple cryptocurrencies an error is returned if any invalid cryptocurrencies are requested or a cryptocurrency does not have matching records in the requested timeframe. If set to true, invalid lookups will be skipped allowing valid cryptocurrencies to still be returned.",
"default": False,
},
"aux": {
"type": "string",
"description": "Optionally specify a comma-separated list of supplemental data fields to return. Pass urls,logo,description,tags,platform,date_added,notice,status to include all auxiliary fields.",
},
},
"required": [],
}
),
types.Tool(
name="get-coin-quotes",
description='''the latest market quote for 1 or more cryptocurrencies. Use the "convert" option to return market values in multiple fiat and cryptocurrency conversions in the same call.''',
inputSchema={
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "One or more comma-separated cryptocurrency CoinMarketCap IDs. Example: 1,2",
},
"slug": {
"type": "string",
"description": "Alternatively pass a comma-separated list of cryptocurrency slugs. Example: \"bitcoin,ethereum\"",
},
"symbol": {
"type": "string",
"description": "Alternatively pass one or more comma-separated cryptocurrency symbols. Example: \"BTC,ETH\"",
},
"convert": {
"type": "string",
"description": "Optionally calculate market quotes in up to 120 currencies at once by passing a comma-separated list of cryptocurrency or fiat currency symbols.",
},
"convert_id": {
"type": "string",
"description": "Optionally calculate market quotes by CoinMarketCap ID instead of symbol. This option is identical to convert outside of ID format.",
},
"aux": {
"type": "string",
"description": "\"num_market_pairs,cmc_rank,date_added,tags,platform,max_supply,circulating_supply,total_supply,is_active,is_fiat\"Optionally specify a comma-separated list of supplemental data fields to return.",
},
"skip_invalid": {
"type": "boolean",
"description": "Pass true to relax request validation rules.",
"default": False,
},
},
"required": [],
},
),
]
async def make_coinmarketcap_request(client: httpx.AsyncClient, url: str) -> dict[str, Any] | None:
"""Make a request to the CoinMarketCap API with proper error handling."""
headers = {
'Accepts': 'application/json',
'X-CMC_PRO_API_KEY': API_KEY,
}
try:
response = await client.get(url, headers=headers, timeout=10.0)
response.raise_for_status()
return response.json()
except Exception:
return None
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""
Handle tool execution requests.
Currently supported tools:
- listing-coins: fetches a list of all cryptocurrencies.
- get-coin-quotes: fetches quotes for a specific cryptocurrency.
- get-coin-info: fetches information for a specific cryptocurrency.
"""
if name == "listing-coins":
request_data = {}
if arguments is not None:
condi_keys = [
"start",
"limit",
"price_min",
"price_max",
"market_cap_min",
"market_cap_max",
"volume_24h_min",
"volume_24h_max",
"circulating_supply_min",
"circulating_supply_max",
"percent_change_24h_min",
"percent_change_24h_max",
"convert",
"convert_id",
"sort",
"sort_dir",
"cryptocurrency_type",
"tag",
"aux",
]
for key in arguments.keys():
if key in condi_keys:
request_data[key] = arguments[key]
async with httpx.AsyncClient() as client:
listing_url = f"https://{API_BASE}/v1/cryptocurrency/listings/latest?{urllib.parse.urlencode(request_data)}"
listing_data = await make_coinmarketcap_request(client, listing_url)
if not listing_data:
return [types.TextContent(type="text", text="Failed to retrieve listing data")]
return [
types.TextContent(
type="text",
text=json.dumps(listing_data)
)
]
elif name == "get-coin-info":
request_data = {}
if arguments is not None:
condi_keys = [
"id",
"slug",
"symbol",
"address",
"skip_invalid",
"aux",
]
for key in arguments.keys():
if key in condi_keys:
request_data[key] = arguments[key]
async with httpx.AsyncClient() as client:
coin_info_url = f"https://{API_BASE}/v2/cryptocurrency/info?{urllib.parse.urlencode(request_data)}"
coin_info_data = await make_coinmarketcap_request(client, coin_info_url)
if not coin_info_data:
return [types.TextContent(type="text", text="Failed to retrieve coin info data")]
return [
types.TextContent(
type="text",
text=json.dumps(coin_info_data)
)
]
elif name == "get-coin-quotes":
request_data = {}
if arguments is not None:
condi_keys = [
"id",
"slug",
"symbol",
"convert",
"convert_id",
"aux",
"skip_invalid",
]
for key in arguments.keys():
if key in condi_keys:
request_data[key] = arguments[key]
async with httpx.AsyncClient() as client:
coin_quotes_url = f"https://{API_BASE}/v2/cryptocurrency/quotes/latest?{urllib.parse.urlencode(request_data)}"
coin_quotes_data = await make_coinmarketcap_request(client, coin_quotes_url)
if not coin_quotes_data:
return [types.TextContent(type="text", text="Failed to retrieve coin quotes data")]
return [
types.TextContent(
type="text",
text=json.dumps(coin_quotes_data)
)
]
else:
raise ValueError(f"Unknown tool: {name}")
async def main(api_key: str):
global API_KEY
API_KEY = api_key
# Run the server using stdin/stdout streams
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="coin-api",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
# This is needed if you'd like to connect to a custom client
if __name__ == "__main__":
#get api key from coinmarketcap
api_key = "xxxxxxx"
asyncio.run(main(api_key))