亚马逊AWS官方博客

基于 MCP 的亚马逊云科技中国区服务 EOL 追踪方案实践

方案背景

亚马逊云科技中国区运营两个区域(北京和宁夏),提供包括 Amazon EKS、Lambda、RDS、Aurora、ElastiCache、OpenSearch、EMR、MSK、Neptune、DMS、Amazon MQ 等在内的众多服务。每个服务都有自己的版本生命周期和 EOL(End of Life,生命周期结束)日期。在云环境运维中,服务版本 EOL 管理是确保系统安全性、合规性和运营卓越的关键环节。

然而在实际运维工作中,我们面临以下挑战:

  • 信息分散:每个服务的 EOL 信息发布在不同的文档页面,难以集中追踪
  • 被动应对:缺乏主动预警机制,常在版本即将过期时才匆忙升级
  • 效率低下:运维人员需要人工查阅多个文档页面,耗时且易遗漏
  • 决策困难:无法快速回答”哪些服务需要优先升级”等问题

为了解决这些痛点,我们构建了一套基于 MCP(Model Context Protocol,模型上下文协议)的自动化 EOL 追踪方案,将版本生命周期管理能力直接集成到 Amazon Q Developer 等 AI 助手中,通过自然语言对话实现主动的基础设施管理。

MCP 技术简介

什么是 MCP?

MCP(Model Context Protocol,模型上下文协议)是一个开放的标准化协议,用于连接 AI 助手与外部数据源和工具。它由 Anthropic 公司提出并开源,旨在解决 AI 应用中数据集成和工具调用的标准化问题。

通过 MCP,AI 助手可以:

  • 访问实时数据:从数据库、API、文件系统等数据源获取最新信息
  • 调用外部工具:执行查询、计算、数据处理等操作
  • 保持上下文:在对话过程中维持对外部资源的访问能力

MCP 的核心优势

  • 标准化集成:一次开发,多个 AI 助手通用。支持 Amazon Q Developer、Claude Desktop 等主流工具
  • 安全可控:基于标准协议,权限清晰,数据访问可追溯
  • 实时更新:直接访问数据源,确保 AI 助手获取的是最新信息,而非训练时的静态数据
  • 降低开发成本:无需为每个 AI 工具单独开发集成方案,通过 MCP 实现统一接入
  • 灵活扩展:可以根据需求添加新的工具和数据源,无需修改 AI 助手本身

为什么选择 MCP?

在 EOL 追踪场景中,MCP 具有独特优势:

  • 自然交互:通过对话方式查询 EOL 信息,无需学习复杂的命令或界面
  • 智能分析:AI 助手可以理解意图,自动选择合适的工具组合完成任务
  • 即时响应:实时访问最新 EOL 数据,确保信息准确性
  • 降低门槛:运维人员无需掌握数据库查询或 API 调用,只需用自然语言提问

技术方案

方案架构

我们设计了一个包含三层的综合解决方案:

第一层:自动数据采集

基于 Amazon Lambda 函数实现无服务器数据采集,每天自动爬取亚马逊云科技中国区官方文档,解析 EOL 信息并存储到 S3。支持以下特性:

  • 从官方文档网页抓取 11+ 个服务的 EOL 数据
  • 监控亚马逊云科技中国区 RSS 订阅源获取最新公告
  • 智能缓存机制(默认 24 小时)减少重复抓取
  • 生成标准化 CSV 报告,便于审计和分析

第二层:MCP Server

基于 TypeScript 构建的 MCP 服务器,通过标准化的 MCP 工具暴露 EOL 数据,实现 AI 助手与 EOL 数据之间的桥接。核心特性包括:

  • 5 个专用 MCP 工具,覆盖查询、检查、预警、报告、建议等场景
  • 多数据源支持(S3、本地缓存、静态回退数据)
  • 7 天本地缓存,保证离线可用性
  • 统一的数据访问接口,简化集成

第三层:自然语言交互

通过 Amazon Q Developer CLI、Claude Desktop 等 MCP 兼容工具,实现自然语言查询,无需构建专门的 Web 界面。

架构示意图

核心代码示例

MCP 工具定义

// 查询即将过期的服务
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "get_expiring_services") {
    const { days, includeExpired } = request.params.arguments;
    
    // 加载 EOL 数据
    await dataManager.ensureDataLoaded();
    
    // 筛选即将过期的版本
    const expiringVersions = [];
    for (const [service, data] of Object.entries(dataManager.getData())) {
      for (const version of data.versions) {
        const daysRemaining = calculateDaysRemaining(version.eol_date);
        if (daysRemaining <= days && (includeExpired || daysRemaining >= 0)) {
          expiringVersions.push({
            service,
            version: version.version,
            eolDate: version.eol_date,
            daysRemaining,
            status: version.status,
            priority: version.priority
          });
        }
      }
    }
    
    // 按剩余天数排序
    expiringVersions.sort((a, b) => a.daysRemaining - b.daysRemaining);
    
    return {
      content: [{ 
        type: "text", 
        text: formatExpiringServices(expiringVersions) 
      }]
    };
  }
});

