kube_executor
Execute kubectl commands or apply YAML configurations securely across multiple Kubernetes clusters using the Multi-Cluster MCP Server for streamlined cluster management.
Instructions
Securely run a kubectl command or apply YAML. Provide either 'command' or 'yaml'.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| cluster | No | The cluster name in a multi-cluster environment. Defaults to the hub cluster. | default |
| command | No | The full kubectl command to execute. Must start with 'kubectl'. | |
| yaml | No | YAML configuration to apply, provided as a string. |
Implementation Reference
- 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 - 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")]: - Helper function to validate that the provided command starts with 'kubectl '.
def is_valid_kubectl_command(command: str) -> bool: return command.strip().startswith("kubectl ")