|
| 1 | +"""Run the Browser Use adapter without installing Browser Use. |
| 2 | +
|
| 3 | +This uses a fake Browser Use-shaped agent so the example is deterministic: |
| 4 | +
|
| 5 | + python examples/browser_use_callback_demo.py |
| 6 | +
|
| 7 | +Then open the BrowserTrace UI: |
| 8 | +
|
| 9 | + browsertrace |
| 10 | +""" |
| 11 | + |
| 12 | +from __future__ import annotations |
| 13 | + |
| 14 | +import asyncio |
| 15 | +import base64 |
| 16 | +import os |
| 17 | + |
| 18 | +from browsertrace import Tracer |
| 19 | +from browsertrace.integrations.browser_use import attach_tracer |
| 20 | + |
| 21 | + |
| 22 | +PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16 |
| 23 | + |
| 24 | + |
| 25 | +class DemoBrowserState: |
| 26 | + def __init__(self, url: str): |
| 27 | + self.url = url |
| 28 | + self.screenshot = base64.b64encode(PNG_BYTES).decode("ascii") |
| 29 | + |
| 30 | + |
| 31 | +class DemoAction: |
| 32 | + def __init__(self, payload: dict): |
| 33 | + self._payload = payload |
| 34 | + |
| 35 | + def model_dump(self, exclude_none: bool = True) -> dict: |
| 36 | + return self._payload |
| 37 | + |
| 38 | + |
| 39 | +class DemoCurrentState: |
| 40 | + def __init__(self, thought: str): |
| 41 | + self.thought = thought |
| 42 | + |
| 43 | + |
| 44 | +class DemoAgentOutput: |
| 45 | + def __init__(self, thought: str, action: dict): |
| 46 | + self.current_state = DemoCurrentState(thought) |
| 47 | + self.action = [DemoAction(action)] |
| 48 | + |
| 49 | + |
| 50 | +class DemoBrowserUseAgent: |
| 51 | + def __init__(self): |
| 52 | + self._new_step_callback = None |
| 53 | + |
| 54 | + def register_new_step_callback(self, callback): |
| 55 | + self._new_step_callback = callback |
| 56 | + |
| 57 | + async def run(self) -> None: |
| 58 | + if self._new_step_callback is None: |
| 59 | + raise RuntimeError("step callback was not registered") |
| 60 | + |
| 61 | + await self._new_step_callback( |
| 62 | + DemoBrowserState("https://example.com/search"), |
| 63 | + DemoAgentOutput( |
| 64 | + "search for the project", |
| 65 | + {"search_google": {"query": "BrowserTrace"}}, |
| 66 | + ), |
| 67 | + 0, |
| 68 | + ) |
| 69 | + await self._new_step_callback( |
| 70 | + DemoBrowserState("https://example.com/results"), |
| 71 | + DemoAgentOutput( |
| 72 | + "open the first useful result", |
| 73 | + {"click": {"selector": "#result-1"}}, |
| 74 | + ), |
| 75 | + 1, |
| 76 | + ) |
| 77 | + |
| 78 | + |
| 79 | +async def main() -> None: |
| 80 | + tracer = Tracer(home=os.environ.get("BROWSERTRACE_HOME")) |
| 81 | + agent = DemoBrowserUseAgent() |
| 82 | + |
| 83 | + with attach_tracer(agent, tracer, name="demo: browser-use callback flow") as bt_run: |
| 84 | + await agent.run() |
| 85 | + |
| 86 | + print(f"BrowserTrace run id: {bt_run.run.id}") |
| 87 | + print("Recorded Browser Use-shaped callback steps: search_google, click") |
| 88 | + print("Open the local UI with: browsertrace") |
| 89 | + |
| 90 | + |
| 91 | +if __name__ == "__main__": |
| 92 | + asyncio.run(main()) |
0 commit comments