node_status
Retrieve status information for CockroachDB cluster nodes to monitor health and performance. Specify a node ID for targeted details or get all nodes' status at once.
Instructions
Get detailed status for a node.
Args:
node_id: Specific node ID (optional, returns all if not specified).
Returns:
Node status information.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| node_id | No |
Implementation Reference
- src/cockroachdb_mcp/server.py:435-449 (handler)MCP tool handler and registration for 'node_status'. Decorated with @mcp.tool(), handles input/output and delegates to cluster.node_status helper.@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)}
- Helper function implementing the core logic of node_status by executing SQL query on crdb_internal.gossip_liveness to retrieve node details.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)}