使用说明

MCP加密服务器

基于Python的MCP服务器提供来自主要交易所的实时加密货币市场数据。

📋 项目概述

该项目实现了一个生产就绪的模型上下文协议（MCP）服务器，用于检索实时和历史加密货币市场数据。它展示了Python web开发的最佳实践，包括RESTful API设计、错误处理、缓存和全面测试。

任务合规性： 通过实时更新、历史查询和广泛的测试覆盖，满足对强大的加密货币数据服务器的所有要求。

🎯 方法和架构

设计理念

该项目遵循 层次结构 关注点明确分离的模式：

API层 (api/app.py)-处理HTTP请求/响应的FastAPI REST端点
业务逻辑层 (crypto_client.py)-核心数据检索和处理逻辑
适配器层 (adapters/ccxt_adapter.py)-Exchange集成抽象
数据层 (models.py)-类型安全数据结构的Pydantic模型

关键设计决策

1. Flask上的FastAPI

为什么？ 现代异步支持，自动API文档（Swagger/OpenAPI），内置验证
好处： 生产准备就绪，样板最少，性能卓越

2. CCXT库

为什么？ 用于100多个加密货币交易所的统一API
好处： 币安、Coinbase、Kraken等单一接口，降低集成复杂度

3. Exchange集成的适配器模式

为什么？ 从业务逻辑中抽象出CCXT实现细节
好处： 易于交换交换库或添加自定义实现

4. 线程安全TTL缓存

为什么？ 减少对交易所的API调用，遵守费率限制
实施： 带有线程锁和自动清理的自定义缓存
好处： 重复查询的性能提高了约50倍

5. 集中错误处理

为什么？ 所有端点的一致错误响应
实施： 映射到HTTP状态代码的自定义异常层次结构
好处： 为API消费者提供清晰、可操作的错误消息

技术栈原理

技术	目的	为什么选择
快速API	Web框架	异步支持、自动文档、验证
仓储管理系统	Exchange API	用于100多个交易所的统一接口
派丹蒂克	数据验证	类型安全，自动验证
Uvicorn	ASGI服务器	高性能，生产就绪
Pytest	测试	行业标准，强大的插件生态系统

🚀 安装说明

先决条件

Python 3.10或更高版本 （3.12+推荐）
点包管理器
Internet连接 （用于交换API访问）
Git （用于克隆存储库）

步骤1：克隆存储库

git clone https://github.com/manishraje28/mcp-crypto-server.git
cd mcp-crypto-server

第二步：创建虚拟环境

Windows PowerShell：

python -m venv .venv
.\.venv\Scripts\Activate.ps1

Linux/macOS：

python3 -m venv .venv
source .venv/bin/activate

步骤3：安装依赖项

pip install -r requirements.txt

预期包裹：

fastapi==0.115.5
ccxt==4.5.19
pydantic==2.11.7
uvicorn==0.35.0
pytest==8.3.4

步骤4：验证安装

# Run tests to verify everything works
pytest -v

# Expected output: All tests passing ✅

步骤5：启动服务器

# Navigate to src directory
cd src

# Start server
python -m uvicorn mcp_crypto_server.api.app:app --host 127.0.0.1 --port 8000

服务器将在以下位置可用：

API基础： http://127.0.0.1:8000
交互式文档： http://127.0.0.1:8000/docs
备选文件： http://127.0.0.1:8000/redoc

步骤6：测试API

选项1：浏览器

http://127.0.0.1:8000/price?symbol=BTC/USDT&exchange=binance

选项2：卷曲

curl "http://127.0.0.1:8000/price?symbol=BTC/USDT&exchange=binance"

选项3:Python

import requests
response = requests.get("http://127.0.0.1:8000/price?symbol=BTC/USDT&exchange=binance")
print(response.json())

🔧 配置

环境变量

创建一个 .env 项目根目录中的文件（可选）：

# Default exchange to use
DEFAULT_EXCHANGE=binance

# Cache TTL in seconds
CACHE_TTL_SECONDS=10

# Server settings
HOST=127.0.0.1
PORT=8000

配置选项

