OCM MCP Server

Instructions

Implementation Reference

• src/tools/clusters.ts:18-65 (handler)

  Main handler function for the 'clusters' tool: lists ManagedCluster CRs from open-cluster-management.io API, formats as table with name, hub accepted, URL, joined, available, age.
  export async function listClusters({ }): Promise<CallToolResult> { const response = await client.list<k8s.KubernetesObject>("cluster.open-cluster-management.io/v1", "ManagedCluster") if (!response || response.items.length == 0) { console.warn("no managed clusters on the current cluster") return { content: [{ type: "text", text: "no managed clusters available on the current cluster" }], } } clusterToServerAPIMap = new Map( response.items.map((item: any) => { const name: string = item.metadata?.name; const server: string = item.spec?.managedClusterClientConfigs?.[0]?.url; return [name, server]; }) ); // Format table header let result = `NAME HUB ACCEPTED MANAGED CLUSTER URLS JOINED AVAILABLE AGE\n`; // Process each cluster and format the output response.items.forEach((item: any) => { const name: string = item.metadata?.name || "Unknown"; const hubAccepted: string = item.spec?.hubAcceptsClient ? "true" : "false"; const server: string = item.spec?.managedClusterClientConfigs?.[0]?.url || "N/A"; // Extract conditions const joinedCondition = item.status?.conditions?.find((c: any) => c.type === "ManagedClusterJoined")?.status || "False"; const availableCondition = item.status?.conditions?.find((c: any) => c.type === "ManagedClusterConditionAvailable")?.status || "False"; // Calculate cluster age const creationTimestamp = item.metadata?.creationTimestamp; const age = creationTimestamp ? getClusterAge(creationTimestamp) : "N/A"; // Append formatted row result += `${name.padEnd(10)} ${hubAccepted.padEnd(14)} ${server.padEnd(80)} ${joinedCondition.padEnd(8)} ${availableCondition.padEnd(10)} ${age}\n`; }); return { content: [{ type: "text", text: result }], } }
• src/tools/clusters.ts:15-17 (schema)

  Description and empty Zod schema (args) for the 'clusters' tool.
  export const listClusterDesc = "Retrieves a list of Kubernetes clusters (also known as managed clusters or spoke clusters)." export const listClustersArgs = {}
• src/index.ts:24-28 (registration)

  Registration of the 'clusters' tool in the TypeScript MCP server.
  "clusters", listClusterDesc, listClustersArgs, // should be a Zod schema, e.g., z.object({...}) async (args, extra) => listClusters(args) // ensure listClusters matches (args, extra) => ... )
• python/multicluster_mcp_server/tools/cluster.py:31-82 (handler)

  Python handler for the 'clusters' tool using MCP decorator: lists ManagedCluster resources and formats table.
  @mcp.tool(description="Retrieves a list of Kubernetes clusters (also known as managed clusters or spoke clusters).") def clusters() -> Annotated[str, Field(description="The managed clusters, also known as spoke clusters.")]: config.load_kube_config() dyn_client = DynamicClient(ApiClient()) try: managed_cluster_res = dyn_client.resources.get( api_version="cluster.open-cluster-management.io/v1", kind="ManagedCluster" ) response = managed_cluster_res.get() items = response.items except Exception as e: return f"Failed to list clusters: {e}" if not items: return "No managed clusters available on the current cluster" header = ( f"{'NAME':<12} {'HUB ACCEPTED':<15} {'MANAGED CLUSTER URLS':<80} " f"{'JOINED':<8} {'AVAILABLE':<10} {'AGE'}" ) result_lines = [header] for item in items: metadata = item.metadata spec = item.spec or {} status = item.status or {} name = metadata.name or "Unknown" hub_accepted = str(spec.get("hubAcceptsClient", False)).lower() server = spec.get("managedClusterClientConfigs", [{}])[0].get("url", "N/A") conditions = status.get("conditions", []) joined = next((c.get("status") for c in conditions if c.get("type") == "ManagedClusterJoined"), "False") available = next((c.get("status") for c in conditions if c.get("type") == "ManagedClusterConditionAvailable"), "False") creation_timestamp = metadata.creationTimestamp age = get_cluster_age(creation_timestamp) if creation_timestamp else "N/A" if generate_kubeconfig: try: kubeconfig_path = setup_cluster_access(cluster=name) cluster_kubeconfig_map[name] = kubeconfig_path except Exception as e: logger.warning(f"Failed to setup access for cluster '{name}': {e}") result_lines.append( f"{name:<12} {hub_accepted:<15} {server:<80} {joined:<8} {available:<10} {age}" ) return "\n".join(result_lines)
• python/multicluster_mcp_server/__main__.py:3-3 (registration)

  Import of the 'clusters' tool for registration in Python MCP server (registration via decorator in source file).
  from multicluster_mcp_server.tools.cluster import clusters