OCI MCP Server

Instructions

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

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