Mcpserver Cloud Import

Epam基础设施漂移探测器MCP服务器

一种模型上下文协议（MCP）服务器，通过使用Terraformer导入云资源并将其与基线Terraform状态文件进行比较，实现基础设施漂移检测。

概述

此MCP服务器通过以下方式提供检测基础设施漂移的工具：

1. 使用Terraformer导入当前部署的云资源
2. 读取基线Terraform状态文件（预期状态）
3. 读取导入的状态文件（实际部署状态）
4. 实现预期与实际基础设施状态之间的比较

建筑

┌─────────────────────────────────────────────────────────────┐
│                    MCP Client (e.g., Claude)                │
└───────────────────────────┬─────────────────────────────────┘
                            │ MCP Protocol (STDIO)
┌───────────────────────────▼─────────────────────────────────┐
│           Epam-InfraDriftDetector MCP Server                │
│  ┌───────────────────────────────────────────────────────┐  │
│  │  Tool 1: import_deployed_resources                    │  │
│  │  - Executes Terraformer CLI                          │  │
│  │  - Imports actual cloud infrastructure                │  │
│  │  - Supports AWS, Azure, GCP                          │  │
│  └───────────────────────────────────────────────────────┘  │
│  ┌───────────────────────────────────────────────────────┐  │
│  │  Tool 2: read_lastdeployed_tfstates                  │  │
│  │  - Reads baseline Terraform state file               │  │
│  │  - Returns intended infrastructure state             │  │
│  └───────────────────────────────────────────────────────┘  │
│  ┌───────────────────────────────────────────────────────┐  │
│  │  Tool 3: read_imported_tfstates                      │  │
│  │  - Reads multiple imported state files               │  │
│  │  - Merges resources from multiple folders            │  │
│  │  - Returns actual infrastructure state               │  │
│  └───────────────────────────────────────────────────────┘  │
└───────────────────────────┬─────────────────────────────────┘
                            │
                            ▼
              ┌─────────────────────────────┐
              │  Cloud Provider APIs        │
              │  - AWS                      │
              │  - Azure                    │
              │  - Google Cloud             │
              └─────────────────────────────┘

特性

1.多云支持

• 亚马逊云服务：完全支持基于区域的资源导入
• Azure：支持订阅和资源组筛选
• 谷歌云（GCP）：基于区域的资源导入

2.智能错误处理

• 区分信息性消息和实际错误
• 提供详细的错误上下文，包括命令、返回代码和输出
• 将配置警告与执行失败分开处理

3.灵活的资源过滤

• 导入特定资源类型（例如。， ec2,vpc,s3,rds)
• 明确请求时支持通配符导入
• 针对特定地区的过滤，实现高效导入

安装

先决条件

# Install Terraformer
# Linux
wget https://github.com/GoogleCloudPlatform/terraformer/releases/download/0.8.24/terraformer-all-linux-amd64
chmod +x terraformer-all-linux-amd64
sudo mv terraformer-all-linux-amd64 /usr/local/bin/terraformer

# macOS
brew install terraformer

# Install Terraform (required by Terraformer)
# Linux
wget https://releases.hashicorp.com/terraform/1.6.0/terraform_1.6.0_linux_amd64.zip
unzip terraform_1.6.0_linux_amd64.zip
sudo mv terraform /usr/local/bin/

# macOS
brew install terraform

# Install Python dependencies
pip install mcp

运行服务器

# Direct execution
python fixed_importaws_mcpserver.py

# As a module (if packaged)
python -m epam_infradriftdetector

MCP工具

工具1： import_deployed_resources

使用Terraformer CLI工具导入当前部署的云资源。

参数

返回值

{
  "status": "success|error|warning",
  "stdout": "Command standard output",
  "stderr": "Command standard error",
  "command": "Executed terraformer command",
  "return_code": 0
}

用法示例

AWS示例：

result = import_deployed_resources(
    cloud="aws",
    resources="vpc,subnet,ec2_instance,sg",
    regions="us-east-1,us-west-2",
    path_output="/data/aws-import"
)

Azure示例：

result = import_deployed_resources(
    cloud="azure",
    resources="virtual_machine,storage_account",
    regions="subscription=12345678-1234-1234-1234-123456789012",
    path_output="/data/azure-import"
)

GCP示例：

result = import_deployed_resources(
    cloud="gcp",
    resources="compute_instance,storage_bucket",
    regions="us-central1,us-east1",
    path_output="/data/gcp-import"
)

云特定行为

亚马逊云服务

• 用途 --regions 多区域支持标志
• 需要配置AWS凭据（AWS CLI或环境变量）
• 例子： terraformer import aws --resources=ec2,vpc --regions=us-east-1 --path-output=/output

Azure

• 使用ARM环境变量进行身份验证：

- ARM_SUBSCRIPTION_ID - ARM_CLIENT_ID - ARM_CLIENT_SECRET - ARM_TENANT_ID

