OCI MCP Server

Instructions

List work requests (async operations) for OKE resources in a compartment.

Args:
    compartment_id: OCID of the compartment
    resource_id: Optional OCID of a specific resource (cluster or node pool) to filter by

Returns:
    List of work requests with their details including:
    - Operation type (create, update, delete, etc.)
    - Status and completion percentage
    - Associated resources
    - Timestamps (accepted, started, finished)

Implementation Reference

• mcp_server_oci/tools/oke.py:395-444 (handler)

  Core handler function that executes the tool logic: lists OKE work requests using OCI ContainerEngineClient.list_work_requests, formats the response into a list of dictionaries with details like id, operation_type, status, resources, percent_complete, and timestamps.

  def list_work_requests(container_engine_client: oci.container_engine.ContainerEngineClient,
                         compartment_id: str,
                         resource_id: Optional[str] = None) -> List[Dict[str, Any]]:
      """
      List work requests in a compartment, optionally filtered by resource.
  
      Args:
          container_engine_client: OCI ContainerEngine client
          compartment_id: OCID of the compartment
          resource_id: Optional OCID of the resource to filter by (cluster or node pool)
  
      Returns:
          List of work requests with their details
      """
      try:
          kwargs = {"compartment_id": compartment_id}
          if resource_id:
              kwargs["resource_id"] = resource_id
  
          work_requests_response = oci.pagination.list_call_get_all_results(
              container_engine_client.list_work_requests,
              **kwargs
          )
  
          work_requests = []
          for wr in work_requests_response.data:
              work_requests.append({
                  "id": wr.id,
                  "operation_type": wr.operation_type,
                  "status": wr.status,
                  "compartment_id": wr.compartment_id,
                  "resources": [
                      {
                          "action_type": res.action_type if hasattr(res, 'action_type') else None,
                          "entity_type": res.entity_type if hasattr(res, 'entity_type') else None,
                          "identifier": res.identifier if hasattr(res, 'identifier') else None,
                          "entity_uri": res.entity_uri if hasattr(res, 'entity_uri') else None,
                      }
                      for res in wr.resources
                  ] if hasattr(wr, 'resources') and wr.resources else [],
                  "percent_complete": wr.percent_complete if hasattr(wr, 'percent_complete') else None,
                  "time_accepted": str(wr.time_accepted) if hasattr(wr, 'time_accepted') and wr.time_accepted else None,
                  "time_started": str(wr.time_started) if hasattr(wr, 'time_started') and wr.time_started else None,
                  "time_finished": str(wr.time_finished) if hasattr(wr, 'time_finished') and wr.time_finished else None,
              })
  
          logger.info(f"Found {len(work_requests)} work requests in compartment {compartment_id}" +
                     (f" for resource {resource_id}" if resource_id else ""))
          return work_requests

• mcp_server_oci/mcp_server.py:1884-1909 (registration)

  MCP tool registration for 'list_oke_work_requests' with wrapper for error handling, logging, and profile management. Calls the handler from tools.oke.list_work_requests.

  @mcp.tool(name="list_oke_work_requests")
  @mcp_tool_wrapper(
      start_msg="Listing OKE work requests in compartment {compartment_id}...",
      error_prefix="Error listing OKE work requests"
  )
  async def mcp_list_oke_work_requests(
      ctx: Context,
      compartment_id: str,
      resource_id: Optional[str] = None
  ) -> List[Dict[str, Any]]:
      """
      List work requests (async operations) for OKE resources in a compartment.
  
      Args:
          compartment_id: OCID of the compartment
          resource_id: Optional OCID of a specific resource (cluster or node pool) to filter by
  
      Returns:
          List of work requests with their details including:
          - Operation type (create, update, delete, etc.)
          - Status and completion percentage
          - Associated resources
          - Timestamps (accepted, started, finished)
      """
      return list_work_requests(oci_clients["container_engine"], compartment_id, resource_id)