Constructing Browser-Utilizing AI Brokers in Python

<br /> python -m venv browser_agent_env <br /> <br /> # macOS / Linux <br /> supply browser_agent_env/bin/activate <br /> <br /> # Home windows <br /> browser_agent_envScriptsactivate

<br /> pip set up playwright <br /> browser-use <br /> langchain <br /> langchain-openai <br /> langgraph <br /> langchain-community <br /> python-dotenv

<br /> playwright set up chromium

<br /> OPENAI_API_KEY=your_openai_api_key_here

<br /> # first_run.py <br /> # Navigate to a URL, take a screenshot, and extract the web page title. <br /> # Conditions: pip set up playwright && playwright set up chromium <br /> # run: python first_run.py <br /> <br /> import asyncio <br /> from playwright.async_api import async_playwright <br /> <br /> async def major(): <br /> async with async_playwright() as p: <br /> # Launch Chromium in headless mode (no seen browser window). <br /> # Set headless=False if you wish to watch it run throughout growth. <br /> browser = await p.chromium.launch(headless=True) <br /> <br /> # A browser context is sort of a recent browser profile. <br /> # It isolates cookies, storage, and cache from different contexts. <br /> context = await browser.new_context( <br /> viewport={“width”: 1280, “peak”: 720}, <br /> user_agent=( <br /> “Mozilla/5.0 (Home windows NT 10.0; Win64; x64) “ <br /> “AppleWebKit/537.36 (KHTML, like Gecko) “ <br /> “Chrome/120.0.0.0 Safari/537.36″ <br /> ) <br /> ) <br /> <br /> web page = await context.new_page() <br /> <br /> # Navigate to the URL and wait till the community is idle. <br /> # “networkidle” means no open community connections for 500ms. <br /> # For quicker pages, “domcontentloaded” is adequate. <br /> await web page.goto(“https://instance.com”, wait_until=”networkidle”) <br /> <br /> # Extract the web page title <br /> title = await web page.title() <br /> print(f”Web page title: {title}”) <br /> <br /> # Extract the textual content content material of the h1 heading <br /> h1 = await web page.text_content(“h1″) <br /> print(f”H1 heading: {h1}”) <br /> <br /> # Take a full-page screenshot and reserve it to disk <br /> await web page.screenshot(path=”screenshot.png”, full_page=True) <br /> print(“Screenshot saved to screenshot.png”) <br /> <br /> await browser.shut() <br /> <br /> asyncio.run(major())

<br /> # form_submit.py <br /> # Full and submit a multi-field login type on a public demo web site. <br /> # Goal: https://the-internet.herokuapp.com/login (public take a look at web site) <br /> # Conditions: pip set up playwright && playwright set up chromium <br /> # run: python form_submit.py <br /> <br /> import asyncio <br /> from playwright.async_api import async_playwright <br /> <br /> async def login_and_verify(username: str, password: str) -> dict: <br /> “”” <br /> Try and log in to a demo web site and return whether or not it succeeded. <br /> Handles: enter filling, button clicking, and consequence verification. <br /> “”” <br /> async with async_playwright() as p: <br /> browser = await p.chromium.launch(headless=True) <br /> context = await browser.new_context() <br /> web page = await context.new_page() <br /> <br /> await web page.goto(“https://the-internet.herokuapp.com/login”) <br /> <br /> # Await the shape to be seen earlier than interacting. <br /> # state=”seen” is the default however makes the intent specific. <br /> await web page.wait_for_selector(“#username”, state=”seen”) <br /> <br /> # fill() clears the sphere first, then sorts the worth. <br /> # It fires the main focus, enter, and alter occasions so as. <br /> await web page.fill(“#username”, username) <br /> await web page.fill(“#password”, password) <br /> <br /> # click on() fires actual mouse occasions — mousedown, mouseup, click on. <br /> # This triggers JavaScript listeners {that a} plain DOM click on misses. <br /> await web page.click on(“button[type=”submit”]”) <br /> <br /> # Await the web page to settle after type submission <br /> await web page.wait_for_load_state(“networkidle”) <br /> <br /> # Verify which consequence aspect appeared <br /> success_el = await web page.query_selector(“.flash.success”) <br /> error_el = await web page.query_selector(“.flash.error”) <br /> <br /> if success_el: <br /> message = await success_el.inner_text() <br /> consequence = {“success”: True, “message”: message.strip()} <br /> elif error_el: <br /> message = await error_el.inner_text() <br /> consequence = {“success”: False, “message”: message.strip()} <br /> else: <br /> consequence = {“success”: False, “message”: “Unknown consequence”} <br /> <br /> await browser.shut() <br /> return consequence <br /> <br /> <br /> async def major(): <br /> # Legitimate credentials for the demo web site <br /> consequence = await login_and_verify(“tomsmith”, “SuperSecretPassword!”) <br /> print(f”Legitimate login: {consequence}”) <br /> <br /> # Invalid credentials to confirm error dealing with <br /> result_fail = await login_and_verify(“wronguser”, “wrongpass”) <br /> print(f”Invalid login: {result_fail}”) <br /> <br /> <br /> asyncio.run(major())

