CockroachDB MCP Server

Instructions

Args:
    node_id: Specific node ID (optional, returns all if not specified).

Returns:
    Node status information.

Implementation Reference

• src/cockroachdb_mcp/tools/cluster.py:104-163 (handler)

  Core handler function that executes the node_status tool logic by querying CockroachDB's crdb_internal.gossip_liveness table for detailed node information.

  async def node_status(node_id: int | None = None) -> dict[str, Any]:
      """Get detailed status for a node or all nodes.
  
      Args:
          node_id: Specific node ID (optional).
  
      Returns:
          Node status information.
      """
      conn = await connection_manager.ensure_connected()
  
      try:
          query = """
              SELECT
                  node_id,
                  address,
                  build_tag,
                  started_at,
                  updated_at,
                  locality,
                  is_live,
                  ranges,
                  leases
              FROM crdb_internal.gossip_liveness
          """
  
          if node_id is not None:
              query += f" WHERE node_id = {node_id}"
  
          query += " ORDER BY node_id"
  
          async with conn.cursor() as cur:
              await cur.execute(query)
              rows = await cur.fetchall()
  
          if node_id is not None and not rows:
              return {"status": "error", "error": f"Node {node_id} not found"}
  
          nodes = []
          for row in rows:
              nodes.append(
                  {
                      "node_id": row.get("node_id"),
                      "address": row.get("address"),
                      "build_tag": row.get("build_tag"),
                      "started_at": str(row.get("started_at")) if row.get("started_at") else None,
                      "updated_at": str(row.get("updated_at")) if row.get("updated_at") else None,
                      "locality": row.get("locality"),
                      "is_live": row.get("is_live"),
                      "ranges": row.get("ranges"),
                      "leases": row.get("leases"),
                  }
              )
  
          if node_id is not None:
              return {"node": nodes[0] if nodes else None}
  
          return {"nodes": nodes, "count": len(nodes)}
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/server.py:435-449 (registration)

  MCP tool registration with @mcp.tool() decorator. This wrapper function registers the tool and delegates to the core implementation in cluster.py.

  @mcp.tool()
  async def node_status(node_id: int | None = None) -> dict[str, Any]:
      """Get detailed status for a node.
  
      Args:
          node_id: Specific node ID (optional, returns all if not specified).
  
      Returns:
          Node status information.
      """
      try:
          return await cluster.node_status(node_id)
      except Exception as e:
          return {"status": "error", "error": str(e)}