Browser Automation

from marsys.agents import BrowserAgent, BrowserAgentMode

# Mode selection with enum (type-safe)
browser_agent = await BrowserAgent.create_safe(
    model_config=config,
    name="scraper",
    mode=BrowserAgentMode.PRIMITIVE,  # Using enum
    goal="Efficiently fetch and extract content from web pages"
)

# Mode selection with string (convenient)
browser_agent = await BrowserAgent.create_safe(
    model_config=config,
    name="scraper",
    mode="primitive",  # Using string
    goal="Efficiently fetch and extract content from web pages"
)

# ADVANCED mode - Visual interaction
browser_agent = await BrowserAgent.create_safe(
    model_config=config,  # Main agent model (Claude Haiku/Sonnet recommended)
    name="navigator",
    mode=BrowserAgentMode.ADVANCED,  # or mode="advanced"
    auto_screenshot=True,  # Enable visual feedback
    vision_model_config=ModelConfig(  # Vision model for screenshot analysis
        type="api",
        provider="openrouter",
        name="google/gemini-3-flash-preview",  # Recommended: fast and cost-effective
        # For complex tasks, use: "google/gemini-3-pro-preview"
        temperature=0,
        thinking_budget=0  # Disable thinking for faster vision responses
    ),
    goal="Navigate and interact with web pages like a human"
)

from marsys.agents import BrowserAgent
from marsys.models import ModelConfig

# PRIMITIVE Mode - Fast content extraction
browser_agent = await BrowserAgent.create_safe(
    model_config=ModelConfig(
        type="api",
        provider="openrouter",
        name="anthropic/claude-opus-4.6",
        temperature=0.3
    ),
    name="web_scraper",
    mode="primitive",  # Simple string mode selection
    goal="Fast web scraping agent for content extraction",
    headless=True,
    tmp_dir="./runs/run-20260206"
)

# ADVANCED Mode - Visual interaction with auto-screenshot
browser_agent_advanced = await BrowserAgent.create_safe(
    model_config=ModelConfig(
        type="api",
        provider="openrouter",
        name="anthropic/claude-opus-4.6",  # Main agent for decision-making and planning
        temperature=0.3
    ),
    name="web_navigator",
    mode="advanced",  # Simple string mode selection
    goal="Expert web automation agent for complex interactions",
    auto_screenshot=True,  # Enable visual feedback
    vision_model_config=ModelConfig(  # Required for auto-screenshot
        type="api",
        provider="openrouter",
        name="google/gemini-3-flash-preview",  # Recommended: fast and cost-effective for browser vision
        # For complex tasks, use: "google/gemini-3-pro-preview"
        temperature=0,
        thinking_budget=0  # Disable thinking for faster vision responses
    ),
    headless=False,
    tmp_dir="./runs/run-20260206"
)

# Always clean up
try:
    # Use the agent
    result = await browser_agent.run("Navigate to example.com and extract the main heading")
finally:
    if browser_agent.browser_tool:
        await browser_agent.browser_tool.close()

browser_agent = await BrowserAgent.create_safe(
    model_config=config,
    name="web_scraper",
    mode="primitive",
    tmp_dir="./runs/run-20260206",
    downloads_subdir="downloads",             # Host folder under tmp_dir
    downloads_virtual_dir="./downloads",      # Path shown to the agent
    fetch_file_tool_name="fetch_file",        # Expose download tool under custom name
)

from marsys.agents import AgentPool

# Create pool of browser agents
browser_pool = AgentPool(
    agent_class=BrowserAgent,
    num_instances=3,
    model_config=config,
    agent_name="BrowserPool",
    headless=True
)

# Parallel scraping
async def scrape_urls(urls: List[str]):
    tasks = []
    for i, url in enumerate(urls):
        async with browser_pool.acquire(f"branch_{i}") as agent:
            task = agent.run(f"Scrape content from {url}")
            tasks.append(task)

    results = await asyncio.gather(*tasks)
    return results

# Cleanup pool
await browser_pool.cleanup()

