trading_agent.tools.browser_session

One desk's Chrome browser, driven by Playwright.

Chrome runs headless with a profile of its own inside the desk's folder, so it never sees the user's logins, cookies or history. It identifies itself as ordinary Chrome rather than HeadlessChrome and hides the automation flag, because Indian finance sites such as Moneycontrol and Zee Business refuse headless browsers with HTTP 403.

Every request the browser makes, including images, scripts and redirects, passes the address guard, so no page can make it fetch anything from this machine or the local network, and file:// pages cannot be opened. Downloads are refused.

A snapshot gives every visible clickable element a number, so a model that cannot see pictures can still say "click 3" or "type into 7".

Typical usage example:

session = BrowserSession(profile_directory, screenshot_directory, guard, browser_configuration) snapshot = await session.open('https://www.nseindia.com/') await session.close()

BrowserError

Bases: Exception

The browser could not do what was asked, for a reason the model can act on.

BrowserSession

BrowserSession(profile_directory, screenshot_directory, guard, browser_configuration)

A desk's browser, started on first use.

Creates the session without starting Chrome.

Parameters:

Name	Type	Description	Default
profile_directory	Path	The desk's own Chrome profile folder.	required
screenshot_directory	Path	Where screenshots are saved.	required
guard	AddressGuard	The check every request must pass.	required
browser_configuration	BrowserConfiguration	Timeouts and limits.	required

Source code in src/trading_agent/tools/browser_session.py

def __init__(
    self,
    profile_directory: Path,
    screenshot_directory: Path,
    guard: address_guard_module.AddressGuard,
    browser_configuration: agent_configuration.BrowserConfiguration,
):
    """Creates the session without starting Chrome.

    Args:
        profile_directory (Path): The desk's own Chrome profile folder.
        screenshot_directory (Path): Where screenshots are saved.
        guard (address_guard_module.AddressGuard): The check every request must pass.
        browser_configuration (agent_configuration.BrowserConfiguration): Timeouts and limits.
    """
    self._profile_directory = profile_directory
    self._screenshot_directory = screenshot_directory
    self._guard = guard
    self._settings = browser_configuration
    self._playwright = None
    self._context = None
    self._current_page = None
    self._host_verdicts = {}
    self._lock = asyncio.Lock()

open async

open(address)

Opens a page in the current tab and returns its snapshot.

Parameters:

Name	Type	Description	Default
address	str	The page's address.	required

Returns:

Type	Description
str	The page's snapshot.

Raises:

Type	Description
BrowserError	The address is refused, or the page could not be opened.
AddressRefusedError	The address is not on the public internet.

Source code in src/trading_agent/tools/browser_session.py

async def open(self, address: str) -> str:
    """Opens a page in the current tab and returns its snapshot.

    Args:
        address (str): The page's address.

    Returns:
        str: The page's snapshot.

    Raises:
        BrowserError: The address is refused, or the page could not be opened.
        address_guard_module.AddressRefusedError: The address is not on the public internet.
    """
    async with self._lock:
        await self._guard.check(address)
        page = await self._page()
        try:
            response = await page.goto(
                address,
                wait_until='domcontentloaded',
                timeout=self._settings.navigation_timeout_seconds * 1000,
            )
        except playwright.async_api.Error as error:
            raise BrowserError(
                f'Could not open {address}: {error.message.splitlines()[0]}'
            ) from error
        await page.wait_for_timeout(_PAGE_SETTLE_MILLISECONDS)
        status = response.status if response is not None else 'unknown'
        return f'HTTP status {status}\n' + await self._snapshot_text(page)

snapshot async

snapshot()

Lists the current page's clickable elements and text.

Returns:

Type	Description
str	The snapshot.

Raises:

Type	Description
BrowserError	No page is open yet.

Source code in src/trading_agent/tools/browser_session.py

async def snapshot(self) -> str:
    """Lists the current page's clickable elements and text.

    Returns:
        str: The snapshot.

    Raises:
        BrowserError: No page is open yet.
    """
    async with self._lock:
        return await self._snapshot_text(await self._open_page())

click async

click(reference)

Clicks a numbered element from the last snapshot.

Parameters:

Name	Type	Description	Default
reference	int	The element's number.	required

Returns:

Type	Description
str	The snapshot after the click.

Raises:

Type	Description
BrowserError	No element has that number, or the click failed.

Source code in src/trading_agent/tools/browser_session.py

async def click(self, reference: int) -> str:
    """Clicks a numbered element from the last snapshot.

    Args:
        reference (int): The element's number.

    Returns:
        str: The snapshot after the click.

    Raises:
        BrowserError: No element has that number, or the click failed.
    """
    async with self._lock:
        page = await self._open_page()
        element = page.locator(f'[{_REFERENCE_ATTRIBUTE}="{reference}"]')
        try:
            await element.click(timeout=_ACTION_TIMEOUT_MILLISECONDS)
        except playwright.async_api.Error as error:
            raise BrowserError(
                f'Could not click element {reference}: {error.message.splitlines()[0]}. Take a new snapshot; the numbers change when the page changes.'
            ) from error
        await self._settle(page)
        return await self._snapshot_text(page)

type_text async

type_text(reference, text, submit)