<br /> # File add <br /> await web page.set_input_files(“#file-upload”, “/path/to/doc.pdf”) <br /> <br /> # Choose dropdown by seen label textual content <br /> await web page.select_option(“#country-select”, label=”Nigeria”) <br /> <br /> # Verify a checkbox <br /> await web page.examine(“#agree-terms”) <br /> <br /> # Deal with a modal dialog (verify/alert) <br /> web page.on(“dialog”, lambda dialog: asyncio.ensure_future(dialog.settle for()))

<br /> # agent_tools.py <br /> # LangGraph agent with three browser instruments: navigate_and_extract, fill_and_submit_form, take_screenshot <br /> # Conditions: pip set up playwright langchain langchain-openai langgraph python-dotenv <br /> # playwright set up chromium <br /> # run: python agent_tools.py <br /> <br /> import asyncio <br /> import os <br /> from dotenv import load_dotenv <br /> from langchain_openai import ChatOpenAI <br /> from langchain.instruments import software <br /> from langchain_core.messages import HumanMessage <br /> from langgraph.prebuilt import create_react_agent <br /> from playwright.async_api import async_playwright <br /> <br /> load_dotenv() <br /> <br /> # ── SHARED BROWSER STATE ────────────────────────────────────────────────────── <br /> # We maintain a single browser occasion alive for the agent’s lifetime. <br /> # Creating and destroying a browser on each software name is gradual and wasteful. <br /> _browser = None <br /> _page = None <br /> _playwright = None <br /> <br /> async def get_page(): <br /> “””Return the shared web page, launching the browser if wanted.””” <br /> world _browser, _page, _playwright <br /> if _browser is None: <br /> _playwright = await async_playwright().begin() <br /> _browser = await _playwright.chromium.launch(headless=True) <br /> context = await _browser.new_context(viewport={“width”: 1280, “peak”: 720}) <br /> _page = await context.new_page() <br /> return _page <br /> <br /> <br /> async def close_browser(): <br /> “””Clear up browser assets when the agent session ends.””” <br /> world _browser, _page, _playwright <br /> if _browser: <br /> await _browser.shut() <br /> await _playwright.cease() <br /> _browser = None <br /> _page = None <br /> _playwright = None <br /> <br /> <br /> # ── BROWSER TOOLS ───────────────────────────────────────────────────────────── <br /> # Observe: these are async instruments (async def). LangChain’s @software decorator helps <br /> # async capabilities straight, and the agent have to be invoked with ainvoke() in order that <br /> # software calls run on the identical occasion loop as an alternative of attempting to begin a second one. <br /> <br /> @software <br /> async def navigate_and_extract(url: str) -> str: <br /> “”” <br /> Navigate to a URL and return the seen textual content content material of the web page. <br /> Use this to go to web sites and browse their content material. <br /> Enter: a full URL string together with https:// (e.g., ‘https://instance.com’). <br /> “”” <br /> web page = await get_page() <br /> await web page.goto(url, wait_until=”domcontentloaded”, timeout=15000) <br /> await web page.wait_for_load_state(“networkidle”) <br /> content material = await web page.inner_text(“physique”) <br /> # Truncate to keep away from flooding the LLM context window <br /> return content material[:3000] if len(content material) > 3000 else content material <br /> <br /> <br /> @software <br /> async def fill_and_submit_form(selector_value_pairs: str) -> str: <br /> “”” <br /> Fill type fields and submit a type on the presently loaded web page. <br /> Enter: a comma-separated string of ‘selector:worth’ pairs ending with ‘submit:button_selector’. <br /> Instance: ‘#electronic mail:consumer@instance.com,#password:secret,submit:button[type=submit]’ <br /> “”” <br /> web page = await get_page() <br /> strive: <br /> pairs = selector_value_pairs.cut up(“,”) <br /> submit_selector = None <br /> <br /> for pair in pairs: <br /> key, val = pair.cut up(“:”, 1) <br /> key = key.strip() <br /> val = val.strip() <br /> if key == “submit”: <br /> submit_selector = val <br /> else: <br /> await web page.fill(key, val) <br /> <br /> if submit_selector: <br /> await web page.click on(submit_selector) <br /> await web page.wait_for_load_state(“networkidle”) <br /> <br /> return f”Kind submitted. Present URL: {web page.url}” <br /> besides Exception as e: <br /> return f”Kind interplay failed: {str(e)}” <br /> <br /> <br /> @software <br /> async def take_screenshot(filename: str) -> str: <br /> “”” <br /> Take a screenshot of the present browser web page and reserve it to a file. <br /> Use this to visually confirm the present state of the web page. <br /> Enter: filename string (e.g., ‘consequence.png’). <br /> “”” <br /> web page = await get_page() <br /> await web page.screenshot(path=filename, full_page=False) <br /> return f”Screenshot saved to {filename}” <br /> <br /> <br /> # ── AGENT SETUP ─────────────────────────────────────────────────────────────── <br /> <br /> llm = ChatOpenAI( <br /> mannequin=”gpt-4o”, <br /> temperature=0, <br /> api_key=os.getenv(“OPENAI_API_KEY”) <br /> ) <br /> <br /> instruments = [navigate_and_extract, fill_and_submit_form, take_screenshot] <br /> <br /> # create_react_agent wires collectively the LLM, the instruments, and the ReAct reasoning loop. <br /> # The agent decides which software to name, calls it, reads the consequence, and continues. <br /> agent = create_react_agent(llm, instruments) <br /> <br /> <br /> # ── DEMO ────────────────────────────────────────────────────────────────────── <br /> <br /> async def major(): <br /> consequence = await agent.ainvoke({ <br /> “messages”: [HumanMessage( <br /> content=( <br /> “Go to https://example.com, read the page content, “ <br /> “then take a screenshot called example.png” <br /> ) <br /> )] <br /> }) <br /> print(consequence[“messages”][-1].content material) <br /> await close_browser() <br /> <br /> <br /> asyncio.run(major())

