Use pyagentbrowser in async applications

AsyncBrowser mirrors the browser, snapshot, ref, query, and capability namespace APIs. Operations that call the engine are awaitable. Handle construction and immutable snapshot lookup remain synchronous.

import asyncio

from agentbrowser import AsyncBrowser, Wait


async def main() -> None:
    async with await AsyncBrowser.launch() as browser:
        await browser.page.open("https://example.com")
        snapshot = await browser.page.observe()
        link = snapshot.one(role="link", name="Learn more")
        result = await link.click(
            wait=Wait.url("*://www.iana.org/*")
        )
        print(result.after.origin)


asyncio.run(main())

Shared semantics

The synchronous and asynchronous surfaces use the same method names, parameters, result models, validation, confirmation behavior, domain policy, and evidence model.

The intentional differences are:

• Engine operations return awaitables.
• Query factories and snapshot lookups stay synchronous because they create handles or inspect immutable data.
• AsyncBrowser.close(*, timeout=None) optionally limits a caller's wait in seconds.
• Native work is serialized through one owner thread so the event loop stays available.

Cancellation and close

Canceling an await skips queued work that has not started. For active work, cancellation interrupts native dispatch and waits for the owner thread to settle the command before propagating CancelledError. The controller can accept subsequent commands and close normally.

Browser effects already sent can still occur. Inspect the current document before retrying a mutation, and capture a new snapshot before using refs.

close() is single-flight. Concurrent callers share the same terminal result or terminal error. Closing rejects queued operations with RuntimeError, requests native shutdown, and joins the owner thread.

The default waits for native cleanup, including Chrome's graceful-exit period. An explicit timeout raises TimeoutError for that caller while shared cleanup continues. Await close() again to retrieve its result. A worker that remains alive after the internal join timeout raises RuntimeError.

Confirmation

AsyncPendingAction.confirm() and deny() are awaitable. Confirmation preserves typed decoding and any remaining wait, snapshot, and diff work for the initiating operation.

from agentbrowser import AsyncBrowser, ConfirmationRequired, SessionOptions


async def publish() -> None:
    session = SessionOptions(confirm_actions=("click",))
    async with await AsyncBrowser.launch(session=session) as browser:
        await browser.page.set_content("<button>Publish</button>")
        snapshot = await browser.page.observe()
        try:
            await snapshot.one(role="button", name="Publish").click()
        except ConfirmationRequired as required:
            await required.pending.confirm()

Use Browser controllers and Capability namespaces for the shared contract. Async counterparts preserve the same order and return shapes.