数据管理策略

class EOLDataManager {
  private async loadData(): Promise<void> {
    // 优先级 1: 从 S3 加载最新数据
    if (this.config.s3Bucket) {
      try {
        await this.loadFromS3();
        return;
      } catch (error) {
        console.warn("S3 数据加载失败,尝试缓存:", error.message);
      }
    }
    
    // 优先级 2: 使用本地缓存(7天有效期)
    try {
      await this.loadFromCache();
      if (this.isCacheValid()) {
        return;
      }
    } catch (error) {
      console.warn("缓存加载失败:", error.message);
    }
    
    // 优先级 3: 使用静态回退数据
    this.loadFallbackData();
  }
  
  private async loadFromS3(): Promise<void> {
    const s3Client = new S3Client({ 
      region: this.config.s3Region || "cn-north-1" 
    });
    
    const command = new GetObjectCommand({
      Bucket: this.config.s3Bucket,
      Key: this.config.s3Key || "eol_mapping/aws_china_eol_latest.csv"
    });
    
    const response = await s3Client.send(command);
    const csvContent = await response.Body!.transformToString();
    
    // 解析 CSV 并转换为 JSON
    this.data = this.parseCSVToEOLData(csvContent);
    
    // 缓存到本地
    await this.saveToCache(this.data);
  }
}

Lambda 数据采集核心逻辑

def scrape_eol_data(service_type: str, url: str) -> List[Dict]:
    """爬取特定服务的 EOL 数据"""
    response = http.request('GET', url)
    html_content = response.data.decode('utf-8')
    
    # 使用正则表达式解析 HTML 表格
    table_pattern = r'<table[^>]*>(.*?)</table>'
    tables = re.findall(table_pattern, html_content, re.DOTALL | re.IGNORECASE)
    
    versions = []
    for table in tables:
        rows = re.findall(r'<tr[^>]*>(.*?)</tr>', table, re.DOTALL | re.IGNORECASE)
        for row in rows[1:]:  # 跳过表头
            cells = re.findall(r'<td[^>]*>(.*?)</td>', row, re.DOTALL | re.IGNORECASE)
            if len(cells) >= 2:
                version_name = clean_html(cells[0])
                eol_date = clean_html(cells[1])
                
                # 计算剩余天数和状态
                days_remaining = calculate_days_remaining(eol_date)
                status = determine_status(days_remaining)
                
                versions.append({
                    'version': version_name,
                    'eol_date': eol_date,
                    'days_remaining': days_remaining,
                    'status': status,
                    'priority': determine_priority(status)
                })
    
    return versions

def lambda_handler(event, context):
    """Lambda 主函数"""
    # 爬取所有服务的 EOL 数据
    all_data = {}
    for service, url in SERVICE_URLS.items():
        try:
            all_data[service] = scrape_eol_data(service, url)
        except Exception as e:
            logger.error(f"爬取 {service} 失败: {str(e)}")
            # 使用静态回退数据
            all_data[service] = FALLBACK_DATA.get(service, [])
    
    # 生成 CSV 报告
    csv_content = generate_csv_report(all_data)
    
    # 上传到 S3
    s3_client.put_object(
        Bucket=S3_BUCKET,
        Key=f"{S3_PREFIX}aws_china_eol_latest.csv",
        Body=csv_content.encode('utf-8-sig'),  # 添加 BOM 兼容 Excel
        ContentType='text/csv; charset=utf-8'
    )
    
    return {'statusCode': 200, 'body': 'EOL data updated successfully'}

EOL 状态判定与优先级算法

def calculate_days_remaining(eol_date_str: str) -> int:
    """计算距离EOL日期的剩余天数"""
    try:
        # 支持多种日期格式: YYYY-MM-DD 或 YYYY-MM
        if len(eol_date_str) == 7:  # YYYY-MM 格式
            eol_date = datetime.strptime(eol_date_str + "-01", "%Y-%m-%d")
            # 移到月末
            next_month = eol_date + timedelta(days=32)
            eol_date = next_month.replace(day=1) - timedelta(days=1)
        else:
            eol_date = datetime.strptime(eol_date_str, "%Y-%m-%d")
        
        today = datetime.now()
        days_remaining = (eol_date - today).days
        return days_remaining
    except Exception as e:
        logger.warning(f"日期解析失败: {eol_date_str}, 错误: {str(e)}")
        return -9999  # 返回特殊值表示未知

