Cryptocurrency Market Data MCP Server

Instructions

Implementation Reference

• src/server.py:265-286 (handler)

  Handler implementation for the 'get-top-volumes' tool. Fetches all tickers from the specified exchange using CCXT, sorts them by 24h base volume (descending), selects the top N (default 5), formats each using format_ticker helper, and returns a formatted text response listing the top pairs.

  elif name == "get-top-volumes":
      limit = int(arguments.get("limit", 5))
      tickers = await exchange.fetch_tickers()
  
      # Sort by volume and get top N
      sorted_tickers = sorted(
          tickers.values(),
          key=lambda x: float(x.get('baseVolume', 0) or 0),
          reverse=True
      )[:limit]
  
      formatted_results = []
      for ticker in sorted_tickers:
          formatted_data = await format_ticker(ticker, exchange_id)
          formatted_results.append(formatted_data)
  
      return [
          types.TextContent(
              type="text",
              text=f"Top {limit} pairs by volume on {exchange_id.upper()}:\n\n" + "\n".join(formatted_results)
          )
      ]

• src/server.py:135-148 (registration)

  Tool registration in the @server.list_tools() handler, defining the tool name, description, and input schema (optional 'limit' number and 'exchange' string from supported exchanges).

  types.Tool(
      name="get-top-volumes",
      description="Get top cryptocurrencies by trading volume from a specific exchange",
      inputSchema={
          "type": "object",
          "properties": {
              "limit": {
                  "type": "number",
                  "description": "Number of pairs to return (default: 5)",
              },
              "exchange": get_exchange_schema()
          }
      },
  ),

• src/server.py:59-66 (schema)

  Shared schema function providing JSON schema for the 'exchange' input parameter used across tools, including get-top-volumes (defines type string, enum of supported exchanges, default 'binance').

  def get_exchange_schema() -> Dict[str, Any]:
      """Get the JSON schema for exchange selection."""
      return {
          "type": "string",
          "description": f"Exchange to use (supported: {', '.join(SUPPORTED_EXCHANGES.keys())})",
          "enum": list(SUPPORTED_EXCHANGES.keys()),
          "default": "binance"
      }

• src/server.py:44-56 (helper)

  Helper function to format individual ticker data (symbol, prices, volume, bid/ask) into a multi-line string, used by the get-top-volumes handler to display each top volume pair.

  async def format_ticker(ticker: Dict[str, Any], exchange_id: str) -> str:
      """Format ticker data into a readable string."""
      return (
          f"Exchange: {exchange_id.upper()}\n"
          f"Symbol: {ticker.get('symbol')}\n"
          f"Last Price: {ticker.get('last', 'N/A')}\n"
          f"24h High: {ticker.get('high', 'N/A')}\n"
          f"24h Low: {ticker.get('low', 'N/A')}\n"
          f"24h Volume: {ticker.get('baseVolume', 'N/A')}\n"
          f"Bid: {ticker.get('bid', 'N/A')}\n"
          f"Ask: {ticker.get('ask', 'N/A')}\n"
          "---"
      )

• src/server.py:31-42 (helper)

  Helper to lazily initialize and cache CCXT exchange instances for supported exchanges, used indirectly by get-top-volumes via the common flow.

  async def get_exchange(exchange_id: str) -> ccxt.Exchange:
      """Get or create an exchange instance."""
      exchange_id = exchange_id.lower()
      if exchange_id not in SUPPORTED_EXCHANGES:
          raise ValueError(f"Unsupported exchange: {exchange_id}")
  
      if exchange_id not in exchange_instances:
          exchange_class = SUPPORTED_EXCHANGES[exchange_id]
          exchange_instances[exchange_id] = exchange_class()
  
      return exchange_instances[exchange_id]