# π 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