ExecuteTerraformCommand
Run Terraform commands (init, plan, validate, apply, destroy) to manage AWS infrastructure from a specified directory with configurable variables and region settings.
Instructions
Execute Terraform workflow commands against an AWS account.
This tool runs Terraform commands (init, plan, validate, apply, destroy) in the
specified working directory, with optional variables and region settings.
Parameters:
command: Terraform command to execute
working_directory: Directory containing Terraform files
variables: Terraform variables to pass
aws_region: AWS region to use
strip_ansi: Whether to strip ANSI color codes from output
Returns:
A TerraformExecutionResult object containing command output and statusInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| command | Yes | Terraform command to execute | |
| working_directory | Yes | Directory containing Terraform files | |
| variables | No | Terraform variables to pass | |
| aws_region | No | AWS region to use | |
| strip_ansi | No | Whether to strip ANSI color codes from output |
Implementation Reference
- Core implementation of ExecuteTerraformCommand tool: executes terraform commands via subprocess, includes security pattern checks using get_dangerous_patterns, cleans ANSI output, parses JSON outputs after apply.
async def execute_terraform_command_impl( request: TerraformExecutionRequest, ) -> TerraformExecutionResult: """Execute Terraform workflow commands against an AWS account. This tool runs Terraform commands (init, plan, validate, apply, destroy) in the specified working directory, with optional variables and region settings. Parameters: request: Details about the Terraform command to execute Returns: A TerraformExecutionResult object containing command output and status """ logger.info(f"Executing 'terraform {request.command}' in {request.working_directory}") # Helper function to clean output text def clean_output_text(text: str) -> str: """Clean output text by removing or replacing problematic Unicode characters. Args: text: The text to clean Returns: Cleaned text with ASCII-friendly replacements """ if not text: return text # First remove ANSI escape sequences (color codes, cursor movement) ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') text = ansi_escape.sub('', text) # Remove C0 and C1 control characters (except common whitespace) control_chars = re.compile(r'[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F-\x9F]') text = control_chars.sub('', text) # Replace HTML entities html_entities = { '->': '->', # Replace HTML arrow '<': '<', # Less than '>': '>', # Greater than '&': '&', # Ampersand } for entity, replacement in html_entities.items(): text = text.replace(entity, replacement) # Replace box-drawing and other special Unicode characters with ASCII equivalents unicode_chars = { '\u2500': '-', # Horizontal line '\u2502': '|', # Vertical line '\u2514': '+', # Up and right '\u2518': '+', # Up and left '\u2551': '|', # Double vertical '\u2550': '-', # Double horizontal '\u2554': '+', # Double down and right '\u2557': '+', # Double down and left '\u255a': '+', # Double up and right '\u255d': '+', # Double up and left '\u256c': '+', # Double cross '\u2588': '#', # Full block '\u25cf': '*', # Black circle '\u2574': '-', # Left box drawing '\u2576': '-', # Right box drawing '\u2577': '|', # Down box drawing '\u2575': '|', # Up box drawing } for char, replacement in unicode_chars.items(): text = text.replace(char, replacement) return text # Set environment variables for AWS region if provided env = os.environ.copy() if request.aws_region: env['AWS_REGION'] = request.aws_region # Security check for command injection allowed_commands = ['init', 'plan', 'validate', 'apply', 'destroy'] if request.command not in allowed_commands: logger.error(f'Invalid Terraform command: {request.command}') return TerraformExecutionResult( command=f'terraform {request.command}', status='error', error_message=f'Invalid Terraform command: {request.command}. Allowed commands are: {", ".join(allowed_commands)}', working_directory=request.working_directory, outputs=None, ) # Check for potentially dangerous characters or command injection attempts dangerous_patterns = get_dangerous_patterns() logger.debug(f'Checking for {len(dangerous_patterns)} dangerous patterns') for pattern in dangerous_patterns: if request.variables: # Check if the pattern is in any of the variable values for var_name, var_value in request.variables.items(): if pattern in str(var_value) or pattern in str(var_name): logger.error( f'Potentially dangerous pattern detected in variable {var_name}: {pattern}' ) return TerraformExecutionResult( command=f'terraform {request.command}', status='error', error_message=f"Security violation: Potentially dangerous pattern '{pattern}' detected in variable '{var_name}'", working_directory=request.working_directory, outputs=None, ) # Build the command cmd = ['terraform', request.command] # Add auto-approve flag for apply and destroy commands to make them non-interactive if request.command in ['apply', 'destroy']: logger.info(f'Adding -auto-approve flag to {request.command} command') cmd.append('-auto-approve') # Add variables only for commands that accept them (plan, apply, destroy) if request.command in ['plan', 'apply', 'destroy'] and request.variables: logger.info(f'Adding {len(request.variables)} variables to {request.command} command') for key, value in request.variables.items(): cmd.append(f'-var={key}={value}') # Execute command try: process = subprocess.run( cmd, cwd=request.working_directory, capture_output=True, text=True, env=env ) # Prepare the result stdout = process.stdout stderr = process.stderr if process.stderr else '' # Clean output text if requested if request.strip_ansi: logger.debug('Cleaning command output text (ANSI codes and control characters)') stdout = clean_output_text(stdout) stderr = clean_output_text(stderr) result = { 'command': f'terraform {request.command}', 'status': 'success' if process.returncode == 0 else 'error', 'return_code': process.returncode, 'stdout': stdout, 'stderr': stderr, 'working_directory': request.working_directory, 'outputs': None, } # Get outputs if this was a successful apply command if request.command == 'apply' and process.returncode == 0: try: logger.info('Getting Terraform outputs') output_process = subprocess.run( ['terraform', 'output', '-json'], cwd=request.working_directory, capture_output=True, text=True, env=env, ) if output_process.returncode == 0 and output_process.stdout: # Get output and clean it if needed output_stdout = output_process.stdout if request.strip_ansi: output_stdout = clean_output_text(output_stdout) # Parse the JSON output raw_outputs = json.loads(output_stdout) # Process outputs to extract values from complex structure processed_outputs = {} for key, value in raw_outputs.items(): # Terraform outputs in JSON format have a nested structure # with 'value', 'type', and sometimes 'sensitive' if isinstance(value, dict) and 'value' in value: processed_outputs[key] = value['value'] else: processed_outputs[key] = value result['outputs'] = processed_outputs logger.info(f'Extracted {len(processed_outputs)} Terraform outputs') except Exception as e: logger.warning(f'Failed to get Terraform outputs: {e}') # Return the output return TerraformExecutionResult(**result) except Exception as e: return TerraformExecutionResult( command=f'terraform {request.command}', status='error', error_message=str(e), working_directory=request.working_directory, outputs=None, ) - awslabs/terraform_mcp_server/server.py:68-101 (registration)Registers the ExecuteTerraformCommand tool using @mcp.tool decorator, defines input parameters with descriptions, constructs request and delegates to impl.
@mcp.tool(name='ExecuteTerraformCommand') async def execute_terraform_command( command: Literal['init', 'plan', 'validate', 'apply', 'destroy'] = Field( ..., description='Terraform command to execute' ), working_directory: str = Field(..., description='Directory containing Terraform files'), variables: Optional[Dict[str, str]] = Field(None, description='Terraform variables to pass'), aws_region: Optional[str] = Field(None, description='AWS region to use'), strip_ansi: bool = Field(True, description='Whether to strip ANSI color codes from output'), ) -> TerraformExecutionResult: """Execute Terraform workflow commands against an AWS account. This tool runs Terraform commands (init, plan, validate, apply, destroy) in the specified working directory, with optional variables and region settings. Parameters: command: Terraform command to execute working_directory: Directory containing Terraform files variables: Terraform variables to pass aws_region: AWS region to use strip_ansi: Whether to strip ANSI color codes from output Returns: A TerraformExecutionResult object containing command output and status """ request = TerraformExecutionRequest( command=command, working_directory=working_directory, variables=variables, aws_region=aws_region, strip_ansi=strip_ansi, ) return await execute_terraform_command_impl(request) - Pydantic model defining input schema for TerraformExecutionRequest used in the tool.
class TerraformExecutionRequest(BaseModel): """Request model for Terraform command execution with parameters. Attributes: command: The Terraform command to execute (init, plan, validate, apply, destroy). directory: Directory containing Terraform configuration files. variables: Optional dictionary of Terraform variables to pass. aws_region: Optional AWS region to use. strip_ansi: Whether to strip ANSI color codes from command output. """ command: Literal['init', 'plan', 'validate', 'apply', 'destroy'] = Field( ..., description='Terraform command to execute' ) working_directory: str = Field(..., description='Directory containing Terraform files') variables: Optional[Dict[str, str]] = Field(None, description='Terraform variables to pass') aws_region: Optional[str] = Field(None, description='AWS region to use') strip_ansi: bool = Field(True, description='Whether to strip ANSI color codes from output') - Pydantic model defining output schema for TerraformExecutionResult returned by the tool.
class TerraformExecutionResult(BaseModel): """Result model for Terraform command execution. Attributes: command: The Terraform command that was executed. status: Execution status (success/error). return_code: The command's return code (0 for success). stdout: Standard output from the Terraform command. stderr: Standard error output from the Terraform command. working_directory: Directory where the command was executed. error_message: Optional error message if execution failed. outputs: Dictionary of output values from Terraform (for apply command). """ command: str status: Literal['success', 'error'] return_code: Optional[int] = None stdout: Optional[str] = None stderr: str = '' working_directory: str error_message: Optional[str] = None outputs: Optional[Dict[str, Any]] = Field( None, description='Terraform outputs (for apply command)' ) - Helper function providing list of dangerous patterns used in execute_terraform_command_impl for security validation against command injection.
def get_dangerous_patterns() -> List[str]: """Get a list of dangerous patterns for command injection detection. Returns: List of dangerous patterns to check for """ # Dangerous patterns that could indicate command injection attempts # Separated by platform for better organization and maintainability patterns = [ '|', ';', '&', '&&', '||', # Command chaining '>', '>>', '<', # Redirection '`', '$(', # Command substitution '--', # Double dash options 'rm', 'mv', 'cp', # Potentially dangerous commands '/bin/', '/usr/bin/', # Path references '../', './', # Directory traversal # Unix/Linux specific dangerous patterns 'sudo', # Privilege escalation 'chmod', 'chown', # File permission changes 'su', # Switch user 'bash', 'sh', 'zsh', # Shell execution 'curl', 'wget', # Network access 'ssh', 'scp', # Remote access 'eval', # Command evaluation 'exec', # Command execution 'source', # Script sourcing # Windows specific dangerous patterns 'cmd', 'powershell', 'pwsh', # Command shells 'net', # Network commands 'reg', # Registry access 'runas', # Privilege escalation 'del', 'rmdir', # File deletion 'start', # Process execution 'taskkill', # Process termination 'sc', # Service control 'schtasks', # Scheduled tasks 'wmic', # WMI commands '%SYSTEMROOT%', '%WINDIR%', # System directories '.bat', '.cmd', '.ps1', # Script files ] return patterns