AWS Terraform MCP Server

Instructions

Implementation Reference

• awslabs/terraform_mcp_server/impl/tools/execute_terragrunt_command.py:29-317 (handler)

  Main handler function that implements the Terragrunt command execution logic, including security checks, command building, subprocess execution, output processing, and result formatting.
  async def execute_terragrunt_command_impl( request: TerragruntExecutionRequest, ) -> TerragruntExecutionResult: """Execute Terragrunt workflow commands against an AWS account. This tool runs Terragrunt commands (init, plan, validate, apply, destroy, run-all) in the specified working directory, with optional variables and region settings. Parameters: request: Details about the Terragrunt command to execute Returns: A TerragruntExecutionResult object containing command output and status """ logger.info(f"Executing 'terragrunt {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 = { '-&gt;': '->', # Replace HTML arrow '&lt;': '<', # Less than '&gt;': '>', # Greater than '&amp;': '&', # 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', 'output', 'run-all'] if request.command not in allowed_commands: logger.error(f'Invalid Terragrunt command: {request.command}') return TerragruntExecutionResult( command=f'terragrunt {request.command}', status='error', error_message=f'Invalid Terragrunt command: {request.command}. Allowed commands are: {", ".join(allowed_commands)}', working_directory=request.working_directory, outputs=None, affected_dirs=None, ) # Validate that terragrunt_config is not used with run-all if request.terragrunt_config and request.command == 'run-all': logger.error('terragrunt_config cannot be used with run-all command') return TerragruntExecutionResult( command=f'terragrunt {request.command}', status='error', error_message='Invalid configuration: --terragrunt-config cannot be used with run-all command', working_directory=request.working_directory, outputs=None, affected_dirs=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 TerragruntExecutionResult( command=f'terragrunt {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, affected_dirs=None, ) # Check terragrunt_config for dangerous patterns if request.terragrunt_config and pattern in str(request.terragrunt_config): logger.error(f'Potentially dangerous pattern detected in terragrunt_config: {pattern}') return TerragruntExecutionResult( command=f'terragrunt {request.command}', status='error', error_message=f"Security violation: Potentially dangerous pattern '{pattern}' detected in terragrunt_config", working_directory=request.working_directory, outputs=None, affected_dirs=None, ) # Also check include_dirs and exclude_dirs for dangerous patterns if request.include_dirs: for dir_path in request.include_dirs: if pattern in str(dir_path): logger.error( f'Potentially dangerous pattern detected in include_dirs: {pattern}' ) return TerragruntExecutionResult( command=f'terragrunt {request.command}', status='error', error_message=f"Security violation: Potentially dangerous pattern '{pattern}' detected in include_dirs", working_directory=request.working_directory, outputs=None, affected_dirs=None, ) if request.exclude_dirs: for dir_path in request.exclude_dirs: if pattern in str(dir_path): logger.error( f'Potentially dangerous pattern detected in exclude_dirs: {pattern}' ) return TerragruntExecutionResult( command=f'terragrunt {request.command}', status='error', error_message=f"Security violation: Potentially dangerous pattern '{pattern}' detected in exclude_dirs", working_directory=request.working_directory, outputs=None, affected_dirs=None, ) # Build the command base_cmd = ['terragrunt'] # Handle run-all command differently if request.command == 'run-all': base_cmd.append('run-all') # The actual terraform command becomes the first argument # Default to 'apply' if not specified in the command base_cmd.append('apply') else: base_cmd.append(request.command) # Add auto-approve flag for apply and destroy commands to make them non-interactive if request.command in ['apply', 'destroy'] or (request.command == 'run-all'): logger.info(f'Adding -auto-approve flag to {request.command} command') base_cmd.append('-auto-approve') # Add terragrunt_config if specified and not using run-all if request.terragrunt_config: logger.info(f'Using custom terragrunt config file: {request.terragrunt_config}') base_cmd.append(f'--terragrunt-config={request.terragrunt_config}') # Add variables only for commands that accept them (plan, apply, destroy, output) if request.command in ['plan', 'apply', 'destroy', 'output', 'run-all'] and request.variables: logger.info(f'Adding {len(request.variables)} variables to {request.command} command') for key, value in request.variables.items(): base_cmd.append(f'-var={key}={value}') # Add include-dirs if specified if request.include_dirs and request.command == 'run-all': for dir_path in request.include_dirs: base_cmd.append(f'--queue-include-dir={dir_path}') # Add exclude-dirs if specified if request.exclude_dirs and request.command == 'run-all': for dir_path in request.exclude_dirs: base_cmd.append(f'--queue-exclude-dir={dir_path}') # Execute command try: process = subprocess.run( base_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) # Extract affected directories for run-all command affected_dirs = None if request.command == 'run-all': affected_dirs = [] # Look for directory paths in the output dir_pattern = re.compile(r'Module at\s+"([^"]+)"') for match in dir_pattern.finditer(stdout): affected_dirs.append(match.group(1)) result = { 'command': f'terragrunt {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, 'affected_dirs': affected_dirs, } # Get outputs if this was a successful apply or output command if ( request.command in ['apply', 'output'] or (request.command == 'run-all') ) and process.returncode == 0: try: logger.info('Getting Terragrunt outputs') output_process = subprocess.run( ['terragrunt', '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)} Terragrunt outputs') except Exception as e: logger.warning(f'Failed to get Terragrunt outputs: {e}') # Return the output return TerragruntExecutionResult(**result) except Exception as e: return TerragruntExecutionResult( command=f'terragrunt {request.command}', status='error', error_message=str(e), working_directory=request.working_directory, outputs=None, affected_dirs=None, )
• awslabs/terraform_mcp_server/server.py:103-156 (registration)

  MCP tool registration with @mcp.tool decorator, defines input parameters with descriptions, creates request model, and delegates to impl.
  @mcp.tool(name='ExecuteTerragruntCommand') async def execute_terragrunt_command( command: Literal['init', 'plan', 'validate', 'apply', 'destroy', 'output', 'run-all'] = Field( ..., description='Terragrunt command to execute' ), working_directory: str = Field(..., description='Directory containing Terragrunt 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'), include_dirs: Optional[List[str]] = Field( None, description='Directories to include in a multi-module run' ), exclude_dirs: Optional[List[str]] = Field( None, description='Directories to exclude from a multi-module run' ), run_all: bool = Field(False, description='Run command on all modules in subdirectories'), terragrunt_config: Optional[str] = Field( None, description='Path to a custom terragrunt config file (not valid with run-all)' ), ) -> TerragruntExecutionResult: """Execute Terragrunt workflow commands against an AWS account. This tool runs Terragrunt commands (init, plan, validate, apply, destroy, run-all) in the specified working directory, with optional variables and region settings. Terragrunt extends Terraform's functionality by providing features like remote state management, dependencies between modules, and the ability to execute Terraform commands on multiple modules at once. Parameters: command: Terragrunt command to execute working_directory: Directory containing Terragrunt files variables: Terraform variables to pass aws_region: AWS region to use strip_ansi: Whether to strip ANSI color codes from output include_dirs: Directories to include in a multi-module run exclude_dirs: Directories to exclude from a multi-module run run_all: Run command on all modules in subdirectories terragrunt_config: Path to a custom terragrunt config file (not valid with run-all) Returns: A TerragruntExecutionResult object containing command output and status """ request = TerragruntExecutionRequest( command=command, working_directory=working_directory, variables=variables, aws_region=aws_region, strip_ansi=strip_ansi, include_dirs=include_dirs, exclude_dirs=exclude_dirs, run_all=run_all, terragrunt_config=terragrunt_config, ) return await execute_terragrunt_command_impl(request)
• awslabs/terraform_mcp_server/models/models.py:321-382 (schema)

  Pydantic models for input (TerragruntExecutionRequest) and output (TerragruntExecutionResult) schema validation and type definitions.
  class TerragruntExecutionRequest(BaseModel): """Request model for Terragrunt command execution with parameters. Attributes: command: The Terragrunt command to execute (init, plan, validate, apply, destroy, etc.). working_directory: Directory containing Terragrunt 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. include_dirs: Optional list of directories to include in a multi-module run. exclude_dirs: Optional list of directories to exclude from a multi-module run. run_all: Whether to run the command in all subdirectories with terragrunt.hcl files. """ command: Literal['init', 'plan', 'validate', 'apply', 'destroy', 'output', 'run-all'] = Field( ..., description='Terragrunt command to execute' ) working_directory: str = Field(..., description='Directory containing Terragrunt 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') include_dirs: Optional[List[str]] = Field( None, description='Directories to include in a multi-module run' ) exclude_dirs: Optional[List[str]] = Field( None, description='Directories to exclude from a multi-module run' ) run_all: bool = Field(False, description='Run command on all modules in subdirectories') terragrunt_config: Optional[str] = Field( None, description='Path to a custom terragrunt config file (not valid with run-all)' ) class TerragruntExecutionResult(BaseModel): """Result model for Terragrunt command execution. Attributes: command: The Terragrunt command that was executed. status: Execution status (success/error). return_code: The command's return code (0 for success). stdout: Standard output from the Terragrunt command. stderr: Standard error output from the Terragrunt command. working_directory: Directory where the command was executed. error_message: Optional error message if execution failed. outputs: Dictionary of output values from Terragrunt (for apply command). affected_dirs: List of directories affected by a run-all 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='Terragrunt outputs (for apply or output command)' ) affected_dirs: Optional[List[str]] = Field( None, description='Directories affected by a run-all command' )