How to Build an AI Agent That Browses the Web

# Create a virtual environment
python -m venv agent_env
source agent_env/bin/activate  # On Windows: agent_env\Scripts\activate

# Install Browser Use
pip install browser-use

# Install a complementary package for structured output
pip install pydantic

# You'll also need a browser installed (Chrome, Edge, or Firefox)

export ANTHROPIC_API_KEY="your-key-here"

from browser_use import Agent, BrowserConfig
from anthropic import Anthropic

async def create_web_agent():
    """Initialize a web-browsing agent."""
    client = Anthropic()

    # Configure the browser (headless for production, windowed for debugging)
    browser_config = BrowserConfig(
        headless=True,  # Set to False to see the browser window
        no_sandbox=True  # Required in Docker/containers
    )

    agent = Agent(
        task="Go to Amazon, search for 'noise-canceling headphones', "
             "and tell me the price of the top result.",
        llm_client=client,
        browser_config=browser_config
    )

    return agent

async def run_agent():
    """Execute the agent and capture results."""
    agent = await create_web_agent()
    result = await agent.run()
    print(f"Agent result: {result}")
    return result

# Run the agent
if __name__ == "__main__":
    import asyncio
    asyncio.run(run_agent())

from browser_use import Agent, BrowserConfig
from anthropic import Anthropic

async def create_authenticated_agent():
    """Create an agent with a pre-authenticated browser session."""
    client = Anthropic()

    # Start with a user data directory (persists cookies and cache)
    browser_config = BrowserConfig(
        user_data_dir="/tmp/browser_session",  # Persists authentication
        headless=False  # Set to True after you've authenticated manually
    )

    agent = Agent(
        task="Check my email inbox and count unread messages.",
        llm_client=client,
        browser_config=browser_config
    )

    return agent

import os
from browser_use import Agent, BrowserConfig
from anthropic import Anthropic

async def create_agent_with_login():
    """Create an agent and instruct it to log in."""
    client = Anthropic()

    # Load credentials from environment (never hardcode)
    username = os.getenv("AGENT_USERNAME")
    password = os.getenv("AGENT_PASSWORD")

    agent = Agent(
        task=f"Log in with username '{username}' and password, "
             "then navigate to the billing page and screenshot the current balance.",
        llm_client=client,
        browser_config=BrowserConfig(headless=True)
    )

    return agent

import asyncio
from browser_use import Agent, BrowserConfig
from anthropic import Anthropic

async def run_with_retry(task: str, max_retries: int = 3):
    """Execute a task with exponential backoff."""
    client = Anthropic()

    for attempt in range(max_retries):
        try:
            agent = Agent(
                task=task,
                llm_client=client,
                browser_config=BrowserConfig(headless=True)
            )
            result = await agent.run()
            return result

        except Exception as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # 1s, 2s, 4s
                print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)
            else:
                print(f"All {max_retries} attempts failed.")
                raise

async def fetch_price_with_fallback(product_name: str):
    """Try primary source, then fallback to secondary source."""
    client = Anthropic()

    # Try primary source first
    primary_task = f"Go to Amazon and find the price for '{product_name}'."

    try:
        agent = Agent(
            task=primary_task,
            llm_client=client,
            browser_config=BrowserConfig(headless=True)
        )
        result = await agent.run()
        return result, "amazon"

    except Exception as e:
        print(f"Primary source failed: {e}. Trying fallback...")

        # Fall back to secondary source
        fallback_task = f"Go to eBay and find the price for '{product_name}'."
        agent = Agent(
            task=fallback_task,
            llm_client=client,
            browser_config=BrowserConfig(headless=True)
        )
        result = await agent.run()
        return result, "ebay"

async def multi_step_workflow():
    """Break complex tasks into steps."""
    client = Anthropic()

    steps = [
        "Navigate to LinkedIn and search for 'AI engineers in San Francisco'",
        "Filter results by companies",
        "Export the list",
        "Save to CSV"
    ]

    results = {}
    for i, step in enumerate(steps):
        try:
            agent = Agent(
                task=step,
                llm_client=client,
                browser_config=BrowserConfig(headless=True)
            )
            results[f"step_{i+1}"] = await agent.run()
        except Exception as e:
            print(f"Step {i+1} failed: {e}")
            results[f"step_{i+1}"] = None
            # Decide whether to continue or abort

    return results

from anthropic import Anthropic

client = Anthropic()

# Define your system prompt with cache control
system_prompt = [
    {
        "type": "text",
        "text": "You are a web-browsing agent. Your task is to extract structured data from websites. "
                "Always be precise. Always verify data before returning."
    },
    {
        "type": "text",
        "text": "Available actions: click(selector), type(text), scroll(), wait(), screenshot()",
        "cache_control": {"type": "ephemeral"}  # This section is cached
    }
]

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system=system_prompt,
    messages=[
        {
            "role": "user",
            "content": "Search for 'python tutorial' on Google and return the top 3 results."
        }
    ]
)

print(response.usage)  # Shows cache hits

# Production code: always use headless=True
browser_config = BrowserConfig(headless=True)

# Local debugging: use headless=False
browser_config = BrowserConfig(headless=False)

async def batch_price_check(products: list[str]):
    """Check prices for multiple products in a single agent session."""
    client = Anthropic()

    task = f"""
    Check prices for these products on Amazon:
    1. {products[0]}
    2. {products[1]}
    3. {products[2]}

    Return a JSON array with product name and price.
    """

    agent = Agent(
        task=task,
        llm_client=client,
        browser_config=BrowserConfig(headless=True)
    )

    result = await agent.run()
    return result

import asyncio
import json
from datetime import datetime
from browser_use import Agent, BrowserConfig
from anthropic import Anthropic

class PriceMonitorAgent:
    def __init__(self):
        self.client = Anthropic()
        self.results = {}

    async def monitor_competitor(self, competitor_url: str, product_name: str):
        """Monitor price on a competitor site."""
        task = f"""
        Visit {competitor_url}
        Search for '{product_name}'
        Extract the price of the first result
        Return as JSON: {{"product": "...", "price": "$...", "timestamp": "..."}}
        """

        try:
            agent = Agent(
                task=task,
                llm_client=self.client,
                browser_config=BrowserConfig(headless=True)
            )
            result = await agent.run()
            self.results[competitor_url] = {
                "result": result,
                "timestamp": datetime.now().isoformat(),
                "status": "success"
            }
        except Exception as e:
            self.results[competitor_url] = {
                "error": str(e),
                "timestamp": datetime.now().isoformat(),
                "status": "failed"
            }

    async def monitor_all(self, competitors: dict[str, str]):
        """Monitor all competitors concurrently."""
        tasks = [
            self.monitor_competitor(url, product)
            for url, product in competitors.items()
        ]
        await asyncio.gather(*tasks)
        return self.results

# Usage
async def main():
    monitor = PriceMonitorAgent()
    competitors = {
        "https://amazon.com": "wireless headphones",
        "https://bestbuy.com": "wireless headphones",
        "https://walmart.com": "wireless headphones"
    }
    results = await monitor.monitor_all(competitors)
    print(json.dumps(results, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

# Bad: relies on specific CSS selectors
browser.click(".price-tag-xyz-2026")

# Good: uses natural language
agent.task = "Click the 'Add to Cart' button and proceed to checkout"

import time

for product in products:
    try:
        # Run agent
        result = await agent.run()
    except Exception as e:
        if "429" in str(e) or "rate limit" in str(e).lower():
            print("Rate limited. Backing off for 1 hour...")
            time.sleep(3600)
    time.sleep(random.uniform(2, 5))  # Random delay between requests