服务器可以通过以下方式配置 src/mcp_crypto_server/config.py:

DEFAULT_EXCHANGE = "binance"  # Default exchange for queries
CACHE_TTL = 10.0              # Cache time-to-live in seconds
SUPPORTED_EXCHANGES = [       # List of supported exchanges
    "binance", "coinbase", "kraken", "bitfinex", "kucoin"
]

📚 假设和约束

假设

互联网连接
- 服务器需要活动的互联网连接才能访问exchange API
- 未实现脱机模式或数据持久性
Exchange可用性
- 假设主要交易所（币安、Coinbase、Kraken）可用
- 如果特定交换减少，则性能会下降
速率限制
- 假设合理的查询率（缓存有帮助）
- 公共终结点不需要身份验证
- CCXT库处理的速率限制
数据新鲜度
- 假设10秒缓存的实时数据是可接受的
- 加载交易所市场时，首次请求可能很慢（2-5秒）
符号格式
- 交易对使用CCXT统一格式： BASE/QUOTE （例如。， BTC/USDT)
- 特定于交易所的符号由CCXT规范化
Python版本
- 为Python 3.10开发的项目+
- 通篇使用类型提示和现代语法

约束条件

仅限公共数据
- 仅实施了公共市场数据端点
- 无交易或账户管理功能
- 不需要API密钥身份验证
同步操作
- 为了简单起见，当前的实现是同步的
- 可以通过async/await增强并发性
内存缓存
- 服务器重启时缓存数据丢失
- 未实现持久存储（Redis、数据库）
- 适合开发和低流量生产
错误处理
- 网络错误、交换机停机时间处理得当
- 无效的符号/交换返回适当的HTTP错误
- 超时处理委托给CCXT库
测试范围
- 核心功能的单元和集成测试
- 不包括负载测试或性能基准测试
- 测试中使用模拟数据以避免交换API依赖关系

特性

REST API终结点：

/price -当前交易价格
/ticker -买入价/卖出价/最新价格
/ohlcv -历史烛台数据
/orderbook -市场深度（买入/卖出）
/top_markets -顶级交易对

技术栈：

REST端点的FastAPI
用于交换集成的CCXT

📊 特性

REST API终结点：

/price -任何交易品种的当前交易价格
/ticker -完整的报价单，包括买入价/卖出价/最新价格
/ohlcv -历史烛台数据（OHLCV）
/orderbook -买卖市场深度
/top_markets -按交易所划分的热门交易对

关键能力：

✅ 实时加密货币价格
✅ 具有多个时间范围（1m至1w）的历史数据
✅ 通过CCXT支持100多个交易所
✅ 线程安全缓存以提高性能
✅ 全面的错误处理
✅ 使用Pydantic进行类型安全数据验证
✅ 自动生成的API文档（Swagger/OpenAPI）

🏗️ 项目结构

mcp-crypto-server/
├── src/
│   └── mcp_crypto_server/
│       ├── api/
│       │   └── app.py              # FastAPI REST endpoints (5 routes)
│       ├── adapters/
│       │   └── ccxt_adapter.py     # CCXT exchange integration
│       ├── cache.py                # Thread-safe TTL cache implementation
│       ├── config.py               # Configuration settings
│       ├── crypto_client.py        # Core business logic layer
│       ├── errors.py               # Custom exception hierarchy
│       ├── models.py               # Pydantic data models (8 models)
│       ├── mcp_tools.py            # MCP protocol tools
│       └── server.py               # MCP server entry point
├── tests/
│   ├── conftest.py                 # Pytest configuration & fixtures
│   ├── test_cache.py               # Cache unit tests
│   ├── test_crypto_client.py       # Client integration tests
│   └── test_mcp_tools.py           # MCP tools tests
├── .env                            # Environment configuration (optional)
├── .gitignore                      # Git ignore rules
├── README.md                       # This file
├── requirements.txt                # Python dependencies
├── pyproject.toml                  # Project metadata
└── run_server.py                   # Convenience server launcher

层职责：

API层： HTTP请求处理、路由、响应格式化
业务逻辑： 数据检索、缓存、验证
适配器层： 抽象了特定于Exchange的实现
数据层： 具有自动验证功能的类型安全模型

