How to Build a Browser Agent

1"""Price Watch: a browser agent that checks competitor prices and flags changes.
2Run: python price_watch.py
3(pip install browser-use; playwright install chromium; set ANTHROPIC_API_KEY)
4"""
5import asyncio
6import json
7from pathlib import Path
8
9from browser_use import Agent, ChatAnthropic
10from pydantic import BaseModel
11
12BASELINE_FILE = Path("price_baseline.json")
13
14URLS = [
15    "https://competitor-a.example.com/product/widget",
16    "https://competitor-b.example.com/items/widget-pro",
17]
18
19llm = ChatAnthropic(model="claude-sonnet-5-5", temperature=0.0)
20
21
22class PriceCheck(BaseModel):
23    title: str
24    price: float
25    in_stock: bool
26
27
28async def check_with_retry(url: str, attempts: int = 2) -> PriceCheck | None:
29    for _ in range(attempts):
30        agent = Agent(
31            task=(
32                f"Go to {url} and read the product title, current price, and "
33                f"whether it's in stock. Report exactly that, nothing else."
34            ),
35            llm=llm,
36            output_model_schema=PriceCheck,
37        )
38        history = await agent.run()
39        if history.is_successful():
40            return history.structured_output
41    return None
42
43
44async def check_all(urls: list[str]) -> dict[str, PriceCheck]:
45    results: dict[str, PriceCheck] = {}
46    for url in urls:
47        result = await check_with_retry(url)
48        if result is not None:
49            results[url] = result
50        else:
51            print(f"[WARN] Failed to check {url} after retries")
52    return results
53
54
55def load_baseline() -> dict:
56    if BASELINE_FILE.exists():
57        return json.loads(BASELINE_FILE.read_text())
58    return {}
59
60
61def save_baseline(current: dict[str, PriceCheck]) -> None:
62    serializable = {url: check.model_dump() for url, check in current.items()}
63    BASELINE_FILE.write_text(json.dumps(serializable, indent=2))
64
65
66def flag_changes(current: dict[str, PriceCheck], baseline: dict) -> list[str]:
67    alerts = []
68    for url, check in current.items():
69        old = baseline.get(url)
70        if old and old["price"] and abs(check.price - old["price"]) / old["price"] > 0.02:
71            alerts.append(f"{url}: {old['price']} -> {check.price} ({check.title})")
72    return alerts
73
74
75async def main():
76    baseline = load_baseline()
77    current = await check_all(URLS)
78
79    alerts = flag_changes(current, baseline)
80    if alerts:
81        print("Price changes detected:")
82        for alert in alerts:
83            print(f"  - {alert}")
84    else:
85        print("No price changes above threshold.")
86
87    save_baseline(current)
88
89
90if __name__ == "__main__":
91    asyncio.run(main())
92