import os import asyncio import json import re import uvicorn # Used for running the FastAPI application from typing import Dict, List, Optional, Any from dataclasses import dataclass from datetime import datetime import logging # FastAPI imports from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse # MCP Server imports import mcp.types as types from mcp.server import Server, NotificationOptions, InitializationOptions # Configure logging # Set a very low level for debugging if needed, usually INFO for production logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # ============================================================================ # TERRAFORM CODE TEMPLATES AND GENERATORS (Existing code, slightly adjusted formatting for consistency) # ============================================================================ class TerraformTemplates: """Static Terraform code templates for different resource types""" @staticmethod def get_provider_block(provider: str, version: str = None) -> str: """Generate provider configuration block""" versions = { "aws": "~> 5.0", "azurerm": "~> 3.0", "google": "~> 4.0", "kubernetes": "~> 2.0" } version = version or versions.get(provider, "latest") if provider == "aws": return f'''terraform {{ required_version = ">= 1.0" required_providers {{ aws = {{ source = "hashicorp/aws" version = "{version}" }} }} }} provider "aws" {{ region = var.aws_region default_tags {{ tags = {{ Project = var.project_name ManagedBy = "Terraform" Environment = var.environment }} }} }}''' elif provider == "azurerm": return f'''terraform {{ required_version = ">= 1.0" required_providers {{ azurerm = {{ source = "hashicorp/azurerm" version = "{version}" }} }} }} provider "azurerm" {{ features {{}} }}''' elif provider == "kubernetes": return f'''terraform {{ required_version = ">= 1.0" required_providers {{ kubernetes = {{ source = "hashicorp/kubernetes" version = "{version}" }} }} }} provider "kubernetes" {{ config_path = "~/.kube/config" }}''' else: return f'''terraform {{ required_version = ">= 1.0" required_providers {{ {provider} = {{ source = "hashicorp/{provider}" version = "{version}" }} }} }} provider "{provider}" {{ # Configure the {provider} provider }}''' @staticmethod def get_common_variables(provider: str) -> str: """Generate common variables for a provider""" if provider == "aws": return '''variable "aws_region" { description = "AWS region" type = string default = "us-west-2" } variable "project_name" { description = "Name of the project" type = string } variable "environment" { description = "Environment name" type = string default = "dev" validation { condition = contains(["dev", "staging", "prod"], var.environment) error_message = "Environment must be dev, staging, or prod." } }''' elif provider == "azurerm": return '''variable "location" { description = "Azure region" type = string default = "East US" } variable "project_name" { description = "Name of the project" type = string } variable "environment" { description = "Environment name" type = string default = "dev" validation { condition = contains(["dev", "staging", "prod"], var.environment) error_message = "Environment must be dev, staging, or prod." } }''' else: return '''variable "project_name" { description = "Name of the project" type = string } variable "environment" { description = "Environment name" type = string default = "dev" }''' class TerraformCodeGenerator: """Pure Terraform code generation without external dependencies""" def __init__(self): self.templates = TerraformTemplates() def generate_vpc_config(self, name: str, options: Dict) -> str: """Generate AWS VPC configuration""" cidr = options.get("cidr", "10.0.0.0/16") azs = options.get("azs", ["us-west-2a", "us-west-2b"]) enable_nat = options.get("enable_nat_gateway", True) return f'''# {name} VPC Infrastructure Configuration {self.templates.get_provider_block("aws")} {self.templates.get_common_variables("aws")} variable "vpc_cidr" {{ description = "CIDR block for VPC" type = string default = "{cidr}" validation {{ condition = can(cidrhost(var.vpc_cidr, 0)) error_message = "VPC CIDR must be a valid IPv4 CIDR block." }} }} # VPC Module module "{name}_vpc" {{ source = "terraform-aws-modules/vpc/aws" name = "${{var.environment}}-{name}-vpc" cidr = var.vpc_cidr azs = {azs} public_subnets = [for k, v in {azs} : cidrsubnet(var.vpc_cidr, 8, k)] private_subnets = [for k, v in {azs} : cidrsubnet(var.vpc_cidr, 8, k + 10)] enable_nat_gateway = {str(enable_nat).lower()} enable_vpn_gateway = false enable_dns_hostnames = true enable_dns_support = true public_subnet_tags = {{ Type = "Public" Tier = "Web" }} private_subnet_tags = {{ Type = "Private" Tier = "Application" }} tags = {{ Name = "${{var.environment}}-{name}-vpc" }} }} # Outputs output "vpc_id" {{ description = "ID of the VPC" value = module.{name}_vpc.vpc_id }} output "vpc_cidr_block" {{ description = "CIDR block of the VPC" value = module.{name}_vpc.vpc_cidr_block }} output "public_subnets" {{ description = "List of IDs of public subnets" value = module.{name}_vpc.public_subnets }} output "private_subnets" {{ description = "List of IDs of private subnets" value = module.{name}_vpc.private_subnets }} output "nat_gateway_ips" {{ description = "List of public Elastic IPs for NAT Gateway" value = module.{name}_vpc.nat_public_ips }}''' def generate_ec2_config(self, name: str, options: Dict) -> str: """Generate AWS EC2 configuration""" instance_type = options.get("instance_type", "t3.micro") key_pair = options.get("key_pair_name", f"{name}-key") ami_filter = options.get("ami_filter", "amzn2-ami-hvm-*-x86_64-gp2") return f'''# {name} EC2 Instance Configuration {self.templates.get_provider_block("aws")} {self.templates.get_common_variables("aws")} variable "instance_type" {{ description = "EC2 instance type" type = string default = "{instance_type}" validation {{ condition = contains([ "t3.nano", "t3.micro", "t3.small", "t3.medium", "t3.large", "t3.xlarge", "t3.2xlarge", "m5.large", "m5.xlarge" ], var.instance_type) error_message = "Instance type must be a valid instance type." }} }} variable "key_pair_name" {{ description = "Name of the AWS key pair" type = string default = "{key_pair}" }} # Data sources data "aws_ami" "app_ami" {{ most_recent = true owners = ["amazon"] filter {{ name = "name" values = ["{ami_filter}"] }} filter {{ name = "virtualization-type" values = ["hvm"] }} }} data "aws_vpc" "default" {{ default = true }} data "aws_subnets" "default" {{ filter {{ name = "vpc-id" values = [data.aws_vpc.default.id] }} }} # EC2 Instance Module module "{name}_instance" {{ source = "terraform-aws-modules/ec2-instance/aws" name = "{name}-instance" instance_type = var.instance_type ami = data.aws_ami.app_ami.id key_name = var.key_pair_name monitoring = true vpc_security_group_ids = [aws_security_group.{name}_sg.id] subnet_id = data.aws_subnets.default.ids[0] create_iam_instance_profile = true iam_role_description = "IAM role for {name} EC2 instance" iam_role_policies = {{ AmazonSSMManagedInstanceCore = "arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore" }} user_data_base64 = base64encode(local.user_data) root_block_device = [ {{ encrypted = true volume_type = "gp3" throughput = 200 volume_size = 20 tags = {{ Name = "{name}-root-block" }} }}, ] tags = {{ Name = "{name}-instance" }} }} # Security Group resource "aws_security_group" "{name}_sg" {{ name_prefix = "{name}-sg" description = "Security group for {name} instance" vpc_id = data.aws_vpc.default.id ingress {{ description = "SSH" from_port = 22 to_port = 22 protocol = "tcp" cidr_blocks = ["10.0.0.0/8"] }} ingress {{ description = "HTTP" from_port = 80 to_port = 80 protocol = "tcp" cidr_blocks = ["0.0.0.0/0"] }} ingress {{ description = "HTTPS" from_port = 443 to_port = 443 protocol = "tcp" cidr_blocks = ["0.0.0.0/0"] }} egress {{ from_port = 0 to_port = 0 protocol = "-1" cidr_blocks = ["0.0.0.0/0"] }} tags = {{ Name = "{name}-security-group" }} }} # User data script locals {{ user_data = <<-EOT #!/bin/bash yum update -y yum install -y httpd systemctl start httpd systemctl enable httpd echo "