Types into a numbered field, replacing what is in it.

Parameters:

Name	Type	Description	Default
reference	int	The field's number from the last snapshot.	required
text	str	What to type.	required
submit	bool	Whether to press Enter afterwards.	required

Returns:

Type	Description
str	The snapshot after typing.

Raises:

Type	Description
BrowserError	No field has that number, or typing failed.

Source code in src/trading_agent/tools/browser_session.py

async def type_text(self, reference: int, text: str, submit: bool) -> str:
    """Types into a numbered field, replacing what is in it.

    Args:
        reference (int): The field's number from the last snapshot.
        text (str): What to type.
        submit (bool): Whether to press Enter afterwards.

    Returns:
        str: The snapshot after typing.

    Raises:
        BrowserError: No field has that number, or typing failed.
    """
    async with self._lock:
        page = await self._open_page()
        field = page.locator(f'[{_REFERENCE_ATTRIBUTE}="{reference}"]')
        try:
            await field.fill(text, timeout=_ACTION_TIMEOUT_MILLISECONDS)
            if submit:
                await field.press(
                    'Enter', timeout=_ACTION_TIMEOUT_MILLISECONDS
                )
        except playwright.async_api.Error as error:
            raise BrowserError(
                f'Could not type into element {reference}: {error.message.splitlines()[0]}. Take a new snapshot and pick a text field.'
            ) from error
        await self._settle(page)
        return await self._snapshot_text(page)

screenshot async

screenshot()

Takes a screenshot of the visible part of the current page.

Returns:

Type	Description
tuple[bytes, Path]	The PNG picture and where it was saved.

Raises:

Type	Description
BrowserError	No page is open yet.

Source code in src/trading_agent/tools/browser_session.py

async def screenshot(self) -> tuple[bytes, Path]:
    """Takes a screenshot of the visible part of the current page.

    Returns:
        tuple[bytes, Path]: The PNG picture and where it was saved.

    Raises:
        BrowserError: No page is open yet.
    """
    async with self._lock:
        page = await self._open_page()
        self._screenshot_directory.mkdir(parents=True, exist_ok=True)
        now = datetime.datetime.now(datetime.UTC)
        path = self._screenshot_directory / f'{now:%Y%m%d-%H%M%S-%f}.png'
        picture = await page.screenshot(path=str(path), type='png')
        return picture, path

back async

back()

Goes back one page in the current tab.

Returns:

Type	Description
str	The snapshot of the previous page.

Raises:

Type	Description
BrowserError	There is no earlier page.

Source code in src/trading_agent/tools/browser_session.py

async def back(self) -> str:
    """Goes back one page in the current tab.

    Returns:
        str: The snapshot of the previous page.

    Raises:
        BrowserError: There is no earlier page.
    """
    async with self._lock:
        page = await self._open_page()
        response = await page.go_back(
            wait_until='domcontentloaded',
            timeout=self._settings.navigation_timeout_seconds * 1000,
        )
        if response is None:
            raise BrowserError('There is no earlier page in this tab.')
        await page.wait_for_timeout(_PAGE_SETTLE_MILLISECONDS)
        return await self._snapshot_text(page)

tabs async

tabs(action, index)

Lists, switches to, or closes tabs.

Parameters:

Name	Type	Description	Default
action	str	"list", "switch" or "close".	required
index	int | None	The tab's number from the list, for "switch" and "close".	required

Returns:

Type	Description
str	The list of tabs after the action, with the current one marked.

Raises:

Type	Description
BrowserError	The tab number does not exist, or no number was given.

Source code in src/trading_agent/tools/browser_session.py

async def tabs(self, action: str, index: int | None) -> str:
    """Lists, switches to, or closes tabs.

    Args:
        action (str): "list", "switch" or "close".
        index (int | None): The tab's number from the list, for "switch" and "close".

    Returns:
        str: The list of tabs after the action, with the current one marked.

    Raises:
        BrowserError: The tab number does not exist, or no number was given.
    """
    async with self._lock:
        await self._page()
        pages = self._context.pages
        if action != 'list':
            if index is None or not 1 <= index <= len(pages):
                raise BrowserError(
                    f'Give a tab number from 1 to {len(pages)}.'
                )
            chosen = pages[index - 1]
            if action == 'switch':
                self._current_page = chosen
                await chosen.bring_to_front()
            if action == 'close':
                await chosen.close()
                if chosen is self._current_page:
                    self._current_page = None
        lines = []
        for number, page in enumerate(self._context.pages, start=1):
            marker = ' (current)' if page is self._current_page else ''
            lines.append(
                f'{number}. {await page.title() or "(untitled)"} · {page.url}{marker}'
            )
        return '\n'.join(lines) or 'No tabs are open.'

close async

close()

Closes Chrome and stops Playwright.

Source code in src/trading_agent/tools/browser_session.py

async def close(self) -> None:
    """Closes Chrome and stops Playwright."""
    if self._context is not None:
        try:
            await self._context.close()
        except playwright.async_api.Error:
            _LOGGER.warning('Chrome did not close cleanly')
    if self._playwright is not None:
        await self._playwright.stop()
    self._context = None
    self._playwright = None
    self._current_page = None