A live window into Decagon Browser Actions. A small Python helper inside each Browser Action streams the hidden browser to this app. By default it records the page itself (rrweb), so the viewer shows a smooth live page with a real cursor. If a site blocks that, the run automatically switches to screenshots after each action.
Pick your space (usually your first name), then copy this into Duet and ask it to add the viewer to your Browser Actions.
# Browser Action Viewer integration (helper v1.1.0)
Goal: every Browser Action tool on this team streams what it does to the Browser Action Viewer at https://view.decagon.tools, so the demo audience can watch it live at **https://view.decagon.tools/s/selftest**. Monitoring must never change what the tool does or make it fail.
## What to set up (once per team)
1. Add the helper below to the tools repo as `lib/bav.py` (or the repo's equivalent shared-library location). Copy it exactly; do not rewrite it.
2. Team secret (optional): `bav_space` = `selftest`. Use it as the space so each SE's demos stay separate. If secrets are awkward, the literal string `"selftest"` is fine.
3. Team secret (optional): `bav_ingest_key`, only if the viewer owner says ingest is locked. Pass it as `ingest_key=`.
## How every Browser Action uses it
- Keep `__BROWSER_TOOL__ = True` and the injected Playwright `page`. Do not create or close pages/browsers.
- Wrap the tool body in `with BavRun(page, tool="<tool_name>", space=<space>, conversation_id=<conversation id>) as run:`.
Get the conversation id from the repo's Conversation API (often `Conversation.get_id()`). It is the channel key: it must be the real Decagon conversation id so every Browser Action in one conversation lands on the same viewer page.
- Use the run's wrappers for each visible action so the viewer captures it: `run.goto(url, label)`, `run.click(selector, label)`, `run.fill(selector, value, label, secret=False)`, `run.type(...)` (character by character, looks more natural), `run.select(selector, value, label)`, `run.check(selector, label)`.
- After any custom Playwright code (waits, reads, multi-step widgets), call `run.step("Short human label")`.
- Pass `secret=True` for passwords and sensitive values. Inputs are masked in replay by default.
- Labels are what the audience reads in the timeline: short, present tense, business language ("Updating employer", not "click #emp-save").
- Keep the tool's own error handling: return the tool's result dict with an `error` field rather than raising. `BavRun` never swallows exceptions; it just marks the run as an error in the viewer.
- Optional: return `run.viewer_url` in the tool result if the agent should share the link.
## Modes (automatic)
- **Live replay (default):** the helper injects a pinned rrweb recorder into every page and sends DOM changes, so the viewer shows a smooth live page with a real cursor.
- **Screenshot mode (fallback):** if the recorder cannot run on the site, the whole run switches to JPEG screenshots after each action, with the cursor drawn from the element's position. Force it with `force_screenshots=True` for canvas-heavy or heavily locked-down sites.
- Optional flags: `mask_inputs=False` to show typed values in replay (demo data only), `inline_images=True` if images on the target site are behind a login and appear broken in replay.
## Example
```python
__BROWSER_TOOL__ = True
from lib.bav import BavRun
def update_employment(employer: str, income: str):
space = Secrets.get("bav_space") or "selftest" # use the repo's secret accessor
try:
with BavRun(page, tool="update_employment", space=space,
conversation_id=Conversation.get_id()) as run:
run.goto("https://example.com/applications/123", "Opening the application")
run.fill("#employer", employer, "Entering employer")
run.fill("#income", income, "Entering annual income")
run.click("#save", "Saving employment")
page.wait_for_selector(".toast-success", timeout=10000)
run.step("Employment saved")
return {"saved": True, "error": None}
except Exception as e:
return {"saved": False, "error": str(e)}
```
Adapt the secret and Conversation accessors to whatever this repo already uses.
## Testing
Open https://view.decagon.tools/s/selftest in a browser, then run the tool (preview chat or a real call). The viewer follows the newest conversation in the space; click **Pin** to stay on one. If nothing appears: check the space string matches, check the tool can reach https://view.decagon.tools, and look for a "Screenshot mode: …" step in the timeline, which explains why replay was unavailable.
## Helper: lib/bav.py (copy exactly)
```python
"""
Decagon Browser Action Viewer helper (v1.1.0).
Streams what a Browser Action does to https://view.decagon.tools so anyone can watch it live.
Replay mode (rrweb) by default; falls back to screenshots for the whole run when the
recorder can't run on the target site. Monitoring never raises and never fails the tool.
Usage inside a Browser Action (Decagon injects the Playwright 'page'):
from lib.bav import BavRun
with BavRun(page, tool="update_employment", space="david",
conversation_id=Conversation.get_id()) as run:
run.goto("https://example.com/login", "Opening the login page")
run.fill("#username", user, "Entering username")
run.fill("#password", password, "Entering password", secret=True)
run.click("button[type=submit]", "Signing in")
run.step("Signed in")
"""
import base64
import json
import time
import uuid
from concurrent.futures import ThreadPoolExecutor, wait
import requests
BAV_BASE_URL = "https://view.decagon.tools"
BAV_HELPER_VERSION = "1.1.0"
_TIMEOUT_S = 5
_FLUSH_S = 6
_MAX_FRAME_BYTES = 350 * 1024
_MAX_BATCH_CHARS = 900_000
_recorder_cache = {}
class BavRun:
def __init__(self, page, tool, space, conversation_id=None, base_url=None,
ingest_key=None, force_screenshots=False, mask_inputs=True,
inline_images=False, enabled=True):
self.page = page
self.tool = str(tool)
self.space = str(space or "default")
self.conversation_id = str(conversation_id or "local-" + uuid.uuid4().hex[:8])
self.base_url = (base_url or BAV_BASE_URL).rstrip("/")
self.ingest_key = ingest_key
self.enabled = bool(enabled)
self.mode = "screenshot" if force_screenshots else "replay"
self._verified = bool(force_screenshots)
self._options = {"maskAllInputs": bool(mask_inputs), "inlineImages": bool(inline_images)}
self.run_id = "%s-%s" % (self.tool[:40], uuid.uuid4().hex[:10])
self._seq = 0
self._pool = ThreadPoolExecutor(max_workers=1)
self._http = requests.Session() # keep-alive: no new TLS handshake per event
self._futures = []
self._closed = False
@property
def viewer_url(self):
return "%s/s/%s/c/%s" % (self.base_url, self.space, self.conversation_id)
# ---- lifecycle -------------------------------------------------------
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc, tb):
if exc is not None:
self.finish("error", "%s: %s" % (exc_type.__name__, exc))
else:
self.finish("ok")
return False # never swallow the tool's own exception
def start(self):
if not self.enabled:
return self
if self.mode == "replay":
self._install_recorder()
self._send({"type": "start", "mode": self.mode, "label": "Started " + self.tool})
return self
def finish(self, status="ok", summary=None):
if self._closed or not self.enabled:
return
self._closed = True
try:
self._capture(label=None)
except Exception:
pass
payload = {"type": "end", "status": "error" if status == "error" else "ok"}
if summary:
payload["summary"] = str(summary)[:300]
self._send(payload)
self.flush()
def flush(self):
try:
wait(self._futures, timeout=_FLUSH_S)
except Exception:
pass
try:
self._pool.shutdown(wait=False, cancel_futures=True)
except TypeError:
self._pool.shutdown(wait=False)
except Exception:
pass
try:
self._http.close()
except Exception:
pass
# ---- actions (perform, then capture) ---------------------------------
def goto(self, url, label=None, wait_until="domcontentloaded", **kwargs):
result = self.page.goto(url, wait_until=wait_until, **kwargs)
self.checkpoint(label or "Opened " + str(url))
return result
def click(self, selector, label=None, **kwargs):
loc = self.page.locator(selector).first
loc.click(**kwargs)
box = self._box_after(loc)
self.checkpoint(label or "Clicked " + selector, action="click", box=box)
def fill(self, selector, value, label=None, secret=False, **kwargs):
loc = self.page.locator(selector).first
loc.fill(str(value), **kwargs)
box = self._box_after(loc)
detail = None if secret else {"value": str(value)[:200]}
self.checkpoint(label or "Typed into " + selector, action="type", box=box, detail=detail)
def type(self, selector, value, label=None, secret=False, delay=40, **kwargs):
loc = self.page.locator(selector).first
if hasattr(loc, "press_sequentially"):
loc.press_sequentially(str(value), delay=delay, **kwargs)
else:
loc.type(str(value), delay=delay, **kwargs)
box = self._box_after(loc)
detail = None if secret else {"value": str(value)[:200]}
self.checkpoint(label or "Typed into " + selector, action="type", box=box, detail=detail)
def select(self, selector, value, label=None, **kwargs):
loc = self.page.locator(selector).first
loc.select_option(value, **kwargs)
box = self._box_after(loc)
self.checkpoint(label or "Selected " + str(value), action="select", box=box,
detail={"value": str(value)[:200]})
def check(self, selector, label=None, **kwargs):
loc = self.page.locator(selector).first
loc.check(**kwargs)
box = self._box_after(loc)
self.checkpoint(label or "Checked " + selector, action="click", box=box)
def step(self, label, status="ok", detail=None):
"""Timeline entry plus a capture. Use after any custom Playwright work."""
self.checkpoint(label, status=status, detail=detail)
def checkpoint(self, label=None, status="ok", action=None, box=None, detail=None):
if not self.enabled or self._closed:
return
try:
step = None
if label:
step = {"label": str(label)[:300], "status": status}
if detail:
step["detail"] = detail
# The step rides along with the capture: one request per action, and the viewer gets
# the screenshot and its label at the same moment.
if not self._capture(label=label, action=action, box=box, step=step) and step:
step["type"] = "step"
self._send(step)
except Exception:
pass
# ---- internals -------------------------------------------------------
def _install_recorder(self):
try:
script = _recorder_cache.get(self.base_url)
if script is None:
res = requests.get(self.base_url + "/api/public/v1/recorder", timeout=_TIMEOUT_S)
res.raise_for_status()
script = res.text
_recorder_cache[self.base_url] = script
full = "(function () { window.__bavOptions = %s;\n%s\n})()" % (
json.dumps(self._options), script)
self.page.add_init_script(script=full)
if str(self.page.url).startswith("http"):
self.page.evaluate(full)
except Exception as e:
self._fallback("Recorder could not load (%s)" % type(e).__name__)
def _fallback(self, reason):
if self.mode != "screenshot":
self.mode = "screenshot"
self._verified = True
self._send({"type": "step", "mode": "screenshot", "status": "ok",
"label": "Screenshot mode: " + reason})
def _verify_replay(self):
if self._verified or not str(self.page.url).startswith("http"):
return
deadline = time.time() + 1.0
state = None
while time.time() < deadline:
try:
state = self.page.evaluate(
"() => window.__bav ? { ok: window.__bav.ok, error: window.__bav.error } : null")
except Exception:
state = None
if state and state.get("ok"):
self._verified = True
return
time.sleep(0.15)
reason = (state or {}).get("error") or "recorder blocked on this site"
self._fallback(str(reason)[:120])
def _capture(self, label=None, action=None, box=None, step=None):
"""Send what the page looks like now. Returns True if something was sent (with step)."""
if self.mode == "replay":
self._verify_replay()
if self.mode == "replay":
events = self.page.evaluate("() => window.__bav ? window.__bav.drain() : []") or []
batches = list(_batches(events))
for i, batch in enumerate(batches):
last = i == len(batches) - 1
self._send({"type": "rrweb", "events": batch, "step": step if last else None})
return bool(batches)
image = self._screenshot()
if image is None:
return False
frame = {"type": "frame", "image": base64.b64encode(image).decode("ascii"),
"mime": "image/jpeg", "label": label, "action": action, "box": box,
"viewport": self.page.viewport_size, "step": step}
self._send(frame)
return True
def _screenshot(self):
for quality in (60, 40, 25):
try:
image = self.page.screenshot(type="jpeg", quality=quality, scale="css")
except Exception:
return None
if len(image) <= _MAX_FRAME_BYTES:
return image
return None
def _box_after(self, loc):
# Screenshots are taken after the action, so measure again. If the element is gone
# (the click navigated), send no box rather than one from the previous page.
try:
if loc.count() == 0:
return None
return loc.bounding_box(timeout=300)
except Exception:
return None
def _send(self, payload):
if not self.enabled:
return
self._seq += 1
body = {k: v for k, v in payload.items() if v is not None}
body.update({"v": 1, "space": self.space, "conversation_id": self.conversation_id,
"run_id": self.run_id, "tool": self.tool, "seq": self._seq})
headers = {"content-type": "application/json"}
if self.ingest_key:
headers["x-bav-key"] = self.ingest_key
url = self.base_url + "/api/public/v1/ingest"
try:
self._futures.append(self._pool.submit(_post, self._http, url, body, headers))
except Exception:
pass
def _post(http, url, body, headers):
try:
http.post(url, data=json.dumps(body), headers=headers, timeout=_TIMEOUT_S)
except Exception:
pass
def _batches(events):
batch, size = [], 0
for event in events:
n = len(json.dumps(event))
if batch and size + n > _MAX_BATCH_CHARS:
yield batch
batch, size = [], 0
batch.append(event)
size += n
if batch:
yield batch
```
https://view.decagon.tools/s/selftest before the call.Add ?embed=1 to a space or conversation URL and the page renders the stage alone: no header, no sidebar, just the live browser with a status chip and the latest step. Drop it in an <iframe> in your demo app; open the same URL without embed for the full view.
<iframe src="https://view.decagon.tools/s/selftest/c/<conversation_id>?embed=1" style="width:100%;aspect-ratio:16/10;border:0"></iframe>
Pin to a conversation id you control (an application number works) so every Browser Action in that demo lands on one page; use /s/<space>?embed=1 to follow whatever ran last.
Same file as in the Duet Markdown. Also served at https://view.decagon.tools/api/public/v1/helper.
"""
Decagon Browser Action Viewer helper (v1.1.0).
Streams what a Browser Action does to https://view.decagon.tools so anyone can watch it live.
Replay mode (rrweb) by default; falls back to screenshots for the whole run when the
recorder can't run on the target site. Monitoring never raises and never fails the tool.
Usage inside a Browser Action (Decagon injects the Playwright 'page'):
from lib.bav import BavRun
with BavRun(page, tool="update_employment", space="david",
conversation_id=Conversation.get_id()) as run:
run.goto("https://example.com/login", "Opening the login page")
run.fill("#username", user, "Entering username")
run.fill("#password", password, "Entering password", secret=True)
run.click("button[type=submit]", "Signing in")
run.step("Signed in")
"""
import base64
import json
import time
import uuid
from concurrent.futures import ThreadPoolExecutor, wait
import requests
BAV_BASE_URL = "https://view.decagon.tools"
BAV_HELPER_VERSION = "1.1.0"
_TIMEOUT_S = 5
_FLUSH_S = 6
_MAX_FRAME_BYTES = 350 * 1024
_MAX_BATCH_CHARS = 900_000
_recorder_cache = {}
class BavRun:
def __init__(self, page, tool, space, conversation_id=None, base_url=None,
ingest_key=None, force_screenshots=False, mask_inputs=True,
inline_images=False, enabled=True):
self.page = page
self.tool = str(tool)
self.space = str(space or "default")
self.conversation_id = str(conversation_id or "local-" + uuid.uuid4().hex[:8])
self.base_url = (base_url or BAV_BASE_URL).rstrip("/")
self.ingest_key = ingest_key
self.enabled = bool(enabled)
self.mode = "screenshot" if force_screenshots else "replay"
self._verified = bool(force_screenshots)
self._options = {"maskAllInputs": bool(mask_inputs), "inlineImages": bool(inline_images)}
self.run_id = "%s-%s" % (self.tool[:40], uuid.uuid4().hex[:10])
self._seq = 0
self._pool = ThreadPoolExecutor(max_workers=1)
self._http = requests.Session() # keep-alive: no new TLS handshake per event
self._futures = []
self._closed = False
@property
def viewer_url(self):
return "%s/s/%s/c/%s" % (self.base_url, self.space, self.conversation_id)
# ---- lifecycle -------------------------------------------------------
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc, tb):
if exc is not None:
self.finish("error", "%s: %s" % (exc_type.__name__, exc))
else:
self.finish("ok")
return False # never swallow the tool's own exception
def start(self):
if not self.enabled:
return self
if self.mode == "replay":
self._install_recorder()
self._send({"type": "start", "mode": self.mode, "label": "Started " + self.tool})
return self
def finish(self, status="ok", summary=None):
if self._closed or not self.enabled:
return
self._closed = True
try:
self._capture(label=None)
except Exception:
pass
payload = {"type": "end", "status": "error" if status == "error" else "ok"}
if summary:
payload["summary"] = str(summary)[:300]
self._send(payload)
self.flush()
def flush(self):
try:
wait(self._futures, timeout=_FLUSH_S)
except Exception:
pass
try:
self._pool.shutdown(wait=False, cancel_futures=True)
except TypeError:
self._pool.shutdown(wait=False)
except Exception:
pass
try:
self._http.close()
except Exception:
pass
# ---- actions (perform, then capture) ---------------------------------
def goto(self, url, label=None, wait_until="domcontentloaded", **kwargs):
result = self.page.goto(url, wait_until=wait_until, **kwargs)
self.checkpoint(label or "Opened " + str(url))
return result
def click(self, selector, label=None, **kwargs):
loc = self.page.locator(selector).first
loc.click(**kwargs)
box = self._box_after(loc)
self.checkpoint(label or "Clicked " + selector, action="click", box=box)
def fill(self, selector, value, label=None, secret=False, **kwargs):
loc = self.page.locator(selector).first
loc.fill(str(value), **kwargs)
box = self._box_after(loc)
detail = None if secret else {"value": str(value)[:200]}
self.checkpoint(label or "Typed into " + selector, action="type", box=box, detail=detail)
def type(self, selector, value, label=None, secret=False, delay=40, **kwargs):
loc = self.page.locator(selector).first
if hasattr(loc, "press_sequentially"):
loc.press_sequentially(str(value), delay=delay, **kwargs)
else:
loc.type(str(value), delay=delay, **kwargs)
box = self._box_after(loc)
detail = None if secret else {"value": str(value)[:200]}
self.checkpoint(label or "Typed into " + selector, action="type", box=box, detail=detail)
def select(self, selector, value, label=None, **kwargs):
loc = self.page.locator(selector).first
loc.select_option(value, **kwargs)
box = self._box_after(loc)
self.checkpoint(label or "Selected " + str(value), action="select", box=box,
detail={"value": str(value)[:200]})
def check(self, selector, label=None, **kwargs):
loc = self.page.locator(selector).first
loc.check(**kwargs)
box = self._box_after(loc)
self.checkpoint(label or "Checked " + selector, action="click", box=box)
def step(self, label, status="ok", detail=None):
"""Timeline entry plus a capture. Use after any custom Playwright work."""
self.checkpoint(label, status=status, detail=detail)
def checkpoint(self, label=None, status="ok", action=None, box=None, detail=None):
if not self.enabled or self._closed:
return
try:
step = None
if label:
step = {"label": str(label)[:300], "status": status}
if detail:
step["detail"] = detail
# The step rides along with the capture: one request per action, and the viewer gets
# the screenshot and its label at the same moment.
if not self._capture(label=label, action=action, box=box, step=step) and step:
step["type"] = "step"
self._send(step)
except Exception:
pass
# ---- internals -------------------------------------------------------
def _install_recorder(self):
try:
script = _recorder_cache.get(self.base_url)
if script is None:
res = requests.get(self.base_url + "/api/public/v1/recorder", timeout=_TIMEOUT_S)
res.raise_for_status()
script = res.text
_recorder_cache[self.base_url] = script
full = "(function () { window.__bavOptions = %s;\n%s\n})()" % (
json.dumps(self._options), script)
self.page.add_init_script(script=full)
if str(self.page.url).startswith("http"):
self.page.evaluate(full)
except Exception as e:
self._fallback("Recorder could not load (%s)" % type(e).__name__)
def _fallback(self, reason):
if self.mode != "screenshot":
self.mode = "screenshot"
self._verified = True
self._send({"type": "step", "mode": "screenshot", "status": "ok",
"label": "Screenshot mode: " + reason})
def _verify_replay(self):
if self._verified or not str(self.page.url).startswith("http"):
return
deadline = time.time() + 1.0
state = None
while time.time() < deadline:
try:
state = self.page.evaluate(
"() => window.__bav ? { ok: window.__bav.ok, error: window.__bav.error } : null")
except Exception:
state = None
if state and state.get("ok"):
self._verified = True
return
time.sleep(0.15)
reason = (state or {}).get("error") or "recorder blocked on this site"
self._fallback(str(reason)[:120])
def _capture(self, label=None, action=None, box=None, step=None):
"""Send what the page looks like now. Returns True if something was sent (with step)."""
if self.mode == "replay":
self._verify_replay()
if self.mode == "replay":
events = self.page.evaluate("() => window.__bav ? window.__bav.drain() : []") or []
batches = list(_batches(events))
for i, batch in enumerate(batches):
last = i == len(batches) - 1
self._send({"type": "rrweb", "events": batch, "step": step if last else None})
return bool(batches)
image = self._screenshot()
if image is None:
return False
frame = {"type": "frame", "image": base64.b64encode(image).decode("ascii"),
"mime": "image/jpeg", "label": label, "action": action, "box": box,
"viewport": self.page.viewport_size, "step": step}
self._send(frame)
return True
def _screenshot(self):
for quality in (60, 40, 25):
try:
image = self.page.screenshot(type="jpeg", quality=quality, scale="css")
except Exception:
return None
if len(image) <= _MAX_FRAME_BYTES:
return image
return None
def _box_after(self, loc):
# Screenshots are taken after the action, so measure again. If the element is gone
# (the click navigated), send no box rather than one from the previous page.
try:
if loc.count() == 0:
return None
return loc.bounding_box(timeout=300)
except Exception:
return None
def _send(self, payload):
if not self.enabled:
return
self._seq += 1
body = {k: v for k, v in payload.items() if v is not None}
body.update({"v": 1, "space": self.space, "conversation_id": self.conversation_id,
"run_id": self.run_id, "tool": self.tool, "seq": self._seq})
headers = {"content-type": "application/json"}
if self.ingest_key:
headers["x-bav-key"] = self.ingest_key
url = self.base_url + "/api/public/v1/ingest"
try:
self._futures.append(self._pool.submit(_post, self._http, url, body, headers))
except Exception:
pass
def _post(http, url, body, headers):
try:
http.post(url, data=json.dumps(body), headers=headers, timeout=_TIMEOUT_S)
except Exception:
pass
def _batches(events):
batch, size = [], 0
for event in events:
n = len(json.dumps(event))
if batch and size + n > _MAX_BATCH_CHARS:
yield batch
batch, size = [], 0
batch.append(event)
size += n
if batch:
yield batch
The helper is the supported client. For anything custom, POST JSON to /api/public/v1/ingest with space, conversation_id, run_id, tool, seq and type one of start, rrweb (events), frame (image base64 JPEG, optional box, viewport, action), step (label, status) or end. Add mode to switch a run between replay and screenshot. Header x-bav-key is required only if the owner set BAV_INGEST_KEY. Data is deleted after 24 hours.
inline_images=True or force_screenshots=True.?delay=2500 to the viewer URL for a larger smoothing buffer.