def determine_status(days_remaining: int) -> str:
    """根据剩余天数判定状态"""
    if days_remaining < 0:
        return "Expired"  # 已过期
    elif days_remaining <= 30:
        return "Critical"  # 危急 (30天内)
    elif days_remaining <= 90:
        return "Warning"  # 警告 (90天内)
    elif days_remaining <= 180:
        return "Notice"  # 提醒 (180天内)
    else:
        return "Normal"  # 正常

def determine_priority(status: str) -> str:
    """根据状态确定优先级"""
    priority_map = {
        "Expired": "Highest",      # 最高优先级
        "Critical": "High",        # 高优先级
        "Warning": "Medium-High",  # 中高优先级
        "Notice": "Medium",        # 中等优先级
        "Normal": "Low"            # 低优先级
    }
    return priority_map.get(status, "Unknown")

CSV 报告生成

def generate_csv_report(all_data: Dict) -> str:
    """生成标准化的CSV报告"""
    import csv
    from io import StringIO
    
    output = StringIO()
    writer = csv.writer(output)
    
    # CSV 表头
    headers = [
        'Service Type', 'Version/Type', 'EOL Date', 'Days Remaining',
        'Status', 'Priority', 'Recommended Action', 'Supported Regions',
        'Official Documentation Link', 'Upgrade Guidance Link',
        'Data Source', 'Last Updated'
    ]
    writer.writerow(headers)
    
    # 写入数据
    for service, versions in all_data.items():
        for version_info in versions:
            row = [
                service,
                version_info['version'],
                version_info['eol_date'],
                version_info['days_remaining'],
                version_info['status'],
                version_info['priority'],
                generate_recommendation(version_info['status']),
                'cn-north-1, cn-northwest-1',
                get_doc_link(service),
                get_upgrade_link(service),
                'AWS China Official Documentation',
                datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            ]
            writer.writerow(row)
    
    return output.getvalue()

def generate_recommendation(status: str) -> str:
    """生成升级建议"""
    recommendations = {
        "Expired": "Immediate upgrade required - Version no longer supported",
        "Critical": "Urgent: Plan upgrade within 30 days",
        "Warning": "Important: Plan upgrade within 3 months",
        "Notice": "Advisory: Consider upgrade in next maintenance window",
        "Normal": "No immediate action required"
    }
    return recommendations.get(status, "Review documentation for details")

RSS 公告解析(增强数据准确性)

def parse_rss_announcements(rss_url: str) -> List[Dict]:
    """解析AWS中国区RSS订阅源,获取EOL相关公告"""
    import feedparser
    
    feed = feedparser.parse(rss_url)
    eol_announcements = []
    
    # EOL 相关关键词
    eol_keywords = ['end of life', 'eol', '生命周期结束', '停止支持', 
                    'deprecation', '弃用', 'end of support']
    
    for entry in feed.entries:
        title = entry.title.lower()
        summary = entry.get('summary', '').lower()
        
        # 检查是否包含EOL关键词
        if any(keyword in title or keyword in summary for keyword in eol_keywords):
            eol_announcements.append({
                'title': entry.title,
                'link': entry.link,
                'published': entry.published,
                'summary': entry.get('summary', '')[:200]  # 前200字符
            })
    
    logger.info(f"从RSS发现 {len(eol_announcements)} 条EOL相关公告")
    return eol_announcements

HTML 清理与数据标准化

def clean_html(text: str) -> str:
    """清理HTML标签和多余空白字符"""
    # 移除HTML标签
    text = re.sub(r'<[^>]+>', '', text)
    # 解码HTML实体
    text = html.unescape(text)
    # 移除多余空白
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

def standardize_version_name(version: str, service_type: str) -> str:
    """标准化版本号格式"""
    version = version.strip()
    
    # 针对不同服务类型的特殊处理
    if service_type == 'EKS':
        # 提取Kubernetes版本号: "Kubernetes 1.28" -> "1.28"
        match = re.search(r'(\d+\.\d+)', version)
        if match:
            return f"Kubernetes {match.group(1)}"
    
    elif service_type == 'Lambda':
        # 标准化运行时名称: "python3.11" -> "Python 3.11"
        if 'python' in version.lower():
            match = re.search(r'(\d+)\.(\d+)', version)
            if match:
                return f"Python {match.group(1)}.{match.group(2)}"
    
    return version

应用场景

