OCI MCP Server

Instructions

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)