Constructing Browser-Utilizing AI Brokers in Python

python –m venv browser_agent_env

# macOS / Linux

supply browser_agent_env/bin/activate

# Home windows

browser_agent_envScriptsactivate

pip set up playwright

            browser–use

            langchain

            langchain–openai

            langgraph

            langchain–neighborhood

            python–dotenv

playwright set up chromium

OPENAI_API_KEY=your_openai_api_key_here

# first_run.py

# Navigate to a URL, take a screenshot, and extract the web page title.

# Stipulations: pip set up playwright && playwright set up chromium

# Learn how to run: python first_run.py

import asyncio

from playwright.async_api import async_playwright

async def foremost():

    async with async_playwright() as p:

        # Launch Chromium in headless mode (no seen browser window).

        # Set headless=False if you wish to watch it run throughout growth.

        browser = await p.chromium.launch(headless=True)

        # A browser context is sort of a recent browser profile.

        # It isolates cookies, storage, and cache from different contexts.

        context = await browser.new_context(

            viewport={“width”: 1280, “peak”: 720},

            user_agent=(

                “Mozilla/5.0 (Home windows NT 10.0; Win64; x64) “

                “AppleWebKit/537.36 (KHTML, like Gecko) “

                “Chrome/120.0.0.0 Safari/537.36”

            )

        )

        web page = await context.new_page()

        # Navigate to the URL and wait till the community is idle.

        # “networkidle” means no open community connections for 500ms.

        # For sooner pages, “domcontentloaded” is ample.

        await web page.goto(“https://instance.com”, wait_until=“networkidle”)

        # Extract the web page title

        title = await web page.title()

        print(f“Web page title: {title}”)

        # Extract the textual content content material of the h1 heading

        h1 = await web page.text_content(“h1”)

        print(f“H1 heading: {h1}”)

        # Take a full-page screenshot and put it aside to disk

        await web page.screenshot(path=“screenshot.png”, full_page=True)

        print(“Screenshot saved to screenshot.png”)

        await browser.shut()

asyncio.run(foremost())

# scrape_books.py

# Scrape ebook titles, costs, and scores from books.toscrape.com

# This can be a authorized scraping sandbox website constructed for observe.

# Stipulations: pip set up playwright && playwright set up chromium

# Learn how to run: python scrape_books.py

import asyncio

import json

from playwright.async_api import async_playwright

async def scrape_books(max_pages: int = 3) -> record[dict]:

    “”“

    Scrape ebook listings from books.toscrape.com throughout a number of pages.

    Returns an inventory of dicts with title, worth, score, and web page quantity.

    ““”

    outcomes = []

    async with async_playwright() as p:

        browser = await p.chromium.launch(headless=True)

        context = await browser.new_context(viewport={“width”: 1280, “peak”: 720})

        web page = await context.new_page()

        for page_num in vary(1, max_pages + 1):

            url = f“https://books.toscrape.com/catalogue/page-{page_num}.html”

            print(f“Scraping web page {page_num}: {url}”)

            await web page.goto(url, wait_until=“domcontentloaded”)

            # Look forward to the product playing cards to be seen earlier than extracting.

            # That is essential on JavaScript-heavy pages the place content material masses after the HTML.

            # timeout=10000 means wait as much as 10 seconds earlier than elevating an error.

            await web page.wait_for_selector(“article.product_pod”, timeout=10000)

            # Get all ebook playing cards on the present web page

            books = await web page.query_selector_all(“article.product_pod”)

            for ebook in books:

                title_el = await ebook.query_selector(“h3 a”)

                title = await title_el.get_attribute(“title”) if title_el else “N/A”

                # Extract worth textual content

                price_el = await ebook.query_selector(“.price_color”)

                worth = await price_el.inner_text() if price_el else “N/A”

                # Extract star score from the CSS class title.

                rating_el = await ebook.query_selector(“p.star-rating”)

                rating_class = await rating_el.get_attribute(“class”) if rating_el else “”

                score = rating_class.change(“star-rating”, “”).strip()

                outcomes.append({

                    “title”: title,

                    “worth”: worth,

                    “score”: score,

                    “web page”: web page_num

                })

            print(f”  Extracted {len(books)} books from web page {page_num}”)

        await browser.shut()

    return outcomes

async def foremost():

    books = await scrape_books(max_pages=2)

    print(f“nTotal books scraped: {len(books)}”)

    print(json.dumps(books[:3], indent=2))

asyncio.run(foremost())

# form_submit.py

# Full and submit a multi-field login type on a public demo website.

# Goal: https://the-internet.herokuapp.com/login (public take a look at website)

# Stipulations: pip set up playwright && playwright set up chromium

# Learn how to run: python form_submit.py

import asyncio

from playwright.async_api import async_playwright

async def login_and_verify(username: str, password: str) -> dict:

    “”“

    Try to log in to a demo website and return whether or not it succeeded.

    Handles: enter filling, button clicking, and outcome verification.

    ““”

    async with async_playwright() as p:

        browser = await p.chromium.launch(headless=True)

        context = await browser.new_context()

        web page = await context.new_page()

        await web page.goto(“https://the-internet.herokuapp.com/login”)

        # Look forward to the shape to be seen earlier than interacting.

        # state=”seen” is the default however makes the intent express.

        await web page.wait_for_selector(“#username”, state=“seen”)

        # fill() clears the sector first, then varieties the worth.

        # It fires the main focus, enter, and alter occasions so as.

        await web page.fill(“#username”, username)

        await web page.fill(“#password”, password)

        # click on() fires actual mouse occasions — mousedown, mouseup, click on.

        # This triggers JavaScript listeners {that a} plain DOM click on misses.

        await web page.click on(“button[type=”submit”]”)

        # Look forward to the web page to settle after type submission

        await web page.wait_for_load_state(“networkidle”)

        # Test which outcome component appeared

        success_el = await web page.query_selector(“.flash.success”)

        error_el = await web page.query_selector(“.flash.error”)

        if success_el:

            message = await success_el.inner_text()

            outcome = {“success”: True, “message”: message.strip()}

        elif error_el:

            message = await error_el.inner_text()

            outcome = {“success”: False, “message”: message.strip()}

        else:

            outcome = {“success”: False, “message”: “Unknown outcome”}

        await browser.shut()

        return outcome

async def foremost():

    # Legitimate credentials for the demo website

    outcome = await login_and_verify(“tomsmith”, “SuperSecretPassword!”)

    print(f“Legitimate login:   {outcome}”)

    # Invalid credentials to confirm error dealing with

    result_fail = await login_and_verify(“wronguser”, “wrongpass”)

    print(f“Invalid login: {result_fail}”)

asyncio.run(foremost())

# File add

await web page.set_input_files(“#file-upload”, “/path/to/doc.pdf”)

# Choose dropdown by seen label textual content

await web page.select_option(“#country-select”, label=“Nigeria”)

# Test a checkbox

await web page.examine(“#agree-terms”)

# Deal with a modal dialog (affirm/alert)

web page.on(“dialog”, lambda dialog: asyncio.ensure_future(dialog.settle for()))

# agent_tools.py

# LangGraph agent with three browser instruments: navigate_and_extract, fill_and_submit_form, take_screenshot

# Stipulations: pip set up playwright langchain langchain-openai langgraph python-dotenv

#                playwright set up chromium

# Learn how to run: python agent_tools.py

import asyncio

import os

from dotenv import load_dotenv

from langchain_openai import ChatOpenAI

from langchain.instruments import instrument

from langchain_core.messages import HumanMessage

from langgraph.prebuilt import create_react_agent

from playwright.async_api import async_playwright

load_dotenv()

# ── SHARED BROWSER STATE ──────────────────────────────────────────────────────

# We maintain a single browser occasion alive for the agent’s lifetime.

# Creating and destroying a browser on each instrument name is gradual and wasteful.

_browser = None

_page = None

_playwright = None

async def get_page():

    “”“Return the shared web page, launching the browser if wanted.”“”

    international _browser, _page, _playwright

    if _browser is None:

        _playwright = await async_playwright().begin()

        _browser = await _playwright.chromium.launch(headless=True)

        context = await _browser.new_context(viewport={“width”: 1280, “peak”: 720})

        _page = await context.new_page()

    return _page

async def close_browser():

    “”“Clear up browser assets when the agent session ends.”“”

    international _browser, _page, _playwright

    if _browser:

        await _browser.shut()

        await _playwright.cease()

        _browser = None

        _page = None

        _playwright = None

# ── BROWSER TOOLS ─────────────────────────────────────────────────────────────

# Word: these are async instruments (async def). LangChain’s @instrument decorator helps

# async features instantly, and the agent should be invoked with ainvoke() in order that

# instrument calls run on the identical occasion loop as an alternative of attempting to begin a second one.

@instrument

async def navigate_and_extract(url: str) -> str:

    “”“

    Navigate to a URL and return the seen textual content content material of the web page.

    Use this to go to web sites and skim their content material.

    Enter: a full URL string together with https:// (e.g., ‘https://instance.com’).

    ““”

    web page = await get_page()

    await web page.goto(url, wait_until=“domcontentloaded”, timeout=15000)

    await web page.wait_for_load_state(“networkidle”)

    content material = await web page.inner_text(“physique”)

    # Truncate to keep away from flooding the LLM context window

    return content material[:3000] if len(content material) > 3000 else content material

@instrument

async def fill_and_submit_form(selector_value_pairs: str) -> str:

    “”“

    Fill type fields and submit a type on the at the moment loaded web page.

    Enter: a comma-separated string of ‘selector:worth’ pairs ending with ‘submit:button_selector’.

    Instance: ‘#e-mail:consumer@instance.com,#password:secret,submit:button[type=submit]’

    ““”

    web page = await get_page()

    strive:

        pairs = selector_value_pairs.break up(“,”)

        submit_selector = None

        for pair in pairs:

            key, val = pair.break up(“:”, 1)

            key = key.strip()

            val = val.strip()

            if key == “submit”:

                submit_selector = val

            else:

                await web page.fill(key, val)

        if submit_selector:

            await web page.click on(submit_selector)

            await web page.wait_for_load_state(“networkidle”)

        return f“Kind submitted. Present URL: {web page.url}”

    besides Exception as e:

        return f“Kind interplay failed: {str(e)}”

@instrument

async def take_screenshot(filename: str) -> str:

    “”“

    Take a screenshot of the present browser web page and put it aside to a file.

    Use this to visually confirm the present state of the web page.

    Enter: filename string (e.g., ‘outcome.png’).

    ““”

    web page = await get_page()

    await web page.screenshot(path=filename, full_page=False)

    return f“Screenshot saved to {filename}”

# ── AGENT SETUP ───────────────────────────────────────────────────────────────

llm = ChatOpenAI(

    mannequin=“gpt-4o”,

    temperature=0,

    api_key=os.getenv(“OPENAI_API_KEY”)

)

instruments = [navigate_and_extract, fill_and_submit_form, take_screenshot]

# create_react_agent wires collectively the LLM, the instruments, and the ReAct reasoning loop.

# The agent decides which instrument to name, calls it, reads the outcome, and continues.

agent = create_react_agent(llm, instruments)

# ── DEMO ──────────────────────────────────────────────────────────────────────

async def foremost():

    outcome = await agent.ainvoke({

        “messages”: [HumanMessage(

            content=(

                “Go to https://example.com, read the page content, “

                “then take a screenshot called example.png”

            )

        )]

    })

    print(outcome[“messages”][–1].content material)

    await close_browser()

asyncio.run(foremost())

# browser_use_agent.py

# A browser-use agent that accepts a pure language activity and completes it

# with none CSS selectors or hardcoded web page construction.

# Stipulations: pip set up browser-use playwright python-dotenv

#                playwright set up chromium

# Learn how to run: python browser_use_agent.py

import asyncio

import os

from dotenv import load_dotenv

from langchain_openai import ChatOpenAI

from browser_use import Agent

load_dotenv()

async def run_browser_task(activity: str) -> str:

    “”“

    Hand a pure language activity to a browser-use agent.

    The agent handles navigation, clicks, and extraction with out selectors.

    ““”

    # temperature=0 retains selections deterministic and reduces hallucinated actions

    llm = ChatOpenAI(

        mannequin=“gpt-4o”,

        temperature=0,

        api_key=os.getenv(“OPENAI_API_KEY”)

    )

    # Agent wraps the browser, the LLM, and the duty loop collectively.

    # max_actions_per_step limits what number of actions the agent takes earlier than

    # re-reading the web page — prevents runaway loops on advanced pages.

    agent = Agent(

        activity=activity,

        llm=llm,

        max_actions_per_step=5

    )

    # run() executes the complete activity loop:

    # learn web page → determine motion → take motion → learn up to date web page → repeat

    outcome = await agent.run()

    # final_result() returns the agent’s extracted content material or conclusion

    return outcome.final_result() or “Process accomplished with no extracted output.”

async def foremost():

    activity = (

        “Go to https://books.toscrape.com and discover the three costliest books “

        “on the primary web page. Return their titles and costs.”

    )

    print(f“Process: {activity}n”)

    output = await run_browser_task(activity)

    print(f“Outcome:n{output}”)

asyncio.run(foremost())

# hard_parts.py — Half 1: Anti-bot stealth launch

# Stipulations: pip set up playwright && playwright set up chromium

# Learn how to run: python hard_parts.py

import asyncio

import json

from pathlib import Path

from playwright.async_api import async_playwright

async def launch_stealth_browser(playwright):

    “”“

    Launch a browser context that appears extra like an actual human session.

    Covers: reasonable viewport, user-agent, locale, timezone, webdriver flag.

    Word: For severe anti-bot targets, think about a paid service like Browserbase.

    ““”

    browser = await playwright.chromium.launch(

        headless=True,

        args=[

            “–disable-blink-features=AutomationControlled”,  # Hides webdriver detection

            “–no-sandbox”,

            “–disable-dev-shm-usage”,

        ]

    )

    context = await browser.new_context(

        viewport={“width”: 1366, “peak”: 768},   # Frequent desktop decision

        user_agent=(

            “Mozilla/5.0 (Home windows NT 10.0; Win64; x64) “

            “AppleWebKit/537.36 (KHTML, like Gecko) “

            “Chrome/124.0.0.0 Safari/537.36”

        ),

        locale=“en-US”,

        timezone_id=“America/New_York”,

        java_script_enabled=True,

    )

    # Take away the ‘webdriver’ property that Playwright injects by default.

    # Bot detection programs examine for this within the browser’s JS setting.

    await context.add_init_script(

        “Object.defineProperty(navigator, ‘webdriver’, {get: () => undefined})”

    )

    return browser, context

# Half 2: Sensible ready methods (add to your scraper or agent instruments)

async def smart_wait_examples(web page):

    “”“

    4 methods to attend for the suitable web page state, with out arbitrary sleeps.

    ““”

    # STRATEGY 1: Look forward to a particular component to seem within the DOM

    # Use when you understand precisely what component alerts content material has loaded

    await web page.wait_for_selector(“.product-list”, state=“seen”, timeout=10000)

    # STRATEGY 2: Look forward to a particular API response

    # Use when the content material comes from an XHR/fetch name you possibly can determine

    async with web page.expect_response(

        lambda r: “/api/merchandise” in r.url and r.standing == 200

    ) as response_info:

        await web page.click on(“#load-more”)

    response = await response_info.worth

    print(f“API responded: {response.standing}”)

    # STRATEGY 3: Look forward to the URL to vary after type submission

    # Use when a profitable submit redirects to a brand new web page

    await web page.wait_for_url(“**/dashboard**”, timeout=10000)

    # STRATEGY 4: Look forward to a JavaScript variable to be set

    # Use when no visible component reliably alerts the prepared state

    await web page.wait_for_function(

        “() => window.__dataLoaded === true”,

        timeout=10000

    )

# Half 3: Session persistence throughout runs

COOKIES_FILE = Path(“session_cookies.json”)

async def save_session(context) -> None:

    “”“Save browser cookies to disk after a profitable login.”“”

    cookies = await context.cookies()

    COOKIES_FILE.write_text(json.dumps(cookies, indent=2))

    print(f“Session saved: {len(cookies)} cookies written.”)

async def load_session(context) -> bool:

    “”“Load saved cookies earlier than navigating. Returns True if session was discovered.”“”

    if not COOKIES_FILE.exists():

        print(“No saved session. Contemporary login required.”)

        return False

    cookies = json.masses(COOKIES_FILE.read_text())

    await context.add_cookies(cookies)

    print(f“Session restored: {len(cookies)} cookies loaded.”)

    return True

# Dockerfile for headless Playwright-based browser agent

# Construct: docker construct -t browser-agent .

# Run:   docker run –rm -e OPENAI_API_KEY=your_key browser-agent

FROM python:3.11–slim

# Set up system dependencies required by Chromium

RUN apt–get replace && apt–get set up –y

    libnss3 libatk1.0–0 libatk–bridge2.0–0 libcups2

    libdrm2 libxkbcommon0 libxcomposite1 libxdamage1

    libxrandr2 libgbm1 libasound2 libpangocairo–1.0–0

    libpango–1.0–0 libcairo2 libx11–6 libxext6 libxfixes3

    fonts–liberation wget ca–certificates

    && rm –rf /var/lib/apt/lists/*

WORKDIR /app

# Set up Python dependencies first (cached layer — solely rebuilds on necessities change)

COPY necessities.txt .

RUN pip set up —no–cache–dir –r necessities.txt

# Set up Playwright browser binaries into the picture

RUN playwright set up chromium

RUN playwright set up–deps chromium

# Copy software code final (adjustments right here do not invalidate the pip/playwright layers)

COPY . .

CMD [“python”, “agent_tools.py”]

necessities.txt:

playwright

browser–use

langchain

langchain–openai

langgraph

python–dotenv

# Parallel scraping with semaphore fee limiting

# Runs as much as 3 browser periods concurrently

import asyncio

from playwright.async_api import async_playwright

async def scrape_url(browser, url: str, semaphore: asyncio.Semaphore) -> dict:

    “”“Scrape a single URL, respecting the concurrency semaphore.”“”

    async with semaphore:

        context = await browser.new_context()

        web page = await context.new_page()

        await web page.goto(url, wait_until=“domcontentloaded”)

        title = await web page.title()

        await context.shut()   # Shut context (not browser) to launch assets

        return {“url”: url, “title”: title}

async def scrape_parallel(urls: record[str], max_concurrent: int = 3) -> record[dict]:

    “”“Scrape an inventory of URLs in parallel, capped at max_concurrent periods.”“”

    semaphore = asyncio.Semaphore(max_concurrent)  # Cap concurrent periods

    async with async_playwright() as p:

        # One browser shared throughout all contexts — less expensive than one browser per URL

        browser = await p.chromium.launch(headless=True)

        duties = [scrape_url(browser, url, semaphore) for url in urls]

        outcomes = await asyncio.collect(*duties)

        await browser.shut()

    return record(outcomes)

# reference_agent.py

# Full browser-using AI agent: navigates, extracts, summarizes.

# Goal: books.toscrape.com (public scraping sandbox)

# Stipulations: pip set up playwright langchain langchain-openai langgraph python-dotenv

#                playwright set up chromium

# Learn how to run: python reference_agent.py

import asyncio

import os

from dotenv import load_dotenv

from langchain_openai import ChatOpenAI

from langchain.instruments import instrument

from langchain_core.messages import HumanMessage, SystemMessage

from langgraph.prebuilt import create_react_agent

from playwright.async_api import async_playwright

load_dotenv()

# ── BROWSER STATE ─────────────────────────────────────────────────────────────

_browser = None

_context = None

_page = None

_playwright = None

async def get_page():

    international _browser, _context, _page, _playwright

    if _browser is None:

        _playwright = await async_playwright().begin()

        _browser = await _playwright.chromium.launch(headless=True)

        _context = await _browser.new_context(

            viewport={“width”: 1280, “peak”: 720},

            user_agent=(

                “Mozilla/5.0 (Home windows NT 10.0; Win64; x64) “

                “AppleWebKit/537.36 (KHTML, like Gecko) “

                “Chrome/120.0.0.0 Safari/537.36”

            )

        )

        # Take away webdriver fingerprint

        await _context.add_init_script(

            “Object.defineProperty(navigator, ‘webdriver’, {get: () => undefined})”

        )

        _page = await _context.new_page()

    return _page

async def teardown():

    international _browser, _playwright

    if _browser:

        await _browser.shut()

        await _playwright.cease()

        _browser = None

        _playwright = None

# ── TOOLS ─────────────────────────────────────────────────────────────────────

@instrument

async def navigate(url: str) -> str:

    “”“

    Navigate the browser to a URL and return the web page’s textual content content material.

    Use when it’s essential to open an internet site or transfer to a brand new web page.

    Enter: full URL with https:// prefix.

    ““”

    web page = await get_page()

    await web page.goto(url, wait_until=“domcontentloaded”, timeout=20000)

    await web page.wait_for_load_state(“networkidle”)

    content material = await web page.inner_text(“physique”)

    return content material[:4000]

@instrument

async def extract_structured(css_selector: str) -> str:

    “”“

    Extract textual content from all components matching a CSS selector on the present web page.

    Use when it’s essential to pull particular components from the loaded web page.

    Enter: legitimate CSS selector string (e.g., ‘h3 a’, ‘.price_color’, ‘article.product_pod’).

    ““”

    web page = await get_page()

    strive:

        await web page.wait_for_selector(css_selector, timeout=5000)

        components = await web page.query_selector_all(css_selector)

        texts = []

        for el in components[:20]:  # Cap at 20 components to maintain output manageable

            textual content = await el.inner_text()

            texts.append(textual content.strip())

        return “n”.be a part of(texts) if texts else “No components discovered.”

    besides Exception as e:

        return f“Extraction failed: {str(e)}”

@instrument

async def get_current_url() -> str:

    “”“Return the URL the browser is at the moment on. No enter required.”“”

    web page = await get_page()

    return web page.url

# ── AGENT ─────────────────────────────────────────────────────────────────────

llm = ChatOpenAI(

    mannequin=“gpt-4o”,

    temperature=0,

    api_key=os.getenv(“OPENAI_API_KEY”)

)

instruments = [navigate, extract_structured, get_current_url]

agent = create_react_agent(llm, instruments)

SYSTEM = (

    “You’re a browser-based analysis agent. You might have entry to an actual browser. “

    “Use navigate() to open pages, extract_structured() to tug particular components, “

    “and get_current_url() to examine the place you’re. “

    “All the time navigate first, then extract. Be concise in your closing reply.”

)

async def run_agent(question: str) -> str:

    outcome = await agent.ainvoke({

        “messages”: [

            SystemMessage(content=SYSTEM),

            HumanMessage(content=query)

        ]

    })

    await teardown()

    return outcome[“messages”][–1].content material

# ── DEMO ──────────────────────────────────────────────────────────────────────

if __name__ == “__main__”:

    question = (

        “Go to https://books.toscrape.com and extract the titles and costs “

        “of the primary 5 books listed. Return them as a structured record.”

    )

    print(f“Question: {question}n”)

    reply = asyncio.run(run_agent(question))

    print(f“Reply:n{reply}”)