CockroachDB MCP Server

Instructions

Implementation Reference

• src/cockroachdb_mcp/server.py:422-433 (handler)

  MCP tool handler for 'list_nodes' registered with @mcp.tool(). Delegates to cluster.list_nodes() with error handling.
  @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)}
• src/cockroachdb_mcp/tools/cluster.py:58-101 (helper)

  Core helper function that executes the SQL query on crdb_internal.gossip_nodes to retrieve and format the list of cluster nodes.
  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)}