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