feat(task-step): browser_connect_cdp hands an agent a CDP address to an env's website browser

src/agent_env/agents/AGENTS.md

@@ -120,8 +120,9 @@ Reference: `enable(MCP_CONFIG_V1, ...)`, and `claude_bridge.py` where it builds
 ### `cdp-config/v1`, `/ext/cdp-config`
 
 For an agent that drives a browser itself. A browser reaches the agent as an endpoint `{name, http_url, ws_url,
-headers?, metadata?}` sent to `add`, and adding a name again replaces it; nothing in agent-env sends one yet (the
-`browser_connect_cdp` task step will). The SDK implements `add`, `list` and `remove`, and passes the endpoints in
+headers?, metadata?}` sent to `add`, and adding a name again replaces it. The `browser_connect_cdp` task step sends
+one, `chromium` by default, with `metadata {browser}`. A reset ends the address, so a task runs `browser_connect_cdp`
+again after `reset_env`, or gives the turn a `retry_config` that reruns it. The SDK implements `add`, `list` and `remove`, and passes the endpoints in
 `request.cdp_endpoints` every turn. Take yours with `request.cdp_endpoint()`, or `request.cdp_endpoint(name)` when
 the task gives several, and connect afresh each turn, since an address can change between turns. With no endpoint, or
 several and no name, it fails the turn as an `infra_error` (`cdp.no_endpoint`, `cdp.ambiguous`), so the agent never

src/agent_env/task/episode/rules.py

@@ -133,6 +133,7 @@ def _rubrics_verifier(step: RubricsVerifierTaskStep, outside: Outside) -> list[s
     "deploy_agent": _deploy_agent,
     "deploy_human_agent": _deploys_an_agent,
     "register_agent_triggers": _needs_the_agent,
+    "browser_connect_cdp": _needs_the_agent,
     "snapshot_agent_state": _needs_the_agent,
     "collect_artifacts": _collect_artifacts,
     "register_env_triggers": _register_env_triggers,

src/agent_env/task_step/registry.py

@@ -27,6 +27,7 @@ def get_task_step_registry() -> dict[str, type["TaskStep"]]:
 
 
 def _builtin_registry() -> dict[str, type["TaskStep"]]:
+    from agent_env.task_step.task_steps.browser_connect_cdp import BrowserConnectCdpTaskStep
     from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep
     from agent_env.task_step.task_steps.computer_recording import (
         EndComputerRecordingTaskStep,
@@ -88,6 +89,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]:
     return {
         _get_type(ReviewTaskStep): ReviewTaskStep,
         _get_type(BuildMcpCliTaskStep): BuildMcpCliTaskStep,
+        _get_type(BrowserConnectCdpTaskStep): BrowserConnectCdpTaskStep,
         _get_type(CollectArtifactsTaskStep): CollectArtifactsTaskStep,
         _get_type(StartComputerRecordingTaskStep): StartComputerRecordingTaskStep,
         _get_type(EndComputerRecordingTaskStep): EndComputerRecordingTaskStep,

src/agent_env/task_step/task_steps/browser_connect_cdp.py

@@ -0,0 +1,171 @@
+"""Task step that hands a deployed agent a CDP address to an env's website browser, through the agent's cdp-config/v1,
+and hides that browser's MCP tools from the agent. A reset ends the address, so a task connects the agent again after
+reset_env, or with a retry_config that reruns this step."""
+
+from __future__ import annotations
+
+import logging
+from typing import ClassVar, Optional
+
+import httpx
+from agentenv_protocol.client import find_extension_method, invoke_extension
+from agentenv_protocol.tool_access import DISABLE, DISABLE_TOOL_EXTENSION_URI, WILDCARD
+from agentenv_protocol.tool_context import DEFAULT_ROLE
+from agentenv_protocol.website_browser import BROWSER
+
+from agent_env.a2a_agent.a2a_agent import A2AAgent
+from agent_env.entity.refs import EntityRef
+from agent_env.env.env import DeployedEnv, DeployedSandboxEnv
+from agent_env.providers.sandbox_providers.local_docker.provider import host_url_for
+from agent_env.providers.sandbox_providers.sandbox_provider import reachable_url
+from agent_env.task_step.context import DeployedAgent, TaskStepContext
+from agent_env.task_step.task_step import TaskStep, TaskStepDependency
+
+logger = logging.getLogger(__name__)
+
+#: Where the step records each connection it makes, one per agent and name: the role its tools were hidden from.
+CONNECTIONS_KEY = "browser_connections"
+#: The name the address is added to the agent's cdp-config under when the step names none.
+DEFAULT_NAME = "chromium"
+#: How long the browser may take to answer cdp, which waits for the browser to be ready.
+CDP_TIMEOUT_S = 180
+#: How long the agent may take to answer cdp-config's add.
+ADD_TIMEOUT_S = 60
+
+
+class BrowserConnectCdpTaskStep(TaskStep):
+    """Hand the agent ``agent_name`` a CDP address to the website browser ``browser`` in the env ``env_id``, the
+    env's only one when unnamed: added to its cdp-config/v1 as ``name``, with every tool the env serves hidden from
+    its role unless ``keep_tools``. The address is the browser's own for that agent, as its sandbox reaches it."""
+
+    type: ClassVar[str] = "browser_connect_cdp"
+    entity_refs = (EntityRef.env("env_id"),)
+
+    def __init__(
+        self,
+        id: str,
+        version: Optional[int],
+        env_id: str,
+        agent_name: Optional[str] = None,
+        browser: Optional[str] = None,
+        name: str = DEFAULT_NAME,
+        keep_tools: bool = False,
+        depends_on: Optional[list[TaskStepDependency]] = None,
+        fail_task_on_error: bool = True,
+    ):
+        super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error)
+        self.env_id = env_id
+        self.agent_name = agent_name or TaskStep.DEFAULT_AGENT_NAME
+        self.browser = browser
+        self.name = name
+        self.keep_tools = keep_tools
+
+    def config_problems(self) -> list[str]:
+        problems = []
+        if not isinstance(self.env_id, str) or not self.env_id:
+            problems.append(f"step {self.id!r}: env_id must be a non-empty string")
+        if not isinstance(self.name, str) or not self.name:
+            problems.append(f"step {self.id!r}: name must be a non-empty string")
+        if not isinstance(self.keep_tools, bool):
+            problems.append(f"step {self.id!r}: keep_tools must be true or false")
+        return problems
+
+    def to_dict(self) -> dict:
+        base = super().to_dict()
+        base.update(env_id=self.env_id, agent_name=self.agent_name, browser=self.browser, name=self.name,
+                    keep_tools=self.keep_tools)
+        return base
+
+    @classmethod
+    def from_dict(cls, data: dict) -> BrowserConnectCdpTaskStep:
+        return cls(
+            **cls._base_from_dict(data),
+            env_id=data["env_id"],
+            agent_name=data.get("agent_name"),
+            browser=data.get("browser"),
+            name=data.get("name") or DEFAULT_NAME,
+            keep_tools=data.get("keep_tools", False),
+        )
+
+    async def execute(self, context: TaskStepContext) -> TaskStepContext:
+        deployed = _deployed_env(context, self.env_id)
+        agent = _deployed_agent(context, self.agent_name)
+        cdp_config = A2AAgent.find_extension(agent.a2a_card or {}, A2AAgent.EXT_CDP_CONFIG)
+        if cdp_config is None or not agent.a2a_url:
+            raise RuntimeError(f"agent {agent.agent_name!r} can't take a CDP address: its card has no "
+                               f"{A2AAgent.EXT_CDP_CONFIG}")
+        browser, card = _browser(deployed, self.browser)
+        address = await invoke_extension(deployed.environment_url, card, BROWSER, {"user": agent.agent_name},
+                                         CDP_TIMEOUT_S, method="cdp", headers=deployed.control_headers)
+        http_url, ws_url = _as_agent_reaches(address, deployed, agent)
+        async with httpx.AsyncClient(timeout=ADD_TIMEOUT_S) as client:
+            added = await client.post(agent.a2a_url + cdp_config["params"]["endpoint"], json={
+                "name": self.name, "http_url": http_url, "ws_url": ws_url,
+                "metadata": {"browser": browser or deployed.environment_card.get("name")}})
+        added.raise_for_status()
+        role = agent.role or DEFAULT_ROLE
+        hidden = None if self.keep_tools else WILDCARD
+        if hidden:
+            await deployed.invoke(DISABLE_TOOL_EXTENSION_URI, DISABLE, {"role": role, "tools": hidden})
+        logger.info(f"agent {agent.agent_name!r} has {self.name!r}, a CDP address to {browser or 'the env'} in env "
+                    f"{deployed.env_id!r}" + (f"; the env's tools are hidden from role {role!r}" if hidden else ""))
+        # The agent's cdp-config now holds this address under its name, so it replaces any connection recorded so.
+        records = context.metadata.setdefault(CONNECTIONS_KEY, [])
+        records[:] = [record for record in records
+                      if (record["agent_name"], record["name"]) != (self.agent_name, self.name)] + [{
+            "step_id": self.id, "env_id": self.env_id, "agent_name": self.agent_name, "browser": browser,
+            "name": self.name, "keep_tools": self.keep_tools, "role": role, "tools_hidden": hidden}]
+        return context
+
+
+def _deployed_env(context: TaskStepContext, env_id: str) -> DeployedEnv:
+    deployed = next((d for d in context.deployed_envs if d.env_id == env_id), None)
+    if deployed is None:
+        raise RuntimeError(f"env {env_id!r} isn't in context.deployed_envs")
+    return deployed
+
+
+def _deployed_agent(context: TaskStepContext, agent_name: str) -> DeployedAgent:
+    agent = next((a for a in context.deployed_agents if a.agent_name == agent_name), None)
+    if agent is None:
+        raise RuntimeError(f"agent {agent_name!r} isn't in context.deployed_agents")
+    return agent
+
+
+def _browser(deployed: DeployedEnv, named: Optional[str]) -> tuple[Optional[str], dict]:
+    """The website browser to connect to, the one named else the env's only one: its name among the env's children,
+    or None for the env itself, and its card. A browser computer's driver takes its app's name, so a child is found
+    among the children, never by name as the driver's own card."""
+    card = deployed.environment_card or {}
+    browsers = [(None, card)] if find_extension_method(card, BROWSER, "cdp") else []
+    browsers += [(child.get("name"), child) for child in card.get("children_environments") or []
+                 if find_extension_method(child, BROWSER, "cdp")]
+    if named is not None:
+        found = next((browser for browser in browsers if browser[0] == named), None)
+        if found is None:
+            raise RuntimeError(f"env {deployed.env_id!r} has no website browser named {named!r}")
+        return found
+    if len(browsers) != 1:
+        raise RuntimeError(f"env {deployed.env_id!r} has {len(browsers) or 'no'} website browsers"
+                           + (": name one as browser" if browsers else ""))
+    return browsers[0]
+
+
+def _as_agent_reaches(address: dict, deployed: DeployedEnv, agent: DeployedAgent) -> tuple[str, str]:
+    """``address``'s URLs as the agent's sandbox reaches them: under the env's URL as the agent reaches that, as
+    deploy_agent gives it MCP URLs. A URL anywhere else, at a gateway's public address say, is kept."""
+    base = deployed.environment_url.rstrip("/")
+    if isinstance(deployed, DeployedSandboxEnv):
+        theirs = reachable_url(base, from_sandbox_type=deployed.sandbox_type, to_sandbox_type=agent.sandbox_type)
+    else:
+        theirs = host_url_for(base, agent.sandbox_type)
+    theirs = theirs.rstrip("/")
+    return _rebased(address["http_url"], base, theirs), _rebased(address["ws_url"], _ws(base), _ws(theirs))
+
+
+def _rebased(url: str, base: str, theirs: str) -> str:
+    return theirs + url[len(base):] if url.startswith(base + "/") else url
+
+
+def _ws(url: str) -> str:
+    return "ws" + url.removeprefix("http") if url.startswith("http") else url

tst/data/a2a_agent/Dockerfile

@@ -1,7 +1,7 @@
 FROM mirror.gcr.io/library/python:3.12-slim@sha256:423ed6ab25b1921a477529254bfeeabf5855151dc2c3141699a1bfc852199fbf
 WORKDIR /app
 COPY agentenv-protocol/ ./agentenv-protocol/
-RUN pip install --no-cache-dir './agentenv-protocol[agent]' && useradd --create-home agent
+RUN pip install --no-cache-dir './agentenv-protocol[agent]' 'websockets>=13' && useradd --create-home agent
 COPY agent.py .
 USER agent
 CMD ["python", "agent.py"]

tst/data/a2a_agent/agent.py

@@ -13,21 +13,29 @@
 It installs skills under ``~/skills/<name>/``, a bundle's files downloaded through their read grants, and a
 ``list-skills`` line in its prompt adds a ``skill <name>: <path>=<sha256 prefix>, ...`` line for each one.
 A ``run <command>`` line has it run that command in its container and add a ``ran <command>: <output>`` line.
+
+A ``cdp <name> <url>`` line has it open the URL in a new tab of the browser its cdp-config names so, over raw CDP, and
+add a ``cdp <name> <digest>: <title>`` line, ``<digest>`` the first 12 hex digits of its address's sha256, so a
+reader can tell one address from another without the address. With no endpoint by that name the turn fails.
 """
 
 import asyncio
 import base64
 import contextlib
 import hashlib
+import itertools
+import json
 import os
 import re
 import signal
 from pathlib import Path
+from collections.abc import Mapping
 from typing import Any
 from urllib.parse import urlsplit
 
 import httpx
 from agentenv_protocol.a2a_agent import (
+    CDP_CONFIG_V1,
     MCP_CONFIG_V1,
     SKILL_CONFIG_V1,
     TRAJECTORY_V1,
@@ -48,10 +56,13 @@
     enable,
     extension,
 )
+from websockets.asyncio.client import connect
 
 SEND_FILE = "send-file "
 LIST_SKILLS = "list-skills"
 RUN = "run "
+CDP = "cdp "
+CDP_LOAD_SECONDS = 30
 RUN_TIMEOUT_SECONDS = 60
 SKILLS_DIR = Path.home() / "skills"
 LIVE_STEPS = re.compile(r"live-steps (\d+)")
@@ -105,6 +116,42 @@ async def _ran(command: str) -> str:
     return f"ran {command}: {output.decode(errors='replace').strip()}"
 
 
+async def _opened(endpoint: Mapping[str, Any], name: str, url: str) -> str:
+    """``url`` opened in a new tab over raw CDP at the browser ``endpoint`` names, as the line that reports it."""
+    digest = hashlib.sha256(endpoint["ws_url"].encode()).hexdigest()[:12]
+    try:
+        async with connect(endpoint["ws_url"], max_size=None, additional_headers=endpoint.get("headers")) as browser:
+            ids = itertools.count(1)
+
+            async def call(method: str, params: dict | None = None, session: str | None = None) -> dict:
+                message_id = next(ids)
+                await browser.send(json.dumps({"id": message_id, "method": method, "params": params or {},
+                                               **({"sessionId": session} if session else {})}))
+                while (answer := json.loads(await browser.recv())).get("id") != message_id:
+                    pass
+                if "error" in answer:
+                    raise RuntimeError(answer["error"].get("message"))
+                return answer.get("result") or {}
+
+            target = (await call("Target.createTarget", {"url": url}))["targetId"]
+            session = (await call("Target.attachToTarget", {"targetId": target, "flatten": True}))["sessionId"]
+            deadline, loaded, title = asyncio.get_running_loop().time() + CDP_LOAD_SECONDS, False, ""
+            while not loaded and asyncio.get_running_loop().time() < deadline:
+                # A new tab is about:blank, already complete, until it commits the URL; a page still navigating may
+                # have no context to evaluate in yet.
+                with contextlib.suppress(RuntimeError):
+                    state, title, href = (await call("Runtime.evaluate", {
+                        "expression": "[document.readyState, document.title, location.href]", "returnByValue": True},
+                        session))["result"]["value"]
+                    loaded = state == "complete" and href != "about:blank"
+                if not loaded:
+                    await asyncio.sleep(0.2)
+            await call("Target.closeTarget", {"targetId": target})
+        return f"cdp {name} {digest}: {title}"
+    except Exception as e:
+        return f"cdp {name} {digest} failed: {type(e).__name__}: {e}"
+
+
 def _skills() -> list[str]:
     """One line per installed skill: each file's path and the first 12 hex digits of its sha256."""
     if not SKILLS_DIR.is_dir():
@@ -121,7 +168,7 @@ def _skills() -> list[str]:
         version="1.0.0",
     ),
     config=EchoAgentConfig,
-    extensions=(MCP_CONFIG_V1, enable(TRAJECTORY_V1, live=True), TRIGGERS_V1),
+    extensions=(MCP_CONFIG_V1, CDP_CONFIG_V1, enable(TRAJECTORY_V1, live=True), TRIGGERS_V1),
 )
 class EchoAgent(AgentEnvAgent):
     async def run(self, request: TaskRequest[EchoAgentConfig]) -> TaskResult:
@@ -137,6 +184,8 @@ async def run(self, request: TaskRequest[EchoAgentConfig]) -> TaskResult:
             reply = "\n".join([reply, *_skills()])
         runs = [line.removeprefix(RUN).strip() for line in prompt.splitlines() if line.startswith(RUN)]
         reply = "\n".join([reply, *[await _ran(command) for command in runs]])
+        opens = [line.removeprefix(CDP).split() for line in prompt.splitlines() if line.startswith(CDP)]
+        reply = "\n".join([reply, *[await _opened(request.cdp_endpoint(name), name, url) for name, url in opens]])
         sends = [line.removeprefix(SEND_FILE).strip() for line in prompt.splitlines() if line.startswith(SEND_FILE)]
         files = [FilePart(uri=uri, name=uri.rsplit("/", 1)[-1], mime_type="text/plain") for uri in sends]
         request.trajectory.append({"type": "echo", "input": prompt, "output": reply})

tst/integration/cli/run_bundle_built_agents_test.py

@@ -37,7 +37,7 @@
 PREFIX = f'''{BASE}
 WORKDIR /app
 COPY agentenv-protocol/ ./agentenv-protocol/
-RUN pip install --no-cache-dir './agentenv-protocol[agent]'
+RUN pip install --no-cache-dir './agentenv-protocol[agent]' 'websockets>=13'
 '''
 
 

tst/integration/cli/run_bundle_installed_agent_test.py

@@ -25,7 +25,7 @@
 DOCKERFILE = f'''{BASE}
 WORKDIR /app
 COPY agentenv-protocol/ ./agentenv-protocol/
-RUN pip install --no-cache-dir './agentenv-protocol[agent]'
+RUN pip install --no-cache-dir './agentenv-protocol[agent]' 'websockets>=13'
 COPY agent.py ./
 CMD ["python", "agent.py"]
 '''