PowerShell Exec MCP Server

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/server.py:912-1007 (handler)

  The handler function for the run_powershell_with_progress tool. It validates input, executes PowerShell code using subprocess, provides progress updates via ctx.report_progress and ctx.info/error, handles timeouts and errors, and returns output with execution metadata.
  async def run_powershell_with_progress( code: str, timeout: Optional[int] = 60, ctx: Optional[Context] = None ) -> str: """Execute PowerShell commands with detailed progress reporting. Args: code: PowerShell code to execute timeout: Command timeout in seconds (1-300, default 60) ctx: MCP context for logging and progress reporting Returns: Command output as string with execution details """ if not ctx: # If no context provided, fall back to basic execution return await execute_powershell(code, timeout) start_time = datetime.now() try: await ctx.info("🔍 Validating PowerShell code...") # Validate timeout if not isinstance(timeout, int) or timeout < 1 or timeout > 300: await ctx.error("❌ Invalid timeout value") raise ValueError("timeout must be between 1 and 300 seconds") # Validate code if not validate_powershell_code(code): await ctx.error("❌ PowerShell code contains potentially dangerous commands") raise ValueError("PowerShell code contains potentially dangerous commands") await ctx.info("✅ Code validation passed") await ctx.info("🚀 Starting PowerShell execution...") # Report progress at start await ctx.report_progress(0, 4, "Initializing PowerShell process") # Create and run process process = await asyncio.create_subprocess_exec( "powershell", "-NoProfile", "-NonInteractive", "-Command", code, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) await ctx.report_progress(1, 4, "Process created, executing command") try: stdout, stderr = await asyncio.wait_for( process.communicate(), timeout=timeout ) await ctx.report_progress(3, 4, "Command execution completed") except asyncio.TimeoutError: process.kill() await ctx.error(f"⏰ Command timed out after {timeout} seconds") raise TimeoutError(f"Command timed out after {timeout} seconds") if process.returncode != 0: error_msg = stderr.decode() if stderr else "Command failed with no error output" await ctx.error(f"❌ PowerShell command failed: {error_msg}") raise RuntimeError(error_msg) result = stdout.decode() if stdout else "" # Calculate execution time end_time = datetime.now() execution_time = (end_time - start_time).total_seconds() await ctx.report_progress(4, 4, "Processing results") await ctx.info(f"✅ Command completed successfully in {execution_time:.2f} seconds") await ctx.info(f"📊 Output size: {len(result)} characters") # Add execution metadata to result metadata = { "execution_time_seconds": execution_time, "output_length": len(result), "exit_code": process.returncode, "timestamp": end_time.isoformat() } if result.strip(): return f"--- PowerShell Output ---\n{result}\n--- Execution Metadata ---\n{json.dumps(metadata, indent=2)}" else: return f"--- No Output Produced ---\n--- Execution Metadata ---\n{json.dumps(metadata, indent=2)}" except Exception as e: await ctx.error(f"❌ Execution failed: {str(e)}") raise
• src/server.py:850-910 (helper)

  Supporting helper function that performs the core PowerShell execution logic, including validation, subprocess management, timeout handling, and error reporting. Called by run_powershell_with_progress when no ctx is provided.
  async def execute_powershell(code: str, timeout: Optional[int] = 60, ctx: Optional[Context] = None) -> str: """Execute PowerShell commands securely. Args: code: PowerShell code to execute timeout: Command timeout in seconds (1-300, default 60) ctx: MCP context for logging and progress reporting Returns: Command output as string """ # Validate timeout if not isinstance(timeout, int) or timeout < 1 or timeout > 300: raise ValueError("timeout must be between 1 and 300 seconds") # Validate code if not validate_powershell_code(code): raise ValueError("PowerShell code contains potentially dangerous commands") if ctx: await ctx.info("Validating PowerShell code...") # Create and run process if ctx: await ctx.info("Starting PowerShell process...") process = await asyncio.create_subprocess_exec( "powershell", "-NoProfile", # Don't load profiles "-NonInteractive", # No interactive prompts "-Command", code, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) try: if ctx: await ctx.info("Executing command...") stdout, stderr = await asyncio.wait_for( process.communicate(), timeout=timeout ) except asyncio.TimeoutError: process.kill() if ctx: await ctx.error(f"Command timed out after {timeout} seconds") raise TimeoutError(f"Command timed out after {timeout} seconds") if process.returncode != 0: error_msg = stderr.decode() if stderr else "Command failed with no error output" if ctx: await ctx.error(f"PowerShell command failed: {error_msg}") raise RuntimeError(error_msg) result = stdout.decode() if stdout else "" if ctx: await ctx.info(f"Command completed successfully, returned {len(result)} characters") return result
• src/server.py:36-61 (helper)

  Helper function that validates PowerShell code against a list of dangerous patterns to prevent harmful commands.
  def validate_powershell_code(code: str) -> bool: """ Validate PowerShell code for potentially harmful commands. Args: code: The PowerShell code to validate Returns: bool: True if code passes validation """ dangerous_patterns = [ r"rm\s+(-r|-f|/s)*\s*/", # Dangerous recursive deletes r"format\s+[a-z]:", # Drive formatting r"Stop-Computer", # System shutdown r"Restart-Computer", # System restart r"Remove-Item.*-Recurse", # Recursive deletion r"Invoke-Expression", # Dynamic code execution r"iex", # Alias for Invoke-Expression r"Start-Process", # Starting new processes r"New-Service", # Creating services r"Set-Service", # Modifying services r"net\s+user", # User account manipulation ] return not any(re.search(pattern, code, re.IGNORECASE) for pattern in dangerous_patterns)