Epam基础设施漂移探测器MCP服务器
一种模型上下文协议(MCP)服务器,通过使用Terraformer导入云资源并将其与基线Terraform状态文件进行比较,实现基础设施漂移检测。
概述
此MCP服务器通过以下方式提供检测基础设施漂移的工具:
- 使用Terraformer导入当前部署的云资源
- 读取基线Terraform状态文件(预期状态)
- 读取导入的状态文件(实际部署状态)
- 实现预期与实际基础设施状态之间的比较
建筑
┌─────────────────────────────────────────────────────────────┐
│ 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_infradriftdetectorMCP工具
工具1: import_deployed_resources
使用Terraformer CLI工具导入当前部署的云资源。
参数
| 参数 | 类型 | 描述 | 示例 |
|---|---|---|---|
cloud | string | 云提供商(aws, azure, gcp) | "aws" |
resources | string | 逗号分隔的资源类型 | "ec2,vpc,s3" |
regions | string | 目标区域(特定于云) | "us-east-1,us-west-2" |
path_output | string | 状态文件的输出目录 | "/tmp/imported-state" |
返回值
{
"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参数assubscription= - 例子:
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状态文件。
参数
| 参数 | 类型 | 描述 | 示例 |
|---|---|---|---|
last_deployed_file_path | string | 基线tfstate文件的路径 | "/baseline/terraform.tfstate" |
返回值
从状态文件返回资源字典列表:
[
{
"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状态文件。
参数
| 参数 | 类型 | 描述 | 示例 |
|---|---|---|---|
post_deployed_tfstates_path | string | 包含导入的tfstate文件的根目录 | "/data/imported-states" |
预期的目录结构
/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")错误处理
服务器实现了复杂的错误检测:
错误类型
- 实际误差 (状态:“错误”):
- 命令执行失败 - 未知标志或无效参数 - 身份验证/授权失败 - 找不到资源错误
- 警告 (状态:“警告”):
- 配置建议(例如,“设置ARM_SUBSCRIPTION_ID环境变量”) - 非关键信息性消息 - 成功执行配置说明
- 成功 (状态:“成功”):
- 命令已成功完成 - 导入的资源没有问题
错误响应结构
{
"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.0Terraformer集成
服务器充当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/bin2.身份验证错误
# 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 list3.导入结果为空
# 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 documentation4.状态文件读取错误
# 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))局限性
- Terraformer依赖关系:需要安装Terraformer CLI并可访问
- 状态文件格式:支持Terraform 0.11+和0.12+格式
- 云提供商支持:仅限于AWS、Azure和GCP(由Terraformer支持)
- 资源类型:导入能力取决于Terraformer对每种资源类型的支持
- 比较逻辑:服务器提供数据读取;漂移比较逻辑必须由MCP客户端实现
安全考虑
- 凭证:切勿在代码中硬编码凭据
- 文件权限:确保状态文件具有适当的读取权限
- 输出路径:验证和清理输出路径以防止目录遍历
- 子流程执行:命令参数已验证,但请谨慎使用
- 云访问:使用最低权限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添加新的云提供商
要添加对新云提供商的支持,请执行以下操作:
- 在中添加特定于云的逻辑
import_deployed_resources() - 处理身份验证要求
- 将区域/参数映射到Terraformer标志
- 更新文档
示例模板:
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)