• 订阅ID可以通过以下方式传递 regions 参数as subscription=
• 例子： terraformer import azure --resources=virtual_machine --path-output=/output

谷歌云

• 用途 --regions 与AWS类似的标志
• 可能需要 --projects 特定项目的标志
• 需要配置GCP凭据（gcloud CLI或服务帐户）
• 例子： terraformer import google --resources=compute_instance --regions=us-central1 --path-output=/output

工具2： read_lastdeployed_tfstates

读取表示预期基础结构状态的基线Terraform状态文件。

参数

返回值

从状态文件返回资源字典列表：

[
    {
        "type": "aws_instance",
        "name": "web_server",
        "provider": "provider[\"registry.terraform.io/hashicorp/aws\"]",
        "instances": [...]
    },
    ...
]

用法示例

baseline_resources = read_lastdeployed_tfstates(
    last_deployed_file_path="/terraform/project/terraform.tfstate"
)

# Access resource types
resource_types = [resource["type"] for resource in baseline_resources]
print(f"Baseline contains: {resource_types}")

工具3： read_imported_tfstates

从导入的云资源中读取并合并多个Terraform状态文件。

参数

预期的目录结构

/data/imported-states/
├── vpc/
│   └── terraform.tfstate
├── ec2/
│   └── terraform.tfstate
├── s3/
│   └── terraform.tfstate
└── rds/
    └── terraform.tfstate

返回值

返回合并资源的字典：

{
    "aws_vpc.vpc-12345": {
        "type": "aws_vpc",
        "primary": {
            "id": "vpc-12345",
            "attributes": {...}
        }
    },
    "aws_instance.i-67890": {
        "type": "aws_instance",
        "primary": {
            "id": "i-67890",
            "attributes": {...}
        }
    }
}

用法示例

imported_resources = read_imported_tfstates(
    post_deployed_tfstates_path="/data/aws-import"
)

# Get all resource keys
resource_ids = list(imported_resources.keys())
print(f"Found {len(resource_ids)} imported resources")

# Access specific resource
vpc_resource = imported_resources.get("aws_vpc.vpc-12345")

漂移检测工作流程

完整示例

# Step 1: Import current cloud state
import_result = import_deployed_resources(
    cloud="aws",
    resources="vpc,subnet,ec2_instance,sg,s3,rds",
    regions="us-east-1",
    path_output="/tmp/current-state"
)

if import_result["status"] == "success":
    print("✅ Successfully imported current infrastructure")
    
    # Step 2: Read baseline state (what should be deployed)
    baseline = read_lastdeployed_tfstates(
        last_deployed_file_path="/terraform/production/terraform.tfstate"
    )
    
    # Step 3: Read imported state (what is actually deployed)
    current = read_imported_tfstates(
        post_deployed_tfstates_path="/tmp/current-state"
    )
    
    # Step 4: Compare and detect drift
    baseline_types = {r["type"] for r in baseline}
    current_types = {r["type"].split(".")[0] for r in current.keys()}
    
    added_resources = current_types - baseline_types
    removed_resources = baseline_types - current_types
    
    if added_resources or removed_resources:
        print("⚠️  Infrastructure drift detected!")
        if added_resources:
            print(f"   Added: {added_resources}")
        if removed_resources:
            print(f"   Removed: {removed_resources}")
    else:
        print("✅ No drift detected")

错误处理

服务器实现了复杂的错误检测：

错误类型

1. 实际误差 （状态：“错误”）：

- 命令执行失败 - 未知标志或无效参数 - 身份验证/授权失败 - 找不到资源错误

1. 警告 （状态：“警告”）：

- 配置建议（例如，“设置ARM_SUBSCRIPTION_ID环境变量”） - 非关键信息性消息 - 成功执行配置说明

1. 成功 （状态：“成功”）：

- 命令已成功完成 - 导入的资源没有问题

错误响应结构

{
  "status": "error",
  "stdout": "Standard output from command",
  "stderr": "Standard error from command",
  "command": "Full terraformer command executed",
  "return_code": 1,
  "error_message": "Detailed error description"
}

配置

环境变量

亚马逊云服务

export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_DEFAULT_REGION="us-east-1"

Azure

export ARM_SUBSCRIPTION_ID="your-subscription-id"
export ARM_CLIENT_ID="your-client-id"
export ARM_CLIENT_SECRET="your-client-secret"
export ARM_TENANT_ID="your-tenant-id"

谷歌云

export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account-key.json"
export GOOGLE_PROJECT="your-project-id"

MCP客户端配置

克劳德桌面(claude_desktop_config.json):

{
  "mcpServers": {
    "epam-infradriftdetector": {
      "command": "python",
      "args": ["/path/to/fixed_importaws_mcpserver.py"],
      "env": {
        "AWS_REGION": "us-east-1"
      }
    }
  }
}

使用Docker：