class NavigationAgent(BrowserAgent):
    """Agent with navigation capabilities."""

    async def navigate_with_history(self, urls: List[str], context):
        """Navigate through multiple pages with history."""
        for url in urls:
            await self.browser_tool.goto(url)
            await self._log_progress(context, LogLevel.INFO, f"Navigated to {url}")

            # Wait for page to load
            await self.browser_tool.wait_for_navigation()

            # Take screenshot for debugging
            await self.browser_tool.screenshot(
                filename=f"{url.replace('/', '_')}.png"
            )

        # Navigate back through history
        for _ in range(len(urls) - 1):
            await self.browser_tool.go_back()
            current = await self.browser_tool.get_url()
            await self._log_progress(context, LogLevel.INFO, f"Back to {current}")

class InteractionAgent(BrowserAgent):
    """Agent for web interactions."""

    async def smart_form_fill(self, form_data: Dict, context):
        """Intelligently fill forms based on field types."""

        for field_name, value in form_data.items():
            # Try different selector strategies
            selectors = [
                f"input[name='{field_name}']",
                f"input[id='{field_name}']",
                f"textarea[name='{field_name}']",
                f"select[name='{field_name}']"
            ]

            for selector in selectors:
                try:
                    # Determine field type and fill appropriately
                    if selector.startswith("select"):
                        await self.browser_tool.select_option(selector, str(value))
                    elif isinstance(value, bool):
                        if value:  # Check if should be checked
                            await self.browser_tool.click(selector)
                    else:
                        await self.browser_tool.fill(selector, str(value))

                    await self._log_progress(
                        context, LogLevel.DEBUG,
                        f"Filled {field_name} with {value}"
                    )
                    break
                except Exception:
                    continue

    async def smart_click(self, text: str, context, element_type: str = "button"):
        """Click element by text content."""

        # XPath to find element by text
        xpath = f"//{element_type}[contains(text(), '{text}')]"

        try:
            await self.browser_tool.wait_for_selector(xpath, timeout=5000, state="visible")
            await self.browser_tool.click(xpath)
            await self._log_progress(context, LogLevel.INFO, f"Clicked '{text}' {element_type}")
        except Exception as e:
            # Fallback to JavaScript click
            script = f"""
            Array.from(document.querySelectorAll('{element_type}')).
                find(el => el.textContent.includes('{text}'))?.click()
            """
            await self.browser_tool.evaluate_javascript(script)

# Search for text on the current page
result = await browser_tool.search_page("quantum computing")
# Returns: "Match 1/5 found and highlighted"
# All matches highlighted in YELLOW, current match in ORANGE

# Navigate to next match - call again with SAME term
result = await browser_tool.search_page("quantum computing")
# Returns: "Match 2/5"
# Scrolls to and highlights next occurrence

# Continue navigating
result = await browser_tool.search_page("quantum computing")
# Returns: "Match 3/5"
# Wraps around after last match back to first

# Navigate to documentation page
await browser_tool.goto("https://docs.example.com/api")

# Search for specific API endpoint
result = await browser_tool.search_page("/api/v2/users")
# Match 1/3 found - scrolls to first occurrence

# Check if it's the right one with screenshot
screenshot = await browser_tool.screenshot()
# Visual: See highlighted text in orange

# Not the right one? Navigate to next match
result = await browser_tool.search_page("/api/v2/users")
# Match 2/3 - scrolls to second occurrence

# Clicking a download link automatically detects the download
result = await browser_tool.mouse_click(x=450, y=300)
# Returns: "Action 'mouse_click' triggered a file download.
#          File 'report.pdf' has been downloaded to: ./downloads/report.pdf"

# Navigating to a PDF URL triggers automatic download
result = await browser_tool.goto("https://example.com/paper.pdf")
# Returns: "Action 'goto' triggered a file download.
#          File 'paper.pdf' has been downloaded to: ./downloads/paper.pdf"

# Pressing Enter on a download button
await browser_tool.mouse_click(x=500, y=400)  # Focus download button
await browser_tool.keyboard_press("Enter")
# Returns: "Action 'keyboard_press' triggered a file download.
#          File 'data.xlsx' has been downloaded to: ./downloads/data.xlsx"

