PowerShell Exec MCP Server

Instructions

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

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
      """