run_powershell_with_progress
Execute PowerShell commands with real-time progress tracking and detailed execution reporting for enterprise automation and system management tasks.
Instructions
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 detailsInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| code | Yes | ||
| timeout | No | ||
| ctx | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/server.py:912-1007 (handler)The primary handler for the 'run_powershell_with_progress' tool. It provides detailed progress reporting using ctx.report_progress(), validation, timeout handling, execution via subprocess, and returns formatted 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 actual PowerShell subprocess execution, validation, and error handling. Called by run_powershell_with_progress when no ctx is provided or as fallback.
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 used by the handler to validate PowerShell code against a list of dangerous patterns before execution.
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) - src/server.py:912-912 (registration)The @mcp.tool() decorator registers the 'run_powershell_with_progress' function as an MCP tool in the FastMCP server.
async def run_powershell_with_progress( - src/server.py:913-926 (schema)Input schema defined by function parameters with type hints and comprehensive docstring describing args and return value. FastMCP infers JSON schema from this.
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 """