OCI MCP Server

Instructions

Implementation Reference

• mcp_server_oci/mcp_server.py:1812-1838 (registration)

  MCP tool registration for 'list_oke_node_pools' including decorator, input parameters (compartment_id required, cluster_id optional), docstring schema, and call to core handler.
  @mcp.tool(name="list_oke_node_pools") @mcp_tool_wrapper( start_msg="Listing node pools in compartment {compartment_id}...", error_prefix="Error listing node pools" ) async def mcp_list_oke_node_pools( ctx: Context, compartment_id: str, cluster_id: Optional[str] = None ) -> List[Dict[str, Any]]: """ List all node pools in a compartment, optionally filtered by cluster. Args: compartment_id: OCID of the compartment cluster_id: Optional OCID of the cluster to filter by Returns: List of node pools with their details including: - Node shape and image information - Kubernetes version - Placement configuration (ADs, subnets) - Node count per subnet - Lifecycle state """ return list_node_pools(oci_clients["container_engine"], compartment_id, cluster_id)
• mcp_server_oci/tools/oke.py:163-222 (handler)

  Core implementation of listing OKE node pools using OCI ContainerEngineClient, paginates results, formats node pool details including config, placement, and metadata.
  def list_node_pools(container_engine_client: oci.container_engine.ContainerEngineClient, compartment_id: str, cluster_id: Optional[str] = None) -> List[Dict[str, Any]]: """ List all node pools in a compartment, optionally filtered by cluster. Args: container_engine_client: OCI ContainerEngine client compartment_id: OCID of the compartment cluster_id: Optional OCID of the cluster to filter by Returns: List of node pools with their details """ try: kwargs = {"compartment_id": compartment_id} if cluster_id: kwargs["cluster_id"] = cluster_id node_pools_response = oci.pagination.list_call_get_all_results( container_engine_client.list_node_pools, **kwargs ) node_pools = [] for np in node_pools_response.data: node_pools.append({ "id": np.id, "name": np.name, "compartment_id": np.compartment_id, "cluster_id": np.cluster_id, "lifecycle_state": np.lifecycle_state, "lifecycle_details": np.lifecycle_details, "kubernetes_version": np.kubernetes_version, "node_image_name": np.node_image_name if hasattr(np, 'node_image_name') else None, "node_shape": np.node_shape, "quantity_per_subnet": np.quantity_per_subnet if hasattr(np, 'quantity_per_subnet') else None, "subnet_ids": np.subnet_ids if hasattr(np, 'subnet_ids') else None, "node_config_details": { "size": np.node_config_details.size if np.node_config_details else None, "placement_configs": [ { "availability_domain": pc.availability_domain, "subnet_id": pc.subnet_id, "capacity_reservation_id": pc.capacity_reservation_id if hasattr(pc, 'capacity_reservation_id') else None, } for pc in np.node_config_details.placement_configs ] if np.node_config_details and hasattr(np.node_config_details, 'placement_configs') and np.node_config_details.placement_configs else [], } if hasattr(np, 'node_config_details') and np.node_config_details else None, "time_created": str(np.time_created) if hasattr(np, 'time_created') and np.time_created else None, }) logger.info(f"Found {len(node_pools)} node pools in compartment {compartment_id}" + (f" for cluster {cluster_id}" if cluster_id else "")) return node_pools except Exception as e: logger.exception(f"Error listing node pools: {e}") raise