CockroachDB MCP Server

Instructions

Returns:
    Cluster health including node count and live nodes.

Implementation Reference

• src/cockroachdb_mcp/server.py:49-59 (handler)

  Handler and registration for the 'cluster_status' tool. Thin wrapper that delegates to cluster.cluster_health() with error handling.

  @mcp.tool()
  async def cluster_status() -> dict[str, Any]:
      """Get the health status of the CockroachDB cluster.
  
      Returns:
          Cluster health including node count and live nodes.
      """
      try:
          return await cluster.cluster_health()
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/tools/cluster.py:10-55 (helper)

  Core helper function implementing cluster health check by querying cluster organization, version, total nodes, live nodes, and determining status.

  async def cluster_health() -> dict[str, Any]:
      """Get the health status of the CockroachDB cluster.
  
      Returns:
          Cluster health information.
      """
      conn = await connection_manager.ensure_connected()
  
      try:
          result: dict[str, Any] = {"status": "healthy"}
  
          async with conn.cursor() as cur:
              # Get cluster settings
              await cur.execute("SHOW CLUSTER SETTING cluster.organization")
              org_row = await cur.fetchone()
              result["organization"] = org_row.get("cluster.organization") if org_row else None
  
              # Get version
              await cur.execute("SELECT version()")
              version_row = await cur.fetchone()
              result["version"] = version_row.get("version") if version_row else None
  
              # Get node count
              await cur.execute("SELECT count(*) as node_count FROM crdb_internal.gossip_nodes")
              node_row = await cur.fetchone()
              result["node_count"] = node_row.get("node_count") if node_row else 0
  
              # Get live node count
              await cur.execute("""
                  SELECT count(*) as live_nodes
                  FROM crdb_internal.gossip_nodes
                  WHERE is_live = true
              """)
              live_row = await cur.fetchone()
              result["live_nodes"] = live_row.get("live_nodes") if live_row else 0
  
              # Check if cluster is healthy
              if result["live_nodes"] < result["node_count"]:
                  result["status"] = "degraded"
                  result["message"] = (
                      f"{result['node_count'] - result['live_nodes']} node(s) are not live"
                  )
  
          return result
      except Exception as e:
          return {"status": "error", "error": str(e)}