Cloud Manage MCP Server

Instructions

Implementation Reference

• main.py:512-524 (handler)

  MCP tool handler for 'power_off_vultr_instance'. Registers the tool and delegates execution to vultr_provider.power_off_instance with security confirmations.

  @mcp.tool()
  def power_off_vultr_instance(
      instance_id: str, 
      ip_confirmation: str = "", 
      name_confirmation: str = "", 
      operation_confirmation: str = ""
  ) -> Dict:
      """
      强制关闭Vultr实例（需要三次确认）
      """
      return vultr_provider.power_off_instance(
          instance_id, ip_confirmation, name_confirmation, operation_confirmation
      )

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

  Core implementation of power operations for Vultr instances, including triple confirmation check, API call to halt the instance via POST /instances/{id}/actions with {'action': 'halt'}.

  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'
          }

• providers/vultr_provider.py:200-212 (helper)

  VultrProvider method that handles power_off logic by calling the shared _execute_power_operation with 'halt' action.

  def power_off_instance(
      self, 
      instance_id: str, 
      ip_confirmation: str = "", 
      name_confirmation: str = "", 
      operation_confirmation: str = ""
  ) -> Dict:
      """
      强制关闭Vultr实例（需要三次确认）
      """
      return self._execute_power_operation(
          instance_id, 'halt', ip_confirmation, name_confirmation, operation_confirmation
      )

• providers/vultr_provider.py:432-432 (registration)

  Global instance creation of VultrProvider, imported and used by main.py tool handlers.

  # 全局实例

• main.py:14-14 (registration)

  Import of vultr_provider module in main.py, enabling tool delegation.

  from providers.vultr_provider import vultr_provider