🚀 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