Cloud Manage MCP Server

Instructions

获取AWS EC2实例的监控数据

Args:
    instance_id (str): EC2实例ID
    hours (int): 获取过去多少小时的数据
    
Returns:
    Dict: 监控数据

Implementation Reference

• main.py:354-367 (handler)

  MCP tool registration and handler for 'get_aws_instance_monitoring'. This thin wrapper delegates the actual logic to the aws_provider instance.

  @mcp.tool()
  def get_aws_instance_monitoring(instance_id: str, hours: int = 1) -> Dict:
      """
      获取AWS EC2实例的监控数据
      
      Args:
          instance_id (str): EC2实例ID
          hours (int): 获取过去多少小时的数据
          
      Returns:
          Dict: 监控数据
      """
      return aws_provider.get_instance_monitoring_data(instance_id, hours)

• providers/aws_provider.py:274-342 (helper)

  Core implementation in AWSProvider class that fetches CloudWatch metrics (CPU, Network, Disk) for the given instance over the specified hours using boto3.

  def get_instance_monitoring_data(self, instance_id: str, hours: int = 1) -> Dict:
      """
      获取实例的监控数据
      
      Args:
          instance_id (str): EC2实例ID
          hours (int): 获取过去多少小时的数据
          
      Returns:
          Dict: 监控数据或错误信息
      """
      if not self.available:
          return {
              'error': f'AWS服务不可用: {getattr(self, "error", "未知错误")}',
              'provider': 'aws'
          }
      
      try:
          end_time = datetime.utcnow()
          start_time = end_time - timedelta(hours=hours)
          
          metrics = ['CPUUtilization', 'NetworkIn', 'NetworkOut', 'DiskReadOps', 'DiskWriteOps']
          monitoring_data = {}
          
          for metric in metrics:
              try:
                  response = self.cloudwatch.get_metric_statistics(
                      Namespace='AWS/EC2',
                      MetricName=metric,
                      Dimensions=[
                          {
                              'Name': 'InstanceId',
                              'Value': instance_id
                          }
                      ],
                      StartTime=start_time,
                      EndTime=end_time,
                      Period=300,  # 5分钟
                      Statistics=['Average', 'Maximum']
                  )
                  
                  monitoring_data[metric] = {
                      'datapoints': len(response['Datapoints']),
                      'data': response['Datapoints']
                  }
              except Exception as e:
                  monitoring_data[metric] = {
                      'error': str(e),
                      'datapoints': 0
                  }
          
          return {
              'provider': 'aws',
              'instance_id': instance_id,
              'time_range': f'{hours}小时',
              'metrics': monitoring_data
          }
          
      except ClientError as e:
          return {
              'error': f'AWS API调用失败: {str(e)}',
              'provider': 'aws'
          }
      except Exception as e:
          return {
              'error': f'获取监控数据时发生错误: {str(e)}',
              'provider': 'aws'
          }