Multi-Cluster MCP Server

Instructions

Implementation Reference

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

  The core handler function `kube_executor` decorated with `@mcp.tool`. It executes kubectl commands or applies YAML manifests on specified clusters, handling kubeconfig setup, validation, and subprocess execution.

  @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)}"

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

  Import of the kube_executor tool in the main entrypoint, which registers it via the decorator when the module is loaded before mcp.run().

  from multicluster_mcp_server.tools.kubectl import kube_executor

• python/multicluster_mcp_server/tools/kubectl.py:23-27 (schema)

  Pydantic-based input schema defined via Annotated Fields in the function signature, including cluster, command, yaml parameters and output type.

  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")]:

• python/multicluster_mcp_server/tools/kubectl.py:10-11 (helper)

  Helper function to validate that the provided command starts with 'kubectl '.

  def is_valid_kubectl_command(command: str) -> bool:
      return command.strip().startswith("kubectl ")