import asyncio
import requests
from browser_use.browser.browser import BrowserSession, BrowserProfile
class AIOSandboxBrowser:
def __init__(self, server_url="http://localhost:8080"):
self.server_url = server_url
self.browser_session = None
# Browser configuration
self.profile = {
"extra_http_headers": {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
},
"ignore_https_errors": True,
"java_script_enabled": True,
"viewport": {"width": 1920, "height": 1080},
}
def get_cdp_url(self):
"""Get CDP WebSocket URL from AIO Sandbox"""
response = requests.get(f"{self.server_url}/cdp/json/version", timeout=10)
response.raise_for_status()
data = response.json()
cdp_url = data.get("webSocketDebuggerUrl")
if cdp_url:
print(f"CDP URL: {cdp_url}")
return cdp_url
else:
raise Exception("Could not get CDP URL from sandbox")
async def initialize(self):
"""Initialize browser session"""
if self.browser_session is None:
cdp_url = self.get_cdp_url()
self.browser_session = BrowserSession(
browser_profile=BrowserProfile(**self.profile),
cdp_url=cdp_url
)
await self.browser_session.start()
print("Browser session initialized")
return self.browser_session
async def navigate_and_screenshot(self, url, screenshot_path="screenshot.png"):
"""Navigate to URL and take screenshot"""
await self.initialize()
# Create new page
page = await self.browser_session.browser_context.new_page()
# Set viewport and headers
await page.set_viewport_size({"width": 1920, "height": 1080})
await page.set_extra_http_headers({
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.9',
})
try:
# Navigate to URL
await page.goto(url, wait_until="networkidle")
# Wait for page to be fully loaded
await page.wait_for_load_state("domcontentloaded")
await page.wait_for_timeout(2000) # Additional wait for rendering
# Take screenshot
await page.screenshot(path=screenshot_path, full_page=True)
# Get page content
content = await page.content()
title = await page.title()
return {
"url": url,
"title": title,
"screenshot": screenshot_path,
"content_length": len(content)
}
finally:
await page.close()
async def extract_page_data(self, url, selectors=None):
"""Extract specific data from page using CSS selectors"""
await self.initialize()
page = await self.browser_session.browser_context.new_page()
try:
await page.goto(url, wait_until="networkidle")
data = {"url": url, "title": await page.title()}
if selectors:
for name, selector in selectors.items():
try:
element = await page.query_selector(selector)
if element:
data[name] = await element.text_content()
else:
data[name] = None
except Exception as e:
data[name] = f"Error: {e}"
return data
finally:
await page.close()
async def close(self):
"""Close browser session"""
if self.browser_session:
await self.browser_session.close()
self.browser_session = None
# Usage example
async def main():
browser = AIOSandboxBrowser()
try:
# Take screenshot of a website
result = await browser.navigate_and_screenshot("https://example.com")
print(f"Screenshot saved: {result}")
# Extract specific data
selectors = {
"heading": "h1",
"description": "p",
"links": "a"
}
data = await browser.extract_page_data("https://example.com", selectors)
print(f"Extracted data: {data}")
finally:
await browser.close()
# Run the example
asyncio.run(main())