场景 1:日常运维检查

业务场景:

运维人员需要快速了解当前环境中哪些服务版本即将过期,以便提前规划升级工作。

传统方式:

  • 手工访问亚马逊云科技中国区文档网站
  • 逐个服务查看版本生命周期页面
  • 在 Excel 中手工记录和计算剩余天数
  • 定期更新(通常每月一次)

使用 MCP 方案:

在 Amazon Q Developer 中输入:

请检查未来 90 天内即将过期的 AWS 服务版本

Amazon Q 自动调用 「get_expiring_services」 工具,返回:

业务提升点:

  • 从每月一次的手工检查提升为随时随地即时查询
  • 查询时间从 30 分钟降低到 5 秒
  • 零遗漏风险,数据每天自动更新

场景 2: 升级规划决策

业务场景:

技术团队需要评估某个具体服务版本的状态,并获得升级建议以制定升级计划。

传统方式:

  • 访问服务文档查看 EOL 日期
  • 查看升级路径文档
  • 评估升级复杂度
  • 在团队会议中讨论

使用 MCP 方案:

在 Amazon Q Developer 中输入:

我们的 RDS MySQL 8.0.40 什么时候过期?应该升级到哪个版本?

Amazon Q 调用 「check_version_status」 和 「get_upgrade_recommendations」 工具,返回:

业务提升点:

  • 从多个文档页面查找信息简化为一次对话
  • 获得可执行的升级建议和文档链接
  • 决策时间从数小时降低到几分钟

场景 3:合规审计准备

业务场景:

在安全合规审计前,需要生成完整的服务版本生命周期报告,证明所有服务处于支持状态。

传统方式:

  • 手动收集所有服务的版本信息
  • 逐一查询 EOL 日期
  • 在 Excel 中制作报表
  • 数据可能已过时(距离上次更新 1-2 周)

使用 MCP 方案:

在 Amazon Q Developer 中输入:

生成一份完整的 AWS 中国区服务 EOL 状态报告

Amazon Q 调用 「generate_eol_report」 工具,返回:

业务提升点:

  • 报告生成从 2-3 小时降低到 10 秒
  • 数据实时性从周级提升到日级
  • 自动格式化,可直接用于审计提交
  • 包含数据来源和更新时间,增强可信度

经验总结

通过本方案的实施,我们在云服务版本生命周期管理方面获得了显著提升:

运维效率提升

  • 自动化程度:从人工查询提升为自动数据采集和更新,运维人员投入时间减少 90%
  • 响应速度:EOL 信息查询从分钟级降低到秒级,决策更加敏捷
  • 数据准确性:每日自动更新,消除人为错误,数据准确率接近 100%

技术方案特点

  • 无服务器架构:Lambda + S3 的组合实现了低成本、高可用的数据采集,无需维护服务器
  • 多层容错机制:S3 → 缓存 → 静态数据的三层架构确保服务始终可用
  • 标准化集成:基于 MCP 协议,可与任何支持 MCP 的 AI 助手集成,不局限于特定工具
  • 自然语言交互:降低使用门槛,运维人员无需学习新的工具界面

推广建议

  • 小范围试点:建议先在少数关键服务(如 EKS、Lambda)上试点,验证方案可行性
  • 数据质量优先:确保 Lambda 爬虫的可靠性,建立监控告警机制
  • 用户培训:提供自然语言查询示例,帮助团队快速上手
  • 持续优化:根据实际使用情况调整 MCP 工具的功能和输出格式

后续扩展方向

  • 告警集成:将 EOL 预警推送到企业微信、钉钉等 IM 工具
  • 自动化升级:结合 Amazon Systems Manager,实现半自动化的版本升级流程
  • 成本分析:统计过期版本可能带来的安全和合规成本
  • 多云支持:扩展支持其他云服务商的 EOL 追踪

________________________________________________________________________________

获取完整方案

本文所述方案已在生产环境中验证并取得良好效果。如果您希望获取完整的技术方案、部署指南或进行技术交流,请联系您的技术客户经理。

*前述特定亚马逊云科技生成式人工智能相关的服务目前在亚马逊云科技海外区域可用。亚马逊云科技中国区域相关云服务由西云数据和光环新网运营,具体信息以中国区域官网为准。

本篇作者

巫佳杰

西云数据技术客户经理,拥有超过 13 年企业通信和云技术领域的客户成功经验。专注于为亚马逊云科技中国区企业客户提供技术支持和解决方案咨询,帮助客户充分发挥云平台的技术价值。擅长云架构设计、运维治理与自动化和云财务管理,热衷于探索 AI 驱动的云管理创新实践。