OCM MCP Server

Instructions

Implementation Reference

• python/multicluster_mcp_server/tools/kubectl.py:22-66 (handler)

  Python MCP tool handler for 'kubectl': executes kubectl commands or applies YAML manifests with cluster-specific kubeconfig handling, validation, and error management.

  @mcp.tool(description="Securely run a kubectl command or apply YAML. Provide either 'command' or 'yaml'.")
  def kube_executor(
      cluster: Annotated[str, Field(description="The cluster name in a multi-cluster environment. Defaults to the hub cluster.")] = "default",
      command: Annotated[Optional[str], Field(description="The full kubectl command to execute. Must start with 'kubectl'.")] = None,
      yaml: Annotated[Optional[str], Field(description="YAML configuration to apply, provided as a string.")] = None,
  ) -> Annotated[str, Field(description="The execution result")]:
      try:
          if not command and not yaml:
              raise ValueError("Either 'command' or 'yaml' must be provided.")
          if command and yaml:
              raise ValueError("Provide only one of 'command' or 'yaml', not both.")
  
          kubeconfig_file = None
          if cluster and cluster != "default":
              kubeconfig_file = get_kubeconfig_file(cluster)
              if not validate_kubeconfig_file(kubeconfig_file):
                  kubeconfig_file = setup_cluster_access(cluster=cluster)
                  if not kubeconfig_file:
                      raise FileNotFoundError(f"KUBECONFIG for cluster '{cluster}' does not exist.")
  
          if command:
              if not isinstance(command, str) or not is_valid_kubectl_command(command):
                  raise ValueError("Invalid command: Only 'kubectl' commands are allowed.")
              final_command = command
          else:
              # Write YAML to a temp file
              if not isinstance(yaml, str) or not yaml.strip():
                  raise ValueError("Invalid YAML content.")
              with tempfile.NamedTemporaryFile("w", delete=False, suffix=".yaml") as temp_file:
                  temp_file.write(yaml)
                  temp_file_path = temp_file.name
              final_command = f"kubectl apply -f {temp_file_path}"
  
          # Add --kubeconfig if needed
          if kubeconfig_file:
              final_command = inject_kubeconfig(final_command, kubeconfig_file)
  
          print(f"[debug] Executing: {final_command}")
          result = subprocess.run(final_command, shell=True, capture_output=True, text=True, timeout=10)
  
          output = result.stdout or result.stderr or "Run kube executor successfully, but no output returned."
          return output
      except Exception as e:
          return f"Error running kube executor: {str(e)}"

• src/tools/kubectl.ts:34-86 (handler)

  TypeScript MCP tool handler for 'kubectl': executes kubectl commands or applies YAML via stdin with cluster kubeconfig injection and validation.

  export async function kubectl({
    command,
    cluster,
    yaml,
  }: {
    command?: string;
    yaml?: string;
    cluster?: string;
  }): Promise<CallToolResult> {
    try {
      if (!command && !yaml) {
        throw new Error("Either 'command' or 'yaml' must be provided.");
      }
      if (command && yaml) {
        throw new Error("Provide only one of 'command' or 'yaml', not both.");
      }
  
      const kubeConfigFile = await getKubeconfigFile(cluster);
  
      let stdout = "";
      let stderr = "";
  
      if (command) {
        if (typeof command !== "string" || !isValidKubectlCommand(command)) {
          throw new Error("Invalid command: Only 'kubectl' commands are allowed.");
        }
  
        const finalCommand = kubeConfigFile
          ? `${command} --kubeconfig=${kubeConfigFile}`
          : command;
  
        const result = await execPromise(finalCommand);
        stdout = result.stdout;
        stderr = result.stderr;
      } else if (yaml) {
        stdout = await applyYaml(yaml, kubeConfigFile)
      }
  
      return {
        content: [{
          type: "text",
          text: stdout?.trim() || stderr?.trim() || "Run kubectl successfully, but no output returned.",
        }],
      };
    } catch (err: any) {
      return {
        content: [{
          type: "text",
          text: `Error running kubectl: ${err.message || String(err)}`,
        }],
      };
    }
  }

• src/index.ts:37-42 (registration)

  TypeScript MCP server registration of the 'kubectl' tool using McpServer.tool().

  server.tool(
    "kubectl",
    kubectlDesc,
    kubectlArgs,
    async (args, extra) => kubectl(args) // ensure connectCluster matches (args, extra) => ...
  )

• src/tools/kubectl.ts:20-32 (schema)

  TypeScript Zod schema definition for 'kubectl' tool arguments and description.

  export const kubectlDesc = "Securely run a kubectl command or apply YAML. Provide either 'command' or 'yaml'.";
  export const kubectlArgs = {
    command: z
      .string()
      .describe("The full kubectl command to execute. Must start with 'kubectl'."),
    yaml: z
      .string()
      .describe("YAML configuration to apply, provided as a string."),
    cluster: z
      .string()
      .describe("The cluster name in a multi-cluster environment. Defaults to the hub cluster.")
      .default("default"),
  };

• src/tools/kubectl.ts:11-12 (helper)

  Helper function to validate if a command is a valid kubectl command (shared pattern in both implementations).

  // Validate that the command starts with "kubectl"
  export function isValidKubectlCommand(command: string): boolean {