| 1 | from __future__ import annotations |
| 2 | |
| 3 | from typing import Any |
| 4 | |
| 5 | from helpers.errors import RepairableException |
| 6 | from plugins._browser.helpers.config import RUNTIME_BACKEND_KEY, get_browser_config |
| 7 | from plugins._browser.helpers.runtime import get_runtime as get_container_runtime |
| 8 | |
| 9 | |
| 10 | DOCKER_BROWSER_RECOVERY_HELP = ( |
| 11 | "To use Agent Zero's internal Docker browser instead, open Browser settings and " |
| 12 | "set Browser location to Internal Docker browser, or run `/browser container` " |
| 13 | "from A0 CLI." |
| 14 | ) |
| 15 | |
| 16 | |
| 17 | async def get_tool_runtime(agent: Any): |
| 18 | context_id = str(agent.context.id) |
| 19 | config = get_browser_config(agent=agent) |
| 20 | backend = str(config.get(RUNTIME_BACKEND_KEY) or "container").strip() |
| 21 | |
| 22 | if backend == "container": |
| 23 | return await get_container_runtime(context_id) |
| 24 | |
| 25 | sid = _select_host_browser_candidate_sid(context_id) |
| 26 | if sid: |
| 27 | from plugins._browser.helpers.connector_runtime import ConnectorBrowserRuntime |
| 28 | |
| 29 | return ConnectorBrowserRuntime(context_id, agent) |
| 30 | |
| 31 | if backend == "host_required": |
| 32 | detail = _host_browser_status_detail(context_id) |
| 33 | message = ( |
| 34 | "Bring Your Own Browser mode is enabled, but no subscribed A0 CLI currently " |
| 35 | "advertises host-browser support" |
| 36 | + (f": {detail}" if detail else ".") |
| 37 | ) |
| 38 | raise RepairableException( |
| 39 | f"{message} Connect A0 CLI to this chat, allow host browser access, and retry. " |
| 40 | f"{DOCKER_BROWSER_RECOVERY_HELP}" |
| 41 | ) |
| 42 | |
| 43 | return await get_container_runtime(context_id) |
| 44 | |
| 45 | |
| 46 | def _select_host_browser_target_sid(context_id: str) -> str | None: |
| 47 | try: |
| 48 | from plugins._a0_connector.helpers.ws_runtime import select_host_browser_target_sid |
| 49 | except ImportError: |
| 50 | return None |
| 51 | return select_host_browser_target_sid(context_id) |
| 52 | |
| 53 | |
| 54 | def _select_host_browser_candidate_sid(context_id: str) -> str | None: |
| 55 | try: |
| 56 | from plugins._a0_connector.helpers.ws_runtime import select_host_browser_candidate_sid |
| 57 | except ImportError: |
| 58 | return _select_host_browser_target_sid(context_id) |
| 59 | return select_host_browser_candidate_sid(context_id) |
| 60 | |
| 61 | |
| 62 | def _host_browser_status_detail(context_id: str) -> str: |
| 63 | try: |
| 64 | from plugins._a0_connector.helpers.ws_runtime import host_browser_metadata_for_context |
| 65 | except ImportError: |
| 66 | return "" |
| 67 | statuses = host_browser_metadata_for_context(context_id) |
| 68 | if not statuses: |
| 69 | return "open A0 CLI and connect it to this Agent Zero chat." |
| 70 | parts = [] |
| 71 | for status in statuses: |
| 72 | parts.append( |
| 73 | f"sid={status.get('sid')} supported={status.get('supported')} " |
| 74 | f"can_prepare={status.get('can_prepare')} enabled={status.get('enabled')} " |
| 75 | f"status={status.get('status') or 'unknown'} " |
| 76 | f"reason={status.get('support_reason') or 'none'}" |
| 77 | ) |
| 78 | return "; ".join(parts) |