CockroachDB MCP Server

Instructions

Returns:
    Region information.

Implementation Reference

• src/cockroachdb_mcp/server.py:451-462 (registration)

  Registration of the 'show_regions' tool using @mcp.tool() decorator. This handler delegates to the cluster module implementation.

  @mcp.tool()
  async def show_regions() -> dict[str, Any]:
      """Show database regions for multi-region clusters.
  
      Returns:
          Region information.
      """
      try:
          return await cluster.show_regions()
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/tools/cluster.py:166-222 (handler)

  Core implementation of show_regions that queries CockroachDB for database regions, primary/secondary regions, survival goal, and cluster regions using crdb_internal tables and SHOW REGIONS.

  async def show_regions() -> dict[str, Any]:
      """Show database regions for multi-region clusters.
  
      Returns:
          Region information.
      """
      conn = await connection_manager.ensure_connected()
  
      try:
          async with conn.cursor() as cur:
              # Get database regions
              await cur.execute("""
                  SELECT
                      database_name,
                      primary_region,
                      secondary_region,
                      regions,
                      survival_goal
                  FROM crdb_internal.databases
                  WHERE database_name = current_database()
              """)
              db_row = await cur.fetchone()
  
              if not db_row:
                  return {
                      "status": "success",
                      "database": connection_manager.current_database,
                      "is_multi_region": False,
                      "message": "Database is not configured for multi-region",
                  }
  
              # Get all regions in the cluster
              await cur.execute("SHOW REGIONS")
              region_rows = await cur.fetchall()
  
          regions = [row.get("region") for row in region_rows if row.get("region")]
  
          return {
              "status": "success",
              "database": db_row.get("database_name"),
              "primary_region": db_row.get("primary_region"),
              "secondary_region": db_row.get("secondary_region"),
              "regions": db_row.get("regions"),
              "survival_goal": db_row.get("survival_goal"),
              "cluster_regions": regions,
              "is_multi_region": bool(db_row.get("primary_region")),
          }
      except Exception as e:
          # Multi-region may not be enabled
          if "unknown function" in str(e).lower() or "regions" in str(e).lower():
              return {
                  "status": "success",
                  "is_multi_region": False,
                  "message": "Multi-region is not enabled for this cluster",
              }
          return {"status": "error", "error": str(e)}