Dataproc MCP Server

Instructions

Args:
    project_id: Google Cloud project ID (optional, uses gcloud config default)
    region: Dataproc region (optional, uses gcloud config default)

Implementation Reference

• src/dataproc_mcp_server/server.py:46-68 (handler)

  MCP tool handler for 'list_clusters' registered via @mcp.tool() decorator. Handles input resolution, calls DataprocClient, and returns result as string.

  @mcp.tool()
  async def list_clusters(
      project_id: str | None = None, region: str | None = None
  ) -> str:
      """List Dataproc clusters in a project and region.
  
      Args:
          project_id: Google Cloud project ID (optional, uses gcloud config default)
          region: Dataproc region (optional, uses gcloud config default)
      """
      resolved = resolve_project_and_region(project_id, region)
      if isinstance(resolved, str):  # Error message
          return resolved
      project_id, region = resolved
  
      client = DataprocClient()
      try:
          result = await client.list_clusters(project_id, region)
          return str(result)
      except Exception as e:
          logger.error("Failed to list clusters", error=str(e))
          return f"Error: {str(e)}"

• src/dataproc_mcp_server/dataproc_client.py:69-111 (helper)

  DataprocClient method that performs the actual Google Cloud Dataproc API call to list clusters and formats the response.

  async def list_clusters(self, project_id: str, region: str) -> dict[str, Any]:
      """List clusters in a project and region."""
      try:
          loop = asyncio.get_event_loop()
          client = self._get_cluster_client(region)
  
          request = types.ListClustersRequest(project_id=project_id, region=region)
  
          # Run in thread pool since the client is sync
          response = await loop.run_in_executor(None, client.list_clusters, request)
  
          clusters = []
          for cluster in response:
              clusters.append(
                  {
                      "name": cluster.cluster_name,
                      "status": cluster.status.state.name,
                      "num_instances": cluster.config.worker_config.num_instances,
                      "machine_type": cluster.config.master_config.machine_type_uri.split(
                          "/"
                      )[-1],
                      "creation_time": cluster.status.state_start_time.isoformat()
                      if cluster.status.state_start_time
                      else None,
                      "zone": cluster.config.gce_cluster_config.zone_uri.split("/")[
                          -1
                      ]
                      if cluster.config.gce_cluster_config.zone_uri
                      else None,
                  }
              )
  
          return {
              "clusters": clusters,
              "total_count": len(clusters),
              "project_id": project_id,
              "region": region,
          }
  
      except Exception as e:
          logger.error("Failed to list clusters", error=str(e))
          raise

• src/dataproc_mcp_server/server.py:22-43 (helper)

  Utility function used by list_clusters handler to resolve project_id and region from inputs or gcloud defaults.

  def resolve_project_and_region(
      project_id: str | None, region: str | None
  ) -> tuple[str, str] | str:
      """Resolve project_id and region from parameters or gcloud config defaults.
  
      Returns:
          Tuple of (project_id, region) if successful, error message string if failed.
      """
      # Resolve project_id
      if project_id is None:
          project_id = get_default_project()
          if project_id is None:
              return "Error: No project_id provided and no default project configured in gcloud. Run 'gcloud config set project PROJECT_ID' or provide project_id parameter."
  
      # Resolve region
      if region is None:
          region = get_default_region()
          if region is None:
              return "Error: No region provided and no default region configured in gcloud. Run 'gcloud config set compute/region REGION' or provide region parameter."
  
      return project_id, region