使用说明
加密MCP服务器
基于Python的模型上下文协议(MCP)服务器,使用CCXT库提供来自主要交易所的实时和历史加密货币市场数据。
🚀 特性
- 实时价格数据:获取任何加密货币交易对的当前价格
- 历史数据:检索OHLCV(打开、高、低、关闭、音量)历史数据
- 市场统计:获取包括24小时变化在内的全面市场摘要
- 订单簿数据:访问当前的买卖订单簿
- 多交换支持:通过CCXT与100多家加密货币交易所合作
- 智能高速缓存:通过内置缓存层减少API调用
- 错误处理:强大的错误处理和验证
- 综合测试:pytest提供全面的测试覆盖
📋 需求
- Python 3.10或更高版本
- pip(Python包管理器)
- 用于访问API的Internet连接
🔧 安装
- 克隆存储库:
git clone <your-repo-url>
cd crypto-mcp-server
- 创建虚拟环境 (推荐):
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
- 安装依赖项:
pip install -r requirements.txt
🏃 用法
运行服务器
启动MCP服务器:
python -m src.crypto_mcp_server.server
可用工具
服务器公开了以下MCP工具:
-
加密价格:获取加密货币对的当前价格
- 参数:
symbol(例如“BTC/USDT”),use_cache(可选)
- 参数:
-
get_multiple_prices:获取多对鞋的价格
- 参数:
symbols(交易对列表)
- 参数:
-
获取历史数据:获取历史OHLCV数据
- 参数:
symbol,timeframe(例如“1d”),limit,use_cache
- 参数:
-
get_market_summary:获取全面的市场统计数据
- 参数:
symbol,use_cache
- 参数:
-
获取订单簿:获取当前订单簿
- 参数:
symbol,limit
- 参数:
-
搜索_符号:搜索可用的交易对
- 参数:
query(例如,“BTC”)
- 参数:
-
get_supportd_exchange:列出所有支持的交易所
- 无需参数
-
clear_cache:清除所有缓存数据
- 无需参数
-
get_ache_stats:获取缓存统计信息
- 无需参数
MCP客户端使用示例
# Example: Get Bitcoin price
{
"tool": "get_crypto_price",
"arguments": {
"symbol": "BTC/USDT"
}
}
# Example: Get historical data
{
"tool": "get_historical_data",
"arguments": {
"symbol": "ETH/USDT",
"timeframe": "1h",
"limit": 24
}
}
🧪 测试
运行所有测试:
pytest
运行覆盖率测试:
pytest --cov=src/crypto_mcp_server --cov-report=html
运行特定测试文件:
pytest tests/test_crypto_api.py -v
测试覆盖率
该项目包括以下方面的综合测试:
- ✅ API包装器功能
- ✅ 缓存层
- ✅ MCP服务器工具
- ✅ 错误处理
- ✅ 边缘案例
📁 项目结构
crypto-mcp-server/
├── src/
│ └── crypto_mcp_server/
│ ├── __init__.py # Package initialization
│ ├── server.py # Main MCP server
│ ├── crypto_api.py # CCXT API wrapper
│ └── cache.py # Caching layer
├── tests/
│ ├── test_server.py # Server tests
│ ├── test_crypto_api.py # API tests
│ └── test_cache.py # Cache tests
├── requirements.txt # Python dependencies
├── README.md # This file
└── .gitignore # Git ignore rules
🔑 关键设计决策
1.交易所选择
- 默认交换:币安
- 依据:币安是规模最大、最可靠的交易所之一,拥有卓越的API支持
- 灵活性:可以配置为使用CCXT支持的100多个交换机中的任何一个
2.缓存策略
- TTL值:
- 实时价格:30秒
- 历史数据:5分钟
- 市场总结:1分钟
- 依据:平衡数据新鲜度与API费率限制合规性
3.错误处理
- 所有API调用都包装在try-chatch块中
- 部分故障时性能良好
- 调试的详细错误消息
4.数据验证
- 符号格式验证
- 时间框架验证
- 限制参数验证
🛠️ 配置
更改默认Exchange
编辑 server.py:
server = CryptoMCPServer(
exchange_id='coinbase', # Change to any supported exchange
cache_ttl=60
)
调整缓存TTL
server = CryptoMCPServer(
exchange_id='binance',
cache_ttl=120 # 2 minutes default cache
)
🔍 假设
- 网络连接:假设稳定的互联网连接
- Exchange可用性:假设目标交换API可以运行
- 速率限制:通过CCXT的enableRateLimit内置速率限制
- 数据格式:假定标准CCXT数据格式
- 无身份验证:使用公共终结点(不需要API密钥)
📊 性能注意事项
- 缓存:将重复查询的API调用减少90%
- 速率限制:由CCXT自动管理
- 并发请求:处理多个同时进行的请求
- 内存使用:具有自动清理功能的内存缓存
🐛 已知限制
- 历史数据:受交易所特定限制的限制
- 实时更新:不是真正的WebSocket流(基于轮询)
- 认证:目前仅支持公共终结点
- 缓存持久性:缓存仅在内存中(非持久)
🔮 未来的增强功能
- []WebSocket支持真正的实时更新
- []支持经过身份验证的端点
- []持久缓存(Redis/SQLite)
- []多交易所聚合
- []CoinMarketCap整合
- []自定义警报和通知
- []投资组合跟踪
📝 许可证
麻省理工学院许可证-欢迎根据您的需求使用此项目。
🙏 致谢
📧 联系
如有疑问或问题,请在GitHub上打开问题。
备注:该项目是作为实习任务的一部分开发的。它展示了Python开发、API集成、测试和MCP协议实现的熟练程度。
