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 for browser vision
        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()

# 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
)

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 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"
        )

# 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")