Cloud Manage MCP Server

Instructions

Implementation Reference

• main.py:571-584 (registration)

  MCP registration of the 'power_on_alibaba_instance' tool. This is the entry point handler that delegates to the AlibabaProvider's power_on_instance method with security confirmation parameters.

  @mcp.tool()
  def power_on_alibaba_instance(
      instance_id: str, 
      ip_confirmation: str = "", 
      name_confirmation: str = "", 
      operation_confirmation: str = ""
  ) -> Dict:
      """
      启动阿里云ECS实例（需要三次确认）
      """
      return alibaba_provider.power_on_instance(
          instance_id, ip_confirmation, name_confirmation, operation_confirmation
      )

• providers/alibaba_provider.py:202-214 (handler)

  Handler method in AlibabaProvider class that handles power-on logic by invoking the generic _execute_power_operation with 'start' action.

  def power_on_instance(
      self, 
      instance_id: str, 
      ip_confirmation: str = "", 
      name_confirmation: str = "", 
      operation_confirmation: str = ""
  ) -> Dict:
      """
      启动ECS实例（需要三次确认）
      """
      return self._execute_power_operation(
          instance_id, 'start', ip_confirmation, name_confirmation, operation_confirmation
      )

• providers/alibaba_provider.py:244-358 (handler)

  Core handler for all power operations on Alibaba ECS instances. Performs triple security confirmation, validates inputs, and executes the actual API call via Alibaba SDK (StartInstance for power-on).

  def _execute_power_operation(
      self, 
      instance_id: str, 
      operation: str, 
      ip_confirmation: str, 
      name_confirmation: str, 
      operation_confirmation: str
  ) -> Dict:
      """
      执行电源操作的通用函数
      """
      if not self.available:
          return {
              'error': f'阿里云服务不可用: {getattr(self, "error", "未知错误")}',
              'provider': 'alibaba'
          }
      
      # 首先获取实例信息
      try:
          request = ecs_models.DescribeInstancesRequest(
              region_id=self.region_id,
              instance_ids=json.dumps([instance_id])
          )
          
          response = self.client.describe_instances(request)
          
          if not response.body.instances or not response.body.instances.instance:
              return {
                  'error': f'未找到ID为 {instance_id} 的ECS实例',
                  'provider': 'alibaba'
              }
          
          instance = response.body.instances.instance[0]
          
          # 格式化实例信息用于确认
          instance_info = self._format_instance_for_confirmation(instance)
          
      except Exception as e:
          return {
              'error': f'获取ECS实例信息时发生错误: {str(e)}',
              'provider': 'alibaba'
          }
      
      # 检查是否提供了确认信息
      if not ip_confirmation or not name_confirmation or not operation_confirmation:
          # 转换操作名称
          operation_mapping = {
              'start': 'power_on',
              'stop': 'power_off',
              'reboot': 'reboot'
          }
          mapped_operation = operation_mapping.get(operation, operation)
          return require_triple_confirmation(instance_info, mapped_operation)
      
      # 验证确认信息
      security = SecurityConfirmation()
      operation_mapping = {
          'start': 'power_on',
          'stop': 'power_off',
          'reboot': 'reboot'
      }
      mapped_operation = operation_mapping.get(operation, operation)
      
      is_valid, error_message = security.validate_power_operation(
          instance_info, mapped_operation, ip_confirmation, name_confirmation, operation_confirmation
      )
      
      if not is_valid:
          return {
              'error': f'确认验证失败: {error_message}',
              'provider': 'alibaba',
              'requires_confirmation': True
          }
      
      # 执行实际操作
      try:
          if operation == 'start':
              request = ecs_models.StartInstanceRequest(
                  instance_id=instance_id
              )
              response = self.client.start_instance(request)
          elif operation == 'stop':
              request = ecs_models.StopInstanceRequest(
                  instance_id=instance_id,
                  force_stop=True
              )
              response = self.client.stop_instance(request)
          elif operation == 'reboot':
              request = ecs_models.RebootInstanceRequest(
                  instance_id=instance_id,
                  force_stop=True
              )
              response = self.client.reboot_instance(request)
          else:
              return {
                  'error': f'不支持的操作类型: {operation}',
                  'provider': 'alibaba'
              }
          
          return {
              'provider': 'alibaba',
              'instance_id': instance_id,
              'operation_success': True,
              'operation': operation,
              'request_id': response.body.request_id,
              'message': f'已成功提交 {operation} 操作',
              'confirmation_validated': True
          }
          
      except Exception as e:
          return {
              'error': f'执行 {operation} 操作时发生错误: {str(e)}',
              'provider': 'alibaba'
          }