<br /> # browser_use_agent.py <br /> # A browser-use agent that accepts a pure language activity and completes it <br /> # with none CSS selectors or hardcoded web page construction. <br /> # Conditions: pip set up browser-use playwright python-dotenv <br /> # playwright set up chromium <br /> # run: python browser_use_agent.py <br /> <br /> import asyncio <br /> import os <br /> from dotenv import load_dotenv <br /> from langchain_openai import ChatOpenAI <br /> from browser_use import Agent <br /> <br /> load_dotenv() <br /> <br /> async def run_browser_task(activity: str) -> str: <br /> “”” <br /> Hand a pure language activity to a browser-use agent. <br /> The agent handles navigation, clicks, and extraction with out selectors. <br /> “”” <br /> # temperature=0 retains selections deterministic and reduces hallucinated actions <br /> llm = ChatOpenAI( <br /> mannequin=”gpt-4o”, <br /> temperature=0, <br /> api_key=os.getenv(“OPENAI_API_KEY”) <br /> ) <br /> <br /> # Agent wraps the browser, the LLM, and the duty loop collectively. <br /> # max_actions_per_step limits what number of actions the agent takes earlier than <br /> # re-reading the web page — prevents runaway loops on advanced pages. <br /> agent = Agent( <br /> activity=activity, <br /> llm=llm, <br /> max_actions_per_step=5 <br /> ) <br /> <br /> # run() executes the total activity loop: <br /> # learn web page → determine motion → take motion → learn up to date web page → repeat <br /> consequence = await agent.run() <br /> <br /> # final_result() returns the agent’s extracted content material or conclusion <br /> return consequence.final_result() or “Activity accomplished with no extracted output.” <br /> <br /> <br /> async def major(): <br /> activity = ( <br /> “Go to https://books.toscrape.com and discover the three costliest books “ <br /> “on the primary web page. Return their titles and costs.” <br /> ) <br /> print(f”Activity: {activity}n”) <br /> output = await run_browser_task(activity) <br /> print(f”Outcome:n{output}”) <br /> <br /> <br /> asyncio.run(major())

