🚀 Crypto MCP Server
📌 Overview
Crypto MCP Server is a Model Context Protocol (MCP) compatible server that provides real-time and historical cryptocurrency market data using ccxt. It exposes three fully tested tools:
get_ticker → Live price & market summary
get_ohclv → Historical candlestick data
stream_ticker → Real-time price streaming (async generator)
The project includes:
Custom error handling
In-memory caching
A lightweight MCP server architecture
Complete test suite (pytest)
Local client for manual testing
✨ Features
🔹 get_ticker
Fetches:
last price
high & low
base volume
price_change_percent
🔹 get_ohclv
Fetches OHLCV candles with:
timestamp
open
high
low
close
volume
🔹 stream_ticker
Real-time streaming ticker with async generator that yields updates every N seconds.
🧩 Project Structure
crypto-mcp-server/ │ ├── server/ │ ├── main.py │ ├── mcp_server.py │ ├── cache.py │ ├── errors.py │ ├── exchanges.py │ └── tools/ │ ├── get_ticker.py │ ├── get_ohclv.py │ └── stream_ticker.py │ ├── tests/ │ ├── test_get_ticker.py │ ├── test_ohclv.py │ └── test_stream_ticker.py │ ├── client_test.py ├── README.md ├── requirements.txt └── .gitignore
🛠️ Tech Stack
Python 3.10+
ccxt for exchange APIs
pytest for testing
asyncio for streaming
MCP server protocol style
📦 Installation
git clone https://github.com/yourusername/crypto-mcp-server cd crypto-mcp-server pip install -r requirements.txt
▶️ Running the MCP Server
From the server/ directory:
python -m server.main
You should see:
Crypto MCP Server running… Registered tools: get_ticker, get_ohclv, stream_ticker
🧪 Running Tests
All tests are under tests/ and cover:
Valid/invalid symbols
Unsupported exchanges
API errors
Streaming behavior
Run:
pytest -vv
📁 Tools Implemented
🔹 get_ticker
Handles:
Invalid symbols
Unsupported exchanges
ccxt API exceptions
Caching responses for 20 seconds
🔹 get_ohclv
Returns OHCLV historical candles
Validates timeframe & limit
Raises custom errors
🔹 stream_ticker
Async generator
Streams live ticker updates
Internal delay using asyncio.sleep()
Full validation & error handling
Caching Layer
cache.py implements simple in-memory TTL cache:
save(key, value, ttl)
get(key)
Prevents extra API calls
🧪 Testing Strategy
Unit Tests (pytest)
Each tool has:
success test
invalid symbol test
invalid exchange test
API failure test
Streaming Tests
Uses asyncio.mark
Simulates multiple ticker updates
Checks generator behavior
🧪 Local Testing Without MCP CLI
Run:
python client_test.py
It prints:
get_ticker result
get_ohclv candles
3 streamed ticker updates
🔍 Example Output
--- Testing get_ticker --- {...}
--- Testing get_ohclv --- {...}
--- Testing stream_ticker --- {...}
📌 Key Learning & Highlights
Designed full MCP-style server
Implemented async streaming tool
Wrote complete test suite
Built caching + error handling abstraction
Validated exchange + symbol inputs safely