• utils/security.py:12-69 (helper)

  Helper function for validating triple confirmation inputs used in power operations.

  @staticmethod
  def validate_power_operation(
      instance_info: Dict,
      operation: str,
      ip_confirmation: str,
      name_confirmation: str,
      operation_confirmation: str
  ) -> Tuple[bool, str]:
      """
      验证电源操作的三次确认
      
      Args:
          instance_info (dict): 实例信息
          operation (str): 操作类型 ('shutdown', 'reboot', 'power_off', 'power_on')
          ip_confirmation (str): 用户确认的IP地址
          name_confirmation (str): 用户确认的实例名称
          operation_confirmation (str): 用户确认的操作类型
          
      Returns:
          Tuple[bool, str]: (是否通过验证, 错误信息)
      """
      errors = []
      
      # 获取实例的实际信息
      actual_ip = instance_info.get('public_ip', instance_info.get('public_ipv4', '未知'))
      actual_name = instance_info.get('name', instance_info.get('instance_name', '未知'))
      
      # 第一次确认：IP地址
      if ip_confirmation.strip() != actual_ip:
          errors.append(f"IP地址确认失败: 期望 '{actual_ip}', 但收到 '{ip_confirmation}'")
      
      # 第二次确认：实例名称
      if name_confirmation.strip() != actual_name:
          errors.append(f"实例名称确认失败: 期望 '{actual_name}', 但收到 '{name_confirmation}'")
      
      # 第三次确认：操作类型
      valid_operations = {
          'shutdown': ['shutdown', '关机', '优雅关机'],
          'power_off': ['power_off', '强制关机', '强制断电'],
          'reboot': ['reboot', '重启', '重新启动'],
          'power_on': ['power_on', '开机', '启动']
      }
      
      operation_confirmed = False
      for op, variations in valid_operations.items():
          if op == operation and operation_confirmation.strip().lower() in [v.lower() for v in variations]:
              operation_confirmed = True
              break
      
      if not operation_confirmed:
          expected_ops = valid_operations.get(operation, [operation])
          errors.append(f"操作确认失败: 期望 '{expected_ops[0]}', 但收到 '{operation_confirmation}'")
      
      if errors:
          return False, "; ".join(errors)
      
      return True, "所有确认项目验证通过"

• utils/security.py:158-190 (helper)

  Helper function that generates the triple confirmation requirement response when user hasn't provided confirmations.

  def require_triple_confirmation(instance_info: Dict, operation: str) -> Dict[str, any]:
      """
      生成三重确认要求的完整信息
      
      Args:
          instance_info (dict): 实例信息
          operation (str): 操作类型
          
      Returns:
          Dict[str, any]: 确认要求信息
      """
      security = SecurityConfirmation()
      
      # 获取确认提示
      prompt = security.create_confirmation_prompt(instance_info, operation)
      
      # 检查操作安全性
      is_safe, safety_level, warnings = security.check_operation_safety(instance_info, operation)
      
      return {
          'requires_confirmation': True,
          'confirmation_prompt': prompt,
          'safety_check': {
              'is_safe': is_safe,
              'safety_level': safety_level,
              'warnings': warnings
          },
          'confirmation_format': {
              'ip_confirmation': f"请输入目标IP地址: {prompt['target_ip']}",
              'name_confirmation': f"请输入实例名称: {prompt['target_name']}", 
              'operation_confirmation': f"请输入操作类型: {prompt['operation_name']}"
          }
      }