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