Live local Playwright browser environment. The environment owns the browser/page and executes each model action as an async Python snippet with ``page``, ``context``, ``browser``, ``playwright``, and ``task`` already available.
| 189 | |
| 190 | |
| 191 | class LocalBrowserEnvironment: |
| 192 | """Live local Playwright browser environment. |
| 193 | |
| 194 | The environment owns the browser/page and executes each model action as an async |
| 195 | Python snippet with ``page``, ``context``, ``browser``, ``playwright``, and |
| 196 | ``task`` already available. |
| 197 | """ |
| 198 | |
| 199 | def __init__(self, *, config_class: type = LocalBrowserEnvironmentConfig, **kwargs): |
| 200 | self.config = config_class(**kwargs) |
| 201 | fields_set = getattr(self.config, "model_fields_set", None) |
| 202 | if fields_set is None: |
| 203 | fields_set = getattr(self.config, "__fields_set__", set()) |
| 204 | self._config_fields_set = set(fields_set) |
| 205 | self.config.local_cdp_url = _resolve_local_cdp_url( |
| 206 | self.config.local_cdp_url, |
| 207 | explicit="local_cdp_url" in self._config_fields_set, |
| 208 | ) |
| 209 | self.config.output_dir = self.config.output_dir.expanduser() |
| 210 | self.config.user_data_dir = _resolve_user_data_dir( |
| 211 | self.config.user_data_dir, |
| 212 | explicit="user_data_dir" in self._config_fields_set, |
| 213 | ) |
| 214 | self._playwright = None |
| 215 | self._browser = None |
| 216 | self._context = None |
| 217 | self._page = None |
| 218 | self._local_cdp_page = None |
| 219 | self._loop: asyncio.AbstractEventLoop | None = None |
| 220 | self._local_cdp_process: subprocess.Popen | None = None |
| 221 | self._connected_over_cdp = False |
| 222 | self._step_index = 0 |
| 223 | self._prepared_task: dict[str, Any] = {} |
| 224 | self._console_history: list[str] = [] |
| 225 | self._step_console: list[str] = [] |
| 226 | self._step_python_code = "" |
| 227 | self._step_python_output = "" |
| 228 | |
| 229 | def _screenshots_dir(self) -> Path: |
| 230 | return self.config.output_dir / "screenshots" |
| 231 | |
| 232 | def _steps_dir(self) -> Path: |
| 233 | return self.config.output_dir / "steps" |
| 234 | |
| 235 | def prepare(self, **kwargs) -> None: |
| 236 | self._prepared_task = dict(kwargs) |
| 237 | self._step_index = 0 |
| 238 | self._console_history = [] |
| 239 | self._step_console = [] |
| 240 | start_url = kwargs.get("start_url") or self.config.start_url |
| 241 | if start_url: |
| 242 | self.config.start_url = str(start_url) |
| 243 | |
| 244 | self.config.output_dir.mkdir(parents=True, exist_ok=True) |
| 245 | self._steps_dir().mkdir(parents=True, exist_ok=True) |
| 246 | self._screenshots_dir().mkdir(parents=True, exist_ok=True) |
| 247 | (self.config.output_dir / "task.json").write_text( |
| 248 | json.dumps(kwargs, indent=2), |
nothing calls this directly
no outgoing calls
no test coverage detected