Skip to main content
Glama

Kubectl MCP Tool

cursor_wrapper.py6.34 kB
#!/usr/bin/env python3 """ Debug wrapper for kubectl-mcp server when run by Cursor. This script captures detailed logs and handles common errors. """ import os import sys import time import traceback import logging import shutil import signal import atexit import json # Set up debug logging to a file log_dir = os.path.expanduser("~/.cursor/logs") os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, "kubectl_mcp_debug.log") logging.basicConfig( level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler(log_file), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger("cursor-wrapper") def check_environment(): """Check environment variables and file permissions.""" logger.info("----- Environment Check -----") logger.info(f"Working directory: {os.getcwd()}") logger.info(f"Python executable: {sys.executable}") logger.info(f"Python version: {sys.version}") logger.info(f"Python path: {sys.path}") # Check kubectl kubectl_path = shutil.which("kubectl") if kubectl_path: logger.info(f"kubectl found at: {kubectl_path}") else: logger.error("kubectl not found in PATH") # Check kubeconfig kube_config = os.environ.get('KUBECONFIG', '~/.kube/config') expanded_path = os.path.expanduser(kube_config) logger.info(f"KUBECONFIG: {kube_config} (expanded: {expanded_path})") if os.path.exists(expanded_path): logger.info(f"Kubernetes config exists at {expanded_path}") # Check permissions try: with open(expanded_path, 'r') as f: logger.info("Successfully opened kubeconfig file") except Exception as e: logger.error(f"Failed to read kubeconfig: {e}") else: logger.error(f"Kubernetes config does not exist at {expanded_path}") # List all environment variables logger.info("Environment variables:") for key, value in sorted(os.environ.items()): if key.lower() in ('path', 'pythonpath', 'kubeconfig'): logger.info(f"{key}: {value}") # Skip logging other env vars for security # Test kubectl try: import subprocess result = subprocess.run( ["kubectl", "version", "--client"], capture_output=True, text=True ) logger.info(f"kubectl version result: {result.returncode}") logger.info(f"kubectl version output: {result.stdout}") if result.stderr: logger.warning(f"kubectl stderr: {result.stderr}") except Exception as e: logger.error(f"Failed to run kubectl: {e}") def run_direct_mcp_server(): """Run a direct FastMCP server for better compatibility with Cursor.""" try: logger.info("Creating direct FastMCP server for Cursor") # Import the modules needed for the server from mcp.server.fastmcp import FastMCP from kubectl_mcp_tool.natural_language import process_query # Create a FastMCP server with explicit name server = FastMCP(name="kubectl-mcp") logger.info("FastMCP server created successfully") # Register the natural language processing tool @server.tool( name="process_natural_language", description="Process natural language query for kubectl", parameters={ "type": "object", "properties": { "query": { "type": "string", "description": "The natural language query to process" } }, "required": ["query"] } ) async def process_natural_language(query: str): """Process natural language query for kubectl.""" logger.info(f"Received query from Cursor: {query}") result = process_query(query) logger.info(f"Result: {result}") return json.dumps(result) # Register a simple ping tool to help with debugging @server.tool( name="kubernetes_ping", description="Ping the kubernetes server to check if it's working" ) async def kubernetes_ping(): """Simple ping tool for testing connectivity.""" logger.info("Ping received from Cursor") return "Kubernetes MCP server is working!" # Start the server using stdio transport logger.info("Starting FastMCP server with stdio transport") import asyncio # Handle clean exit def cleanup(): logger.info("Shutting down MCP server") atexit.register(cleanup) # Run the server asyncio.run(server.run_stdio_async()) except Exception as e: logger.error(f"Error running direct MCP server: {e}") logger.error(traceback.format_exc()) raise def run_mcp_server(): """Run the MCP server with error handling - old version.""" try: logger.info("Importing kubectl_mcp_tool.cli.cli...") from kubectl_mcp_tool.cli import cli logger.info("Starting MCP server...") sys.argv = [sys.argv[0], "serve"] cli.main() except ImportError as e: logger.error(f"Import error: {e}") logger.error(f"Python path: {sys.path}") # Try to import the module using an absolute path try: project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, project_root) logger.info(f"Added {project_root} to sys.path") from kubectl_mcp_tool.cli import cli sys.argv = [sys.argv[0], "serve"] cli.main() except Exception as inner_e: logger.error(f"Failed second import attempt: {inner_e}") logger.error(traceback.format_exc()) except Exception as e: logger.error(f"Error starting MCP server: {e}") logger.error(traceback.format_exc()) if __name__ == "__main__": logger.info("===== Starting kubectl-mcp debug wrapper =====") check_environment() # Use the direct FastMCP implementation for better Cursor compatibility run_direct_mcp_server()

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/rohitg00/kubectl-mcp-server'

If you have feedback or need assistance with the MCP directory API, please join our Discord server