# List all files in the downloads directory
downloads = await browser_tool.list_downloads()
# Returns a formatted list with sizes and paths

# PDFs are NEVER displayed in browser - always downloaded
await browser_tool.goto("https://research.org/paper.pdf")
# Automatically downloads to ./downloads/paper.pdf
# Browser stays on previous page

# search_page() does NOT work with PDFs
# Instead, use file operation tools on the downloaded file

browser_tool = await BrowserTool.create_safe(
    downloads_path="/custom/path/downloads",  # Custom host download directory
    temp_dir="/custom/tmp",  # Custom temp directory (default: ./tmp)
    downloads_virtual_dir="./downloads",  # Virtual path returned to agents
)

class ScraperAgent(BrowserAgent):
    """Advanced web scraping agent."""

    async def extract_structured_data(self, url: str, schema: Dict, context):
        """Extract data according to schema."""

        await self.browser_tool.goto(url)
        await self.browser_tool.wait_for_navigation()

        # Extract based on schema
        extracted_data = {}

        for field_name, config in schema.items():
            selector = config.get('selector')
            attribute = config.get('attribute')
            multiple = config.get('multiple', False)

            try:
                if multiple:
                    # Extract from multiple elements via JS
                    if attribute:
                        script = f"""
                        Array.from(document.querySelectorAll({selector!r}))
                            .map(el => el.getAttribute({attribute!r}))
                        """
                    else:
                        script = f"""
                        Array.from(document.querySelectorAll({selector!r}))
                            .map(el => (el.textContent || '').trim())
                        """
                    extracted_data[field_name] = await self.browser_tool.evaluate_javascript(script)
                else:
                    # Extract from single element
                    if attribute:
                        value = await self.browser_tool.get_attribute(
                            selector, attribute
                        )
                    else:
                        value = await self.browser_tool.get_text(selector)

                    extracted_data[field_name] = value

            except Exception as e:
                await self._log_progress(
                    context, LogLevel.WARNING,
                    f"Failed to extract {field_name}: {e}"
                )
                extracted_data[field_name] = None

        return extracted_data

    async def extract_table_data(self, table_selector: str, context):
        """Extract data from HTML tables."""

        script = f"""
        () => {{
            const table = document.querySelector('{table_selector}');
            if (!table) return null;

            const headers = Array.from(table.querySelectorAll('th'))
                .map(th => th.textContent.trim());

            const rows = Array.from(table.querySelectorAll('tbody tr'))
                .map(tr => {{
                    const cells = Array.from(tr.querySelectorAll('td'));
                    const rowData = {{}};
                    cells.forEach((td, i) => {{
                        rowData[headers[i] || `col_${i}`] = td.textContent.trim();
                    }});
                    return rowData;
                }});

            return {{headers, rows}};
        }}
        """

        return await self.browser_tool.evaluate_javascript(script)

class PaginationAgent(BrowserAgent):
    """Handle paginated content."""

    async def scrape_all_pages(
        self,
        start_url: str,
        item_selector: str,
        next_button_selector: str,
        max_pages: int = 10,
        context = None
    ):
        """Scrape data across multiple pages."""

        all_items = []
        current_page = 1

        await self.browser_tool.goto(start_url)

        while current_page <= max_pages:
            # Wait for items to load
            await self.browser_tool.wait_for_selector(
                item_selector, timeout=10000, state="visible"
            )

            # Extract items from current page
            items = await self.browser_tool.evaluate_javascript(f"""
                Array.from(document.querySelectorAll('{item_selector}'))
                    .map(el => el.textContent.trim())
            """)

            all_items.extend(items)
            await self._log_progress(
                context, LogLevel.INFO,
                f"Page {current_page}: Extracted {len(items)} items"
            )

            # Check for next page
            try:
                await self.browser_tool.wait_for_selector(
                    next_button_selector, timeout=2000, state="visible"
                )
                await self.browser_tool.click(next_button_selector)
                await self.browser_tool.wait_for_navigation()
                current_page += 1
            except Exception:
                break

        return all_items

