Multi-Cluster MCP Server

Instructions

Implementation Reference

• python/multicluster_mcp_server/tools/connect.py:222-259 (handler)

  Python MCP tool handler for 'connect_cluster': decorates the function and implements logic to create ManagedServiceAccount, RBAC via ManifestWork, retrieve token secret, and generate kubeconfig file.

  @mcp.tool(description="Generates the 'KUBECONFIG' for the managed cluster and binds it to the specified ClusterRole (default: cluster-admin).")
  def connect_cluster(
      cluster: Annotated[str, Field(description="The target cluster where the ServiceAccount will be created for the KUBECONFIG.")],
      cluster_role: Annotated[str, Field(description="The ClusterRole defining permissions to access the cluster.")] = "cluster-admin",
  ) -> Annotated[str, Field(description="A message indicating the kubeconfig file or failure of the operation.")]:
      return setup_cluster_access(cluster, cluster_role=cluster_role)
  
  def setup_cluster_access(cluster: str, cluster_role: str = "cluster-admin", mcp_server: str = server_name):
      logger.debug(f"Setting up ManagedServiceAccount and RBAC for cluster: {cluster}")
      
      msa_result = create_or_update_managed_service_account(cluster, mcp_server)
      if not msa_result:
          logger.error("Failed to set up ManagedServiceAccount. Skipping RBAC setup.")
          return None
      
      rbac_result = create_or_update_rbac(cluster, mcp_server, cluster_role)
      if not rbac_result:
          logger.error("RBAC (ManifestWork) setup failed.")
          return None
        
      server_url = get_managed_cluster_url(cluster_name=cluster)
      if not server_url:
          logger.error(f"API server URL not found for ManagedCluster '{cluster}'.")
          return None
      
      token_secret = get_secret_with_timeout(cluster, mcp_server)
      if not token_secret:
        logger.error(f"Failed to get the service account token for cluster: {cluster}")
        return None
      
      kubeconfig_path_or_error = generate_kubeconfig_file_from_secret(token_secret, server_url, mcp_server)
      if not kubeconfig_path_or_error.startswith("/tmp/"):
          logger.error(kubeconfig_path_or_error)
          return None
        
      logger.debug(f"Generate the kubeconfig file: {kubeconfig_path_or_error}")
      return kubeconfig_path_or_error

• src/tools/clusters.ts:82-211 (handler)

  TypeScript MCP tool handler for 'connect_cluster': async function that applies ManagedServiceAccount, ManagedClusterAddOn, ClusterRoleBinding via ManifestWork, polls for token secret, and generates KUBECONFIG.

  export async function connectCluster({ cluster, clusterRole = "cluster-admin" }: {
    cluster: string, clusterRole?: string
  }): Promise<CallToolResult> {
    // https://open-cluster-management.io/docs/getting-started/integration/managed-serviceaccount/
    const mcpServerName = "multicluster-mcp-server"
    const msa = {
      apiVersion: 'authentication.open-cluster-management.io/v1beta1',
      kind: 'ManagedServiceAccount',
      metadata: {
        name: mcpServerName,
        namespace: cluster,
      },
      spec: {
        rotation: {},
      },
    }
  
    const mca = {
      apiVersion: 'addon.open-cluster-management.io/v1alpha1',
      kind: 'ManagedClusterAddOn',
      metadata: {
        name: "managed-serviceaccount",
        namespace: cluster,
      },
    }
  
    let result = `Successfully connected to cluster ${cluster} using ServiceAccount ${mcpServerName}, with the ${clusterRole} ClusterRole assigned.`;
  
    let isErrored = false
    try {
  
      const [applyMsa, getMca, getClusters] = await Promise.all([
        client.patch<k8s.KubernetesObject>(
          msa,
          undefined,
          undefined,
          mcpServerName,
          true,
          k8s.PatchStrategy.ServerSideApply
        ),
        client.read(mca),
        listClusters({})
      ]);
  
  
      if (!applyMsa) {
        console.warn(`Patched ManagedServiceAccount ${msa.metadata.namespace}/${msa.metadata.name} with empty response`);
      }
  
      const saNamespace = (getMca as any)?.status?.namespace;
      if (!saNamespace) {
        throw new Error(`ManagedServiceAccount ${mca.metadata.namespace}/${mca.metadata.name} not found in the cluster`);
      }
  
      const clusterRoleBinding = {
        apiVersion: "rbac.authorization.k8s.io/v1",
        kind: "ClusterRoleBinding",
        metadata: {
          name: `${mcpServerName}-clusterrolebinding`,
        },
        roleRef: {
          apiGroup: "rbac.authorization.k8s.io",
          kind: "ClusterRole",
          name: clusterRole, // default clusterRole name for kubernetes admin - "cluster-admin"
        },
        subjects: [
          {
            kind: "ServiceAccount",
            name: mcpServerName,
            namespace: saNamespace,
          },
        ],
      };
  
      // create manifestWork to binding the clusterRole into the serviceAccount
      const bindingPermissionManifestWork = {
        apiVersion: 'work.open-cluster-management.io/v1',
        kind: 'ManifestWork',
        metadata: {
          name: mcpServerName,
          namespace: cluster,
        },
        spec: {
          workload: {
            manifests: [
              clusterRoleBinding,
            ]
          }
        },
      }
  
      const [tokenSecret, applyRBACManifest, appliedStatusErrMessage] = await Promise.all([
        getSecretWithRetry(cluster, mcpServerName),
        // createKubeConfigFile(acmMCPServer, cluster),
        client.patch<k8s.KubernetesObject>(
          bindingPermissionManifestWork, undefined, undefined, mcpServerName, true,
          k8s.PatchStrategy.ServerSideApply),
        // get the status
        manifestWorkAppliedErrorMessage(client, mcpServerName, cluster)
      ]);
  
      // error token
      if (typeof tokenSecret == 'string') {
        throw new Error(tokenSecret)
      }
  
      // error status
      if (appliedStatusErrMessage != "") {
        throw new Error(appliedStatusErrMessage)
      }
  
      const kubeConfigErrMessage = generateKubeconfig(tokenSecret, clusterToServerAPIMap);
      if (kubeConfigErrMessage) {
        throw new Error(kubeConfigErrMessage)
      }
  
    } catch (err: any) {
      isErrored = true
      result = `Failed to generate KUBECONFIG for ${cluster}: ${err}`
    }
  
    // return manifestsResponse
    return {
      content: [{
        type: "text",
        text: result
      }],
      isErrored: isErrored
    }
  }

• src/index.ts:31-35 (registration)

  Explicit registration of the 'connect_cluster' tool in the MCP server using server.tool() with description, args schema, and handler.

    "connect_cluster",
    connectClusterDesc,
    connectClusterArgs,
    async (args, extra) => connectCluster(args) // ensure connectCluster matches (args, extra) => ...
  )

• src/tools/clusters.ts:75-80 (schema)

  Zod schema definition for connect_cluster tool parameters (cluster and optional clusterRole) and tool description.

  export const connectClusterArgs = {
    cluster: z.string().describe("The target cluster where the ServiceAccount will be created for the KUBECONFIG."),
    clusterRole: z.string().default('cluster-admin').describe("The ClusterRole defining permissions to access the cluster")
  }
  
  export const connectClusterDesc = "Generates the KUBECONFIG for the managed cluster and binds it to the specified ClusterRole (default: cluster-admin)."

• python/multicluster_mcp_server/__main__.py:4-4 (registration)

  Import of connect_cluster tool in __main__.py, which triggers auto-registration via @mcp.tool decorator when mcp.run() is called.

  from multicluster_mcp_server.tools.connect import connect_cluster