crypto-orderbook-mcp

Instructions

Implementation Reference

• main.py:78-127 (handler)

  The @mcp.tool()-decorated handler function that implements the core logic for calculating order book depth, imbalance, and mid-price using the fetched order book data.
  @mcp.tool() async def calculate_orderbook(exchange_id: str, symbol: str, depth_percentage: float = 1.0, ctx: Context = None) -> Dict: """ Calculate the order book depth and imbalance for a given trading pair on a specified exchange. Args: exchange_id: The exchange identifier (e.g., 'binance', 'kraken') symbol: The trading pair (e.g., 'BTC/USDT') depth_percentage: Percentage range from mid-price to calculate depth and imbalance (default: 1.0%) Returns: Dictionary containing bid depth, ask depth, imbalance, mid-price, and timestamp. """ exchange, order_book, mid_price, price_range, error = await fetch_order_book_data(exchange_id, symbol, depth_percentage, ctx) if error: return {"error": error} try: # Calculate depth and imbalance, handling both [price, vol] and [price, vol, 0] formats bids = order_book.get('bids', []) asks = order_book.get('asks', []) bid_volume = sum(entry[1] for entry in bids if len(entry) >= 2 and entry[0] >= mid_price - price_range) ask_volume = sum(entry[1] for entry in asks if len(entry) >= 2 and entry[0] <= mid_price + price_range) # Calculate imbalance: (bid_volume - ask_volume) / (bid_volume + ask_volume) total_volume = bid_volume + ask_volume if total_volume == 0: await ctx.error("Zero total volume in order book") return {"error": "Zero total volume in order book"} imbalance = (bid_volume - ask_volume) / total_volume return { "exchange": exchange_id, "symbol": symbol, "bid_depth": bid_volume, "ask_depth": ask_volume, "imbalance": imbalance, "mid_price": mid_price, "timestamp": order_book['timestamp'] or int(asyncio.get_event_loop().time() * 1000) } except Exception as e: await ctx.error(f"Error calculating order book metrics: {str(e)}") return {"error": f"Error calculating order book metrics: {str(e)}"} finally: if exchange: await exchange.close()
• main.py:14-77 (helper)

  Helper function that handles input validation, initializes the ccxt exchange, fetches the order book, and computes mid-price and price range used by the calculate_orderbook handler.
  async def fetch_order_book_data(exchange_id: str, symbol: str, depth_percentage: float, ctx: Context) -> Tuple[Optional[ccxt.Exchange], Optional[Dict], Optional[float], Optional[float], Optional[str]]: """ Common function to validate inputs, fetch order book, and calculate mid-price and price range. Args: exchange_id: The exchange identifier (e.g., 'binance', 'kraken') symbol: The trading pair (e.g., 'BTC/USDT') depth_percentage: Percentage range from mid-price (default: 1.0%) ctx: MCP context for error reporting Returns: Tuple containing (exchange object, order book, mid_price, price_range, error) or (None, None, None, None, error) on error. """ # Validate exchange if exchange_id.lower() not in SUPPORTED_EXCHANGES: await ctx.error(f"Unsupported exchange: {exchange_id}") return None, None, None, None, f"Unsupported exchange: {exchange_id}" # Validate depth percentage if depth_percentage <= 0 or depth_percentage > 10: await ctx.error("Depth percentage must be between 0 and 10") return None, None, None, None, "Depth percentage must be between 0 and 10" # Initialize exchange try: exchange = getattr(ccxt, exchange_id.lower())() except AttributeError: await ctx.error(f"Failed to initialize exchange: {exchange_id}") return None, None, None, None, f"Failed to initialize exchange: {exchange_id}" try: # Validate symbol try: markets = await exchange.load_markets() if symbol not in markets: await ctx.error(f"Invalid symbol {symbol} for exchange {exchange_id}") return None, None, None, None, f"Invalid symbol {symbol} for exchange {exchange_id}" except Exception as e: await ctx.error(f"Error validating symbol {symbol}: {str(e)}") return None, None, None, None, f"Error validating symbol {symbol}: {str(e)}" # Fetch order book try: order_book = await exchange.fetch_order_book(symbol, limit=100) except ccxt.BaseError as e: await ctx.error(f"Failed to fetch order book for {symbol}: {str(e)}") return None, None, None, None, f"Failed to fetch order book for {symbol}: {str(e)}" # Calculate mid-price bids = order_book.get('bids', []) asks = order_book.get('asks', []) if not bids or not asks: await ctx.error("Empty order book") return None, None, None, None, "Empty order book" mid_price = (bids[0][0] + asks[0][0]) / 2 price_range = mid_price * (depth_percentage / 100) return exchange, order_book, mid_price, price_range, None except Exception as e: await ctx.error(f"Error fetching order book data: {str(e)}") return None, None, None, None, f"Error fetching order book data: {str(e)}"
• main.py:78-78 (registration)

  The @mcp.tool() decorator registers the calculate_orderbook function as an MCP tool.
  @mcp.tool()