node_status
Retrieve node status information from a CockroachDB cluster to monitor health and performance. Specify a node ID for targeted details or get all nodes' status.
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 |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- 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)}