Dataproc MCP Server

Instructions

Args:
    project_id: Google Cloud project ID
    region: Dataproc region
    cluster_name: Cluster name (optional)
    job_states: Filter by job states

Implementation Reference

• src/dataproc_mcp_server/server.py:204-230 (handler)

  The primary MCP tool handler for 'list_jobs', registered via @mcp.tool() decorator. It invokes the DataprocClient's list_jobs method and handles errors.

  @mcp.tool()
  async def list_jobs(
      project_id: str,
      region: str,
      cluster_name: str | None = None,
      job_states: list[str] | None = None,
  ) -> str:
      """List jobs in a Dataproc cluster.
  
      Args:
          project_id: Google Cloud project ID
          region: Dataproc region
          cluster_name: Cluster name (optional)
          job_states: Filter by job states
      """
      client = DataprocClient()
      try:
          result = await client.list_jobs(
              project_id=project_id,
              region=region,
              cluster_name=cluster_name,
              job_states=job_states or [],
          )
          return str(result)
      except Exception as e:
          logger.error("Failed to list jobs", error=str(e))
          return f"Error: {str(e)}"

• src/dataproc_mcp_server/dataproc_client.py:329-379 (helper)

  Supporting helper function in DataprocClient that performs the actual Google Cloud API call to list jobs, applies filtering, and formats the response.

  async def list_jobs(
      self,
      project_id: str,
      region: str,
      cluster_name: str | None = None,
      job_states: list[str] | None = None,
  ) -> dict[str, Any]:
      """List jobs in a region."""
      try:
          loop = asyncio.get_event_loop()
          client = self._get_job_client(region)
  
          request = types.ListJobsRequest(
              project_id=project_id,
              region=region,
              cluster_name=cluster_name,
              job_state_matcher=types.ListJobsRequest.StateMatcherType.ALL,
          )
  
          response = await loop.run_in_executor(None, client.list_jobs, request)
  
          jobs = []
          for job in response:
              # Filter by states if provided
              if job_states and job.status.state.name not in job_states:
                  continue
  
              jobs.append(
                  {
                      "job_id": job.reference.job_id,
                      "cluster_name": job.placement.cluster_name,
                      "status": job.status.state.name,
                      "job_type": self._get_job_type(job),
                      "submission_time": job.status.state_start_time.isoformat()
                      if job.status.state_start_time
                      else None,
                      "driver_output_uri": job.driver_output_resource_uri,
                  }
              )
  
          return {
              "jobs": jobs,
              "total_count": len(jobs),
              "project_id": project_id,
              "region": region,
              "cluster_name": cluster_name,
          }
  
      except Exception as e:
          logger.error("Failed to list jobs", error=str(e))
          raise