代币价格mcp
消息控制协议(MCP)服务器,用于从多个区块链来源检索加密货币代币价格。使用fastmcp构建,支持来自CoinGecko、链上DEX聚合器和区块链价格提要的实时价格数据。
特性
- MCP服务器,用于存储具有多个数据源的代币价格数据
- 支持多种传输协议(stdio、SSE、WebSocket)
- CoinGecko API实时价格信息
- 来自DEX聚合器(Uniswap、PancakeSwap等)的链上价格数据
- Chainlink价格馈送集成
- 易于使用的客户端,用于与MCP服务器交互
- 运行服务器的命令行界面
- 与Claude Desktop集成
- 带TTL的全面缓存
- 速率限制和断路器模式
- 多链支持(以太坊、BSC、Polygon、Arbitrum等)
快速开始
# Install the package
pip install token-prices-mcp
# Run the server (stdio transport for Claude Desktop)
token-prices-mcp --transport stdio
# Or run with SSE transport
token-prices-mcp --transport sse --host 127.0.0.1 --port 8000基本用法
具有网络传输的客户端
import asyncio
import os
from token_prices_mcp.client import TokenPricesMCPClient
async def main():
# Set the server URL (or use environment variable)
server_url = "http://localhost:8000/sse"
os.environ["TOKEN_PRICES_MCP_URL"] = server_url
async with TokenPricesMCPClient() as client:
# Get token price from CoinGecko
btc_price = await client.get_token_price("bitcoin")
print("BTC Price:", btc_price)
# Get multiple token prices
prices = await client.get_multiple_prices(["ethereum", "bitcoin", "cardano"])
print("Multiple prices:", prices)
# Get token price with historical data
eth_data = await client.get_token_data("ethereum", include_24h_change=True)
print("ETH Data:", eth_data)
# Get price from specific DEX
uniswap_price = await client.get_dex_price("0xA0b86a33E6D0bBa464", "uniswap_v3")
print("Uniswap price:", uniswap_price)
if __name__ == "__main__":
asyncio.run(main())服务器配置
from token_prices_mcp.server import create_server
import asyncio
# Create the server with custom configuration
server = create_server(
name="Token Prices API",
cache_ttl=300, # 5 minutes cache
rate_limit=100 # 100 requests per minute
)
# Run the server with desired transport
if __name__ == "__main__":
asyncio.run(
server.run(
transport="stdio", # or "sse"
host="127.0.0.1", # for network transports
port=8000 # for network transports
)
)可用工具
价格数据
get_token_price(token_id: str, vs_currency: str = "usd"):获取当前代币价格get_multiple_prices(token_ids: list, vs_currency: str = "usd"):获取多种代币价格get_token_data(token_id: str, include_24h_change: bool = False):获取详细的令牌数据get_price_history(token_id: str, days: int = 7):获取历史价格数据search_tokens(query: str):按名称或符号搜索令牌
DEX集成
get_dex_price(token_address: str, dex: str, chain: str = "ethereum"):从特定DEX获取价格get_liquidity_pools(token_address: str, chain: str = "ethereum"):获取流动性池信息get_dex_volume(token_address: str, timeframe: str = "24h"):从DEX获取交易量
链上数据
get_chainlink_price(pair: str):从Chainlink价格源获取价格get_token_supply(token_address: str, chain: str = "ethereum"):获取令牌供应信息get_token_holders(token_address: str, chain: str = "ethereum"):获取顶级代币持有者get_market_cap(token_id: str):获取实时市值
市场分析
get_trending_tokens(limit: int = 10):获取热门代币get_top_gainers(limit: int = 10, timeframe: str = "24h"):获得最高收益者get_top_losers(limit: int = 10, timeframe: str = "24h"):获得最大的失败者get_market_overview():获取整体市场统计数据
资源
prices://tokens:获取所有支持的令牌prices://tokens/{token_id}:获取特定令牌信息prices://markets:获取所有市场信息prices://chains:获得支持的区块链网络prices://dexes:获取支持的DEX平台
环境变量
# CoinGecko API (optional, for higher rate limits)
COINGECKO_API_KEY=your_api_key_here
# Ethereum RPC endpoint (for on-chain data)
ETHEREUM_RPC_URL=https://mainnet.infura.io/v3/your_project_id
# BSC RPC endpoint
BSC_RPC_URL=https://bsc-dataseed.binance.org/
# Polygon RPC endpoint
POLYGON_RPC_URL=https://polygon-rpc.com/
# Arbitrum RPC endpoint
ARBITRUM_RPC_URL=https://arb1.arbitrum.io/rpc
# Cache settings
CACHE_TTL=300
RATE_LIMIT_PER_MINUTE=100
# Server settings
TOKEN_PRICES_MCP_URL=http://localhost:8000/sse命令行界面
# Show help
token-prices-mcp --help
# Run with stdio transport (for Claude Desktop)
token-prices-mcp --transport stdio
# Run with SSE transport
token-prices-mcp --transport sse --host localhost --port 8000
# Run with custom cache TTL
token-prices-mcp --transport stdio --cache-ttl 600
# Run with debug logging
token-prices-mcp --transport stdio --debug支撑链
- 以太坊
- 币安智能链(BSC)
- 多边形
- 仲裁
- 乐观主义
- 雪崩
- 幻影
- 克洛诺斯
- 极光
支持的DEX
- Uniswap V2/V3
- PancakeSwap V2/V3
- SushiSwap 的
- QuickSwap
- 商人乔
- SpookySwap
- 曲线金融
- 平衡器
- 1英寸
数据源
- CoinGecko API:代币价格和市场数据的主要来源
- 链上DEX数据:来自去中心化交易所的实时价格
- Chainlink价格反馈:去中心化的oracle价格数据
- 区块链RPC:直接链上代币和合约数据
缓存策略
- 默认情况下,价格数据缓存5分钟
- 市场数据缓存10分钟
- 令牌元数据缓存1小时
- 历史数据缓存24小时
- 每种数据类型可配置TTL
速率限制
- CoinGecko API:100次请求/分钟(免费层)
- 链上调用:每RPC 50个请求/分钟
- 指数回退的内部速率限制
- 失败请求的断路器模式
发展
# Clone the repository
git clone https://github.com/tokenprices/token-prices-mcp.git
cd token-prices-mcp
# Install development dependencies
pip install -e ".[dev]"
# Install pre-commit hooks
pre-commit install
# Run tests
pytest
# Run type checking
pyright src/token_prices_mcp
# Run linting
ruff check .例子
检查 examples/ 更多使用示例的目录:
claude_desktop_server.py:使用克劳德桌面的stdio传输运行服务器claude_desktop_client.py:用于连接到stdio服务器的客户端sse_server.py:使用SSE传输运行服务器sse_client.py:用于连接到SSE服务器的客户端price_monitoring.py:实时价格监控示例dex_arbitrage.py:DEX价格比较示例
API费率限制
CoinGecko(免费版)
- 100个请求/分钟
- 1000个请求/小时
CoinGecko(专业版)
- 500个请求/分钟
- 10000次请求/小时
链上RPC
- 因供应商而异
- 内置速率限制和重试逻辑
错误处理
服务器实现了全面的错误处理:
- 指数回退自动重试
- 发生故障的API端点的断路器
- API不可用时回退到缓存数据
- 服务质量严重下降
- 详细的错误记录和监控
许可证
Apache许可证2.0
贡献
- 分叉存储库
- 创建要素分支
- 进行更改
- 添加新功能的测试
- 确保所有测试通过
- 提交拉取请求
支持
- 文档:
- 问题:
- 社区: 讨论