<br /> # hard_parts.py — Half 1: Anti-bot stealth launch <br /> # Conditions: pip set up playwright && playwright set up chromium <br /> # run: python hard_parts.py <br /> <br /> import asyncio <br /> import json <br /> from pathlib import Path <br /> from playwright.async_api import async_playwright <br /> <br /> async def launch_stealth_browser(playwright): <br /> “”” <br /> Launch a browser context that appears extra like an actual human session. <br /> Covers: practical viewport, user-agent, locale, timezone, webdriver flag. <br /> Observe: For severe anti-bot targets, contemplate a paid service like Browserbase. <br /> “”” <br /> browser = await playwright.chromium.launch( <br /> headless=True, <br /> args=[ <br /> “–disable-blink-features=AutomationControlled”, # Hides webdriver detection <br /> “–no-sandbox”, <br /> “–disable-dev-shm-usage”, <br /> ] <br /> ) <br /> <br /> context = await browser.new_context( <br /> viewport={“width”: 1366, “peak”: 768}, # Widespread desktop decision <br /> user_agent=( <br /> “Mozilla/5.0 (Home windows NT 10.0; Win64; x64) “ <br /> “AppleWebKit/537.36 (KHTML, like Gecko) “ <br /> “Chrome/124.0.0.0 Safari/537.36″ <br /> ), <br /> locale=”en-US”, <br /> timezone_id=”America/New_York”, <br /> java_script_enabled=True, <br /> ) <br /> <br /> # Take away the ‘webdriver’ property that Playwright injects by default. <br /> # Bot detection programs examine for this within the browser’s JS surroundings. <br /> await context.add_init_script( <br /> “Object.defineProperty(navigator, ‘webdriver’, {get: () => undefined})” <br /> ) <br /> <br /> return browser, context

<br /> # Half 2: Good ready methods (add to your scraper or agent instruments) <br /> <br /> async def smart_wait_examples(web page): <br /> “”” <br /> 4 methods to attend for the correct web page state, with out arbitrary sleeps. <br /> “”” <br /> # STRATEGY 1: Await a selected aspect to look within the DOM <br /> # Use when you understand precisely what aspect alerts content material has loaded <br /> await web page.wait_for_selector(“.product-list”, state=”seen”, timeout=10000) <br /> <br /> # STRATEGY 2: Await a selected API response <br /> # Use when the content material comes from an XHR/fetch name you may determine <br /> async with web page.expect_response( <br /> lambda r: “/api/merchandise” in r.url and r.standing == 200 <br /> ) as response_info: <br /> await web page.click on(“#load-more”) <br /> response = await response_info.worth <br /> print(f”API responded: {response.standing}”) <br /> <br /> # STRATEGY 3: Await the URL to alter after type submission <br /> # Use when a profitable submit redirects to a brand new web page <br /> await web page.wait_for_url(“**/dashboard**”, timeout=10000) <br /> <br /> # STRATEGY 4: Await a JavaScript variable to be set <br /> # Use when no visible aspect reliably alerts the prepared state <br /> await web page.wait_for_function( <br /> “() => window.__dataLoaded === true”, <br /> timeout=10000 <br /> )

<br /> # Half 3: Session persistence throughout runs <br /> <br /> COOKIES_FILE = Path(“session_cookies.json”) <br /> <br /> async def save_session(context) -> None: <br /> “””Save browser cookies to disk after a profitable login.””” <br /> cookies = await context.cookies() <br /> COOKIES_FILE.write_text(json.dumps(cookies, indent=2)) <br /> print(f”Session saved: {len(cookies)} cookies written.”) <br /> <br /> <br /> async def load_session(context) -> bool: <br /> “””Load saved cookies earlier than navigating. Returns True if session was discovered.””” <br /> if not COOKIES_FILE.exists(): <br /> print(“No saved session. Contemporary login required.”) <br /> return False <br /> cookies = json.hundreds(COOKIES_FILE.read_text()) <br /> await context.add_cookies(cookies) <br /> print(f”Session restored: {len(cookies)} cookies loaded.”) <br /> return True

