aiowhitebit-mcp

aiowhitebit-mcp

用于 WhiteBit 加密货币交易所 API 的消息控制协议（MCP）服务和客户端实现。基于 aiowhitebit 库和 fastmcp 构建。

功能

• 面向 WhiteBit API 公共端点的 MCP 服务
• 支持多种传输协议（stdio、SSE、WebSocket）
• 易于使用的客户端，可与 MCP 服务交互
• 用于运行服务的命令行界面
• 与 Claude Desktop 集成
• 通过 WebSocket 获取实时市场数据
• 全面的测试覆盖和类型检查
• 现代化开发工具（ruff、pyright、pre-commit）
• 支持磁盘持久化的缓存
• 限流和熔断器模式

aiowhitebit 0.3.0 集成新增功能

• BookTicker WebSocket 流：实时最佳买入价／卖出价更新
• 期货资金费率历史记录：访问期货市场的历史资金费率
• 增强的 WebSocket 事件：所有 WebSocket 事件现在都包含可选元数据（event_time、update_id）

快速开始

# Install the package
pip install aiowhitebit-mcp

# Run the server (stdio transport for Claude Desktop)
aiowhitebit-mcp --transport stdio

# Or run with SSE transport
aiowhitebit-mcp --transport sse --host 127.0.0.1 --port 8000

基本用法

使用网络传输的客户端

import asyncio
import os
from aiowhitebit_mcp.client import WhiteBitMCPClient

async def main():
    # Set the server URL (or use environment variable)
    server_url = "http://localhost:8000/sse"
    os.environ["WHITEBIT_MCP_URL"] = server_url

    async with WhiteBitMCPClient() as client:
        # Get market info
        btc_usdt = await client.get_market_resource("BTC_USDT")
        print("BTC/USDT Market Info:", btc_usdt)

        # Get real-time price via WebSocket
        price = await client.get_last_price("BTC_USDT")
        print("Current BTC/USDT price:", price)

        # Get order book
        orderbook = await client.get_orderbook("BTC_USDT")
        print("Order book:", orderbook)

        # Get funding history for futures market
        funding_history = await client.get_funding_history("BTC_USDT")
        print("Funding history:", funding_history)

        # Subscribe to BookTicker for real-time best bid/ask
        subscription = await client.bookticker_subscribe("BTC_USDT")
        print("BookTicker subscription:", subscription)

if __name__ == "__main__":
    asyncio.run(main())

服务配置

from aiowhitebit_mcp.server import create_server
import asyncio

# Create the server with custom configuration
server = create_server(
    name="WhiteBit API"
)

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

可用工具

公共 API

• get_server_time()：获取当前服务时间
• get_market_info()：获取所有市场信息
• get_orderbook(market: str)：获取订单簿
• get_recent_trades(market: str, limit: int = 100)：获取近期交易记录
• get_ticker(market: str)：获取行情信息
• get_fee(market: str)：获取交易手续费
• get_server_status()：获取服务状态
• get_asset_status_list()：获取所有资产的状态
• get_funding_history(market: str)：获取期货市场的资金费率历史记录

WebSocket API

• get_last_price(market: str)：获取实时价格
• get_market_depth(market: str)：获取实时订单簿
• bookticker_subscribe(market: str)：订阅 BookTicker 流，获取最佳买入价／卖出价
• bookticker_unsubscribe(market: str)：取消订阅 BookTicker 流
• connect_websocket()：连接到 WebSocket API
• disconnect_websocket()：断开与 WebSocket API 的连接

资源

• whitebit://markets：获取所有市场信息
• whitebit://markets/{market}：获取指定市场信息
• whitebit://assets：获取所有资产信息
• whitebit://assets/{asset}：获取指定资产信息

命令行界面

# Show help
aiowhitebit-mcp --help

# Run with stdio transport (for Claude Desktop)
aiowhitebit-mcp --transport stdio

# Run with SSE transport
aiowhitebit-mcp --transport sse --host localhost --port 8000

开发

# Clone the repository
git clone https://github.com/yourusername/aiowhitebit-mcp.git
cd aiowhitebit-mcp

# Install development dependencies
pip install -e ".[dev]"

# Install pre-commit hooks
pre-commit install

# Run tests
pytest

# Run type checking
pyright src/aiowhitebit_mcp

# Run linting
ruff check .

示例

如需查看更多用法示例，请查看 examples/ 目录：

• claude_desktop_server.py：使用 stdio 传输为 Claude Desktop 运行服务
• claude_desktop_client.py：连接到 stdio 服务的客户端
• sse_server.py：使用 SSE 传输运行服务
• sse_client.py：连接到 SSE 服务的客户端

许可证

Apache License 2.0