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