CockroachDB MCP Server

Instructions

Implementation Reference

• src/cockroachdb_mcp/tools/cluster.py:58-101 (handler)

  Core implementation of list_nodes tool: queries crdb_internal.gossip_nodes table, processes rows into structured node list with status and counts.
  async def list_nodes() -> dict[str, Any]: """List all nodes in the CockroachDB cluster. Returns: List of cluster nodes with their status. """ conn = await connection_manager.ensure_connected() try: async with conn.cursor() as cur: await cur.execute(""" SELECT node_id, address, locality, is_live, CASE WHEN is_live THEN 'live' ELSE 'dead' END as status FROM crdb_internal.gossip_nodes ORDER BY node_id """) rows = await cur.fetchall() nodes = [] for row in rows: nodes.append( { "node_id": row.get("node_id"), "address": row.get("address"), "locality": row.get("locality"), "is_live": row.get("is_live"), "status": row.get("status"), } ) live_count = sum(1 for n in nodes if n["is_live"]) return { "nodes": nodes, "total_count": len(nodes), "live_count": live_count, "dead_count": len(nodes) - live_count, } except Exception as e: return {"status": "error", "error": str(e)}
• src/cockroachdb_mcp/server.py:422-433 (registration)

  Registers the list_nodes tool in the FastMCP server, delegating to the implementation in tools.cluster.
  @mcp.tool() async def list_nodes() -> dict[str, Any]: """List all nodes in the CockroachDB cluster. Returns: List of cluster nodes with their status. """ try: return await cluster.list_nodes() except Exception as e: return {"status": "error", "error": str(e)}