Cloud Manage MCP Server

Instructions

Implementation Reference

• main.py:498-510 (registration)

  MCP tool registration for 'power_on_vultr_instance'. This decorator registers the function as an MCP tool.

  @mcp.tool()
  def power_on_vultr_instance(
      instance_id: str, 
      ip_confirmation: str = "", 
      name_confirmation: str = "", 
      operation_confirmation: str = ""
  ) -> Dict:
      """
      开启Vultr实例（需要三次确认）
      """
      return vultr_provider.power_on_instance(
          instance_id, ip_confirmation, name_confirmation, operation_confirmation
      )

• providers/vultr_provider.py:186-198 (handler)

  Provider-level handler for powering on a Vultr instance. Delegates to the shared _execute_power_operation helper with 'start' action.

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

• providers/vultr_provider.py:228-334 (helper)

  Core helper function that handles instance retrieval, triple confirmation validation using utils.security, and executes the Vultr API call for power operations (start, halt, reboot). This is the exact implementation of the power-on logic.

  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'Vultr服务不可用: {getattr(self, "error", "未知错误")}',
              'provider': 'vultr'
          }
      
      # 首先获取实例信息
      try:
          response = requests.get(f'{self.base_url}/instances/{instance_id}', headers=self.headers, timeout=10)
          
          if response.status_code == 404:
              return {
                  'error': f'未找到ID为 {instance_id} 的Vultr实例',
                  'provider': 'vultr'
              }
          
          if response.status_code != 200:
              return {
                  'error': f'获取实例信息失败: {response.status_code} - {response.text}',
                  'provider': 'vultr'
              }
          
          data = response.json()
          instance = data.get('instance', {})
          
          # 格式化实例信息用于确认
          instance_info = self._format_instance_for_confirmation(instance)
          
      except Exception as e:
          return {
              'error': f'获取Vultr实例信息时发生错误: {str(e)}',
              'provider': 'vultr'
          }
      
      # 检查是否提供了确认信息
      if not ip_confirmation or not name_confirmation or not operation_confirmation:
          # 转换操作名称
          operation_mapping = {
              'start': 'power_on',
              'halt': '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',
          'halt': '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': 'vultr',
              'requires_confirmation': True
          }
      
      # 执行实际操作
      try:
          operation_data = {'action': operation}
          response = requests.post(
              f'{self.base_url}/instances/{instance_id}/actions',
              headers=self.headers,
              json=operation_data,
              timeout=10
          )
          
          if response.status_code not in [200, 202, 204]:
              return {
                  'error': f'执行 {operation} 操作失败: {response.status_code} - {response.text}',
                  'provider': 'vultr'
              }
          
          return {
              'provider': 'vultr',
              'instance_id': instance_id,
              'operation_success': True,
              'operation': operation,
              'message': f'已成功提交 {operation} 操作',
              'confirmation_validated': True
          }
          
      except Exception as e:
          return {
              'error': f'执行 {operation} 操作时发生错误: {str(e)}',
              'provider': 'vultr'
          }