class DynamicContentAgent(BrowserAgent):
    """Handle JavaScript-heavy sites."""

    async def wait_for_ajax_content(
        self,
        content_selector: str,
        timeout: int = 30,
        context = None
    ):
        """Wait for AJAX content to load."""

        # Wait for a selector that indicates content has loaded
        await self.browser_tool.wait_for_selector(
            content_selector, timeout=timeout * 1000, state="visible"
        )

    async def infinite_scroll_scrape(
        self,
        item_selector: str,
        target_count: int,
        context = None
    ):
        """Handle infinite scroll patterns."""

        items_found = 0
        no_new_items_count = 0
        max_no_new = 3

        while items_found < target_count:
            # Count current items
            current_items = await self.browser_tool.evaluate_javascript(
                f"document.querySelectorAll('{item_selector}').length"
            )

            if current_items == items_found:
                no_new_items_count += 1
                if no_new_items_count >= max_no_new:
                    break
            else:
                no_new_items_count = 0
                items_found = current_items

            # Scroll to bottom
            await self.browser_tool.evaluate_javascript(
                "window.scrollTo(0, document.body.scrollHeight)"
            )

            # Wait for potential new content
            await asyncio.sleep(2)

            await self._log_progress(
                context, LogLevel.DEBUG,
                f"Found {items_found} items, target: {target_count}"
            )

        # Extract all items
        return await self.browser_tool.evaluate_javascript(f"""
            Array.from(document.querySelectorAll('{item_selector}'))
                .map(el => el.textContent.trim())
        """)

from marsys.agents import BrowserAgent

# Create agent with existing session state
agent = await BrowserAgent.create_safe(
    model_config=config,
    name="AuthenticatedBrowser",
    mode="advanced",
    session_path="./sessions/linkedin_session.json",  # Load existing session
    headless=True
)

# Browser is now initialized with saved cookies and localStorage
# Already logged in to LinkedIn, Google, etc.
await agent.run("Go to linkedin.com/feed and extract posts")

# Save via BrowserAgent tool invocation
result = await agent.run("Save the current session to ./sessions/my_session.json")
# Returns a success message with cookie/origin counts

# You can save additional checkpoints as needed
result = await agent.run("Save the current session to ./sessions/backup.json")

{
  "cookies": [
    {
      "name": "session_id",
      "value": "abc123",
      "domain": ".example.com",
      "path": "/",
      "expires": 1735689600,
      "httpOnly": true,
      "secure": true
    }
  ],
  "origins": [
    {
      "origin": "https://example.com",
      "localStorage": [
        {"name": "user_token", "value": "xyz789"}
      ]
    }
  ]
}

class AuthAgent(BrowserAgent):
    """Handle authentication flows."""

    async def login_with_cookies(
        self,
        login_url: str,
        cookies: List[Dict],
        context = None
    ):
        """Login using saved cookies."""

        # Navigate to site
        await self.browser_tool.goto(login_url)

        # Set cookies
        for cookie in cookies:
            await self.browser_tool.context.add_cookies([cookie])

        # Refresh to apply cookies
        await self.browser_tool.reload()

        # Verify login success
        return await self.verify_login_status(context)

    async def handle_2fa(
        self,
        code_input_selector: str,
        get_2fa_code: Callable,
        context = None
    ):
        """Handle two-factor authentication."""

        # Wait for 2FA input
        await self.browser_tool.wait_for_selector(
            code_input_selector, timeout=30000, state="visible"
        )

        # Get 2FA code (from email, SMS, authenticator, etc.)
        code = await get_2fa_code()

        # Enter code
        await self.browser_tool.fill(code_input_selector, code)

        # Submit (usually auto-submits, but can click submit if needed)
        await self.browser_tool.press_key("Enter")

        # Wait for redirect after successful 2FA
        await self.browser_tool.wait_for_navigation()