<br /> # Dockerfile for headless Playwright-based browser agent <br /> # Construct: docker construct -t browser-agent . <br /> # Run: docker run –rm -e OPENAI_API_KEY=your_key browser-agent <br /> <br /> FROM python:3.11-slim <br /> <br /> # Set up system dependencies required by Chromium <br /> RUN apt-get replace && apt-get set up -y <br /> libnss3 libatk1.0-0 libatk-bridge2.0-0 libcups2 <br /> libdrm2 libxkbcommon0 libxcomposite1 libxdamage1 <br /> libxrandr2 libgbm1 libasound2 libpangocairo-1.0-0 <br /> libpango-1.0-0 libcairo2 libx11-6 libxext6 libxfixes3 <br /> fonts-liberation wget ca-certificates <br /> && rm -rf /var/lib/apt/lists/* <br /> <br /> WORKDIR /app <br /> <br /> # Set up Python dependencies first (cached layer — solely rebuilds on necessities change) <br /> COPY necessities.txt . <br /> RUN pip set up –no-cache-dir -r necessities.txt <br /> <br /> # Set up Playwright browser binaries into the picture <br /> RUN playwright set up chromium <br /> RUN playwright install-deps chromium <br /> <br /> # Copy utility code final (modifications right here do not invalidate the pip/playwright layers) <br /> COPY . . <br /> <br /> CMD [“python”, “agent_tools.py”] <br /> <br /> necessities.txt: <br /> playwright <br /> browser-use <br /> langchain <br /> langchain-openai <br /> langgraph <br /> python-dotenv

<br /> # Parallel scraping with semaphore price limiting <br /> # Runs as much as 3 browser classes concurrently <br /> <br /> import asyncio <br /> from playwright.async_api import async_playwright <br /> <br /> async def scrape_url(browser, url: str, semaphore: asyncio.Semaphore) -> dict: <br /> “””Scrape a single URL, respecting the concurrency semaphore.””” <br /> async with semaphore: <br /> context = await browser.new_context() <br /> web page = await context.new_page() <br /> await web page.goto(url, wait_until=”domcontentloaded”) <br /> title = await web page.title() <br /> await context.shut() # Shut context (not browser) to launch assets <br /> return {“url”: url, “title”: title} <br /> <br /> <br /> async def scrape_parallel(urls: record[str], max_concurrent: int = 3) -> record[dict]: <br /> “””Scrape a listing of URLs in parallel, capped at max_concurrent classes.””” <br /> semaphore = asyncio.Semaphore(max_concurrent) # Cap concurrent classes <br /> <br /> async with async_playwright() as p: <br /> # One browser shared throughout all contexts — less expensive than one browser per URL <br /> browser = await p.chromium.launch(headless=True) <br /> duties = [scrape_url(browser, url, semaphore) for url in urls] <br /> outcomes = await asyncio.collect(*duties) <br /> await browser.shut() <br /> <br /> return record(outcomes)

