Pull Request: Add Dry-Run Mode to Kubectl MCP Tool

README.md

@@ -316,6 +316,70 @@ python -m python_tests.test_all_features
 └── start_mcp_server.sh       # Server startup script
 ```
 
+---
+
+## Dry-Run Mode Verification
+
+To ensure that dry-run mode is working as expected, follow these steps:
+
+### Step 1. Start the Server in Dry-Run Mode
+
+Make sure your MCP server is running with the dry-run flag enabled. In your terminal, run:
+
+```bash
+python3 or python  -m kubectl_mcp_tool.cli serve --transport stdio --dry-run
+```
+
+You should see a log message similar to:
+
+```
+<frozen runpy>:128: RuntimeWarning: 'kubectl_mcp_tool.cli.__main__' found in sys.modules after import of package 'kubectl_mcp_tool.cli', but prior to execution of 'kubectl_mcp_tool.cli.__main__'; this may result in unpredictable behaviour
+2025-03-26 12:03:13,184 - INFO - Starting MCP server
+```
+
+This confirms that the server is up and running in dry-run mode.
+
+### Step 2. Test a Mutating Command
+
+A sample script (`create_pod.py`) is provided (located at create_pod.py filepath ) to simulate a mutating operation. Open a new terminal (or a Python shell) in the same virtual environment and run the following code:
+
+```python
+import asyncio
+from kubectl_mcp_tool.mcp_kubectl_tool import create_pod
+
+pod_yaml = """
+apiVersion: v1
+kind: Pod
+metadata:
+  name: test-pod
+spec:
+  containers:
+  - name: test-container
+    image: nginx:latest
+"""
+
+# Run the create_pod command asynchronously with a specified namespace
+result = asyncio.run(create_pod(pod_yaml, namespace="default"))
+print(result)
+```
+
+### Expected Output
+
+Since dry-run mode is enabled, instead of actually applying the pod configuration, the output should indicate that the command was simulated. For example, you might see something like:
+
+```
+Dry-run mode: Command simulated: kubectl apply -f /tmp/tmpabcd.yaml
+```
+
+This confirms that the mutating operation (create pod) was intercepted by the dry-run logic and was not executed on your Kubernetes cluster.
+
+
+```
+Run CTLR-C to  shutdown your terminal
+```
+---
+
+
 ## Contributing
 
 Contributions are welcome! Please feel free to submit a Pull Request.

create_pod.py

@@ -0,0 +1,17 @@
+import asyncio
+from kubectl_mcp_tool.mcp_kubectl_tool import create_pod
+
+pod_yaml = """
+apiVersion: v1
+kind: Pod
+metadata:
+  name: test-pod
+spec:
+  containers:
+  - name: test-container
+    image: nginx:latest
+"""
+
+# Run the create_pod command asynchronously with a specified namespace
+result = asyncio.run(create_pod(pod_yaml, namespace="default"))
+print(result)

kubectl_mcp_tool/cli/cli.py

@@ -74,6 +74,11 @@ def main():
         console_handler.setLevel(logging.DEBUG)
         logger.info("Debug mode enabled")
 
+    # Set dry-run mode if requested (for mutating commands)
+    if args.command == "serve" and args.dry_run:
+        os.environ["KUBECTL_MCP_DRY_RUN"] = "1"
+        logger.info("Dry-run mode enabled: Mutating operations will be simulated")
+
     exit_code = 0
 
     try:

kubectl_mcp_tool/mcp_kubectl_tool.py

@@ -16,6 +16,8 @@
 from rich.panel import Panel
 from rich.text import Text
 
+MUTATING_COMMANDS = {"apply", "delete", "create", "scale"}
+
 # Configure logging
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 logger = logging.getLogger("kubectl-mcp-tool")
@@ -24,8 +26,18 @@
 console = Console()
 
 def _run_kubectl_command(command: List[str]) -> str:
-    """Run a kubectl command and return the output."""
+    """Run a kubectl command and return the output.
+
+    If dry-run mode is enabled (via environment variable), and the command is mutating,
+    simulate the command execution.
+    """
+    dry_run = os.getenv("KUBECTL_MCP_DRY_RUN")
+    if dry_run and command[0] in MUTATING_COMMANDS:
+        simulated_command = "kubectl " + " ".join(command)
+        logger.info(f"Dry-run mode: Simulating command: {simulated_command}")
+        return f"Dry-run mode: Command simulated: {simulated_command}"
     try:
+
         result = subprocess.run(
             ["kubectl"] + command,
             capture_output=True,

kubectl_mcp_tool/simple_server.py

@@ -1,3 +1,115 @@
+# #!/usr/bin/env python3
+# """
+# MCP server implementation for kubectl with comprehensive Kubernetes operations support.
+# """
+
+# import asyncio
+# import logging
+# import os
+# import sys
+# from typing import Dict, Any, Optional, List# Try alternative imports (hypothetical examples)
+
+# from fastmcp.server import FastMCP
+# from fastmcp.tools import Tool
+# from fastmcp import  Parameter, ParameterType
+# from kubernetes import client, config
+
+# # Configure logging
+# log_file = os.path.expanduser("~/kubectl-mcp.log")
+# logging.basicConfig(
+#     level=logging.INFO,
+#     format='%(asctime)s - %(levelname)s - %(message)s',
+#     handlers=[
+#         logging.FileHandler(log_file),
+#         logging.StreamHandler(sys.stdout)
+#     ]
+# )
+# logger = logging.getLogger('kubectl-mcp')
+
+# class KubectlServer(FastMCP):
+#     def __init__(self):
+#         super().__init__()
+#         self.name = 'kubectl-mcp'
+#         self.version = '0.1.0'
+#         self.add_tool(self.get_pods, name="get_pods")
+#         self.add_tool(self.get_namespaces, name="get_namespaces")
+#         try:
+
+#             config.load_kube_config()
+#             self.v1 = client.CoreV1Api()
+#             self.apps_v1 = client.AppsV1Api()
+#             logger.info("Successfully initialized Kubernetes client")
+#         except Exception as e:
+#             logger.error(f"Failed to initialize Kubernetes client: {e}")
+#             self.v1 = None
+#             self.apps_v1 = None
+
+#     async def get_tools(self):
+#         return [
+#             Tool(
+#                 name="get_pods",
+#                 description="List all pods in a namespace",
+#                 parameters=[
+#                     Parameter(
+#                         name="namespace",
+#                         type=ParameterType.STRING,
+#                         description="Kubernetes namespace",
+#                         required=True
+#                     )
+#                 ]
+#             ),
+#             Tool(
+#                 name="get_namespaces",
+#                 description="List all namespaces",
+#                 parameters=[]
+#             )
+#         ]
+
+#     async def call_tool(self, tool_name: str, parameters: dict):
+#         logger.info(f'Calling tool {tool_name} with params {parameters}')
+#         try:
+#             if tool_name == "get_pods":
+#                 return await self.get_pods(parameters["namespace"])
+#             elif tool_name == "get_namespaces":
+#                 return await self.get_namespaces()
+#             else:
+#                 raise ValueError(f"Unknown tool: {tool_name}")
+#         except Exception as e:
+#             logger.error(f'Error calling tool {tool_name}: {e}')
+#             return {"error": str(e)}
+
+#     async def get_pods(self, namespace: str):
+#         if not self.v1:
+#             return {"error": "Kubernetes client not initialized"}
+#         try:
+#             pods = self.v1.list_namespaced_pod(namespace)
+#             return {"pods": [pod.metadata.name for pod in pods.items]}
+#         except Exception as e:
+#             logger.error(f"Error getting pods: {e}")
+#             return {"error": str(e)}
+
+#     async def get_namespaces(self):
+#         if not self.v1:
+#             return {"error": "Kubernetes client not initialized"}
+#         try:
+#             namespaces = self.v1.list_namespace()
+#             return {"namespaces": [ns.metadata.name for ns in namespaces.items]}
+#         except Exception as e:
+#             logger.error(f"Error getting namespaces: {e}")
+#             return {"error": str(e)}
+
+# async def main():
+#     logger.info("Starting kubectl MCP server")
+#     server = KubectlServer()
+#     try:
+#         await server.run_stdio_async()
+#     except Exception as e:
+#         logger.error(f"Server error: {e}")
+
+# if __name__ == "__main__":
+#     asyncio.run(main()) 
+
+
 #!/usr/bin/env python3
 """
 MCP server implementation for kubectl with comprehensive Kubernetes operations support.
@@ -9,8 +121,45 @@
 import sys
 from typing import Dict, Any, Optional, List
 
-from fastmcp.server import Server
-from fastmcp.models import Tool, Parameter, ParameterType
+# Use our local definitions instead of importing from fastmcp
+from enum import Enum
+
+class ParameterType(Enum):
+    STRING = "string"
+    INTEGER = "integer"
+    BOOLEAN = "boolean"
+
+class Parameter:
+    def __init__(self, name: str, type: ParameterType, description: str, required: bool):
+        self.name = name
+        self.type = type
+        self.description = description
+        self.required = required
+
+    def dict(self):
+        return {
+            "name": self.name,
+            "type": self.type.value,
+            "description": self.description,
+            "required": self.required,
+        }
+
+# Import Tool from fastmcp.models (if it exists) or define a minimal version if needed.
+try:
+    from fastmcp.models import Tool
+except ImportError:
+    class Tool:
+        def __init__(self, name: str, description: str, parameters: list):
+            self.name = name
+            self.description = description
+            self.parameters = parameters
+
+# Import FastMCP for the server base class
+try:
+    from fastmcp.server import FastMCP
+except ImportError:
+    raise ImportError("FastMCP package not found. Ensure it is installed correctly.")
+
 from kubernetes import client, config
 
 # Configure logging
@@ -25,10 +174,9 @@
 )
 logger = logging.getLogger('kubectl-mcp')
 
-class KubectlServer(Server):
+class KubectlServer(FastMCP):
     def __init__(self):
-        super().__init__()
-        self.name = 'kubectl-mcp'
+        super().__init__(name='kubectl-mcp')
         self.version = '0.1.0'
         try:
             config.load_kube_config()
@@ -98,9 +246,9 @@ async def main():
     logger.info("Starting kubectl MCP server")
     server = KubectlServer()
     try:
-        await server.stdio_server()
+        await server.run_stdio_async()
     except Exception as e:
         logger.error(f"Server error: {e}")
 
 if __name__ == "__main__":
-    asyncio.run(main()) 
+    asyncio.run(main())

requirements.txt

@@ -27,3 +27,5 @@ build>=0.10.0
 setuptools>=67.0.0
 wheel>=0.41.0
 twine>=4.0.0
+
+fastmcp>=0.4.1