list_clusters
Retrieve a list of Dataproc clusters within a specified Google Cloud project and region to manage and monitor cloud data processing resources efficiently.
Instructions
List Dataproc clusters in a project and region.
Args:
project_id: Google Cloud project ID
region: Dataproc region (e.g., us-central1)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | Yes | ||
| region | Yes |
Implementation Reference
- src/dataproc_mcp_server/server.py:46-68 (handler)MCP tool handler for list_clusters: resolves project/region defaults, creates DataprocClient, calls its list_clusters method, and returns result as string or error.@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)}"
- DataprocClient.list_clusters: core implementation that queries Google Cloud Dataproc API via ClusterControllerClient.list_clusters, extracts cluster details into a structured dict.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
- Helper function to resolve project_id and region from inputs or gcloud defaults, returning tuple or error string.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
- src/dataproc_mcp_server/server.py:46-46 (registration)FastMCP @mcp.tool() decorator registers the list_clusters function as an MCP tool.@mcp.tool()