Add one-step OpenBrowser control handoff

Author: federicodeponte

ax_browser_broker/api.py

@@ -179,6 +179,11 @@ class OpenBrowserOpenRequest(BaseModel):
     identity_id: str | None = None
     ttl_seconds: int = Field(default=300, ge=60, le=14400)
     wait_until: str = "domcontentloaded"
+    verify: bool = True
+    screenshot: bool = False
+    control: bool = False
+    control_owner: str | None = None
+    control_ttl_seconds: int = Field(default=900, ge=60, le=3600)
 
 
 class OpenBrowserAuthBatchRequest(BaseModel):
@@ -1930,7 +1935,37 @@ async def openbrowser_open(request: OpenBrowserOpenRequest, _auth: str = Depends
         if isinstance(error, HTTPException):
             raise
         raise _http_error(error) from error
-    return {"lease": lease_obj, "navigation": navigation}
+    result: dict[str, Any] = {"lease": lease_obj, "navigation": navigation}
+    lease_id = str(lease_obj["lease_id"])
+    if request.verify:
+        try:
+            snapshot = await browser_snapshot(LeaseIdRequest(lease_id=lease_id))
+            result["snapshot"] = {
+                "title": snapshot.get("title"),
+                "url": snapshot.get("url"),
+                "bodyText": str(snapshot.get("bodyText") or "")[:1200],
+                "element_count": len(snapshot.get("elements") or []),
+                "slot": snapshot.get("slot"),
+            }
+        except Exception as error:
+            result["verification_error"] = str(error)
+    if request.screenshot:
+        try:
+            shot = await browser_screenshot(ScreenshotRequest(lease_id=lease_id, full_page=False))
+            result["screenshot"] = {key: value for key, value in shot.items() if key != "base64"}
+        except Exception as error:
+            result["screenshot_error"] = str(error)
+    if request.control:
+        control = await lease_control_request(
+            LeaseControlRequest(
+                lease_id=lease_id,
+                owner=request.control_owner or request.owner,
+                ttl_seconds=request.control_ttl_seconds,
+            )
+        )
+        result["control"] = control
+        result["portal_url"] = control.get("portal_url")
+    return result
 
 
 @app.post("/openbrowser/v1/browser/navigate")

ax_browser_broker/cli.py

@@ -65,6 +65,20 @@ def _print(data: dict[str, Any]) -> int:
     return 0
 
 
+def _compact_snapshot(snapshot: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "title": snapshot.get("title"),
+        "url": snapshot.get("url"),
+        "bodyText": str(snapshot.get("bodyText") or "")[:1200],
+        "element_count": len(snapshot.get("elements") or []),
+        "slot": snapshot.get("slot"),
+    }
+
+
+def _compact_screenshot(screenshot: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in screenshot.items() if key != "base64"}
+
+
 def cmd_status(_args: argparse.Namespace) -> int:
     return _print(_request("GET", "/status"))
 
@@ -75,19 +89,43 @@ def cmd_docs(args: argparse.Namespace) -> int:
 
 
 def cmd_open(args: argparse.Namespace) -> int:
-    return _print(
-        _request(
+    result = _request(
+        "POST",
+        "/openbrowser/v1/open",
+        {
+            "owner": args.owner,
+            "identity_id": args.identity,
+            "url": args.url,
+            "ttl_seconds": args.ttl,
+        },
+        auth=True,
+    )
+    lease_id = str((result.get("lease") or {}).get("lease_id") or "")
+    if lease_id and not args.no_verify:
+        snapshot = _request("POST", "/openbrowser/v1/browser/snapshot", {"lease_id": lease_id}, auth=True)
+        result["snapshot"] = _compact_snapshot(snapshot)
+    if lease_id and args.screenshot:
+        screenshot = _request(
+            "POST",
+            "/openbrowser/v1/browser/screenshot",
+            {"lease_id": lease_id, "full_page": False},
+            auth=True,
+        )
+        result["screenshot"] = _compact_screenshot(screenshot)
+    if lease_id and args.control:
+        control = _request(
             "POST",
-            "/openbrowser/v1/open",
+            "/openbrowser/v1/lease-control/request",
             {
-                "owner": args.owner,
-                "identity_id": args.identity,
-                "url": args.url,
-                "ttl_seconds": args.ttl,
+                "owner": args.control_owner or args.owner,
+                "lease_id": lease_id,
+                "ttl_seconds": args.control_ttl,
             },
             auth=True,
         )
-    )
+        result["control"] = control
+        result["portal_url"] = control.get("portal_url")
+    return _print(result)
 
 
 def cmd_auth(args: argparse.Namespace) -> int:
@@ -141,6 +179,11 @@ def build_parser() -> argparse.ArgumentParser:
     open_cmd.add_argument("--identity", default=None)
     open_cmd.add_argument("--owner", default="openbrowser-cli")
     open_cmd.add_argument("--ttl", type=int, default=900)
+    open_cmd.add_argument("--control", action="store_true", help="Return a temporary human-control portal URL")
+    open_cmd.add_argument("--control-owner", default=None)
+    open_cmd.add_argument("--control-ttl", type=int, default=900)
+    open_cmd.add_argument("--screenshot", action="store_true", help="Capture a compact screenshot receipt without base64 output")
+    open_cmd.add_argument("--no-verify", action="store_true", help="Skip the post-navigation snapshot receipt")
     open_cmd.set_defaults(func=cmd_open)
 
     auth = sub.add_parser("auth", help="Create an auth handoff or active lease-control response")

ax_browser_broker/docs.py

@@ -7,6 +7,7 @@
     "quickstart": {
         "title": "OpenBrowser Broker Quickstart",
         "steps": [
+            "For simple user handoff requests such as 'open Lovable for me', call browser_open_control with owner, url, and optional identity_id. It opens the page, verifies state, and returns the control URL.",
             "Call browser_lease with owner and optional identity_id.",
             "Immediately call browser_snapshot or browser_screenshot to see the current page state before doing anything else.",
             "Do NOT call browser_navigate if the current page is already meaningful (e.g. after a human auth handoff the browser is on the target page). Only navigate when the current page is blank, a new tab, or unrelated to the task.",
@@ -16,6 +17,7 @@
             "Call broker_audit after browser-agent work.",
         ],
         "examples": [
+            {"tool": "browser_open_control", "args": {"owner": "agent-name", "identity_id": "work-main", "url": "https://example.com"}},
             {"tool": "browser_lease", "args": {"owner": "agent-name", "identity_id": "work-main"}},
             {"tool": "browser_snapshot", "args": {"lease_id": "<lease_id>"}},
             {"tool": "browser_navigate", "args": {"lease_id": "<lease_id>", "url": "https://example.com"}},
@@ -104,10 +106,12 @@
         "commands": [
             "openbrowser status",
             "openbrowser docs quickstart",
+            "openbrowser open https://example.com --identity work-main --control",
             "openbrowser auth https://example.com/login --identity work-main --owner agent-name",
         ],
         "notes": [
             "The CLI talks to the local broker API and reads the local server-side API key file when needed.",
+            "Use open --control for simple 'open this for me' requests; it returns a verified control URL in one command.",
             "Use it for status, docs, auth handoffs, active lease-control links, and quick smoke checks.",
             "Use Broker MCP directly for normal click/type/screenshot workflows when tools are available.",
         ],

ax_browser_broker/mcp_server.py

@@ -73,6 +73,39 @@ def browser_navigate(lease_id: str, url: str) -> dict[str, Any]:
     return _request("POST", "/browser/navigate", {"lease_id": lease_id, "url": url})
 
 
+@mcp.tool()
+def browser_open_control(
+    owner: str,
+    url: str,
+    identity_id: str | None = None,
+    ttl_seconds: int = 900,
+    control_ttl_seconds: int = 900,
+) -> dict[str, Any]:
+    """Open a URL, verify the page, and return a human-control link in one call."""
+    lease_obj = browser_lease(owner=owner, ttl_seconds=ttl_seconds, identity_id=identity_id)
+    lease_id = str(lease_obj["lease_id"])
+    try:
+        navigation = browser_navigate(lease_id=lease_id, url=url)
+    except Exception:
+        browser_release(lease_id)
+        raise
+    snapshot = browser_snapshot(lease_id)
+    control = lease_control_request(lease_id=lease_id, owner=owner, ttl_seconds=control_ttl_seconds)
+    return {
+        "lease": lease_obj,
+        "navigation": navigation,
+        "snapshot": {
+            "title": snapshot.get("title"),
+            "url": snapshot.get("url"),
+            "bodyText": str(snapshot.get("bodyText") or "")[:1200],
+            "element_count": len(snapshot.get("elements") or []),
+            "slot": snapshot.get("slot"),
+        },
+        "control": control,
+        "portal_url": control.get("portal_url"),
+    }
+
+
 @mcp.tool()
 def browser_snapshot(lease_id: str) -> dict[str, Any]:
     """Return text and interactive element snapshot for the active page."""

ax_browser_broker/remote_mcp_server.py

@@ -110,6 +110,35 @@ def browser_open(owner: str, url: str, identity_id: str | None = None, ttl_secon
     return _request("POST", "/open", {"owner": owner, "url": url, "identity_id": identity_id, "ttl_seconds": ttl_seconds})
 
 
+@mcp.tool()
+def browser_open_control(
+    owner: str,
+    url: str,
+    identity_id: str | None = None,
+    ttl_seconds: int = 900,
+    control_ttl_seconds: int = 900,
+    screenshot: bool = False,
+) -> dict[str, Any]:
+    """Open a URL, verify the page, and return a human-control link in one call."""
+    result = browser_open(owner=owner, url=url, identity_id=identity_id, ttl_seconds=ttl_seconds)
+    lease_id = str((result.get("lease") or {}).get("lease_id") or result.get("lease_id") or "")
+    if lease_id:
+        snapshot = browser_snapshot(lease_id)
+        result["snapshot"] = {
+            "title": snapshot.get("title"),
+            "url": snapshot.get("url"),
+            "bodyText": str(snapshot.get("bodyText") or "")[:1200],
+            "element_count": len(snapshot.get("elements") or []),
+            "slot": snapshot.get("slot"),
+        }
+        if screenshot:
+            result["screenshot"] = {key: value for key, value in browser_screenshot(lease_id).items() if key != "base64"}
+        control = lease_control_request(lease_id=lease_id, owner=owner, ttl_seconds=control_ttl_seconds)
+        result["control"] = control
+        result["portal_url"] = control.get("portal_url")
+    return result
+
+
 @mcp.tool()
 def browser_navigate(lease_id: str, url: str) -> dict[str, Any]:
     """Navigate a leased browser session to a URL."""

tests/test_api.py

@@ -708,6 +708,70 @@ async def fake_release(lease_id):
     assert released == ["lease-open"]
 
 
+def test_openbrowser_open_can_return_verified_control_link(monkeypatch) -> None:
+    monkeypatch.setenv("OPENBROWSER_API_KEYS", "test-openbrowser-key")
+
+    async def fake_create_lease(_request):
+        return {"lease_id": "lease-open", "name": "pool-b", "identity_id": "chrome-work"}
+
+    async def fake_browser_navigate(_request):
+        return {"lease_id": "lease-open", "slot": "pool-b", "url": "https://lovable.dev/dashboard", "title": "Home | Lovable"}
+
+    async def fake_browser_snapshot(_request):
+        return {
+            "lease_id": "lease-open",
+            "slot": "pool-b",
+            "title": "Home | Lovable",
+            "url": "https://lovable.dev/dashboard",
+            "bodyText": "A" * 1400,
+            "elements": [{"selector": "button"}],
+        }
+
+    async def fake_browser_screenshot(_request):
+        return {
+            "lease_id": "lease-open",
+            "slot": "pool-b",
+            "path": "/tmp/shot.png",
+            "mime_type": "image/png",
+            "base64": "secret-image-data",
+        }
+
+    async def fake_lease_control(_request):
+        return {
+            "token": "control-token",
+            "lease_id": "lease-open",
+            "portal_url": "https://browser.example.com/auth/lease-control/control-token",
+        }
+
+    monkeypatch.setattr(api, "create_lease", fake_create_lease)
+    monkeypatch.setattr(api, "browser_navigate", fake_browser_navigate)
+    monkeypatch.setattr(api, "browser_snapshot", fake_browser_snapshot)
+    monkeypatch.setattr(api, "browser_screenshot", fake_browser_screenshot)
+    monkeypatch.setattr(api, "lease_control_request", fake_lease_control)
+    client = TestClient(api.app)
+
+    response = client.post(
+        "/openbrowser/v1/open",
+        json={
+            "owner": "pytest-open",
+            "url": "https://lovable.dev",
+            "identity_id": "chrome-work",
+            "control": True,
+            "screenshot": True,
+        },
+        headers={"authorization": "Bearer test-openbrowser-key"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["portal_url"].endswith("/auth/lease-control/control-token")
+    assert data["snapshot"]["title"] == "Home | Lovable"
+    assert data["snapshot"]["bodyText"] == "A" * 1200
+    assert data["snapshot"]["element_count"] == 1
+    assert data["screenshot"]["path"] == "/tmp/shot.png"
+    assert "base64" not in data["screenshot"]
+
+
 def test_lease_failure_records_telemetry(monkeypatch) -> None:
     events = []
     monkeypatch.setattr(api, "lease", lambda *_args, **_kwargs: (_ for _ in ()).throw(api.LeaseError("No healthy free browser slots")))

tests/test_cli.py

@@ -50,6 +50,66 @@ def fake_request(method, path, body=None, auth=False):
     assert json.loads(capsys.readouterr().out) == {"ok": True}
 
 
+def test_cli_open_control_sends_one_step_handoff_payload(monkeypatch, capsys) -> None:
+    calls = []
+
+    def fake_request(method, path, body=None, auth=False):
+        calls.append((method, path, body, auth))
+        if path == "/openbrowser/v1/open":
+            return {"lease": {"lease_id": "lease-cli"}, "navigation": {"url": "https://lovable.dev/dashboard"}}
+        if path == "/openbrowser/v1/browser/snapshot":
+            return {"title": "Home | Lovable", "url": "https://lovable.dev/dashboard", "bodyText": "ok", "elements": [], "slot": "pool-b"}
+        if path == "/openbrowser/v1/browser/screenshot":
+            return {"path": "/tmp/shot.png", "base64": "image-data"}
+        if path == "/openbrowser/v1/lease-control/request":
+            return {"portal_url": "https://browser.example.com/auth/lease-control/tok"}
+        raise AssertionError(path)
+
+    monkeypatch.setattr(cli, "_request", fake_request)
+
+    assert (
+        cli.main(
+            [
+                "open",
+                "https://lovable.dev",
+                "--identity",
+                "chrome-work",
+                "--owner",
+                "pytest-open",
+                "--control",
+                "--screenshot",
+            ]
+        )
+        == 0
+    )
+
+    assert calls == [
+        (
+            "POST",
+            "/openbrowser/v1/open",
+            {
+                "owner": "pytest-open",
+                "identity_id": "chrome-work",
+                "url": "https://lovable.dev",
+                "ttl_seconds": 900,
+            },
+            True,
+        ),
+        ("POST", "/openbrowser/v1/browser/snapshot", {"lease_id": "lease-cli"}, True),
+        ("POST", "/openbrowser/v1/browser/screenshot", {"lease_id": "lease-cli", "full_page": False}, True),
+        (
+            "POST",
+            "/openbrowser/v1/lease-control/request",
+            {"owner": "pytest-open", "lease_id": "lease-cli", "ttl_seconds": 900},
+            True,
+        ),
+    ]
+    output = json.loads(capsys.readouterr().out)
+    assert output["portal_url"].endswith("/tok")
+    assert output["snapshot"]["title"] == "Home | Lovable"
+    assert "base64" not in output["screenshot"]
+
+
 def test_cli_prints_compact_http_errors(monkeypatch, capsys) -> None:
     def fake_request(_request, timeout=None):
         raise urllib.error.HTTPError(

tests/test_docs.py

@@ -14,7 +14,7 @@ def test_docs_topics_and_quickstart() -> None:
 
     quickstart = docs("quickstart")
     assert quickstart["topic"] == "quickstart"
-    assert quickstart["examples"][0]["tool"] == "browser_lease"
+    assert quickstart["examples"][0]["tool"] == "browser_open_control"
 
     identities = docs("identities")
     assert any("not overwritten" in item for item in identities["facts"])