Aim to make A Model Context Protocol (MCP) server for collecting, processing, and serving cryptocurrency market data including real-time prices, historical data, and market information. This server can be used both as a traditional REST API server and as an MCP server for AI assistants.
The remaining components are largely analogous to the principles of a RESTful API; however, the subsequent segment of code is critical for the operation of the MCP server.
#!/usr/bin/env python3
"""
Cryptocurrency Market Data MCP Server
This MCP server provides tools for fetching cryptocurrency market data,
including real-time prices, historical data, and market information.
"""
import asyncio
import json
import logging
from typing import Any, Dict, List, Optional
from datetime import datetime
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool,
TextContent,
ImageContent,
EmbeddedResource,
LoggingLevel
)
from services.market_data_service import MarketDataService
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Initialize services
market_data_service = MarketDataService()
# Create MCP server
server = Server("crypto-market-server")
@server.list_tools()
async def list_tools() -> List[Tool]:
"""List available tools for cryptocurrency market data."""
return [
Tool(
name="get_crypto_price",
description="Get current price for a specific cryptocurrency",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH, ADA)"
},
"exchange": {
"type": "string",
"description": "Exchange name (optional, defaults to binance)",
"default": "binance"
}
},
"required": ["symbol"]
}
),
Tool(
name="get_multiple_prices",
description="Get current prices for multiple cryptocurrencies",
inputSchema={
"type": "object",
"properties": {
"symbols": {
"type": "array",
"items": {"type": "string"},
"description": "List of cryptocurrency symbols"
},
"exchange": {
"type": "string",
"description": "Exchange name (optional, defaults to binance)",
"default": "binance"
}
},
"required": ["symbols"]
}
),
Tool(
name="get_historical_data",
description="Get historical price data for a cryptocurrency",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Cryptocurrency symbol"
},
"timeframe": {
"type": "string",
"description": "Timeframe (1m, 5m, 15m, 1h, 4h, 1d)",
"default": "1d"
},
"limit": {
"type": "integer",
"description": "Number of data points to retrieve",
"default": 100
},
"exchange": {
"type": "string",
"description": "Exchange name (optional, defaults to binance)",
"default": "binance"
}
},
"required": ["symbol"]
}
),
Tool(
name="get_market_summary",
description="Get market summary for top cryptocurrencies",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Number of cryptocurrencies to include",
"default": 10
},
"exchange": {
"type": "string",
"description": "Exchange name (optional, defaults to binance)",
"default": "binance"
}
}
}
),
Tool(
name="get_exchange_info",
description="Get information about available exchanges",
inputSchema={
"type": "object",
"properties": {
"exchange": {
"type": "string",
"description": "Specific exchange name (optional)"
}
}
}
),
Tool(
name="search_markets",
description="Search for available trading pairs on an exchange",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query for trading pairs"
},
"exchange": {
"type": "string",
"description": "Exchange name (optional, defaults to binance)",
"default": "binance"
}
},
"required": ["query"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls for cryptocurrency market data."""
try:
if name == "get_crypto_price":
symbol = arguments["symbol"].upper()
exchange = arguments.get("exchange", "binance")
price_data = await market_data_service.get_price(symbol, exchange)
if not price_data:
return [TextContent(
type="text",
text=f"Could not fetch price data for {symbol} on {exchange}"
)]
result = {
"symbol": price_data["symbol"],
"price": price_data["price"],
"change_24h": price_data.get("change_24h", 0),
"volume_24h": price_data.get("volume_24h", 0),
"exchange": exchange,
"last_updated": price_data.get("last_updated", datetime.now().isoformat())
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
elif name == "get_multiple_prices":
symbols = [s.upper() for s in arguments["symbols"]]
exchange = arguments.get("exchange", "binance")
prices = await market_data_service.get_multiple_prices(symbols, exchange)
if not prices:
return [TextContent(
type="text",
text=f"Could not fetch price data for symbols on {exchange}"
)]
return [TextContent(
type="text",
text=json.dumps(prices, indent=2)
)]
elif name == "get_historical_data":
symbol = arguments["symbol"].upper()
timeframe = arguments.get("timeframe", "1d")
limit = arguments.get("limit", 100)
exchange = arguments.get("exchange", "binance")
historical_data = await market_data_service.get_historical_data(
symbol, timeframe, limit, exchange
)
if not historical_data:
return [TextContent(
type="text",
text=f"Could not fetch historical data for {symbol} on {exchange}"
)]
return [TextContent(
type="text",
text=json.dumps(historical_data, indent=2)
)]
elif name == "get_market_summary":
limit = arguments.get("limit", 10)
exchange = arguments.get("exchange", "binance")
summary = await market_data_service.get_market_summary(limit, exchange)
if not summary:
return [TextContent(
type="text",
text=f"Could not fetch market summary from {exchange}"
)]
return [TextContent(
type="text",
text=json.dumps(summary, indent=2)
)]
elif name == "get_exchange_info":
exchange_name = arguments.get("exchange")
if exchange_name:
info = await market_data_service.get_exchange_info(exchange_name)
else:
info = await market_data_service.get_available_exchanges()
return [TextContent(
type="text",
text=json.dumps(info, indent=2)
)]
elif name == "search_markets":
query = arguments["query"]
exchange = arguments.get("exchange", "binance")
markets = await market_data_service.search_markets(query, exchange)
if not markets:
return [TextContent(
type="text",
text=f"No markets found for query '{query}' on {exchange}"
)]
return [TextContent(
type="text",
text=json.dumps(markets, indent=2)
)]
else:
return [TextContent(
type="text",
text=f"Unknown tool: {name}"
)]
except Exception as e:
logger.error(f"Error in tool {name}: {str(e)}")
return [TextContent(
type="text",
text=f"Error executing {name}: {str(e)}"
)]
async def main():
"""Main entry point for the MCP server."""
logger.info("Starting Cryptocurrency Market Data MCP Server")
# Initialize market data service
try:
await market_data_service.initialize()
logger.info("Market data service initialized successfully")
except Exception as e:
logger.error(f"Failed to initialize market data service: {e}")
return
# Run the server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
An additional critical component is, undoubtedly, the execution of the MCP server script to ensure mcp server is setup and ready to use properly, it’s quite verbose though
#!/usr/bin/env python3
"""
Setup script for the Cryptocurrency Market Data MCP Server
"""
import os
import sys
import subprocess
import json
from pathlib import Path
def install_dependencies():
"""Install required dependencies"""
print("Installing dependencies...")
try:
subprocess.check_call([sys.executable, "-m", "pip", "install", "-r", "requirements.txt"])
print("✓ Dependencies installed successfully")
return True
except subprocess.CalledProcessError as e:
print(f"✗ Failed to install dependencies: {e}")
return False
def create_env_file():
"""Create .env file if it doesn't exist"""
env_file = Path(".env")
if not env_file.exists():
print("Creating .env file...")
env_content = """# Cryptocurrency Market Data MCP Server Configuration
# Exchanges to use (comma-separated)
EXCHANGES=binance,coinbasepro,kraken
# API Keys (optional - leave empty for public endpoints only)
# BINANCE_API_KEY=your_binance_api_key
# BINANCE_SECRET=your_binance_secret
# COINBASEPRO_API_KEY=your_coinbase_api_key
# COINBASEPRO_SECRET=your_coinbase_secret
# KRAKEN_API_KEY=your_kraken_api_key
# KRAKEN_SECRET=your_kraken_secret
# Server configuration
PORT=8000
"""
with open(env_file, "w") as f:
f.write(env_content)
print("✓ .env file created")
else:
print("✓ .env file already exists")
def test_mcp_server():
"""Test if the MCP server can be imported"""
print("Testing MCP server...")
try:
# Try to import the MCP server
import mcp_server
print("✓ MCP server can be imported successfully")
return True
except ImportError as e:
print(f"✗ Failed to import MCP server: {e}")
return False
def show_usage_instructions():
"""Show usage instructions"""
print("\n" + "="*60)
print("🚀 Cryptocurrency Market Data MCP Server Setup Complete!")
print("="*60)
print("\nUsage Options:")
print("\n1. Run as MCP Server:")
print(" python mcp_server.py")
print("\n2. Test MCP Server:")
print(" python test_mcp.py")
print("\n3. Run as REST API Server:")
print(" python main.py")
print("\n4. Configuration:")
print(" Edit .env file to configure exchanges and API keys")
print("\nMCP Tools Available:")
print(" - get_crypto_price")
print(" - get_multiple_prices")
print(" - get_historical_data")
print(" - get_market_summary")
print(" - get_exchange_info")
print(" - search_markets")
print("\nFor more information, see README.md")
def main():
"""Main setup function"""
print("Setting up Cryptocurrency Market Data MCP Server...")
print("-" * 50)
# Check if we're in the right directory
if not Path("mcp_server.py").exists():
print("✗ mcp_server.py not found. Please run this script from the project root directory.")
sys.exit(1)
success = True
# Install dependencies
if not install_dependencies():
success = False
# Create .env file
create_env_file()
# Test MCP server
if not test_mcp_server():
success = False
if success:
show_usage_instructions()
else:
print("\n✗ Setup completed with errors. Please check the output above.")
sys.exit(1)
if __name__ == "__main__":
main()
As is included in above mcp_server script, the test_mcp_server is necessary as the user need to ensure once they set up the server config json file, they expect to run and retrieve what they request.
#!/usr/bin/env python3
"""
Test script for the Cryptocurrency Market Data MCP Server
"""
import asyncio
import json
import subprocess
import sys
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def test_mcp_server():
"""Test the MCP server functionality"""
# Start the MCP server
server_params = StdioServerParameters(
command="python",
args=["mcp_server.py"],
env=None
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the session
await session.initialize()
# List available tools
print("Available tools:")
tools = await session.list_tools()
for tool in tools.tools:
print(f"- {tool.name}: {tool.description}")
print("\n" + "="*50 + "\n")
# Test getting Bitcoin price
print("Testing: Get Bitcoin price")
try:
result = await session.call_tool(
"get_crypto_price",
{"symbol": "BTC", "exchange": "binance"}
)
print("Result:")
for content in result.content:
if hasattr(content, 'text'):
print(content.text)
except Exception as e:
print(f"Error: {e}")
print("\n" + "="*50 + "\n")
# Test getting multiple prices
print("Testing: Get multiple cryptocurrency prices")
try:
result = await session.call_tool(
"get_multiple_prices",
{"symbols": ["BTC", "ETH", "ADA"], "exchange": "binance"}
)
print("Result:")
for content in result.content:
if hasattr(content, 'text'):
print(content.text)
except Exception as e:
print(f"Error: {e}")
print("\n" + "="*50 + "\n")
# Test market summary
print("Testing: Get market summary")
try:
result = await session.call_tool(
"get_market_summary",
{"limit": 5, "exchange": "binance"}
)
print("Result:")
for content in result.content:
if hasattr(content, 'text'):
print(content.text)
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
asyncio.run(test_mcp_server())
Naixian Zhang 