Cloud Manage MCP Server

Instructions

Implementation Reference

• main.py:527-539 (handler)

  MCP tool handler and registration for 'reboot_vultr_instance'. This function defines the tool interface, input schema via type hints, and delegates to the Vultr provider's reboot_instance method with security confirmation parameters.
  def reboot_vultr_instance( instance_id: str, ip_confirmation: str = "", name_confirmation: str = "", operation_confirmation: str = "" ) -> Dict: """ 重启Vultr实例（需要三次确认） """ return vultr_provider.reboot_instance( instance_id, ip_confirmation, name_confirmation, operation_confirmation )
• providers/vultr_provider.py:214-227 (helper)

  VultrProvider.reboot_instance method: wrapper that calls the shared _execute_power_operation with 'reboot' action, handling input parameters for security checks.
  def reboot_instance( self, instance_id: str, ip_confirmation: str = "", name_confirmation: str = "", operation_confirmation: str = "" ) -> Dict: """ 重启Vultr实例（需要三次确认） """ return self._execute_power_operation( instance_id, 'reboot', ip_confirmation, name_confirmation, operation_confirmation )
• providers/vultr_provider.py:228-334 (handler)

  Core handler logic for all Vultr power operations (including reboot). Fetches instance info, enforces triple confirmation security via utils.security, validates inputs, and executes the API call to Vultr's instances/{id}/actions endpoint.
  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' }