OCI MCP Server

Instructions

Implementation Reference

• mcp_server_oci/tools/oke.py:224-352 (handler)

  Core handler function that retrieves and formats detailed information about an OKE node pool using the OCI Container Engine client, including node configurations, individual nodes, eviction settings, and tags.
  def get_node_pool(container_engine_client: oci.container_engine.ContainerEngineClient, node_pool_id: str) -> Dict[str, Any]: """ Get details of a specific node pool. Args: container_engine_client: OCI ContainerEngine client node_pool_id: OCID of the node pool Returns: Details of the node pool """ try: np = container_engine_client.get_node_pool(node_pool_id).data # Format node config details node_config_details = None if hasattr(np, 'node_config_details') and np.node_config_details: placement_configs = [] if hasattr(np.node_config_details, 'placement_configs') and np.node_config_details.placement_configs: for pc in np.node_config_details.placement_configs: placement_configs.append({ "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, "fault_domains": pc.fault_domains if hasattr(pc, 'fault_domains') else None, }) node_config_details = { "size": np.node_config_details.size, "placement_configs": placement_configs, "nsg_ids": np.node_config_details.nsg_ids if hasattr(np.node_config_details, 'nsg_ids') else None, "kms_key_id": np.node_config_details.kms_key_id if hasattr(np.node_config_details, 'kms_key_id') else None, "is_pv_encryption_in_transit_enabled": np.node_config_details.is_pv_encryption_in_transit_enabled if hasattr(np.node_config_details, 'is_pv_encryption_in_transit_enabled') else None, "freeform_tags": np.node_config_details.freeform_tags if hasattr(np.node_config_details, 'freeform_tags') else None, "defined_tags": np.node_config_details.defined_tags if hasattr(np.node_config_details, 'defined_tags') else None, } # Format node shape config node_shape_config = None if hasattr(np, 'node_shape_config') and np.node_shape_config: node_shape_config = { "ocpus": np.node_shape_config.ocpus if hasattr(np.node_shape_config, 'ocpus') else None, "memory_in_gbs": np.node_shape_config.memory_in_gbs if hasattr(np.node_shape_config, 'memory_in_gbs') else None, } # Format node source details node_source_details = None if hasattr(np, 'node_source_details') and np.node_source_details: node_source_details = { "source_type": np.node_source_details.source_type, "image_id": np.node_source_details.image_id if hasattr(np.node_source_details, 'image_id') else None, "boot_volume_size_in_gbs": np.node_source_details.boot_volume_size_in_gbs if hasattr(np.node_source_details, 'boot_volume_size_in_gbs') else None, } # Format initial node labels initial_node_labels = [] if hasattr(np, 'initial_node_labels') and np.initial_node_labels: for label in np.initial_node_labels: initial_node_labels.append({ "key": label.key if hasattr(label, 'key') else None, "value": label.value if hasattr(label, 'value') else None, }) # Format node eviction node pool settings node_eviction_node_pool_settings = None if hasattr(np, 'node_eviction_node_pool_settings') and np.node_eviction_node_pool_settings: node_eviction_node_pool_settings = { "eviction_grace_duration": np.node_eviction_node_pool_settings.eviction_grace_duration if hasattr(np.node_eviction_node_pool_settings, 'eviction_grace_duration') else None, "is_force_delete_after_grace_duration": np.node_eviction_node_pool_settings.is_force_delete_after_grace_duration if hasattr(np.node_eviction_node_pool_settings, 'is_force_delete_after_grace_duration') else None, } # Format node pool cycling details node_pool_cycling_details = None if hasattr(np, 'node_pool_cycling_details') and np.node_pool_cycling_details: node_pool_cycling_details = { "maximum_unavailable": np.node_pool_cycling_details.maximum_unavailable if hasattr(np.node_pool_cycling_details, 'maximum_unavailable') else None, "maximum_surge": np.node_pool_cycling_details.maximum_surge if hasattr(np.node_pool_cycling_details, 'maximum_surge') else None, "is_node_cycling_enabled": np.node_pool_cycling_details.is_node_cycling_enabled if hasattr(np.node_pool_cycling_details, 'is_node_cycling_enabled') else None, } node_pool_details = { "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_id": np.node_image_id if hasattr(np, 'node_image_id') else None, "node_image_name": np.node_image_name if hasattr(np, 'node_image_name') else None, "node_shape": np.node_shape, "node_shape_config": node_shape_config, "node_source_details": node_source_details, "node_config_details": node_config_details, "initial_node_labels": initial_node_labels, "ssh_public_key": np.ssh_public_key if hasattr(np, 'ssh_public_key') else None, "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, "nodes": [ { "id": node.id if hasattr(node, 'id') else None, "name": node.name if hasattr(node, 'name') else None, "availability_domain": node.availability_domain if hasattr(node, 'availability_domain') else None, "subnet_id": node.subnet_id if hasattr(node, 'subnet_id') else None, "lifecycle_state": node.lifecycle_state if hasattr(node, 'lifecycle_state') else None, "fault_domain": node.fault_domain if hasattr(node, 'fault_domain') else None, "private_ip": node.private_ip if hasattr(node, 'private_ip') else None, "public_ip": node.public_ip if hasattr(node, 'public_ip') else None, "node_error": { "code": node.node_error.code if node.node_error and hasattr(node.node_error, 'code') else None, "message": node.node_error.message if node.node_error and hasattr(node.node_error, 'message') else None, } if hasattr(node, 'node_error') and node.node_error else None, } for node in np.nodes ] if hasattr(np, 'nodes') and np.nodes else [], "node_eviction_node_pool_settings": node_eviction_node_pool_settings, "node_pool_cycling_details": node_pool_cycling_details, "freeform_tags": np.freeform_tags if hasattr(np, 'freeform_tags') else None, "defined_tags": np.defined_tags if hasattr(np, 'defined_tags') else None, "time_created": str(np.time_created) if hasattr(np, 'time_created') and np.time_created else None, } logger.info(f"Retrieved details for node pool {node_pool_id}") return node_pool_details except Exception as e: logger.exception(f"Error getting node pool details: {e}") raise
• mcp_server_oci/mcp_server.py:1840-1863 (registration)

  MCP tool registration for 'get_oke_node_pool' with wrapper that handles context, logging, error handling, and calls the core get_node_pool function.
  @mcp.tool(name="get_oke_node_pool") @mcp_tool_wrapper( start_msg="Getting details for node pool {node_pool_id}...", error_prefix="Error getting node pool details" ) async def mcp_get_oke_node_pool(ctx: Context, node_pool_id: str) -> Dict[str, Any]: """ Get detailed information about a specific node pool. Args: node_pool_id: OCID of the node pool Returns: Detailed node pool information including: - Node configuration (shape, image, SSH keys) - Individual node details (IPs, state, fault domains) - Placement configuration across ADs - Node eviction settings - Node pool cycling details - Initial node labels - Security settings (NSGs, encryption) """ return get_node_pool(oci_clients["container_engine"], node_pool_id)