OCM MCP Server

Instructions

Implementation Reference

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

  The core handler function for the 'connect_cluster' tool. It creates a ManagedServiceAccount, applies ClusterRoleBinding via ManifestWork, retrieves the service account token secret, and generates a KUBECONFIG for the specified cluster.

  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/tools/clusters.ts:75-80 (schema)

  Zod schema and description for the 'connect_cluster' tool inputs.

  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)."

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

  Registration of the 'connect_cluster' tool on the MCP server using server.tool().

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

• python/multicluster_mcp_server/tools/connect.py:223-228 (handler)

  Python implementation of the connect_cluster tool handler, decorated with @mcp.tool and delegating to setup_cluster_access.

  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)

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

  Import of the connect_cluster tool in the main entrypoint, where tools decorated with @mcp.tool are automatically registered when mcp.run() is called.

  from multicluster_mcp_server.tools.cluster import clusters
  from multicluster_mcp_server.tools.connect import connect_cluster
  from multicluster_mcp_server.tools.kubectl import kube_executor
  from multicluster_mcp_server.tools.prometheus import prometheus