基于Bedrock Agentcore 实现智能成本分析与告警系统

def detect_cost_anomalies(days: int = 7) -> str:
    """
    检测AWS成本异常，使用机器学习算法识别异常模式
    """
    try:
        ce_client = boto3.client("ce")
        
        # 计算时间范围
        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=days)
        
        # 获取异常检测结果
        response = ce_client.get_anomalies(
            DateInterval={
                'StartDate': start_date.strftime('%Y-%m-%d'),
                'EndDate': end_date.strftime('%Y-%m-%d')
            },
            MaxResults=50
        )
        
        anomalies = response.get('Anomalies', [])
        
        if not anomalies:
            return f"过去{days}天未检测到成本异常"
        
        # 分析异常详情
        analysis = []
        total_impact = 0
        
        for anomaly in anomalies:
            impact = float(anomaly.get('Impact', {}).get('MaxImpact', 0))
            total_impact += impact
            
            # 获取异常根因分析
            root_causes = []
            for cause in anomaly.get('RootCauses', []):
                service = cause.get('Service', 'Unknown')
                region = cause.get('Region', 'Unknown')
                usage_type = cause.get('UsageType', 'Unknown')
                root_causes.append(f"{service} ({region}) - {usage_type}")
            
            analysis.append({
                'date': anomaly.get('AnomalyStartDate'),
                'impact': impact,
                'score': anomaly.get('AnomalyScore', {}).get('MaxScore', 0),
                'root_causes': root_causes
            })
        
        # 生成分析报告
        report = f"检测到 {len(anomalies)} 个成本异常\n"
        report += f"总影响金额: ${total_impact:.2f}\n\n"
        
        for i, item in enumerate(analysis[:3], 1):
            report += f"异常 {i}:\n"
            report += f" 日期: {item['date']}\n"
            report += f" 影响: ${item['impact']:.2f}\n"
            report += f" 异常分数: {item['score']:.1f}\n"
            report += f" 根因: {', '.join(item['root_causes'][:2])}\n\n"
        
        return report
        
    except Exception as e:
        return f"异常检测失败: {str(e)}"

def get_multi_account_costs(days: int = 30) -> str:
# 获取所有关联账户
        try:
            accounts_response = org_client.list_accounts()
            all_accounts = {
                acc['Id']: acc['Name'] 
                for acc in accounts_response['Accounts']
            }
        except Exception:
            # 如果不是组织主账户，使用当前账户
            sts_client = boto3.client("sts")
            current_account = sts_client.get_caller_identity()['Account']
            all_accounts = {current_account: "Current Account"}
        
        # 按账户分组获取成本数据
        response = ce_client.get_cost_and_usage(
            TimePeriod={
                'Start': start_date.strftime('%Y-%m-%d'),
                'End': end_date.strftime('%Y-%m-%d')
            },
            Granularity='MONTHLY',
            Metrics=['UnblendedCost'],
            GroupBy=[
                {'Type': 'DIMENSION', 'Key': 'LINKED_ACCOUNT'}
            ]
        )
      # 按成本排序
        sorted_accounts = sorted(
            account_costs.items(), 
            key=lambda x: x[1], 
            reverse=True
        )
        report = f"多账户成本分析 (过去{days}天)\n"
        report += f"总成本: ${total_cost:.2f}\n"
        report += f"账户数量: {len(account_costs)}\n\n"
        
    return report

其他工具功能说明