mcp-crypto-server

anjali04853/mcp-crypto-server

3.1

If you are the rightful owner of mcp-crypto-server and would like to certify it and/or have it hosted online, please leave a comment on the right or send an email to dayong@mcphub.com.

MCP Cryptocurrency Market Data Server is a robust server for retrieving real-time and historical cryptocurrency market data from major exchanges using CCXT.

MCP Cryptocurrency Market Data Server

A robust, production-ready Model Context Protocol (MCP) server for retrieving real-time and historical cryptocurrency market data from major exchanges using CCXT.

🚀 Features

Core Functionality

  • Real-time Market Data: Fetch current ticker prices, volumes, and 24h changes
  • Historical Data: Retrieve OHLCV (Open, High, Low, Close, Volume) candlestick data
  • Order Book Data: Get real-time order book with bids and asks
  • Multi-Exchange Support: Binance, Coinbase, Kraken, Bybit, and OKX
  • Batch Operations: Fetch multiple symbols concurrently for efficiency
  • Symbol Search: Find trading pairs across exchanges

Advanced Features

  • Intelligent Caching: TTL-based caching with hit/miss statistics
  • Error Handling: Comprehensive error handling with retry logic
  • Rate Limiting: Built-in rate limit compliance via CCXT
  • Health Monitoring: Server health checks and diagnostics
  • Async/Await: Fully asynchronous for high performance
  • Type Safety: Strongly typed with dataclasses

📋 Requirements

  • Python 3.8+
  • Internet connection for API access
  • No API keys required for public endpoints

🔧 Installation

  1. Clone the repository:
git clone <your-repo-url>
cd mcp-crypto-server
  1. Create a virtual environment:
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
  1. Install dependencies:
pip install -r requirements.txt

🎯 Quick Start

Basic Usage

import asyncio
from mcp_crypto_server import MCPCryptoServer

async def main():
    # Initialize server
    server = MCPCryptoServer(cache_ttl=60)
    await server.initialize()
    
    try:
        # Get current BTC price
        ticker = await server.get_ticker('BTC/USDT', 'binance')
        print(f"BTC Price: ${ticker.price:,.2f}")
        
        # Get historical data
        historical = await server.get_historical_data(
            'BTC/USDT', 
            timeframe='1h', 
            limit=24
        )
        print(f"Fetched {len(historical)} hourly candles")
        
    finally:
        await server.cleanup()

asyncio.run(main())

Running the Example

python mcp_crypto_server.py

📚 API Documentation

Initialization

server = MCPCryptoServer(cache_ttl=60)
await server.initialize()

Get Ticker Data

Fetch current market data for a trading pair:

ticker = await server.get_ticker(
    symbol='BTC/USDT',
    exchange='binance'
)

Returns: MarketData object with:

  • symbol: Trading pair
  • exchange: Exchange name
  • price: Current price
  • volume: 24h volume
  • high: 24h high
  • low: 24h low
  • timestamp: Data timestamp
  • change_24h: 24h percentage change

Get Multiple Tickers

Fetch multiple tickers concurrently:

tickers = await server.get_multiple_tickers(
    symbols=['BTC/USDT', 'ETH/USDT', 'BNB/USDT'],
    exchange='binance'
)

Get Historical Data

Fetch OHLCV candlestick data:

historical = await server.get_historical_data(
    symbol='BTC/USDT',
    timeframe='1h',  # '1m', '5m', '15m', '1h', '4h', '1d', etc.
    limit=100,
    exchange='binance',
    since=datetime.now() - timedelta(days=7)  # Optional
)

Returns: List of HistoricalData objects with:

  • timestamp: Candle timestamp
  • open: Opening price
  • high: Highest price
  • low: Lowest price
  • close: Closing price
  • volume: Trading volume

Get Order Book

Fetch current order book:

orderbook = await server.get_orderbook(
    symbol='BTC/USDT',
    exchange='binance',
    limit=20
)

Get Market Summary

Get comprehensive market overview:

summary = await server.get_market_summary(
    symbols=['BTC/USDT', 'ETH/USDT', 'BNB/USDT'],
    exchange='binance'
)

Get Price Range

Analyze price statistics over a period:

price_range = await server.get_price_range(
    symbol='BTC/USDT',
    days=7,
    exchange='binance'
)

Search Symbols

Find trading pairs matching a query:

results = await server.search_symbols(
    query='BTC',
    exchange='binance'
)

Server Health Check

Get server status and diagnostics:

health = await server.get_server_health()

Cache Management

# Get cache statistics
stats = server.get_cache_stats()

# Clear cache
server.clear_cache()

🧪 Testing

Run All Tests

pytest test_mcp_server.py -v

Run with Coverage

pytest test_mcp_server.py --cov=mcp_crypto_server --cov-report=html

Run Specific Test Classes

# Test only caching
pytest test_mcp_server.py::TestCryptoDataCache -v

# Test only server functionality
pytest test_mcp_server.py::TestMCPCryptoServer -v

# Test error handling
pytest test_mcp_server.py::TestErrorHandling -v

Test Coverage

The test suite includes:

  • ✅ Unit Tests: Individual function testing
  • ✅ Integration Tests: End-to-end workflow testing
  • ✅ Error Handling Tests: Edge cases and failure scenarios
  • ✅ Concurrency Tests: Parallel operations
  • ✅ Performance Tests: Cache efficiency validation
  • ✅ Data Structure Tests: Model validation

