Spaces:
Sleeping
Sleeping
File size: 7,557 Bytes
d7b3d84 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 |
"""Cloud browser service integration for browser-use.
This module provides integration with the browser-use cloud browser service.
When cloud_browser=True, it automatically creates a cloud browser instance
and returns the CDP URL for connection.
"""
import logging
import os
import httpx
from browser_use.browser.cloud.views import CloudBrowserAuthError, CloudBrowserError, CloudBrowserResponse, CreateBrowserRequest
from browser_use.sync.auth import CloudAuthConfig
logger = logging.getLogger(__name__)
class CloudBrowserClient:
"""Client for browser-use cloud browser service."""
def __init__(self, api_base_url: str = 'https://api.browser-use.com'):
self.api_base_url = api_base_url
self.client = httpx.AsyncClient(timeout=30.0)
self.current_session_id: str | None = None
async def create_browser(
self, request: CreateBrowserRequest, extra_headers: dict[str, str] | None = None
) -> CloudBrowserResponse:
"""Create a new cloud browser instance. For full docs refer to https://docs.cloud.browser-use.com/api-reference/v-2-api-current/browsers/create-browser-session-browsers-post
Args:
request: CreateBrowserRequest object containing browser creation parameters
Returns:
CloudBrowserResponse: Contains CDP URL and other browser info
"""
url = f'{self.api_base_url}/api/v2/browsers'
# Try to get API key from environment variable first, then auth config
api_token = os.getenv('BROWSER_USE_API_KEY')
if not api_token:
# Fallback to auth config file
try:
auth_config = CloudAuthConfig.load_from_file()
api_token = auth_config.api_token
except Exception:
pass
if not api_token:
raise CloudBrowserAuthError(
'No authentication token found. Please set BROWSER_USE_API_KEY environment variable to authenticate with the cloud service. You can also create an API key at https://cloud.browser-use.com/new-api-key'
)
headers = {'X-Browser-Use-API-Key': api_token, 'Content-Type': 'application/json', **(extra_headers or {})}
# Convert request to dictionary and exclude unset fields
request_body = request.model_dump(exclude_unset=True)
try:
logger.info('🌤️ Creating cloud browser instance...')
response = await self.client.post(url, headers=headers, json=request_body)
if response.status_code == 401:
raise CloudBrowserAuthError(
'Authentication failed. Please make sure you have set BROWSER_USE_API_KEY environment variable to authenticate with the cloud service. You can also create an API key at https://cloud.browser-use.com/new-api-key'
)
elif response.status_code == 403:
raise CloudBrowserAuthError('Access forbidden. Please check your browser-use cloud subscription status.')
elif not response.is_success:
error_msg = f'Failed to create cloud browser: HTTP {response.status_code}'
try:
error_data = response.json()
if 'detail' in error_data:
error_msg += f' - {error_data["detail"]}'
except Exception:
pass
raise CloudBrowserError(error_msg)
browser_data = response.json()
browser_response = CloudBrowserResponse(**browser_data)
# Store session ID for cleanup
self.current_session_id = browser_response.id
logger.info(f'🌤️ Cloud browser created successfully: {browser_response.id}')
logger.debug(f'🌤️ CDP URL: {browser_response.cdpUrl}')
# Cyan color for live URL
logger.info(f'\033[36m🔗 Live URL: {browser_response.liveUrl}\033[0m')
return browser_response
except httpx.TimeoutException:
raise CloudBrowserError('Timeout while creating cloud browser. Please try again.')
except httpx.ConnectError:
raise CloudBrowserError('Failed to connect to cloud browser service. Please check your internet connection.')
except Exception as e:
if isinstance(e, (CloudBrowserError, CloudBrowserAuthError)):
raise
raise CloudBrowserError(f'Unexpected error creating cloud browser: {e}')
async def stop_browser(
self, session_id: str | None = None, extra_headers: dict[str, str] | None = None
) -> CloudBrowserResponse:
"""Stop a cloud browser session.
Args:
session_id: Session ID to stop. If None, uses current session.
Returns:
CloudBrowserResponse: Updated browser info with stopped status
Raises:
CloudBrowserAuthError: If authentication fails
CloudBrowserError: If stopping fails
"""
if session_id is None:
session_id = self.current_session_id
if not session_id:
raise CloudBrowserError('No session ID provided and no current session available')
url = f'{self.api_base_url}/api/v2/browsers/{session_id}'
# Try to get API key from environment variable first, then auth config
api_token = os.getenv('BROWSER_USE_API_KEY')
if not api_token:
# Fallback to auth config file
try:
auth_config = CloudAuthConfig.load_from_file()
api_token = auth_config.api_token
except Exception:
pass
if not api_token:
raise CloudBrowserAuthError(
'No authentication token found. Please set BROWSER_USE_API_KEY environment variable to authenticate with the cloud service. You can also create an API key at https://cloud.browser-use.com/new-api-key'
)
headers = {'X-Browser-Use-API-Key': api_token, 'Content-Type': 'application/json', **(extra_headers or {})}
request_body = {'action': 'stop'}
try:
logger.info(f'🌤️ Stopping cloud browser session: {session_id}')
response = await self.client.patch(url, headers=headers, json=request_body)
if response.status_code == 401:
raise CloudBrowserAuthError(
'Authentication failed. Please make sure you have set the BROWSER_USE_API_KEY environment variable to authenticate with the cloud service.'
)
elif response.status_code == 404:
# Session already stopped or doesn't exist - treating as error and clearing session
logger.debug(f'🌤️ Cloud browser session {session_id} not found (already stopped)')
# Clear current session if it was this one
if session_id == self.current_session_id:
self.current_session_id = None
raise CloudBrowserError(f'Cloud browser session {session_id} not found')
elif not response.is_success:
error_msg = f'Failed to stop cloud browser: HTTP {response.status_code}'
try:
error_data = response.json()
if 'detail' in error_data:
error_msg += f' - {error_data["detail"]}'
except Exception:
pass
raise CloudBrowserError(error_msg)
browser_data = response.json()
browser_response = CloudBrowserResponse(**browser_data)
# Clear current session if it was this one
if session_id == self.current_session_id:
self.current_session_id = None
logger.info(f'🌤️ Cloud browser session stopped: {browser_response.id}')
logger.debug(f'🌤️ Status: {browser_response.status}')
return browser_response
except httpx.TimeoutException:
raise CloudBrowserError('Timeout while stopping cloud browser. Please try again.')
except httpx.ConnectError:
raise CloudBrowserError('Failed to connect to cloud browser service. Please check your internet connection.')
except Exception as e:
if isinstance(e, (CloudBrowserError, CloudBrowserAuthError)):
raise
raise CloudBrowserError(f'Unexpected error stopping cloud browser: {e}')
async def close(self):
"""Close the HTTP client and cleanup any active sessions."""
# Try to stop current session if active
if self.current_session_id:
try:
await self.stop_browser()
except Exception as e:
logger.debug(f'Failed to stop cloud browser session during cleanup: {e}')
await self.client.aclose()
|