Hello from {name}

" > /var/www/html/index.html # Install CloudWatch agent wget https://s3.amazonaws.com/amazoncloudwatch-agent/amazon_linux/amd64/latest/amazon-cloudwatch-agent.rpm rpm -U ./amazon-cloudwatch-agent.rpm EOT }} # Outputs output "instance_id" {{ description = "ID of the EC2 instance" value = module.{name}_instance.id }} output "instance_public_ip" {{ description = "Public IP address of the EC2 instance" value = module.{name}_instance.public_ip }} output "instance_private_ip" {{ description = "Private IP address of the EC2 instance" value = module.{name}_instance.private_ip }} output "security_group_id" {{ description = "ID of the security group" value = aws_security_group.{name}_sg.id }}''' def generate_s3_config(self, name: str, options: Dict) -> str: """Generate AWS S3 bucket configuration""" versioning = options.get("versioning", True) encryption = options.get("encryption", True) public_access = options.get("block_public_access", True) return f'''# {name} S3 Bucket Configuration {self.templates.get_provider_block("aws")} {self.templates.get_common_variables("aws")} variable "bucket_name" {{ description = "Name of the S3 bucket (will be prefixed with random string)" type = string default = "{name}-bucket" }} variable "enable_versioning" {{ description = "Enable versioning for the S3 bucket" type = bool default = {str(versioning).lower()} }} # Random suffix for bucket name uniqueness resource "random_string" "bucket_suffix" {{ length = 8 special = false upper = false }} # S3 Bucket Module module "{name}_s3_bucket" {{ source = "terraform-aws-modules/s3-bucket/aws" bucket = "${{var.bucket_name}}-${{random_string.bucket_suffix.result}}" # Versioning versioning = {{ enabled = var.enable_versioning }} # Server-side encryption server_side_encryption_configuration = {{ rule = {{ apply_server_side_encryption_by_default = {{ sse_algorithm = "AES256" }} }} }} # Public access block block_public_acls = {str(public_access).lower()} block_public_policy = {str(public_access).lower()} ignore_public_acls = {str(public_access).lower()} restrict_public_buckets = {str(public_access).lower()} # Lifecycle configuration lifecycle_configuration = {{ rule = [ {{ id = "delete_incomplete_multipart_uploads" status = "Enabled" abort_incomplete_multipart_upload = {{ days_after_initiation = 7 }} }}, {{ id = "transition_to_ia" status = "Enabled" transition = [ {{ days = 30 storage_class = "STANDARD_IA" }}, {{ days = 90 storage_class = "GLACIER" }} ] }} ] }} tags = {{ Name = "${{var.bucket_name}}-${{random_string.bucket_suffix.result}}" }} }} # Bucket policy for additional security resource "aws_s3_bucket_policy" "{name}_bucket_policy" {{ bucket = module.{name}_s3_bucket.s3_bucket_id policy = jsonencode({{ Version = "2012-10-17" Statement = [ {{ Sid = "DenyInsecureConnections" Effect = "Deny" Principal = "*" Action = "s3:*" Resource = [ module.{name}_s3_bucket.s3_bucket_arn, "${{module.{name}_s3_bucket.s3_bucket_arn}}/*" ] Condition = {{ Bool = {{ "aws:SecureTransport" = "false" }} }} }} ] }}) }} # Outputs output "bucket_name" {{ description = "Name of the S3 bucket" value = module.{name}_s3_bucket.s3_bucket_id }} output "bucket_arn" {{ description = "ARN of the S3 bucket" value = module.{name}_s3_bucket.s3_bucket_arn }} output "bucket_domain_name" {{ description = "Bucket domain name" value = module.{name}_s3_bucket.s3_bucket_bucket_domain_name }} output "bucket_regional_domain_name" {{ description = "Bucket regional domain name" value = module.{name}_s3_bucket.s3_bucket_bucket_regional_domain_name }}''' def generate_eks_config(self, name: str, options: Dict) -> str: """Generate AWS EKS cluster configuration""" node_instance_type = options.get("node_instance_type", "t3.medium") min_nodes = options.get("min_nodes", 1) max_nodes = options.get("max_nodes", 3) desired_nodes = options.get("desired_nodes", 2) k8s_version = options.get("kubernetes_version", "1.28") return f'''# {name} EKS Cluster Configuration {self.templates.get_provider_block("aws")} {self.templates.get_common_variables("aws")} variable "cluster_name" {{ description = "Name of the EKS cluster" type = string default = "{name}-eks-cluster" }} variable "kubernetes_version" {{ description = "Kubernetes version" type = string default = "{k8s_version}" }} variable "node_instance_type" {{ description = "Instance type for EKS nodes" type = string default = "{node_instance_type}" }} variable "node_group_min_size" {{ description = "Minimum number of nodes" type = number default = {min_nodes} }} variable "node_group_max_size" {{ description = "Maximum number of nodes" type = number default = {max_nodes} }} variable "node_group_desired_size" {{ description = "Desired number of nodes" type = number default = {desired_nodes} }} # Data sources data "aws_availability_zones" "available" {{ filter {{ name = "opt-in-status" values = ["opt-in-not-required"] }} }} data "aws_caller_identity" "current" {{}} # VPC for EKS module "vpc" {{ source = "terraform-aws-modules/vpc/aws" name = "${{var.cluster_name}}-vpc" cidr = "10.0.0.0/16" azs = slice(data.aws_availability_zones.available.names, 0, 3) public_subnets = ["10.0.1.0/24", "10.0.2.0/24", "10.0.3.0/24"] private_subnets = ["10.0.11.0/24", "10.0.12.0/24", "10.0.13.0/24"] enable_nat_gateway = true single_nat_gateway = true enable_dns_hostnames = true enable_dns_support = true public_subnet_tags = {{ "kubernetes.io/role/elb" = "1" }} private_subnet_tags = {{ "kubernetes.io/role/internal-elb" = "1" }} tags = {{ "kubernetes.io/cluster/${{var.cluster_name}}" = "shared" }} }} # EKS Cluster module "eks" {{ source = "terraform-aws-modules/eks/aws" cluster_name = var.cluster_name cluster_version = var.kubernetes_version vpc_id = module.vpc.vpc_id subnet_ids = module.vpc.private_subnets cluster_endpoint_public_access = true cluster_addons = {{ coredns = {{ most_recent = true }} kube-proxy = {{ most_recent = true }} vpc-cni = {{ most_recent = true }} aws-ebs-csi-driver = {{ most_recent = true }} }} eks_managed_node_groups = {{ main = {{ name = "${{var.cluster_name}}-nodes" instance_types = [var.node_instance_type] min_size = var.node_group_min_size max_size = var.node_group_max_size desired_size = var.node_group_desired_size disk_size = 50 labels = {{ Environment = var.environment NodeGroup = "main" }} tags = {{ Name = "${{var.cluster_name}}-node" }} }} }} # Cluster access entry access_entries = {{ admin = {{ kubernetes_groups = [] principal_arn = data.aws_caller_identity.current.arn policy_associations = {{ admin = {{ policy_arn = "arn:aws:eks::aws:cluster-access-policy/AmazonEKSClusterAdminPolicy" access_scope = {{ type = "cluster" }} }} }} }} }} tags = {{ Environment = var.environment Terraform = "true" }} }} # Outputs output "cluster_endpoint" {{ description = "Endpoint for EKS control plane" value = module.eks.cluster_endpoint }} output "cluster_security_group_id" {{ description = "Security group ID attached to the cluster control plane" value = module.eks.cluster_security_group_id }} output "cluster_name" {{ description = "Kubernetes Cluster Name" value = module.eks.cluster_name }} output "cluster_arn" {{ description = "The Amazon Resource Name (ARN) of the cluster" value = module.eks.cluster_arn }} output "cluster_certificate_authority_data" {{ description = "Base64 encoded certificate data required to communicate with the cluster" value = module.eks.cluster_certificate_authority_data }} output "configure_kubectl" {{ description = "Configure kubectl: make sure you're logged in with the correct AWS profile and run the following command to update your kubeconfig" value = "aws eks --region ${{var.aws_region}} update-kubeconfig --name ${{module.eks.cluster_name}}" }} output "vpc_id" {{ description = "ID of the VPC where the cluster security group belongs" value = module.vpc.vpc_id }}''' def generate_azure_vnet_config(self, name: str, options: Dict) -> str: """Generate Azure Virtual Network configuration""" address_space = options.get("address_space", ["10.0.0.0/16"]) return f'''# {name} Azure Virtual Network Configuration {self.templates.get_provider_block("azurerm")} {self.templates.get_common_variables("azurerm")} variable "address_space" {{ description = "Address space for the virtual network" type = list(string) default = {address_space} }} # Resource Group resource "azurerm_resource_group" "{name}_rg" {{ name = "${{var.environment}}-{name}-rg" location = var.location tags = {{ Environment = var.environment Project = var.project_name }} }} # Virtual Network Module module "{name}_vnet" {{ source = "Azure/vnet/azurerm" resource_group_name = azurerm_resource_group.{name}_rg.name location = azurerm_resource_group.{name}_rg.location vnet_name = "${{var.environment}}-{name}-vnet" address_space = var.address_space subnet_prefixes = ["10.0.1.0/24", "10.0.2.0/24", "10.0.3.0/24"] subnet_names = ["subnet1", "subnet2", "subnet3"] tags = {{ Environment = var.environment Project = var.project_name }} }} # Outputs output "vnet_id" {{ description = "The ID of the Virtual Network" value = module.{name}_vnet.vnet_id }} output "vnet_name" {{ description = "The name of the Virtual Network" value = module.{name}_vnet.vnet_name }} output "subnet_ids" {{ description = "The IDs of the subnets" value = module.{name}_vnet.vnet_subnets }} output "resource_group_name" {{ description = "The name of the resource group" value = azurerm_resource_group.{name}_rg.name }}''' # ============================================================================ # TERRAFORM VALIDATION AND UTILITIES (Existing code) # ============================================================================ class TerraformValidator: """Terraform configuration validation utilities""" @staticmethod def validate_syntax(config: str) -> List[str]: """Basic Terraform syntax validation""" issues = [] # Check for terraform block if not re.search(r'terraform\s*\{', config): issues.append("⚠️ Missing terraform {} block") # Check for provider block if not re.search(r'provider\s+"[^"]+"\s*\{', config): issues.append("⚠️ Missing provider configuration") # Check for version constraints if not re.search(r'version\s*=\s*"[^"]+"', config): issues.append("⚠️ Missing provider version constraints") # Check resource naming conventions resources = re.findall(r'resource\s+"([^"]+)"\s+"([^"]+)"', config) for resource_type, resource_name in resources: if re.search(r'[A-Z-]', resource_name): issues.append(f"⚠️ Resource '{resource_name}' should use lowercase with underscores") # Check for variable descriptions variables = re.findall(r'variable\s+"([^"]+)"\s*\{', config) for var_name in variables: var_block = re.search(f'variable\\s+"{var_name}"\\s*\\{{([^}}]+)\\}}', config, re.DOTALL) if var_block and 'description' not in var_block.group(1): issues.append(f"💡 Variable '{var_name}' missing description") return issues @staticmethod def validate_best_practices(config: str) -> List[str]: """Check Terraform best practices""" suggestions = [] # Check for tags if 'tags' not in config.lower(): suggestions.append("💡 Consider adding tags to resources") # Check for remote state if 'backend' not in config: suggestions.append("💡 Consider using remote state storage") # Check for data sources vs hardcoded values if re.search(r'ami-[a-f0-9]+', config): suggestions.append("💡 Consider using data sources instead of hardcoded AMI IDs") # Check for variable validation if 'variable' in config and 'validation' not in config: suggestions.append("💡 Consider adding validation rules to variables") # Check for outputs if 'resource' in config and 'output' not in config: suggestions.append("💡 Consider adding outputs for important resource attributes") return suggestions class TerraformWorkflows: """Terraform deployment workflow generators""" @staticmethod def get_basic_workflow() -> List[str]: """Get basic Terraform workflow commands""" return [ "terraform init", "terraform validate", "terraform fmt", "terraform plan", "terraform apply", "terraform output" ] @staticmethod def get_production_workflow(backend_type: str = "s3") -> List[str]: """Get production-ready workflow commands""" commands = [ f"# Production Terraform Workflow with {backend_type.upper()} backend", "", "# 1. Initialize with backend configuration", "terraform init", "", "# 2. Validate configuration", "terraform validate", "", "# 3. Format code", "terraform fmt", "", "# 4. Security scan (optional)", "# tfsec .", "", "# 5. Plan and save", "terraform plan -out=tfplan", "", "# 6. Review plan output carefully", "terraform show tfplan", "", "# 7. Apply saved plan", "terraform apply tfplan", "", "# 8. Verify outputs", "terraform output", "", "# 9. Clean up plan file", "rm tfplan" ] if backend_type == "s3": commands.insert(4, "# Configure S3 backend if not done") commands.insert(5, "# terraform init -backend-config=backend.hcl") commands.insert(6, "") return commands @staticmethod def get_module_workflow() -> List[str]: """Get module development workflow""" return [ "# Module Development Workflow", "", "# 1. Initialize module", "terraform init", "", "# 2. Validate module", "terraform validate", "", "# 3. Format module code", "terraform fmt -recursive", "", "# 4. Generate documentation", "# terraform-docs markdown . > README.md", "", "# 5. Test module (in examples/ directory)", "cd examples/basic", "terraform init", "terraform plan", "", "# 6. Clean up test", "terraform destroy", "cd ../..", "", "# 7. Tag version", "git tag v1.0.0", "git push --tags" ] class TerraformModuleConverter: """Convert Terraform configurations to reusable modules""" @staticmethod def extract_variables(config: str) -> str: """Extract hardcoded values and convert to variables""" variables = [] # Common patterns to extract patterns = { r'"(t[23]\.[a-z]+)"': ('instance_type', 'string', 'EC2 instance type'), r'"(\d+\.\d+\.\d+\.\d+/\d+)"': ('cidr_block', 'string', 'CIDR block'), r'"(us-[a-z]+-\d+[a-z]?)"': ('aws_region', 'string', 'AWS region'), r'"(eastus|westus|centralus)"': ('location', 'string', 'Azure location'), } for pattern, (var_name, var_type, description) in patterns.items(): if re.search(pattern, config): variables.append(f'''variable "{var_name}" {{ description = "{description}" type = {var_type} default = # Add appropriate default }}''') return '\n\n'.join(variables) @staticmethod def generate_outputs(config: str, resource_name: str) -> str: """Generate outputs for a module""" outputs = [] # Extract resource types from config resources = re.findall(r'resource\s+"([^"]+)"\s+"([^"]+)"', config) modules = re.findall(r'module\s+"([^"]+)"\s*\{', config) for resource_type, name in resources: if 'aws_instance' in resource_type: outputs.extend([ f'output "{name}_id" {{\n description = "ID of the EC2 instance"\n value = {resource_type}.{name}.id\n}}', f'output "{name}_public_ip" {{\n description = "Public IP of the instance"\n value = {resource_type}.{name}.public_ip\n}}' ]) elif 'aws_s3_bucket' in resource_type: outputs.extend([ f'output "{name}_name" {{\n description = "Name of the S3 bucket"\n value = {resource_type}.{name}.id\n}}', f'output "{name}_arn" {{\n description = "ARN of the S3 bucket"\n value = {resource_type}.{name}.arn\n}}' ]) for module_name in modules: outputs.append(f'output "{module_name}_outputs" {{\n description = "All outputs from {module_name} module"\n value = module.{module_name}\n}}') return '\n\n'.join(outputs) # ============================================================================ # TERRAFORM AGENT HANDLERS (Core Logic, adapted from CleanTerraformMCPServer) # This class now holds the business logic, separate from the MCP Server instance # and FastAPI endpoints. # ============================================================================ class TerraformAgentHandlers: """Handles the core logic for Terraform tool operations.""" def __init__(self): self.code_generator = TerraformCodeGenerator() self.validator = TerraformValidator() self.workflows = TerraformWorkflows() self.module_converter = TerraformModuleConverter() async def _handle_generate_config(self, args: dict) -> list[types.TextContent]: """Handle Terraform configuration generation""" resource_type = args.get("resource_type") provider = args.get("provider") name = args.get("name") options = args.get("options", {}) # Generate configuration based on resource type if resource_type == "vpc" and provider == "aws": config = self.code_generator.generate_vpc_config(name, options) elif resource_type == "ec2" and provider == "aws": config = self.code_generator.generate_ec2_config(name, options) elif resource_type == "s3" and provider == "aws": config = self.code_generator.generate_s3_config(name, options) elif resource_type == "eks" and provider == "aws": config = self.code_generator.generate_eks_config(name, options) elif resource_type == "vnet" and provider == "azurerm": config = self.code_generator.generate_azure_vnet_config(name, options) else: return [types.TextContent(type="text", text=f"❌ Configuration generation for {resource_type} on {provider} not yet implemented")] # Basic validation validation_issues = self.validator.validate_syntax(config) best_practices = self.validator.validate_best_practices(config) response = f"""# 🚀 Terraform Configuration Generated ## 📋 Summary - **Resource Type**: {resource_type.upper()} - **Provider**: {provider} - **Name**: {name} - **Generated**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ## 🔧 Configuration ```hcl {config} ``` ## ✅ Validation Results """ if validation_issues: response += "### Issues Found:\n" for issue in validation_issues: response += f"- {issue}\n" else: response += "- ✅ No syntax issues found\n" if best_practices: response += "\n### Best Practice Suggestions:\n" for suggestion in best_practices: response += f"- {suggestion}\n" response += f""" ## 🚀 Next Steps 1. Save configuration to `main.tf` 2. Run `terraform init` to initialize 3. Run `terraform plan` to review changes 4. Run `terraform apply` to create resources ## 📦 Recommended Module Structure ``` {name}-infrastructure/ ├── main.tf # Main configuration ├── variables.tf # Input variables ├── outputs.tf # Output values ├── versions.tf # Provider versions └── README.md # Documentation ``` """ return [types.TextContent(type="text", text=response)] async def _handle_validate_config(self, args: dict) -> list[types.TextContent]: """Handle configuration validation""" config_content = args.get("config_content") check_best_practices = args.get("check_best_practices", True) strict_mode = args.get("strict_mode", False) # Syntax validation syntax_issues = self.validator.validate_syntax(config_content) # Best practices validation best_practice_suggestions = [] if check_best_practices: best_practice_suggestions = self.validator.validate_best_practices(config_content) response = "# 📋 Terraform Configuration Validation\n\n" # Syntax validation results response += "## ✅ Syntax Validation\n\n" if not syntax_issues: response += "✅ **No syntax issues found**\n\n" else: response += f"Found {len(syntax_issues)} syntax issues:\n\n" for issue in syntax_issues: response += f"- {issue}\n" response += "\n" # Best practices results if check_best_practices: response += "## 💡 Best Practices Review\n\n" if not best_practice_suggestions: response += "✅ **Configuration follows best practices**\n\n" else: response += f"Found {len(best_practice_suggestions)} suggestions:\n\n" for suggestion in best_practice_suggestions: response += f"- {suggestion}\n" response += "\n" # Overall score total_issues = len(syntax_issues) total_suggestions = len(best_practice_suggestions) if check_best_practices else 0 if total_issues == 0 and total_suggestions == 0: response += "## 🏆 Overall Assessment\n\n" response += "**Excellent!** Your configuration is clean and follows best practices.\n" elif total_issues == 0: response += "## ✅ Overall Assessment\n\n" response += "**Good!** No syntax errors, but consider the suggestions above.\n" else: response += "## ⚠️ Overall Assessment\n\n" response += "**Needs attention.** Please address the syntax issues before deployment.\n" response += "\n## 🔧 Recommended Actions\n\n" if total_issues > 0: response += "1. **Fix syntax issues** - Address all syntax problems first\n" response += "2. **Run terraform validate** - Validate with Terraform CLI\n" response += "3. **Run terraform fmt** - Format the code consistently\n" if check_best_practices and total_suggestions > 0: response += "4. **Review best practices** - Consider implementing suggestions\n" response += "5. **Test thoroughly** - Always test in development environment first\n" return [types.TextContent(type="text", text=response)] async def _handle_deployment_workflow(self, args: dict) -> list[types.TextContent]: """Handle deployment workflow generation""" workflow_type = args.get("workflow_type", "basic") backend_type = args.get("backend_type", "local") environment = args.get("environment", "development") if workflow_type == "basic": commands = self.workflows.get_basic_workflow() elif workflow_type == "production": commands = self.workflows.get_production_workflow(backend_type) elif workflow_type == "module_development": commands = self.workflows.get_module_workflow() else: return [types.TextContent(type="text", text=f"❌ Unknown workflow type: {workflow_type}")] response = f"""# 🚀 Terraform Deployment Workflow ## 📋 Configuration - **Workflow Type**: {workflow_type.title()} - **Backend**: {backend_type.upper()} - **Environment**: {environment.title()} - **Generated**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ## 🔄 Commands ```bash {chr(10).join(commands)} ``` ## 🛡️ Security Considerations """ if environment == "production": response += """### Production Environment - ✅ Use remote state storage - ✅ Enable state locking - ✅ Implement approval workflows - ✅ Run security scans - ✅ Backup state files regularly - ✅ Use least privilege access - ✅ Monitor all changes """ response += """### General Security - 🔒 Never commit sensitive data to version control - 🔒 Use environment variables for secrets - 🔒 Enable encryption for state files - 🔒 Implement proper IAM policies - 🔒 Regular security reviews ## 📚 Additional Resources - [Terraform Best Practices](https://www.terraform.io/docs/cloud/guides/recommended-practices/index.html) - [State Management](https://www.terraform.io/docs/language/state/index.html) - [Security Guidelines](https://www.terraform.io/docs/cloud/guides/recommended-practices/part1.html) """ return [types.TextContent(type="text", text=response)] async def _handle_convert_to_module(self, args: dict) -> list[types.TextContent]: """Handle module conversion""" config_content = args.get("config_content") module_name = args.get("module_name") extract_variables = args.get("extract_variables", True) generate_examples = args.get("generate_examples", True) response = f"""# 📦 Module Conversion: {module_name} Converting Terraform configuration to a reusable module structure. ## 📁 Module Structure ``` modules/{module_name}/ ├── main.tf # Main configuration ├── variables.tf # Input variables ├── outputs.tf # Output values ├── versions.tf # Provider requirements ├── README.md # Documentation └── examples/ └── basic/ ├── main.tf # Example usage └── README.md # Example documentation ``` """ # Generate variables.tf if extract_variables: variables = self.module_converter.extract_variables(config_content) response += f"""## variables.tf ```hcl {variables} variable "tags" {{ description = "A map of tags to assign to the resource" type = map(string) default = {{}} }} ``` """ # Generate main.tf (modified config) response += f"""## main.tf ```hcl {config_content} ``` """ # Generate outputs.tf outputs = self.module_converter.generate_outputs(config_content, module_name) response += f"""## outputs.tf ```hcl {outputs} ``` """ # Generate versions.tf response += f"""## versions.tf ```hcl terraform {{ required_version = ">= 1.0" required_providers {{ aws = {{ source = "hashicorp/aws" version = ">= 5.0" }} }} }} ``` """ # Generate example usage if generate_examples: response += f"""## examples/basic/main.tf ```hcl module "{module_name}" {{ source = "../../" project_name = "example" environment = "dev" # Add module-specific variables here tags = {{ Environment = "development" Project = "example" }} }} output "{module_name}_outputs" {{ description = "All outputs from the {module_name} module" value = module.{module_name} }} ``` """ response += f"""## 🚀 Next Steps 1. **Create module directory structure** 2. **Save each file in appropriate location** 3. **Test module with examples** 4. **Add validation rules to variables** 5. **Update documentation** 6. **Version tag for release** ## 🧪 Testing the Module ```bash # Navigate to example cd examples/basic # Initialize and test terraform init terraform plan terraform apply # Clean up terraform destroy ``` """ return [types.TextContent(type="text", text=response)] async def _handle_format_code(self, args: dict) -> list[types.TextContent]: """Handle code formatting""" config_content = args.get("config_content") sort_blocks = args.get("sort_blocks", True) # Basic formatting improvements formatted_config = config_content # Normalize indentation (2 spaces) lines = formatted_config.split('\n') formatted_lines = [] indent_level = 0 for line in lines: stripped = line.strip() # Decrease indent for closing braces if stripped == '}': indent_level = max(0, indent_level - 1) # Add indentation if stripped: formatted_lines.append(' ' * indent_level + stripped) else: formatted_lines.append('') # Increase indent for opening braces if stripped.endswith('{'): indent_level += 1 formatted_config = '\n'.join(formatted_lines) response = f"""# 🎨 Terraform Code Formatting ## ✨ Formatted Configuration ```hcl {formatted_config} ``` ## 📋 Formatting Applied - ✅ Normalized indentation (2 spaces) - ✅ Consistent spacing - ✅ Proper block alignment - ✅ Standard Terraform formatting ## 🔧 Additional Formatting Tips 1. **Use `terraform fmt`** - Run the official formatter 2. **Configure editor** - Set up auto-formatting in your IDE 3. **Pre-commit hooks** - Automatically format before commits 4. **Consistent naming** - Use lowercase with underscores 5. **Group related blocks** - Keep similar resources together """ return [types.TextContent(type="text", text=response)] async def _handle_generate_docs(self, args: dict) -> list[types.TextContent]: """Handle documentation generation""" config_content = args.get("config_content") module_name = args.get("module_name", "terraform-module") include_examples = args.get("include_examples", True) # Extract information from config resources = re.findall(r'resource\s+"([^"]+)"\s+"([^"]+)"', config_content) variables = re.findall(r'variable\s+"([^"]+)"\s*\{', config_content) outputs = re.findall(r'output\s+"([^"]+)"\s*\{', config_content) modules = re.findall(r'module\s+"([^"]+)"\s*\{', config_content) response = f"""# 📚 Terraform Documentation: {module_name} ## 📋 Module Overview This Terraform configuration manages infrastructure resources with the following components: """ # Resources section if resources: response += f"""## 🏗️ Resources Created | Resource Type | Name | Description | |---------------|------|-------------| """ for resource_type, name in resources: response += f"| `{resource_type}` | `{name}` | {resource_type.replace('_', ' ').title()} resource |\n" response += "\n" # Modules section if modules: response += f"""## 📦 Modules Used | Module Name | Description | |-------------|-------------| """ for module in modules: response += f"| `{module}` | External module for {module.replace('_', ' ')} |\n" response += "\n" # Variables section if variables: response += f"""## 📥 Input Variables | Name | Description | Type | Default | |------|-------------|------|---------| """ for var in variables: response += f"| `{var}` | Description for {var} | `string` | `null` |\n" response += "\n" # Outputs section if outputs: response += f"""## 📤 Outputs | Name | Description | |------|-------------| """ for output in outputs: response += f"| `{output}` | Output description for {output} |\n" response += "\n" # Usage examples if include_examples: response += f"""## 🚀 Usage Examples ### Basic Usage ```hcl module "{module_name}" {{ source = "./modules/{module_name}" # Required variables project_name = "my-project" environment = "production" tags = {{ Environment = "production" Team = "infrastructure" Project = "my-project" }} }} ``` """ response += f"""## 📋 Requirements | Name | Version | |------|---------| | terraform | >= 1.0 | | aws | >= 5.0 | ## 🔧 Installation & Setup 1. **Clone or download** this module 2. **Configure variables** in `terraform.tfvars` 3. **Initialize Terraform**: `terraform init` 4. **Plan deployment**: `terraform plan` 5. **Apply configuration**: `terraform apply` """ return [types.TextContent(type="text", text=response)] # ============================================================================ # FASTAPI APP AND MCP SERVER INSTANCE # ============================================================================ # Create FastAPI app app = FastAPI(title="Terraform MCP Server") # Create an instance of the TerraformAgentHandlers to manage tool logic terraform_agent_handlers = TerraformAgentHandlers() # Create MCP server instance mcp_server_instance = Server("terraform-mcp-server") # ============================================================================ # MCP SERVER HANDLERS (Decorated on the global mcp_server_instance) # ============================================================================ @mcp_server_instance.list_tools() async def list_tools_for_mcp(): # Renamed to avoid conflict with `list_tools` in the Linux example """List all available Terraform tools""" logger.debug("MCP ListToolsRequest received.") return [ types.Tool( name="generate_terraform_config", description="Generate complete Terraform configuration for infrastructure resources", inputSchema={ "type": "object", "properties": { "resource_type": { "type": "string", "description": "Type of infrastructure resource", "enum": ["vpc", "ec2", "s3", "eks", "rds", "vnet", "compute", "aks"] }, "provider": { "type": "string", "description": "Cloud provider", "enum": ["aws", "azurerm", "google"] }, "name": { "type": "string", "description": "Name for the infrastructure resources" }, "options": { "type": "object", "description": "Resource-specific configuration options", "properties": { "cidr": {"type": "string", "description": "CIDR block for VPC"}, "azs": {"type": "array", "items": {"type": "string"}, "description": "Availability zones"}, "instance_type": {"type": "string", "description": "EC2 instance type"}, "key_pair_name": {"type": "string", "description": "SSH key pair name"}, "enable_nat_gateway": {"type": "boolean", "description": "Enable NAT gateway"}, "versioning": {"type": "boolean", "description": "Enable S3 versioning"}, "encryption": {"type": "boolean", "description": "Enable encryption"}, "node_instance_type": {"type": "string", "description": "EKS node instance type"}, "min_nodes": {"type": "integer", "description": "Minimum nodes"}, "max_nodes": {"type": "integer", "description": "Maximum nodes"}, "desired_nodes": {"type": "integer", "description": "Desired nodes"}, "kubernetes_version": {"type": "string", "description": "Kubernetes version"} } } }, "required": ["resource_type", "provider", "name"] } ), types.Tool( name="validate_terraform_config", description="Validate Terraform configuration syntax and best practices", inputSchema={ "type": "object", "properties": { "config_content": { "type": "string", "description": "Terraform configuration content to validate" }, "check_best_practices": { "type": "boolean", "description": "Include best practices validation", "default": True }, "strict_mode": { "type": "boolean", "description": "Enable strict validation rules", "default": False } }, "required": ["config_content"] } ), types.Tool( name="get_deployment_workflow", description="Generate step-by-step Terraform deployment workflow", inputSchema={ "type": "object", "properties": { "workflow_type": { "type": "string", "description": "Type of deployment workflow", "enum": ["basic", "production", "module_development"], "default": "basic" }, "backend_type": { "type": "string", "description": "Backend storage type", "enum": ["local", "s3", "azurerm", "gcs"], "default": "local" }, "environment": { "type": "string", "description": "Target environment", "enum": ["development", "staging", "production"], "default": "development" } } } ), types.Tool( name="convert_to_module", description="Convert Terraform configuration to reusable module structure", inputSchema={ "type": "object", "properties": { "config_content": { "type": "string", "description": "Terraform configuration to convert" }, "module_name": { "type": "string", "description": "Name for the module" }, "extract_variables": { "type": "boolean", "description": "Extract hardcoded values as variables", "default": True }, "generate_examples": { "type": "boolean", "description": "Generate usage examples", "default": True } }, "required": ["config_content", "module_name"] } ), types.Tool( name="format_terraform_code", description="Format and standardize Terraform code", inputSchema={ "type": "object", "properties": { "config_content": { "type": "string", "description": "Terraform configuration to format" }, "sort_blocks": { "type": "boolean", "description": "Sort resource blocks alphabetically", "default": True } }, "required": ["config_content"] } ), types.Tool( name="generate_terraform_docs", description="Generate documentation for Terraform configuration", inputSchema={ "type": "object", "properties": { "config_content": { "type": "string", "description": "Terraform configuration to document" }, "module_name": { "type": "string", "description": "Name of the module" }, "include_examples": { "type": "boolean", "description": "Include usage examples", "default": True } }, "required": ["config_content"] } ) ] @mcp_server_instance.call_tool() async def call_tool_for_mcp(name: str, arguments: dict) -> list[types.TextContent]: # Renamed """Handle tool calls from MCP client via SSE""" logger.info(f"MCP CallToolRequest received for tool: {name}") try: if name == "generate_terraform_config": return await terraform_agent_handlers._handle_generate_config(arguments) elif name == "validate_terraform_config": return await terraform_agent_handlers._handle_validate_config(arguments) elif name == "get_deployment_workflow": return await terraform_agent_handlers._handle_deployment_workflow(arguments) elif name == "convert_to_module": return await terraform_agent_handlers._handle_convert_to_module(arguments) elif name == "format_terraform_code": return await terraform_agent_handlers._handle_format_code(arguments) elif name == "generate_terraform_docs": return await terraform_agent_handlers._handle_generate_docs(arguments) else: return [types.TextContent(type="text", text=f"❌ Unknown tool: {name}")] except Exception as e: logger.exception(f"Error executing Terraform tool {name}") return [types.TextContent(type="text", text=f"❌ Error executing {name}: {str(e)}")] # ============================================================================ # FASTAPI WEB ENDPOINTS # ============================================================================ @app.get("/") async def root(): """Status endpoint""" logger.info("GET / requested - sending status.") return { "service": "terraform-mcp-server", "status": "running", "mcp_endpoint": "/mcp/sse", "tools": len(await list_tools_for_mcp()), # Dynamically list tools "categories": ["infrastructure_generation", "validation", "workflow_management", "module_conversion", "documentation"] } @app.get("/health") async def health(): """Health check endpoint for Hugging Face Spaces""" logger.debug("GET /health requested - sending healthy status.") return {"status": "healthy", "service": "terraform-mcp-server"} @app.get("/mcp/sse") async def mcp_sse_endpoint(request: Request): """MCP SSE endpoint for agent connections""" logger.info("GET /mcp/sse requested - starting SSE stream.") async def event_stream(): # MCP uses this internally for server initialization and handling client requests # The mcp_server_instance will manage the SSE communication within its run() method. # This wrapper is needed to hook mcp_server_instance.run() into FastAPI's StreamingResponse. # Create dummy readers/writers for mcp_server_instance.run() # In a real-world scenario, mcp.server might provide an HTTP/SSE specific run method # or expect a custom transport. For this setup, we'll try to adapt. # The MCP library's Server.run() method expects streams, so we need to mock or # adapt it to FastAPI's request/response cycle. # However, the most direct approach for MCP and FastAPI is to let MCP handle the HTTP # part if it has a specific HTTP/SSE server implementation, or have FastAPI # call MCP's core logic directly (as I did in the Gradio example). # Given the Linux example, the MCP server is directly serving the /mcp/sse. # This implies mcp.server.Server has a mechanism to do this, or the StreamingResponse # content generator is the key. # Let's align with the Linux code provided: # The Linux example *does not* call `mcp_app.run()` inside the SSE endpoint directly. # Instead, it uses `mcp_app.list_tools()` and `mcp_app.call_tool()` as decorators # on global functions, and the FastAPI endpoint just sends keepalives. # The communication over SSE is typically initiated by a client *sending* an MCP message # to the server, and the server *sending* responses back. # For a simple SSE endpoint that only sends keepalives and doesn't handle incoming MCP calls via SSE directly # (which is what your Linux /mcp/sse seems to do with just keepalives): try: while True: if await request.is_disconnected(): logger.info("Client disconnected from /mcp/sse.") break # Send a keepalive signal to maintain the connection yield f"data: {json.dumps({'type': 'keepalive'})}\n\n" await asyncio.sleep(30) # Send keepalive every 30 seconds except asyncio.CancelledError: logger.info("SSE event stream cancelled.") except Exception as e: logger.error(f"Error in SSE event stream: {e}", exc_info=True) return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "Access-Control-Allow-Origin": "*", # Important for CORS if consumed by a frontend } ) # ============================================================================ # MAIN ENTRY POINT FOR FASTAPI/UVICORN # ============================================================================ def main(): """Main entry point to run the FastAPI application with Uvicorn.""" port = int(os.getenv("PORT", 7860)) # Default to 7860, used by Hugging Face Spaces logger.info(f"🚀 Starting Terraform MCP Server with FastAPI/Uvicorn on port {port}") print(f"📊 Status Endpoint: http://0.0.0.0:{port}/") print(f"❤️ Health Check: http://0.0.0.0:{port}/health") print(f"🔗 MCP SSE Endpoint: http://0.0.0.0:{port}/mcp/sse") print(f"🛠️ Available Tools: generate_terraform_config, validate_terraform_config, get_deployment_workflow, convert_to_module, format_terraform_code, generate_terraform_docs") # Run the FastAPI application using Uvicorn # This will bind to the specified host and port, handling HTTP requests. uvicorn.run(app, host="0.0.0.0", port=port, log_level="info") if __name__ == "__main__": main()