Target Coverage: 85%+ code coverage

🏗️ Architecture

Design Principles

  1. Async-First: All I/O operations are asynchronous for maximum throughput
  2. Caching Layer: Intelligent TTL-based caching reduces API calls
  3. Error Resilience: Comprehensive error handling with graceful degradation
  4. Type Safety: Strong typing with dataclasses for reliability
  5. Separation of Concerns: Clear separation between data fetching, caching, and business logic

Project Structure

mcp-crypto-server/
├── mcp_crypto_server.py      # Main server implementation
├── test_mcp_server.py         # Comprehensive test suite
├── requirements.txt           # Python dependencies
├── README.md                  # Documentation
├── .gitignore                # Git ignore rules
└── examples/                  # Usage examples (optional)

Key Components

  1. MCPCryptoServer: Main server class managing exchange connections and data retrieval
  2. CryptoDataCache: TTL-based caching system with statistics
  3. MarketData: Data structure for current market information
  4. HistoricalData: Data structure for OHLCV candlestick data

🔍 Implementation Details

Supported Exchanges

  • Binance: World's largest crypto exchange
  • Coinbase: US-based regulated exchange
  • Kraken: European exchange with strong security
  • Bybit: Derivatives and spot trading
  • OKX: Global exchange with diverse offerings

Caching Strategy

  • TTL (Time To Live): Configurable expiration (default: 60 seconds)
  • LRU Eviction: Automatic removal of least recently used items
  • Hit Rate Tracking: Performance monitoring
  • Selective Caching: Different TTL for different data types

Error Handling

  • Network timeouts
  • Invalid symbols/exchanges
  • Rate limit compliance
  • Exchange-specific errors
  • Graceful degradation

🎨 Use Cases

  1. Trading Bots: Real-time price monitoring and decision making
  2. Portfolio Trackers: Multi-exchange portfolio valuation
  3. Market Analysis: Historical data analysis and backtesting
  4. Price Alerts: Monitoring price thresholds
  5. Research: Cryptocurrency market research and analysis
  6. Dashboards: Real-time market visualization

📊 Performance

Benchmarks

  • Cached Response: <1ms
  • Uncached Response: 100-500ms (depends on exchange)
  • Batch Fetching: 10 symbols in ~200ms (with cache)
  • Concurrent Requests: Supports 100+ concurrent operations

Optimization Tips

  1. Use batch operations (get_multiple_tickers) for efficiency
  2. Adjust cache TTL based on your use case
  3. Implement connection pooling for high-frequency requests
  4. Use timeframe aggregation for historical data

🔒 Security Considerations

  • No API keys stored (uses public endpoints only)
  • Rate limiting built-in to prevent account bans
  • Input validation on all user-provided data
  • No sensitive data in logs

🚧 Limitations

  1. Public Data Only: No access to private account data
  2. Rate Limits: Subject to exchange rate limits
  3. Market Hours: Some exchanges have maintenance windows
  4. Data Delays: Minor delays possible during high volatility

🛠️ Development

Code Style

# Format code
black mcp_crypto_server.py test_mcp_server.py

# Sort imports
isort mcp_crypto_server.py test_mcp_server.py

# Lint
flake8 mcp_crypto_server.py test_mcp_server.py

# Type check
mypy mcp_crypto_server.py

Contributing

  1. Fork the repository
  2. Create a feature branch
  3. Write tests for new functionality
  4. Ensure all tests pass
  5. Submit a pull request

📝 Assumptions

  1. Internet Connectivity: Stable internet connection required
  2. Exchange Availability: Exchanges are operational and accessible
  3. Rate Limits: Reasonable request frequency (handled by CCXT)
  4. Data Accuracy: Exchange-provided data is accurate
  5. Python Environment: Python 3.8+ with standard libraries
  6. Public Endpoints: Using public endpoints that don't require authentication

🐛 Troubleshooting

Common Issues

"Exchange not supported" error:

  • Check if exchange is in SUPPORTED_EXCHANGES list
  • Verify exchange initialization succeeded

"Symbol not found" error:

  • Ensure correct symbol format (e.g., 'BTC/USDT' not 'BTCUSDT')
  • Check if symbol exists on the exchange

Rate limit errors:

  • Reduce request frequency
  • Increase cache TTL
  • Use batch operations

Connection timeouts:

  • Check internet connectivity
  • Verify exchange is not under maintenance
  • Increase timeout settings

📈 Future Enhancements

  • WebSocket support for real-time streaming
  • More exchanges (Bitfinex, Huobi, etc.)
  • Advanced indicators (RSI, MACD, etc.)
  • Database persistence
  • REST API wrapper
  • Authentication for private endpoints
  • Multi-timeframe analysis
  • Alert system

📄 License

MIT License - feel free to use in your projects

👤 Author

Created as an internship assignment submission for cryptocurrency market data analysis

🙏 Acknowledgments

  • CCXT: Comprehensive cryptocurrency exchange library
  • Python Community: Amazing async/await ecosystem
  • Exchange APIs: Reliable data providers

📞 Support

For issues or questions:

  1. Check the troubleshooting section
  2. Review test cases for usage examples
  3. Consult CCXT documentation for exchange-specific details

Last Updated: November 2025