class ResilientBrowserAgent(BrowserAgent):
    """Browser agent with enhanced error handling."""

    async def retry_operation(
        self,
        operation: Callable,
        max_retries: int = 3,
        backoff_factor: float = 2.0,
        context = None
    ):
        """Execute operation with exponential backoff retry."""

        last_error = None
        wait_time = 1.0

        for attempt in range(max_retries):
            try:
                result = await operation()
                if attempt > 0:
                    await self._log_progress(
                        context, LogLevel.INFO,
                        f"Operation succeeded on attempt {attempt + 1}"
                    )
                return result

            except Exception as e:
                last_error = e
                await self._log_progress(
                    context, LogLevel.WARNING,
                    f"Attempt {attempt + 1} failed: {e}"
                )

                if attempt < max_retries - 1:
                    await asyncio.sleep(wait_time)
                    wait_time *= backoff_factor

        raise Exception(f"Operation failed after {max_retries} attempts: {last_error}")

    async def safe_extract(
        self,
        selector: str,
        default: Any = None,
        context = None
    ):
        """Safely extract element with fallback."""

        try:
            text = await self.browser_tool.get_text(selector)
            if text:
                return text.strip()
        except Exception as e:
            await self._log_progress(
                context, LogLevel.DEBUG,
                f"Failed to extract {selector}: {e}"
            )

        return default

class OptimizedBrowserAgent(BrowserAgent):
    """Optimized browser agent for faster scraping."""

    async def setup_fast_scraping(self, context = None):
        """Configure browser for fast text scraping."""

        # Block unnecessary resources
        await self.browser_tool.context.route("**/*", lambda route:
            route.abort() if route.request.resource_type in
            ["image", "stylesheet", "font", "media"]
            else route.continue_()
        )

        # Disable JavaScript if not needed
        await self.browser_tool.context.set_javascript_enabled(False)

        await self._log_progress(
            context, LogLevel.INFO,
            "Optimized browser for fast scraping"
        )

    async def parallel_scrape(
        self,
        urls: List[str],
        extractor: Callable,
        max_concurrent: int = 5,
        context = None
    ):
        """Scrape multiple URLs in parallel."""

        semaphore = asyncio.Semaphore(max_concurrent)

        async def scrape_with_limit(url):
            async with semaphore:
                try:
                    await self.browser_tool.goto(url)
                    return await extractor(self.browser_tool)
                except Exception as e:
                    await self._log_progress(
                        context, LogLevel.ERROR,
                        f"Failed to scrape {url}: {e}"
                    )
                    return None

        tasks = [scrape_with_limit(url) for url in urls]
        results = await asyncio.gather(*tasks)

        return [r for r in results if r is not None]

# GOOD - Wait for specific conditions
await browser_tool.wait_for_selector("#content", timeout=10000, state="visible")
await browser_tool.wait_for_navigation()

# BAD - Fixed delays
await asyncio.sleep(5)  # Unreliable and slow

# GOOD - Specific, stable selectors
await browser_tool.click("[data-testid='submit-button']")
await browser_tool.click("#unique-id")

# BAD - Fragile selectors
await browser_tool.click("div > span:nth-child(3)")

# GOOD - Always cleanup
browser_agent = await BrowserAgent.create_safe(
    model_config=config,
    name="CleanupExample",
    mode="advanced",
    headless=True,
)
try:
    # Use agent
    result = await browser_agent.run(task)
finally:
    await browser_agent.browser_tool.close()

# BAD - Leaving browsers open
browser_agent = await BrowserAgent.create_safe(
    model_config=config,
    name="LeakyBrowser",
    mode="advanced",
    headless=True,
)
result = await browser_agent.run(task)
# Browser left running!

# GOOD - Detailed error context
try:
    await browser_tool.click(selector)
except Exception as e:
    await self._log_progress(
        context, LogLevel.ERROR,
        f"Failed to click {selector} on {await browser_tool.get_url()}: {e}"
    )
    # Take screenshot for debugging
    await browser_tool.screenshot("error_screenshot.png")

# BAD - Generic error handling
try:
    await browser_tool.click(selector)
except:
    print("Click failed")