PowerShell Exec MCP Server

Instructions

Args:
    logname: Name of the event log (System, Application, Security, etc.)
    newest: Number of most recent events to retrieve (default 10)
    level: Filter by event level (1: Critical, 2: Error, 3: Warning, 4: Information)
    timeout: Command timeout in seconds (1-300, default 60)

Implementation Reference

• src/server.py:132-146 (handler)

  The handler function for the 'get_event_logs' MCP tool. It constructs a PowerShell command using Get-EventLog with optional filtering by newest events and level, selects key fields, formats output as JSON, and executes it securely with timeout.

  @mcp.tool()
  async def get_event_logs(logname: str, newest: Optional[int] = 10, level: Optional[int] = None, timeout: Optional[int] = 60) -> str:
      """Get Windows event logs.
      
      Args:
          logname: Name of the event log (System, Application, Security, etc.)
          newest: Number of most recent events to retrieve (default 10)
          level: Filter by event level (1: Critical, 2: Error, 3: Warning, 4: Information)
          timeout: Command timeout in seconds (1-300, default 60)
      """
      code = f"Get-EventLog -LogName {logname} -Newest {newest}"
      if level:
          code = f"{code} | Where-Object {{ $_.EntryType -eq {level} }}"
      code = f"{code} | Select-Object TimeGenerated, EntryType, Source, Message"
      return await execute_powershell(format_json_output(code), timeout)

• src/server.py:132-132 (registration)

  The @mcp.tool() decorator registers the get_event_logs function as an MCP tool in the FastMCP server instance.

  @mcp.tool()

• src/server.py:30-34 (helper)

  Helper function to ensure PowerShell output is formatted as JSON by appending '| ConvertTo-Json' if missing.

  def format_json_output(code: str) -> str:
      """Add JSON formatting to PowerShell code if not present."""
      if not code.strip().lower().endswith('| convertto-json'):
          code = f"{code} | ConvertTo-Json"
      return code

• src/server.py:850-909 (helper)

  Core helper function that executes the constructed PowerShell code securely, with code validation against dangerous patterns, timeout enforcement, and proper error handling.

  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