OCI MCP Server

Instructions

Implementation Reference

• mcp_server_oci/mcp_server.py:1788-1810 (handler)

  MCP handler function for 'get_oke_cluster' tool. Applies wrapper for logging/error handling and delegates to helper get_cluster.
  @mcp.tool(name="get_oke_cluster") @mcp_tool_wrapper( start_msg="Getting details for OKE cluster {cluster_id}...", error_prefix="Error getting OKE cluster details" ) async def mcp_get_oke_cluster(ctx: Context, cluster_id: str) -> Dict[str, Any]: """ Get detailed information about a specific OKE cluster. Args: cluster_id: OCID of the cluster Returns: Detailed cluster information including: - Kubernetes version and available upgrades - Cluster endpoints (public/private) - Network configuration (VCN, subnets, CIDR blocks) - Add-ons configuration (dashboard, tiller) - Image policy settings - Cluster metadata and options """ return get_cluster(oci_clients["container_engine"], cluster_id)
• mcp_server_oci/mcp_server.py:1788-1788 (registration)

  Registers the get_oke_cluster tool using FastMCP @mcp.tool decorator.
  @mcp.tool(name="get_oke_cluster")
• mcp_server_oci/tools/oke.py:59-161 (helper)

  Helper function implementing the core logic to retrieve and format OKE cluster details using OCI ContainerEngineClient.
  def get_cluster(container_engine_client: oci.container_engine.ContainerEngineClient, cluster_id: str) -> Dict[str, Any]: """ Get details of a specific OKE cluster. Args: container_engine_client: OCI ContainerEngine client cluster_id: OCID of the cluster Returns: Details of the cluster """ try: cluster = container_engine_client.get_cluster(cluster_id).data # Format endpoint config endpoint_config = None if cluster.endpoint_config: endpoint_config = { "subnet_id": cluster.endpoint_config.subnet_id, "nsg_ids": cluster.endpoint_config.nsg_ids, "is_public_ip_enabled": cluster.endpoint_config.is_public_ip_enabled, } # Format endpoints endpoints = None if cluster.endpoints: endpoints = { "kubernetes": cluster.endpoints.kubernetes if hasattr(cluster.endpoints, 'kubernetes') else None, "public_endpoint": cluster.endpoints.public_endpoint if hasattr(cluster.endpoints, 'public_endpoint') else None, "private_endpoint": cluster.endpoints.private_endpoint if hasattr(cluster.endpoints, 'private_endpoint') else None, "vcn_hostname_endpoint": cluster.endpoints.vcn_hostname_endpoint if hasattr(cluster.endpoints, 'vcn_hostname_endpoint') else None, } # Format cluster metadata metadata = None if cluster.metadata: metadata = { "time_created": str(cluster.metadata.time_created) if hasattr(cluster.metadata, 'time_created') and cluster.metadata.time_created else None, "created_by_user_id": cluster.metadata.created_by_user_id if hasattr(cluster.metadata, 'created_by_user_id') else None, "created_by_work_request_id": cluster.metadata.created_by_work_request_id if hasattr(cluster.metadata, 'created_by_work_request_id') else None, "time_updated": str(cluster.metadata.time_updated) if hasattr(cluster.metadata, 'time_updated') and cluster.metadata.time_updated else None, "updated_by_user_id": cluster.metadata.updated_by_user_id if hasattr(cluster.metadata, 'updated_by_user_id') else None, "updated_by_work_request_id": cluster.metadata.updated_by_work_request_id if hasattr(cluster.metadata, 'updated_by_work_request_id') else None, } # Format options options = None if cluster.options: options = { "service_lb_subnet_ids": cluster.options.service_lb_subnet_ids if hasattr(cluster.options, 'service_lb_subnet_ids') else None, "kubernetes_network_config": { "pods_cidr": cluster.options.kubernetes_network_config.pods_cidr if cluster.options.kubernetes_network_config else None, "services_cidr": cluster.options.kubernetes_network_config.services_cidr if cluster.options.kubernetes_network_config else None, } if hasattr(cluster.options, 'kubernetes_network_config') and cluster.options.kubernetes_network_config else None, "add_ons": { "is_kubernetes_dashboard_enabled": cluster.options.add_ons.is_kubernetes_dashboard_enabled if cluster.options.add_ons and hasattr(cluster.options.add_ons, 'is_kubernetes_dashboard_enabled') else None, "is_tiller_enabled": cluster.options.add_ons.is_tiller_enabled if cluster.options.add_ons and hasattr(cluster.options.add_ons, 'is_tiller_enabled') else None, } if hasattr(cluster.options, 'add_ons') and cluster.options.add_ons else None, "admission_controller_options": { "is_pod_security_policy_enabled": cluster.options.admission_controller_options.is_pod_security_policy_enabled if cluster.options.admission_controller_options and hasattr(cluster.options.admission_controller_options, 'is_pod_security_policy_enabled') else None, } if hasattr(cluster.options, 'admission_controller_options') and cluster.options.admission_controller_options else None, "persistent_volume_config": { "defined_tags": cluster.options.persistent_volume_config.defined_tags if cluster.options.persistent_volume_config and hasattr(cluster.options.persistent_volume_config, 'defined_tags') else None, "freeform_tags": cluster.options.persistent_volume_config.freeform_tags if cluster.options.persistent_volume_config and hasattr(cluster.options.persistent_volume_config, 'freeform_tags') else None, } if hasattr(cluster.options, 'persistent_volume_config') and cluster.options.persistent_volume_config else None, "service_lb_config": { "defined_tags": cluster.options.service_lb_config.defined_tags if cluster.options.service_lb_config and hasattr(cluster.options.service_lb_config, 'defined_tags') else None, "freeform_tags": cluster.options.service_lb_config.freeform_tags if cluster.options.service_lb_config and hasattr(cluster.options.service_lb_config, 'freeform_tags') else None, } if hasattr(cluster.options, 'service_lb_config') and cluster.options.service_lb_config else None, } cluster_details = { "id": cluster.id, "name": cluster.name, "compartment_id": cluster.compartment_id, "lifecycle_state": cluster.lifecycle_state, "lifecycle_details": cluster.lifecycle_details, "vcn_id": cluster.vcn_id, "kubernetes_version": cluster.kubernetes_version, "time_created": str(cluster.time_created) if cluster.time_created else None, "time_updated": str(cluster.time_updated) if cluster.time_updated else None, "endpoint_config": endpoint_config, "endpoints": endpoints, "metadata": metadata, "options": options, "available_kubernetes_upgrades": cluster.available_kubernetes_upgrades if hasattr(cluster, 'available_kubernetes_upgrades') else None, "image_policy_config": { "is_policy_enabled": cluster.image_policy_config.is_policy_enabled if cluster.image_policy_config and hasattr(cluster.image_policy_config, 'is_policy_enabled') else None, } if hasattr(cluster, 'image_policy_config') and cluster.image_policy_config else None, "cluster_pod_network_options": cluster.cluster_pod_network_options if hasattr(cluster, 'cluster_pod_network_options') else None, "type": cluster.type if hasattr(cluster, 'type') else None, "freeform_tags": cluster.freeform_tags if hasattr(cluster, 'freeform_tags') else None, "defined_tags": cluster.defined_tags if hasattr(cluster, 'defined_tags') else None, } logger.info(f"Retrieved details for OKE cluster {cluster_id}") return cluster_details except Exception as e: logger.exception(f"Error getting OKE cluster details: {e}") raise