<br /> # reference_agent.py <br /> # Full browser-using AI agent: navigates, extracts, summarizes. <br /> # Goal: books.toscrape.com (public scraping sandbox) <br /> # Conditions: pip set up playwright langchain langchain-openai langgraph python-dotenv <br /> # playwright set up chromium <br /> # run: python reference_agent.py <br /> <br /> import asyncio <br /> import os <br /> from dotenv import load_dotenv <br /> from langchain_openai import ChatOpenAI <br /> from langchain.instruments import software <br /> from langchain_core.messages import HumanMessage, SystemMessage <br /> from langgraph.prebuilt import create_react_agent <br /> from playwright.async_api import async_playwright <br /> <br /> load_dotenv() <br /> <br /> # ── BROWSER STATE ───────────────────────────────────────────────────────────── <br /> _browser = None <br /> _context = None <br /> _page = None <br /> _playwright = None <br /> <br /> async def get_page(): <br /> world _browser, _context, _page, _playwright <br /> if _browser is None: <br /> _playwright = await async_playwright().begin() <br /> _browser = await _playwright.chromium.launch(headless=True) <br /> _context = await _browser.new_context( <br /> viewport={“width”: 1280, “peak”: 720}, <br /> user_agent=( <br /> “Mozilla/5.0 (Home windows NT 10.0; Win64; x64) “ <br /> “AppleWebKit/537.36 (KHTML, like Gecko) “ <br /> “Chrome/120.0.0.0 Safari/537.36″ <br /> ) <br /> ) <br /> # Take away webdriver fingerprint <br /> await _context.add_init_script( <br /> “Object.defineProperty(navigator, ‘webdriver’, {get: () => undefined})” <br /> ) <br /> _page = await _context.new_page() <br /> return _page <br /> <br /> <br /> async def teardown(): <br /> world _browser, _playwright <br /> if _browser: <br /> await _browser.shut() <br /> await _playwright.cease() <br /> _browser = None <br /> _playwright = None <br /> <br /> <br /> # ── TOOLS ───────────────────────────────────────────────────────────────────── <br /> <br /> @software <br /> async def navigate(url: str) -> str: <br /> “”” <br /> Navigate the browser to a URL and return the web page’s textual content content material. <br /> Use when it’s essential to open an internet site or transfer to a brand new web page. <br /> Enter: full URL with https:// prefix. <br /> “”” <br /> web page = await get_page() <br /> await web page.goto(url, wait_until=”domcontentloaded”, timeout=20000) <br /> await web page.wait_for_load_state(“networkidle”) <br /> content material = await web page.inner_text(“physique”) <br /> return content material[:4000] <br /> <br /> <br /> @software <br /> async def extract_structured(css_selector: str) -> str: <br /> “”” <br /> Extract textual content from all parts matching a CSS selector on the present web page. <br /> Use when it’s essential to pull particular parts from the loaded web page. <br /> Enter: legitimate CSS selector string (e.g., ‘h3 a’, ‘.price_color’, ‘article.product_pod’). <br /> “”” <br /> web page = await get_page() <br /> strive: <br /> await web page.wait_for_selector(css_selector, timeout=5000) <br /> parts = await web page.query_selector_all(css_selector) <br /> texts = [] <br /> for el in parts[:20]: # Cap at 20 parts to maintain output manageable <br /> textual content = await el.inner_text() <br /> texts.append(textual content.strip()) <br /> return “n”.be part of(texts) if texts else “No parts discovered.” <br /> besides Exception as e: <br /> return f”Extraction failed: {str(e)}” <br /> <br /> <br /> @software <br /> async def get_current_url() -> str: <br /> “””Return the URL the browser is presently on. No enter required.””” <br /> web page = await get_page() <br /> return web page.url <br /> <br /> <br /> # ── AGENT ───────────────────────────────────────────────────────────────────── <br /> <br /> llm = ChatOpenAI( <br /> mannequin=”gpt-4o”, <br /> temperature=0, <br /> api_key=os.getenv(“OPENAI_API_KEY”) <br /> ) <br /> <br /> instruments = [navigate, extract_structured, get_current_url] <br /> agent = create_react_agent(llm, instruments) <br /> <br /> SYSTEM = ( <br /> “You’re a browser-based analysis agent. You could have entry to an actual browser. “ <br /> “Use navigate() to open pages, extract_structured() to tug particular parts, “ <br /> “and get_current_url() to examine the place you’re. “ <br /> “All the time navigate first, then extract. Be concise in your closing reply.” <br /> ) <br /> <br /> <br /> async def run_agent(question: str) -> str: <br /> consequence = await agent.ainvoke({ <br /> “messages”: [ <br /> SystemMessage(content=SYSTEM), <br /> HumanMessage(content=query) <br /> ] <br /> }) <br /> await teardown() <br /> return consequence[“messages”][-1].content material <br /> <br /> <br /> # ── DEMO ────────────────────────────────────────────────────────────────────── <br /> <br /> if __name__ == “__main__”: <br /> question = ( <br /> “Go to https://books.toscrape.com and extract the titles and costs “ <br /> “of the primary 5 books listed. Return them as a structured record.” <br /> ) <br /> print(f”Question: {question}n”) <br /> reply = asyncio.run(run_agent(question)) <br /> print(f”Reply:n{reply}”)