🌐 API使用示例

获取比特币价格

GET http://127.0.0.1:8000/price?symbol=BTC/USDT&exchange=binance

答复：

{
  "exchange": "binance",
  "symbol": "BTC/USDT",
  "price": 96254.61,
  "timestamp_ms": 1763230278013
}

获取以太坊股票代码

GET http://127.0.0.1:8000/ticker?symbol=ETH/USDT&exchange=binance

答复：

{
  "exchange": "binance",
  "symbol": "ETH/USDT",
  "bid": 3654.2,
  "ask": 3654.25,
  "last": 3654.22,
  "timestamp_ms": 1763225099558,
  "info_source": "ccxt"
}

获取历史烛台数据

GET http://127.0.0.1:8000/ohlcv?symbol=BTC/USDT&timeframe=1h&limit=10&exchange=binance

答复：

{
  "exchange": "binance",
  "symbol": "BTC/USDT",
  "timeframe": "1h",
  "points": [
    {
      "timestamp_ms": 1731686400000,
      "open": 96000.0,
      "high": 96500.0,
      "low": 95800.0,
      "close": 96200.0,
      "volume": 1234.56
    }
  ]
}

获取订单深度

GET http://127.0.0.1:8000/orderbook?symbol=BTC/USDT&depth=5&exchange=binance

答复：

{
  "exchange": "binance",
  "symbol": "BTC/USDT",
  "bids": [
    {"price": 96750.0, "amount": 0.5},
    {"price": 96749.5, "amount": 1.2}
  ],
  "asks": [
    {"price": 96751.0, "amount": 0.6},
    {"price": 96751.5, "amount": 1.1}
  ]
}

获取顶级市场

GET http://127.0.0.1:8000/top_markets?exchange=binance&quote_asset=USDT&limit=5

答复：

{
  "exchange": "binance",
  "markets": [
    {
      "symbol": "BTC/USDT",
      "price": 96750.5,
      "quote_asset": "USDT",
      "base_asset": "BTC"
    }
  ]
}

🧪 测试

运行所有测试

pytest -v

运行特定测试文件

pytest tests/test_cache.py -v

跑步有保障

pytest --cov=src/mcp_crypto_server --cov-report=html

测试覆盖区域

✅ 缓存操作（获取、设置、过期、清理）
✅ 当前价格检索
✅ 票务数据验证
✅ 历史OHLCV数据
✅ 订单簿查询
✅ 顶级市场列表
✅ 无效输入的错误处理
✅ 交换和符号验证

📖 交互式API文档

服务器运行后，访问自动生成的文档：

Swagger用户界面：

http://127.0.0.1:8000/docs

重新记录：

http://127.0.0.1:8000/redoc

特征：

✅ 直接在浏览器中尝试所有端点
✅ 请参阅请求/响应模式
✅ 查看参数说明
✅ 使用不同输入进行测试

最后更新时间： 2025年11月16日
版本： 1.0.0
Python版本： 3.10+

使用说明

MCP加密服务器

📋 项目概述

🎯 方法和架构

设计理念

关键设计决策

1. Flask上的FastAPI

2. CCXT库

3. Exchange集成的适配器模式

4. 线程安全TTL缓存

5. 集中错误处理

技术栈原理

🚀 安装说明

先决条件

步骤1：克隆存储库

第二步：创建虚拟环境

步骤3：安装依赖项

步骤4：验证安装

步骤5：启动服务器

步骤6：测试API

🔧 配置

环境变量

配置选项

📚 假设和约束

假设

约束条件

特性

📊 特性

🏗️ 项目结构

🌐 API使用示例

获取比特币价格

获取以太坊股票代码

获取历史烛台数据

获取订单深度

获取顶级市场

🧪 测试

运行所有测试

运行特定测试文件

跑步有保障

测试覆盖区域

📖 交互式API文档

相关 MCP

加密货币交易所集成服务

互联网流量洞察与网络分析MCP服务

实时金融新闻分析服务

金融数据服务

加密货币查询服务

股票数据服务