CockroachDB MCP Server

Instructions

Args:
    table: Optional table to filter ranges.
    limit: Maximum ranges to return.

Returns:
    Range distribution information.

Implementation Reference

• src/cockroachdb_mcp/tools/cluster.py:412-480 (handler)

  Core handler function that executes the show_ranges tool logic by querying CockroachDB's crdb_internal.ranges_no_leases table for range distribution information.

  async def show_ranges(table: str | None = None, limit: int = 50) -> dict[str, Any]:
      """Show range distribution in the cluster.
  
      Args:
          table: Optional table to filter ranges.
          limit: Maximum ranges to return.
  
      Returns:
          Range distribution information.
      """
      conn = await connection_manager.ensure_connected()
  
      try:
          if table:
              # Parse table name (schema ignored - CockroachDB ranges use just table_name)
              if "." in table:
                  _schema, table_name = table.rsplit(".", 1)
              else:
                  table_name = table
  
              query = f"""
                  SELECT
                      range_id,
                      start_pretty,
                      end_pretty,
                      lease_holder,
                      replicas,
                      range_size_mb
                  FROM crdb_internal.ranges_no_leases
                  WHERE table_name = '{table_name}'
                  LIMIT {limit}
              """
          else:
              query = f"""
                  SELECT
                      range_id,
                      database_name,
                      table_name,
                      start_pretty,
                      end_pretty,
                      lease_holder,
                      replicas,
                      range_size_mb
                  FROM crdb_internal.ranges_no_leases
                  LIMIT {limit}
              """
  
          async with conn.cursor() as cur:
              await cur.execute(query)
              rows = await cur.fetchall()
  
          ranges = []
          for row in rows:
              range_info: dict[str, Any] = {
                  "range_id": row.get("range_id"),
                  "start": row.get("start_pretty"),
                  "end": row.get("end_pretty"),
                  "lease_holder": row.get("lease_holder"),
                  "replicas": row.get("replicas"),
                  "size_mb": row.get("range_size_mb"),
              }
              if not table:
                  range_info["database"] = row.get("database_name")
                  range_info["table"] = row.get("table_name")
              ranges.append(range_info)
  
          return {"ranges": ranges, "count": len(ranges), "table_filter": table}
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/server.py:534-549 (registration)

  Registration of the show_ranges tool via @mcp.tool() decorator. This thin wrapper delegates execution to the core handler in cluster.py.

  @mcp.tool()
  async def show_ranges(table: str | None = None, limit: int = 50) -> dict[str, Any]:
      """Show range distribution in the cluster.
  
      Args:
          table: Optional table to filter ranges.
          limit: Maximum ranges to return.
  
      Returns:
          Range distribution information.
      """
      try:
          return await cluster.show_ranges(table, limit)
      except Exception as e:
          return {"status": "error", "error": str(e)}