{
  "mcpServers": {
    "epam-infradriftdetector": {
      "command": "docker",
      "args": [
        "run",
        "-i",
        "--rm",
        "-v", "${HOME}/.aws:/root/.aws:ro",
        "-v", "/data/terraform-states:/terraform-state:rw",
        "-e", "AWS_REGION=us-east-1",
        "epam-infradriftdetector:latest"
      ]
    }
  }
}

技术细节

框架

使用 FastMCP 从 mcp.server.fastmcp 该软件包提供：

• 使用装饰器简化工具注册
• 自动参数验证
• 用于MCP协议的内置STDIO传输
• 类型安全功能签名

依赖项

mcp>=0.9.0

Terraformer集成

服务器充当Terraformer CLI的包装器：

• 将Terraformer作为子流程执行
• 捕获stdout和stderr以进行结果报告
• 处理特定于云的命令变化
• 解析返回代码和输出以进行错误检测

状态文件格式

地形0.12+格式（基线）：

{
  "version": 4,
  "terraform_version": "1.6.0",
  "resources": [
    {
      "type": "aws_instance",
      "name": "example",
      "provider": "provider[\"registry.terraform.io/hashicorp/aws\"]",
      "instances": [...]
    }
  ]
}

地形0.11格式（导入）：

{
  "version": 3,
  "modules": [
    {
      "resources": {
        "aws_instance.i-12345": {
          "type": "aws_instance",
          "primary": {
            "id": "i-12345",
            "attributes": {...}
          }
        }
      }
    }
  ]
}

最佳实践

1.资源过滤

• 始终使用特定的资源筛选器 在可能的情况下
• 仅限使用 * （所有资源）在明确请求时
• 例子： "vpc,ec2_instance,sg" 而不是 "*"

2.输出组织

• 为每个导入会话创建单独的输出目录
• 在目录名中使用时间戳进行跟踪
• 例子： /data/imports/2024-11-11-15-30

3.多地区进口

• 按顺序导入区域以更好地跟踪错误
• 每个区域使用单独的输出目录
• 所有导入完成后的汇总结果

4.状态文件管理

• 保持基线状态文件的版本控制
• 存档导入的状态文件以进行历史分析
• 定期清理旧的导入目录

5.错误恢复

• 检查 status 处理结果前的字段
• 记录完整命令和输出以进行调试
• 使用详细日志记录重试失败的导入

故障排除

常见问题

1.未找到Terraformer

# Error: terraformer: command not found
# Solution: Verify installation path
which terraformer
# Add to PATH if needed
export PATH=$PATH:/usr/local/bin

2.身份验证错误

# AWS: Check credentials
aws sts get-caller-identity

# Azure: Verify environment variables
echo $ARM_SUBSCRIPTION_ID
echo $ARM_CLIENT_ID

# GCP: Check service account
gcloud auth list

3.导入结果为空

# Possible causes:
# - No resources in specified region
# - Incorrect resource type names
# - Insufficient permissions

# Solution: Enable verbose logging
# Add --verbose flag in the code or check Terraformer documentation

4.状态文件读取错误

# Error: Failed to load tfstate file
# Solution: Verify file path and permissions
import os
print(os.path.exists("/path/to/terraform.tfstate"))
print(os.access("/path/to/terraform.tfstate", os.R_OK))

局限性

1. Terraformer依赖关系：需要安装Terraformer CLI并可访问
2. 状态文件格式：支持Terraform 0.11+和0.12+格式
3. 云提供商支持：仅限于AWS、Azure和GCP（由Terraformer支持）
4. 资源类型：导入能力取决于Terraformer对每种资源类型的支持
5. 比较逻辑：服务器提供数据读取；漂移比较逻辑必须由MCP客户端实现

安全考虑

1. 凭证：切勿在代码中硬编码凭据
2. 文件权限：确保状态文件具有适当的读取权限
3. 输出路径：验证和清理输出路径以防止目录遍历
4. 子流程执行：命令参数已验证，但请谨慎使用
5. 云访问：使用最低权限IAM策略访问云提供商

贡献

编码结构

fixed_importaws_mcpserver.py
├── FastMCP initialization
├── import_deployed_resources() - Cloud resource import
├── read_lastdeployed_tfstates() - Baseline state reader
├── read_imported_tfstates() - Current state reader
└── main() - Server entry point

添加新的云提供商

要添加对新云提供商的支持，请执行以下操作：

1. 在中添加特定于云的逻辑 import_deployed_resources()
2. 处理身份验证要求
3. 将区域/参数映射到Terraformer标志
4. 更新文档

示例模板：

elif cloud.lower() == "newcloud":
    # Add newcloud-specific parameters
    if regions:
        command.extend(["--regions", regions])
    # Add other flags as needed

许可证

\[在此处指定您的许可证\]

支持

对于问题和疑问：

• 在存储库中创建问题
• 联系方式：\[您的支持电子邮件\]

版本历史记录

• 1.0.0 （2024-11-11）：首次发布

- AWS、Azure、GCP支持 - 多区域导入能力 - 智能错误处理 - 状态文件读取和合并

______________________________________________________________________

内置于❤️ 使用模型上下文协议（MCP）