diff --git a/backend/scripts/probe_agentcore_browser.py b/backend/scripts/probe_agentcore_browser.py new file mode 100644 index 00000000..1e29cfed --- /dev/null +++ b/backend/scripts/probe_agentcore_browser.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python +"""Smoke-test the AgentCore Browser tool against real AWS. + +Verifies the parts unit tests cannot: that AgentCore's automation WebSocket +accepts our SigV4 headers, that it exposes a browser-level CDP endpoint where +`Target.getTargets` works, and that a real page navigates and evaluates. + +Usage: + cd backend + AWS_PROFILE=dev-ai BROWSER_ID= \ + uv run python scripts/probe_agentcore_browser.py + + # or let it discover the browser from the account: + AWS_PROFILE=dev-ai uv run python scripts/probe_agentcore_browser.py --discover + +Options: + --url URL Page to load (default: https://example.com) + --discover List custom browsers in the account and use the first + --keep Leave the session running and print the live-view URL + (otherwise the session is stopped on exit) + +Costs a browser session for as long as it runs. Stops it on the way out +unless --keep is passed. +""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s") +logger = logging.getLogger("probe") + + +def _discover_browser_id(region: str) -> str | None: + import boto3 + + client = boto3.client("bedrock-agentcore-control", region_name=region) + try: + response = client.list_browsers(maxResults=20) + except Exception as exc: # noqa: BLE001 + logger.error("list_browsers failed: %s", exc) + return None + summaries = response.get("browserSummaries", []) or response.get("browsers", []) + for summary in summaries: + identifier = summary.get("browserId") or summary.get("browserIdentifier") + name = summary.get("name", "") + logger.info("found browser: %s (%s)", identifier, name) + if identifier and not str(identifier).startswith("aws."): + return identifier + return None + + +async def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--url", default="https://example.com") + parser.add_argument("--discover", action="store_true") + parser.add_argument("--keep", action="store_true") + args = parser.parse_args() + + region = os.environ.get("AWS_REGION", "us-west-2") + + if args.discover: + discovered = _discover_browser_id(region) + if discovered: + os.environ["BROWSER_ID"] = discovered + logger.info("using discovered browser: %s", discovered) + + identifier = os.environ.get("BROWSER_ID") or "aws.browser.v1" + logger.info("region=%s browser=%s", region, identifier) + + from bedrock_agentcore.tools.browser_client import BrowserClient + from agents.builtin_tools.browser.cdp_client import CdpSession + + client = BrowserClient(region=region) + session_id = await asyncio.to_thread( + client.start, + identifier=identifier, + session_timeout_seconds=300, + viewport={"width": 1280, "height": 800}, + ) + logger.info("started session %s", session_id) + + cdp = None + try: + ws_url, headers = await asyncio.to_thread(client.generate_ws_headers) + logger.info("connecting to %s", ws_url.split("/sessions/")[0] + "/sessions/...") + cdp = await CdpSession.connect(ws_url, headers) + print("\n[1/5] CDP connected and attached to a page target ✅") + + await cdp.navigate(args.url) + print(f"[2/5] Navigated to {args.url} ✅") + + title = await cdp.evaluate("document.title") + print(f"[3/5] document.title = {title!r} ✅") + + text = await cdp.evaluate( + "(() => (document.querySelector('main') || document.body).innerText)()" + ) + preview = (text or "")[:200].replace("\n", " ") + print(f"[4/5] Page text ({len(text or '')} chars): {preview}... ✅") + + # The WebMCP hook: prove an init script runs before page scripts. + await cdp.add_init_script("window.__probe_init__ = 'ran';") + await cdp.navigate(args.url) + marker = await cdp.evaluate("window.__probe_init__ || 'MISSING'") + status = "✅" if marker == "ran" else "❌" + print(f"[5/5] Init script before page load: {marker!r} {status}") + + if args.keep: + url = await asyncio.to_thread(client.generate_live_view_url) + print(f"\nLive view (session left running): {url}") + return 0 + + print("\nAll checks passed.") + return 0 + except Exception as exc: # noqa: BLE001 + logger.error("probe failed: %s", exc, exc_info=True) + return 1 + finally: + if cdp is not None: + await cdp.close() + if not args.keep: + await asyncio.to_thread(client.stop) + logger.info("stopped session %s", session_id) + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/backend/scripts/seed_bootstrap_data.py b/backend/scripts/seed_bootstrap_data.py index 8477eb54..f22c191d 100644 --- a/backend/scripts/seed_bootstrap_data.py +++ b/backend/scripts/seed_bootstrap_data.py @@ -402,6 +402,23 @@ def seed_default_models( "isPublic": False, "forwardAuthToken": False, }, + { + "toolId": "browse_web", + "displayName": "Web Browser", + "description": ( + "Browse the web in a real Chrome browser: navigate pages, read " + "JavaScript-rendered content, fill forms, and click through " + "multi-step flows." + ), + "category": "browser", + # Deliberately off by default. Each session bills an AgentCore Browser + # session on top of model tokens, and a browsing transcript is a large + # per-turn payload — this is opt-in per user, granted per role. + "enabledByDefault": False, + "protocol": "local", + "isPublic": False, + "forwardAuthToken": False, + }, { "toolId": "generate_diagram_and_validate", "displayName": "Code Interpreter", diff --git a/backend/src/agents/builtin_tools/__init__.py b/backend/src/agents/builtin_tools/__init__.py index 3c7ea2ed..b06157a1 100644 --- a/backend/src/agents/builtin_tools/__init__.py +++ b/backend/src/agents/builtin_tools/__init__.py @@ -2,9 +2,11 @@ This package contains tools that leverage AWS Bedrock capabilities: - Code Interpreter: Execute Python code for diagrams and charts +- Browser: Drive a real Chrome browser in the AgentCore Browser sandbox - Spreadsheet Analysis: Analyze tabular data via Code Interpreter (factory-produced, not in registry) """ +from .browser import browse_web from .code_interpreter_diagram_tool import generate_diagram_and_validate from .spreadsheet_analysis import make_list_spreadsheets_tool, make_analyze_tool @@ -12,5 +14,6 @@ # Factory-produced tools (make_list_spreadsheets_tool, make_analyze_tool) are created # per-request with context and injected via extra_tools — not registered here. __all__ = [ + 'browse_web', 'generate_diagram_and_validate', ] diff --git a/backend/src/agents/builtin_tools/browser/__init__.py b/backend/src/agents/builtin_tools/browser/__init__.py new file mode 100644 index 00000000..b3fbecdc --- /dev/null +++ b/backend/src/agents/builtin_tools/browser/__init__.py @@ -0,0 +1,5 @@ +"""AgentCore Browser tool package.""" + +from .browse_tool import browse_web + +__all__ = ["browse_web"] diff --git a/backend/src/agents/builtin_tools/browser/browse_tool.py b/backend/src/agents/builtin_tools/browser/browse_tool.py new file mode 100644 index 00000000..ffcdcf54 --- /dev/null +++ b/backend/src/agents/builtin_tools/browser/browse_tool.py @@ -0,0 +1,293 @@ +"""`browse_web` — drive a real browser in the AgentCore Browser sandbox. + +One action per call, against a browser session scoped to the conversation +(see `session_pool`). The model navigates, reads, interacts, and closes. + +Cost posture. Every action's output is capped before it reaches the model, +because a browsing transcript is otherwise the classic unbounded per-turn +payload the cost tenet exists to catch: page text is truncated, evaluated +results are truncated, and a screenshot is only ever taken when the model +explicitly asks — vision tokens are the single most expensive thing this tool +can produce, so it is never automatic. Prefer `extract_text` over `screenshot` +for reading; prefer `evaluate` over both for structured extraction. + +Interaction is done through JS in the page rather than synthesized input +events. That covers clicking, filling and reading for ordinary pages, keeps +the CDP surface small, and is what AWS's own guidance recommends for +extraction. Pages that require true trusted input (drag, some canvas widgets, +a few anti-bot forms) are a known limitation. +""" + +from __future__ import annotations + +import base64 +import json +import logging +import os +from typing import Any, Dict, Optional +from urllib.parse import urlparse + +from strands import tool +from strands.types.tools import ToolContext + +from .cdp_client import CdpError +from . import session_pool + +logger = logging.getLogger(__name__) + +# Output budgets. Roughly: 8000 chars ≈ 2k tokens of page text per action. +MAX_TEXT_CHARS = int(os.environ.get("BROWSER_MAX_TEXT_CHARS", 8000)) +MAX_EVAL_CHARS = int(os.environ.get("BROWSER_MAX_EVAL_CHARS", 4000)) +MAX_LINKS = int(os.environ.get("BROWSER_MAX_LINKS", 50)) + +_ALLOWED_SCHEMES = ("http", "https") + +# Page text, preferring the main content region over site chrome. +_EXTRACT_TEXT_JS = """ +(() => { + const el = document.querySelector('main') || document.querySelector('article') || document.body; + return el ? el.innerText : ''; +})() +""" + +_EXTRACT_LINKS_JS = """ +(() => { + const seen = new Set(); + const out = []; + for (const a of document.querySelectorAll('a[href]')) { + const href = a.href; + if (!href || seen.has(href)) continue; + if (!href.startsWith('http')) continue; + seen.add(href); + out.push({ text: (a.innerText || '').trim().slice(0, 120), href }); + if (out.length >= %d) break; + } + return out; +})() +""" + + +def _ok(text: str, extra: Optional[list] = None) -> Dict[str, Any]: + content = [{"text": text}] + if extra: + content.extend(extra) + return {"content": content, "status": "success"} + + +def _err(text: str) -> Dict[str, Any]: + return {"content": [{"text": text}], "status": "error"} + + +def _truncate(value: str, limit: int, label: str) -> str: + if len(value) <= limit: + return value + return ( + value[:limit] + + f"\n\n[{label} truncated at {limit} chars of {len(value)}. " + "Use `evaluate` with a selector to extract only what you need.]" + ) + + +def _tool_enabled() -> bool: + """Kill switch. Default on, per house style.""" + return os.environ.get("BROWSER_TOOL_ENABLED", "true").strip().lower() != "false" + + +def _validate_url(url: str) -> Optional[str]: + """Return an error string if the URL is not a safe public target.""" + parsed = urlparse(url) + if parsed.scheme not in _ALLOWED_SCHEMES: + return f"❌ Only http/https URLs are supported (got '{parsed.scheme or 'none'}')." + host = (parsed.hostname or "").lower() + if not host: + return "❌ URL has no host." + # The browser runs in AWS with PUBLIC network mode, so it cannot reach our + # VPC — but blocking the obvious loopback/metadata targets keeps the tool + # honest if the network mode ever changes. + if host in ("localhost", "127.0.0.1", "::1", "169.254.169.254", "metadata.google.internal"): + return "❌ Refusing to browse loopback or instance-metadata addresses." + return None + + +def _js_string(value: str) -> str: + """Embed a Python string as a JS literal.""" + return json.dumps(value) + + +@tool(context=True) +async def browse_web( + action: str, + tool_context: ToolContext, + url: Optional[str] = None, + selector: Optional[str] = None, + text: Optional[str] = None, + script: Optional[str] = None, +) -> Dict[str, Any]: + """Browse the web in a real Chrome browser: navigate pages, read content, fill forms, click. + + Use this when a task needs a live, interactive page — content behind + JavaScript rendering, a multi-step flow, a search you must click through. + For a single static page, `fetch_url_content` is cheaper and faster; reach + for this tool when that one is not enough. + + The browser session persists across calls within the conversation, so you + can navigate then interact then read. Call `close` when finished. + + Args: + action: One of: + `navigate` — go to `url` and return the page title and text. + `extract_text` — return the current page's visible text. + `extract_links` — return the current page's links. + `click` — click the element matching `selector`. + `type` — set `text` into the input matching `selector`. + `evaluate` — run `script` (JavaScript) and return its value. + The best tool for structured extraction, e.g. + `[...document.querySelectorAll('h2')].map(e => e.innerText)`. + `screenshot` — capture the viewport as an image. Expensive in + tokens; only use it when you must SEE the layout. + `live_view` — get a URL where the user can watch the session. + `close` — end the browser session. + url: Target URL for `navigate` (http/https). + selector: CSS selector for `click` and `type`. + text: Text to enter for `type`. + script: JavaScript expression for `evaluate`. + + Returns: + ToolResult with the action's output, truncated to a token budget. + """ + if not _tool_enabled(): + return _err("❌ The browser tool is disabled in this environment.") + + action = (action or "").strip().lower() + agent = getattr(tool_context, "agent", None) + if agent is None: + return _err("❌ Browser tool has no agent context.") + + if action == "close": + stopped = await session_pool.release(agent) + return _ok("✅ Browser session closed." if stopped else "No browser session was open.") + + if action == "navigate": + if not url: + return _err("❌ `navigate` requires `url`.") + invalid = _validate_url(url) + if invalid: + return _err(invalid) + + try: + live = await session_pool.acquire(agent) + except Exception as exc: # noqa: BLE001 - surface the reason, not a traceback + logger.error("browser: could not start session: %s", exc, exc_info=True) + return _err(f"❌ Could not start a browser session: {exc}") + + async with live.lock: + try: + return await _dispatch(live, action, url, selector, text, script, agent) + except CdpError as exc: + return _err(f"❌ {exc}") + except Exception as exc: # noqa: BLE001 + logger.error("browser: action '%s' failed: %s", action, exc, exc_info=True) + return _err(f"❌ Browser action '{action}' failed: {exc}") + + +async def _dispatch( + live: Any, + action: str, + url: Optional[str], + selector: Optional[str], + text: Optional[str], + script: Optional[str], + agent: Any, +) -> Dict[str, Any]: + cdp = live.cdp + + if action == "navigate": + await cdp.navigate(url) # type: ignore[arg-type] + title = await cdp.evaluate("document.title") + current = await cdp.evaluate("location.href") + body = await cdp.evaluate(_EXTRACT_TEXT_JS) or "" + return _ok( + f"✅ Navigated to {current}\nTitle: {title}\n\n" + + _truncate(str(body), MAX_TEXT_CHARS, "Page text") + ) + + if action == "extract_text": + body = await cdp.evaluate(_EXTRACT_TEXT_JS) or "" + return _ok(_truncate(str(body), MAX_TEXT_CHARS, "Page text")) + + if action == "extract_links": + links = await cdp.evaluate(_EXTRACT_LINKS_JS % MAX_LINKS) or [] + if not links: + return _ok("No links found on this page.") + lines = [f"- {item.get('text') or '(no text)'} → {item.get('href')}" for item in links] + return _ok(f"{len(lines)} link(s):\n" + "\n".join(lines)) + + if action == "click": + if not selector: + return _err("❌ `click` requires `selector`.") + clicked = await cdp.evaluate( + f""" + (() => {{ + const el = document.querySelector({_js_string(selector)}); + if (!el) return false; + el.click(); + return true; + }})() + """ + ) + if not clicked: + return _err(f"❌ No element matches selector `{selector}`.") + current = await cdp.evaluate("location.href") + return _ok(f"✅ Clicked `{selector}`. Current URL: {current}") + + if action == "type": + if not selector or text is None: + return _err("❌ `type` requires `selector` and `text`.") + # Set the value and fire the events frameworks listen for, so Angular + # / React state updates rather than silently keeping the old value. + typed = await cdp.evaluate( + f""" + (() => {{ + const el = document.querySelector({_js_string(selector)}); + if (!el) return false; + el.focus(); + el.value = {_js_string(text)}; + el.dispatchEvent(new Event('input', {{ bubbles: true }})); + el.dispatchEvent(new Event('change', {{ bubbles: true }})); + return true; + }})() + """ + ) + if not typed: + return _err(f"❌ No element matches selector `{selector}`.") + return _ok(f"✅ Entered text into `{selector}`.") + + if action == "evaluate": + if not script: + return _err("❌ `evaluate` requires `script`.") + value = await cdp.evaluate(script) + try: + rendered = json.dumps(value, ensure_ascii=False, indent=2) + except (TypeError, ValueError): + rendered = str(value) + return _ok(_truncate(rendered, MAX_EVAL_CHARS, "Result")) + + if action == "screenshot": + encoded = await cdp.screenshot() + if not encoded: + return _err("❌ Screenshot returned no data.") + return _ok( + "✅ Screenshot of the current viewport:", + [{"image": {"format": "png", "source": {"bytes": base64.b64decode(encoded)}}}], + ) + + if action == "live_view": + view_url = await session_pool.live_view_url(agent) + if not view_url: + return _err("❌ No live view available for this session.") + return _ok(f"Watch the browser session here (expires shortly):\n{view_url}") + + return _err( + f"❌ Unknown action '{action}'. Valid actions: navigate, extract_text, " + "extract_links, click, type, evaluate, screenshot, live_view, close." + ) diff --git a/backend/src/agents/builtin_tools/browser/cdp_client.py b/backend/src/agents/builtin_tools/browser/cdp_client.py new file mode 100644 index 00000000..5fabcd91 --- /dev/null +++ b/backend/src/agents/builtin_tools/browser/cdp_client.py @@ -0,0 +1,242 @@ +"""Minimal Chrome DevTools Protocol client for the AgentCore Browser. + +AgentCore's automation endpoint is a SigV4-signed WebSocket that speaks CDP. +`bedrock_agentcore.tools.browser_client.BrowserClient` handles the session +lifecycle and request signing; this module handles the protocol on top of it. + +Why raw CDP instead of Playwright: `websockets` is already in the image, and +the actions a browsing agent needs — navigate, evaluate, extract, screenshot — +are a handful of CDP commands. Playwright would add its bundled Node driver to +the inference-api container for auto-waiting and a selector engine we mostly +don't use, because JS evaluated in the page does the same job. If richer +interaction (frames, file chooser, real input events) is ever needed, this +module is the seam to swap. + +Nothing here is specific to the AgentCore Browser except `connect()`. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from typing import Any, Dict, Optional + +from websockets.asyncio.client import connect as ws_connect + +logger = logging.getLogger(__name__) + +# CDP command timeout. Navigation uses its own, longer budget. +_COMMAND_TIMEOUT_SECONDS = 30.0 + +# A screenshot frame can be a few MB; the default websockets frame cap (1 MiB) +# is too small for `Page.captureScreenshot`. +_MAX_FRAME_BYTES = 16 * 1024 * 1024 + + +class CdpError(RuntimeError): + """A CDP command returned an error, or the connection failed.""" + + +class CdpSession: + """One CDP connection, attached to a single page target. + + Commands are multiplexed over the socket by request id. A background + reader resolves futures; events are dropped (we poll instead of + subscribing, which keeps the state machine to one path). + """ + + def __init__(self, websocket: Any) -> None: + self._ws = websocket + self._next_id = 0 + self._pending: Dict[int, asyncio.Future] = {} + self._reader: Optional[asyncio.Task] = None + self._page_session_id: Optional[str] = None + self._closed = False + + # -- lifecycle --------------------------------------------------------- + + @classmethod + async def connect(cls, ws_url: str, headers: Dict[str, str]) -> "CdpSession": + """Open the automation socket and attach to a page target.""" + websocket = await ws_connect( + ws_url, + additional_headers=headers, + max_size=_MAX_FRAME_BYTES, + open_timeout=30, + ) + session = cls(websocket) + session._reader = asyncio.create_task(session._read_loop()) + await session._attach_to_page() + return session + + async def close(self) -> None: + """Close the socket and cancel the reader. Safe to call twice.""" + if self._closed: + return + self._closed = True + if self._reader is not None: + self._reader.cancel() + try: + await self._ws.close() + except Exception: # noqa: BLE001 - teardown is best effort + logger.debug("cdp: websocket close failed", exc_info=True) + for future in self._pending.values(): + if not future.done(): + future.set_exception(CdpError("CDP connection closed")) + self._pending.clear() + + @property + def closed(self) -> bool: + return self._closed + + # -- protocol ---------------------------------------------------------- + + async def _read_loop(self) -> None: + try: + async for raw in self._ws: + try: + message = json.loads(raw) + except (TypeError, ValueError): + continue + message_id = message.get("id") + if message_id is None: + continue # an event; we don't subscribe to any + future = self._pending.pop(message_id, None) + if future is not None and not future.done(): + future.set_result(message) + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 - surfaced on the next command + logger.info("cdp: read loop ended (%s)", exc) + for future in self._pending.values(): + if not future.done(): + future.set_exception(CdpError(f"CDP connection lost: {exc}")) + self._pending.clear() + + async def command( + self, + method: str, + params: Optional[Dict[str, Any]] = None, + *, + session_scoped: bool = True, + timeout: float = _COMMAND_TIMEOUT_SECONDS, + ) -> Dict[str, Any]: + """Send one CDP command and return its `result`. + + `session_scoped` targets the attached page; pass False for + browser-level commands (`Target.*`). + """ + if self._closed: + raise CdpError("CDP session is closed") + + self._next_id += 1 + message_id = self._next_id + payload: Dict[str, Any] = {"id": message_id, "method": method} + if params: + payload["params"] = params + if session_scoped and self._page_session_id: + payload["sessionId"] = self._page_session_id + + future: asyncio.Future = asyncio.get_running_loop().create_future() + self._pending[message_id] = future + try: + await self._ws.send(json.dumps(payload)) + message = await asyncio.wait_for(future, timeout=timeout) + except asyncio.TimeoutError as exc: + self._pending.pop(message_id, None) + raise CdpError(f"{method} timed out after {timeout:.0f}s") from exc + finally: + self._pending.pop(message_id, None) + + if "error" in message: + detail = message["error"].get("message", "unknown error") + raise CdpError(f"{method} failed: {detail}") + return message.get("result", {}) + + async def _attach_to_page(self) -> None: + """Find a page target and attach, so later commands address a tab.""" + targets = await self.command("Target.getTargets", session_scoped=False) + pages = [ + t for t in targets.get("targetInfos", []) if t.get("type") == "page" + ] + if not pages: + created = await self.command( + "Target.createTarget", {"url": "about:blank"}, session_scoped=False + ) + target_id = created["targetId"] + else: + target_id = pages[0]["targetId"] + + attached = await self.command( + "Target.attachToTarget", + {"targetId": target_id, "flatten": True}, + session_scoped=False, + ) + self._page_session_id = attached["sessionId"] + await self.command("Page.enable") + await self.command("Runtime.enable") + + # -- page operations --------------------------------------------------- + + async def evaluate(self, expression: str, *, timeout: float = _COMMAND_TIMEOUT_SECONDS) -> Any: + """Evaluate JS in the page and return the value by value. + + Raises `CdpError` if the expression throws — the message carries the + page's own exception text, which is what the model needs to correct + itself. + """ + result = await self.command( + "Runtime.evaluate", + { + "expression": expression, + "returnByValue": True, + "awaitPromise": True, + "userGesture": True, + }, + timeout=timeout, + ) + exception = result.get("exceptionDetails") + if exception: + text = exception.get("exception", {}).get("description") or exception.get("text") + raise CdpError(f"Page threw: {text}") + return result.get("result", {}).get("value") + + async def navigate(self, url: str, *, timeout: float = 45.0) -> None: + """Navigate and wait for the document to finish loading. + + Polls `document.readyState` rather than subscribing to lifecycle + events — one code path, and a page that never fires `load` (long-poll + connections, some SPAs) still returns once the DOM is usable. + """ + result = await self.command("Page.navigate", {"url": url}, timeout=timeout) + if result.get("errorText"): + raise CdpError(f"Navigation failed: {result['errorText']}") + + deadline = asyncio.get_running_loop().time() + timeout + while asyncio.get_running_loop().time() < deadline: + try: + state = await self.evaluate("document.readyState", timeout=10.0) + except CdpError: + state = None # mid-navigation context swap; retry + if state in ("interactive", "complete"): + return + await asyncio.sleep(0.25) + logger.info("cdp: navigation to %s did not settle within %ss", url, timeout) + + async def add_init_script(self, source: str) -> None: + """Install a script that runs before any page script, on every document. + + The WebMCP shim hook: this is the CDP equivalent of Playwright's + `add_init_script`, and it is what would let a page's + `document.modelContext` declarations be collected before the app + bootstraps. Unused today; see `docs/specs/webmcp-host-spa-tools.md`. + """ + await self.command("Page.addScriptToEvaluateOnNewDocument", {"source": source}) + + async def screenshot(self) -> str: + """Capture the viewport as base64 PNG.""" + result = await self.command( + "Page.captureScreenshot", {"format": "png"}, timeout=45.0 + ) + return result.get("data", "") diff --git a/backend/src/agents/builtin_tools/browser/session_pool.py b/backend/src/agents/builtin_tools/browser/session_pool.py new file mode 100644 index 00000000..d2ec789a --- /dev/null +++ b/backend/src/agents/builtin_tools/browser/session_pool.py @@ -0,0 +1,254 @@ +"""Browser session lifecycle, keyed per conversation. + +One AgentCore Browser session is reused across the tool calls of a +conversation: starting a session costs seconds and dollars, and a browsing +task is inherently multi-step. + +Where the key lives matters. Per the "one session can be served by more than +one agent" rule in CLAUDE.md, nothing durable may be cached on an agent +instance — so the *identity* of the browser session (its id) is stored on the +Strands `agent.state`, exactly as `app_context_dispatch` stores app context, +while the live socket is a process-local lookup keyed by that id. A second +agent instance for the same conversation reads the same id from state; if the +socket isn't in this process it reconnects to the same remote session rather +than starting a second one. + +`AgentState.get()` deep-copies, so the bag is read-modify-written wholesale +and every value is JSON-serializable. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import time +from dataclasses import dataclass, field +from typing import Any, Dict, Optional, Tuple + +from .cdp_client import CdpError, CdpSession + +logger = logging.getLogger(__name__) + +STATE_KEY = "browser_tool" +_SESSION_SUBKEY = "session" + +# AgentCore caps session_timeout at 8h; we deliberately stay near the low end. +# An abandoned session bills until its TTL expires, and a browsing task that +# needs more than 15 minutes of wall clock is a task that should be re-scoped. +SESSION_TIMEOUT_SECONDS = int(os.environ.get("BROWSER_SESSION_TIMEOUT_SECONDS", 900)) + +# Local reap: drop sockets unused for this long, and stop the remote session +# with them. Shorter than the remote TTL so we usually release first. +IDLE_REAP_SECONDS = int(os.environ.get("BROWSER_IDLE_REAP_SECONDS", 600)) + +DEFAULT_VIEWPORT = {"width": 1280, "height": 800} + + +@dataclass +class _LiveSession: + """A connected browser session owned by this process.""" + + session_id: str + identifier: str + client: Any # bedrock_agentcore BrowserClient + cdp: CdpSession + last_used: float = field(default_factory=time.monotonic) + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + + +_live: Dict[str, _LiveSession] = {} +_pool_lock = asyncio.Lock() + + +def _browser_identifier() -> str: + """Our custom browser from PlatformStack, else the AWS-managed one.""" + return os.environ.get("BROWSER_ID") or "aws.browser.v1" + + +def _region() -> str: + return os.environ.get("AWS_REGION", "us-west-2") + + +def _read_state(agent: Any) -> Optional[Dict[str, Any]]: + state = getattr(agent, "state", None) + if state is None: + return None + bag = state.get(STATE_KEY) or {} + entry = bag.get(_SESSION_SUBKEY) + return entry if isinstance(entry, dict) else None + + +def _write_state(agent: Any, entry: Optional[Dict[str, Any]]) -> None: + state = getattr(agent, "state", None) + if state is None: + return + bag = dict(state.get(STATE_KEY) or {}) + if entry is None: + bag.pop(_SESSION_SUBKEY, None) + else: + bag[_SESSION_SUBKEY] = entry + try: + state.set(STATE_KEY, bag) + except ValueError: + logger.warning("browser: session state not serializable; not persisted") + + +async def _start_remote_session() -> Tuple[Any, str, str]: + """Start an AgentCore browser session. Returns (client, identifier, id).""" + from bedrock_agentcore.tools.browser_client import BrowserClient + + identifier = _browser_identifier() + client = BrowserClient(region=_region()) + # boto3 is synchronous; keep it off the event loop. + session_id = await asyncio.to_thread( + client.start, + identifier=identifier, + session_timeout_seconds=SESSION_TIMEOUT_SECONDS, + viewport=DEFAULT_VIEWPORT, + ) + logger.info( + "browser: started session %s on %s (ttl=%ss)", + session_id, identifier, SESSION_TIMEOUT_SECONDS, + ) + return client, identifier, session_id + + +async def _connect(client: Any) -> CdpSession: + ws_url, headers = await asyncio.to_thread(client.generate_ws_headers) + return await CdpSession.connect(ws_url, headers) + + +async def _reap_idle() -> None: + """Close sockets unused past the idle window, stopping the remote session.""" + now = time.monotonic() + stale = [ + sid for sid, live in _live.items() + if now - live.last_used > IDLE_REAP_SECONDS + ] + for sid in stale: + live = _live.pop(sid, None) + if live is None: + continue + logger.info("browser: reaping idle session %s", sid) + await _teardown(live) + + +async def _teardown(live: _LiveSession) -> None: + try: + await live.cdp.close() + except Exception: # noqa: BLE001 - teardown is best effort + logger.debug("browser: cdp close failed", exc_info=True) + try: + await asyncio.to_thread(live.client.stop) + except Exception: # noqa: BLE001 + logger.debug("browser: remote stop failed", exc_info=True) + + +async def acquire(agent: Any) -> _LiveSession: + """Return this conversation's live session, starting or reconnecting it. + + Reconnection matters: the state may name a remote session this process + has never seen (a second cached agent, or a container that restarted). + Reconnecting is strictly cheaper than starting a second browser. + """ + async with _pool_lock: + await _reap_idle() + + entry = _read_state(agent) + if entry: + session_id = entry.get("sessionId") + live = _live.get(session_id) if session_id else None + if live is not None and not live.cdp.closed: + live.last_used = time.monotonic() + return live + if session_id and entry.get("identifier"): + reconnected = await _try_reconnect(entry) + if reconnected is not None: + return reconnected + + client, identifier, session_id = await _start_remote_session() + try: + cdp = await _connect(client) + except Exception: + await asyncio.to_thread(client.stop) + raise + + live = _LiveSession( + session_id=session_id, identifier=identifier, client=client, cdp=cdp + ) + _live[session_id] = live + _write_state( + agent, + { + "sessionId": session_id, + "identifier": identifier, + "startedAt": time.time(), + }, + ) + return live + + +async def _try_reconnect(entry: Dict[str, Any]) -> Optional[_LiveSession]: + """Re-attach to a remote session named in state. None if it's gone.""" + from bedrock_agentcore.tools.browser_client import BrowserClient + + session_id = entry["sessionId"] + identifier = entry["identifier"] + client = BrowserClient(region=_region()) + client.identifier = identifier + client.session_id = session_id + try: + cdp = await _connect(client) + except Exception as exc: # noqa: BLE001 - expired/stopped session + logger.info("browser: cannot reconnect to %s (%s)", session_id, exc) + return None + + logger.info("browser: reconnected to session %s", session_id) + live = _LiveSession( + session_id=session_id, identifier=identifier, client=client, cdp=cdp + ) + _live[session_id] = live + return live + + +async def release(agent: Any) -> bool: + """Stop this conversation's browser session. Idempotent.""" + entry = _read_state(agent) + _write_state(agent, None) + if not entry: + return False + session_id = entry.get("sessionId") + live = _live.pop(session_id, None) if session_id else None + if live is not None: + await _teardown(live) + return True + if session_id and entry.get("identifier"): + # Not ours to close locally, but still billing remotely. + from bedrock_agentcore.tools.browser_client import BrowserClient + + client = BrowserClient(region=_region()) + client.identifier = entry["identifier"] + client.session_id = session_id + try: + await asyncio.to_thread(client.stop) + return True + except Exception: # noqa: BLE001 + logger.debug("browser: remote stop failed", exc_info=True) + return False + + +async def live_view_url(agent: Any) -> Optional[str]: + """Pre-signed live-view URL for the running session, if there is one.""" + entry = _read_state(agent) + if not entry: + return None + live = _live.get(entry.get("sessionId", "")) + client = live.client if live else None + if client is None: + return None + try: + return await asyncio.to_thread(client.generate_live_view_url) + except Exception: # noqa: BLE001 + logger.debug("browser: live view url failed", exc_info=True) + return None diff --git a/backend/src/agents/main_agent/tools/tool_catalog.py b/backend/src/agents/main_agent/tools/tool_catalog.py index 587a1387..a7ab5617 100644 --- a/backend/src/agents/main_agent/tools/tool_catalog.py +++ b/backend/src/agents/main_agent/tools/tool_catalog.py @@ -75,6 +75,15 @@ def to_dict(self) -> dict: icon="calculator", ), + # --- Built-in Tools (Browser) --- + "browse_web": ToolMetadata( + tool_id="browse_web", + name="Web Browser", + description="Browse the web in a real Chrome browser: navigate pages, read JavaScript-rendered content, fill forms, and click through multi-step flows.", + category=ToolCategory.SEARCH, + icon="globe-alt", + ), + # --- Built-in Tools (Code Interpreter) --- "generate_diagram_and_validate": ToolMetadata( tool_id="generate_diagram_and_validate", diff --git a/backend/tests/agents/builtin_tools/browser/__init__.py b/backend/tests/agents/builtin_tools/browser/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/agents/builtin_tools/browser/test_browse_tool.py b/backend/tests/agents/builtin_tools/browser/test_browse_tool.py new file mode 100644 index 00000000..93ff8908 --- /dev/null +++ b/backend/tests/agents/builtin_tools/browser/test_browse_tool.py @@ -0,0 +1,381 @@ +"""Tests for the AgentCore Browser tool. + +No AWS: the CDP layer is exercised against a fake websocket that speaks the +protocol, and the tool layer against a fake CDP session. What these protect is +the framing (ids, sessionId routing, error surfacing) and the output budgets — +the two things a live smoke test is least likely to catch, because a happy-path +browse looks fine right up until a page returns 400KB of text. +""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +import pytest + +from agents.builtin_tools.browser import browse_tool +from agents.builtin_tools.browser.browse_tool import browse_web +from agents.builtin_tools.browser.cdp_client import CdpError, CdpSession + + +def _call(tool, **kwargs): + """Reach the underlying callable inside the Strands tool wrapper.""" + inner = getattr(tool, "_tool_func", None) or getattr(tool, "func", None) or tool + if hasattr(inner, "__wrapped__"): + inner = inner.__wrapped__ + return inner(**kwargs) + + +# -------------------------------------------------------------------------- +# CDP framing +# -------------------------------------------------------------------------- + + +class FakeWebSocket: + """Answers CDP commands from a scripted table.""" + + def __init__(self, responses: Dict[str, Any], *, page_targets: bool = True) -> None: + self._responses = responses + self._page_targets = page_targets + self.sent: List[dict] = [] + self._outbox: asyncio.Queue = asyncio.Queue() + self.closed = False + + async def send(self, raw: str) -> None: + message = json.loads(raw) + self.sent.append(message) + method = message["method"] + if method == "Target.getTargets": + infos = [{"targetId": "T1", "type": "page"}] if self._page_targets else [] + result = {"targetInfos": infos} + elif method == "Target.attachToTarget": + result = {"sessionId": "S1"} + elif method in self._responses: + entry = self._responses[method] + if isinstance(entry, Exception): + await self._outbox.put( + {"id": message["id"], "error": {"message": str(entry)}} + ) + return + result = entry + else: + result = {} + await self._outbox.put({"id": message["id"], "result": result}) + + def __aiter__(self): + return self + + async def __anext__(self): + item = await self._outbox.get() + return json.dumps(item) + + async def close(self) -> None: + self.closed = True + + +async def _session(ws: FakeWebSocket) -> CdpSession: + session = CdpSession(ws) + session._reader = asyncio.create_task(session._read_loop()) + await session._attach_to_page() + return session + + +@pytest.mark.asyncio +async def test_attaches_to_page_and_scopes_later_commands() -> None: + ws = FakeWebSocket({"Runtime.evaluate": {"result": {"value": "hello"}}}) + session = await _session(ws) + + value = await session.evaluate("1+1") + + assert value == "hello" + target_cmds = [m for m in ws.sent if m["method"].startswith("Target.")] + assert [m["method"] for m in target_cmds] == [ + "Target.getTargets", + "Target.attachToTarget", + ] + # Target.* must be browser-scoped, page commands session-scoped. + assert all("sessionId" not in m for m in target_cmds) + evaluate = [m for m in ws.sent if m["method"] == "Runtime.evaluate"][0] + assert evaluate["sessionId"] == "S1" + await session.close() + + +@pytest.mark.asyncio +async def test_creates_a_target_when_none_exists() -> None: + ws = FakeWebSocket({"Target.createTarget": {"targetId": "NEW"}}, page_targets=False) + session = await _session(ws) + + assert any(m["method"] == "Target.createTarget" for m in ws.sent) + await session.close() + + +@pytest.mark.asyncio +async def test_page_exception_surfaces_as_cdp_error() -> None: + ws = FakeWebSocket( + { + "Runtime.evaluate": { + "exceptionDetails": { + "exception": {"description": "ReferenceError: nope is not defined"} + } + } + } + ) + session = await _session(ws) + + with pytest.raises(CdpError, match="ReferenceError"): + await session.evaluate("nope()") + await session.close() + + +@pytest.mark.asyncio +async def test_protocol_error_surfaces_with_method_name() -> None: + ws = FakeWebSocket({"Page.navigate": RuntimeError("Cannot navigate to invalid URL")}) + session = await _session(ws) + + with pytest.raises(CdpError, match="Page.navigate failed"): + await session.command("Page.navigate", {"url": "::"}) + await session.close() + + +@pytest.mark.asyncio +async def test_close_is_idempotent_and_fails_pending_commands() -> None: + ws = FakeWebSocket({}) + session = await _session(ws) + + await session.close() + await session.close() # must not raise + + assert session.closed + with pytest.raises(CdpError, match="closed"): + await session.command("Runtime.evaluate") + + +# -------------------------------------------------------------------------- +# URL validation +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "url", + [ + "file:///etc/passwd", + "http://169.254.169.254/latest/meta-data/", + "http://localhost:8000/admin", + "http://127.0.0.1/", + "ftp://example.com/x", + ], +) +def test_unsafe_urls_are_refused(url: str) -> None: + assert browse_tool._validate_url(url) is not None + + +@pytest.mark.parametrize("url", ["https://example.com", "http://example.com/a?b=c"]) +def test_public_urls_are_allowed(url: str) -> None: + assert browse_tool._validate_url(url) is None + + +# -------------------------------------------------------------------------- +# Tool dispatch and output budgets +# -------------------------------------------------------------------------- + + +@dataclass +class FakeCdp: + values: Dict[str, Any] = field(default_factory=dict) + navigated: List[str] = field(default_factory=list) + closed: bool = False + screenshot_data: str = "" + + async def navigate(self, url: str, **_: Any) -> None: + self.navigated.append(url) + + async def evaluate(self, expression: str, **_: Any) -> Any: + for needle, value in self.values.items(): + if needle in expression: + return value + return None + + async def screenshot(self) -> str: + return self.screenshot_data + + +@dataclass +class FakeLive: + cdp: FakeCdp + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + + +class FakeState: + def __init__(self) -> None: + self._data: Dict[str, Any] = {} + + def get(self, key: str) -> Any: + return json.loads(json.dumps(self._data.get(key))) if key in self._data else None + + def set(self, key: str, value: Any) -> None: + self._data[key] = json.loads(json.dumps(value)) + + +class FakeAgent: + def __init__(self) -> None: + self.state = FakeState() + + +class FakeContext: + def __init__(self, agent: FakeAgent) -> None: + self.agent = agent + + +@pytest.fixture +def patched_pool(monkeypatch): + """Replace the session pool so no AWS call is attempted.""" + cdp = FakeCdp() + live = FakeLive(cdp=cdp) + + async def _acquire(_agent): + return live + + monkeypatch.setattr(browse_tool.session_pool, "acquire", _acquire) + return live + + +@pytest.mark.asyncio +async def test_navigate_returns_title_and_text(patched_pool) -> None: + patched_pool.cdp.values = { + "document.title": "Example Domain", + "location.href": "https://example.com/", + "querySelector('main')": "Body copy here.", + } + + result = await _call( + browse_web, + action="navigate", + url="https://example.com", + tool_context=FakeContext(FakeAgent()), + ) + + assert result["status"] == "success" + text = result["content"][0]["text"] + assert "Example Domain" in text + assert "Body copy here." in text + assert patched_pool.cdp.navigated == ["https://example.com"] + + +@pytest.mark.asyncio +async def test_navigate_refuses_metadata_url_without_touching_the_pool(monkeypatch) -> None: + called = False + + async def _acquire(_agent): + nonlocal called + called = True + raise AssertionError("pool must not be touched") + + monkeypatch.setattr(browse_tool.session_pool, "acquire", _acquire) + + result = await _call( + browse_web, + action="navigate", + url="http://169.254.169.254/latest/meta-data/", + tool_context=FakeContext(FakeAgent()), + ) + + assert result["status"] == "error" + assert not called + + +@pytest.mark.asyncio +async def test_page_text_is_truncated_to_the_budget(patched_pool, monkeypatch) -> None: + monkeypatch.setattr(browse_tool, "MAX_TEXT_CHARS", 100) + patched_pool.cdp.values = {"querySelector('main')": "x" * 5000} + + result = await _call( + browse_web, action="extract_text", tool_context=FakeContext(FakeAgent()) + ) + + text = result["content"][0]["text"] + assert len(text) < 400 + assert "truncated" in text + + +@pytest.mark.asyncio +async def test_click_reports_a_missing_selector_as_error(patched_pool) -> None: + patched_pool.cdp.values = {"querySelector": False} + + result = await _call( + browse_web, + action="click", + selector="#nope", + tool_context=FakeContext(FakeAgent()), + ) + + assert result["status"] == "error" + assert "#nope" in result["content"][0]["text"] + + +@pytest.mark.asyncio +async def test_type_requires_selector_and_text(patched_pool) -> None: + result = await _call( + browse_web, action="type", selector="#q", tool_context=FakeContext(FakeAgent()) + ) + assert result["status"] == "error" + + +@pytest.mark.asyncio +async def test_evaluate_renders_structured_results(patched_pool) -> None: + patched_pool.cdp.values = {"headings": ["One", "Two"]} + + result = await _call( + browse_web, + action="evaluate", + script="headings", + tool_context=FakeContext(FakeAgent()), + ) + + assert result["status"] == "success" + assert "One" in result["content"][0]["text"] + + +@pytest.mark.asyncio +async def test_screenshot_returns_image_bytes(patched_pool) -> None: + import base64 + + patched_pool.cdp.screenshot_data = base64.b64encode(b"PNGDATA").decode() + + result = await _call( + browse_web, action="screenshot", tool_context=FakeContext(FakeAgent()) + ) + + assert result["content"][1]["image"]["source"]["bytes"] == b"PNGDATA" + + +@pytest.mark.asyncio +async def test_unknown_action_lists_valid_actions(patched_pool) -> None: + result = await _call( + browse_web, action="teleport", tool_context=FakeContext(FakeAgent()) + ) + + assert result["status"] == "error" + assert "navigate" in result["content"][0]["text"] + + +@pytest.mark.asyncio +async def test_kill_switch_short_circuits(monkeypatch) -> None: + monkeypatch.setenv("BROWSER_TOOL_ENABLED", "false") + + async def _acquire(_agent): + raise AssertionError("pool must not be touched when disabled") + + monkeypatch.setattr(browse_tool.session_pool, "acquire", _acquire) + + result = await _call( + browse_web, + action="navigate", + url="https://example.com", + tool_context=FakeContext(FakeAgent()), + ) + + assert result["status"] == "error" + assert "disabled" in result["content"][0]["text"] diff --git a/backend/tests/test_seed_system_admin_jwt.py b/backend/tests/test_seed_system_admin_jwt.py index 41f70939..c40d928b 100644 --- a/backend/tests/test_seed_system_admin_jwt.py +++ b/backend/tests/test_seed_system_admin_jwt.py @@ -13,6 +13,7 @@ ) from seed_bootstrap_data import ( # noqa: E402 + DEFAULT_TOOLS, EXAMPLE_SKILL_ID, seed_default_role, seed_example_skills, @@ -131,7 +132,7 @@ def test_creates_default_tools(self, dynamodb_table): """Creates the default tool entries.""" result = seed_default_tools(TABLE_NAME, REGION) - assert result.created == 9 + assert result.created == len(DEFAULT_TOOLS) assert result.failed == 0 # Verify fetch_url_content @@ -230,6 +231,21 @@ def test_creates_default_tools(self, dynamodb_table): assert item["GSI1PK"] == "CATEGORY#document" assert item["GSI1SK"] == "TOOL#workspace_files" + # Verify browse_web. enabledByDefault MUST stay False: each session + # bills an AgentCore Browser session on top of model tokens, so this + # is opt-in per user and granted per role. + resp = dynamodb_table.get_item( + Key={"PK": "TOOL#browse_web", "SK": "METADATA"} + ) + item = resp["Item"] + assert item["toolId"] == "browse_web" + assert item["displayName"] == "Web Browser" + assert item["category"] == "browser" + assert item["protocol"] == "local" + assert item["enabledByDefault"] is False + assert item["GSI1PK"] == "CATEGORY#browser" + assert item["GSI1SK"] == "TOOL#browse_web" + # Verify create_excel_spreadsheet (single toggle for the whole Excel toolset) resp = dynamodb_table.get_item( Key={"PK": "TOOL#create_excel_spreadsheet", "SK": "METADATA"} @@ -264,7 +280,7 @@ def test_skips_existing_tools(self, dynamodb_table): result = seed_default_tools(TABLE_NAME, REGION) - assert result.skipped == 9 + assert result.skipped == len(DEFAULT_TOOLS) assert result.created == 0 def test_partial_skip(self, dynamodb_table): @@ -278,7 +294,7 @@ def test_partial_skip(self, dynamodb_table): result = seed_default_tools(TABLE_NAME, REGION) - assert result.created == 8 + assert result.created == len(DEFAULT_TOOLS) - 1 assert result.skipped == 1