miniwob with mcp browser backend, first draft

Author: ollmer

experiments/run_miniwob.py

@@ -0,0 +1,25 @@
+import logging
+import os
+
+from agentlab.agents.tapeagent.agent import TapeAgentArgs, load_config
+from agentlab.backends.browser.mcp_playwright import MCPPlaywright
+from agentlab.benchmarks.miniwob import MiniWobBenchmark
+from agentlab.experiments.study import make_study
+
+fmt = "%(asctime)s - %(levelname)s - %(name)s:%(lineno)d - %(funcName)s() - %(message)s"
+logging.basicConfig(level=logging.INFO, force=True, format=fmt, handlers=[logging.StreamHandler()])
+
+if __name__ == "__main__":
+    config = load_config("gaia_l1")
+    study = make_study(
+        benchmark=MiniWobBenchmark(backend=MCPPlaywright()),
+        agent_args=TapeAgentArgs(agent_name=config.name, config=config),
+        comment=config.comment,
+        logging_level=logging.INFO,
+        logging_level_stdout=logging.INFO,
+    )
+    if os.environ.get("AGENTLAB_DEBUG"):
+        study.exp_args_list = study.exp_args_list[:3]
+        study.run(n_jobs=1, n_relaunch=1, parallel_backend="sequential")
+    else:
+        study.run(n_jobs=config.n_jobs, n_relaunch=1, parallel_backend=config.parallel_backend)

experiments/test_mcp.py

@@ -0,0 +1,42 @@
+from tapeagents.environment import FunctionCall
+from tapeagents.mcp import ToolCallAction
+
+from agentlab.backends.browser.mcp_playwright import MCPPlaywright
+from agentlab.benchmarks.miniwob.task import get_miniwob_tasks
+
+
+def main():
+    tasks = get_miniwob_tasks()
+    task = tasks[0]
+    setup_js = task.get_setup_js()
+
+    backend = MCPPlaywright()
+    print("="*100)
+    # 1. goto task url
+    print("URL: ", task.url)
+    obs = backend.call_tool("browser_navigate", {"url": task.url})
+    print("------")
+    print(obs)
+    print("-"*100)
+
+    # 2. eval js
+    obs = backend.run_js(setup_js)
+    print("------")
+    print(obs)
+    print("-"*100)
+
+    # 3. validate
+    print("\n\nVALIDATE")
+    js = task.get_task_validate_js()
+    print(js)
+    obs = backend.run_js(js)
+    print("------")
+    print(obs)
+    print("-"*100)
+
+if __name__ == "__main__":
+    main()
+
+
+
+    

src/agentlab/backends/browser/base.py

@@ -0,0 +1,40 @@
+from tapeagents.environment import FunctionCall
+from tapeagents.mcp import MCPEnvironment, ToolCallAction
+from tapeagents.tool_calling import as_openai_tool
+
+
+class BrowserBackend():
+    def run_js(self, js: str):
+        raise NotImplementedError
+
+    def call_tool(self, tool_name: str, arguments: dict) -> str:
+        raise NotImplementedError
+
+    def tools_description(self) -> str:
+        raise NotImplementedError
+
+    def tools(self) -> list[dict]:
+        raise NotImplementedError
+
+
+class MCPBrowserBackend(BrowserBackend):
+    def __init__(self, config_path: str):
+        self.config_path = config_path
+        self.mcp = MCPEnvironment(config_path=self.config_path)
+        self.mcp.initialize()
+
+    def call_tool(self, tool_name: str, arguments: dict) -> str:
+        action = ToolCallAction(
+            function=FunctionCall(name=tool_name, arguments=arguments)
+        )
+        tool_result = self.mcp.step(action)
+        return tool_result.content.content[0].text
+
+
+    def tools_description(self) -> str:
+        return self.mcp.tools_description()
+
+    def tools(self) -> list[dict]:
+        actions = self.mcp.actions()
+        tools = [as_openai_tool(a).model_dump() for a in actions]
+        return tools

src/agentlab/backends/browser/env.py

@@ -0,0 +1,82 @@
+import logging
+import time
+from typing import Any, Literal
+
+from tapeagents.core import Action, Observation, StopStep
+
+from agentlab.backends.browser.base import BrowserBackend
+from agentlab.benchmarks.abstract_env import AbstractEnv
+from agentlab.benchmarks.miniwob.task import AbstractWebTask
+
+logger = logging.getLogger(__name__)
+
+
+class PageObservation(Observation):
+    kind: Literal["page_observation"] = "page_observation"
+    content: str
+
+class BrowserAction(Action):
+    kind: Literal["browser_action"] = "browser_action"
+    name: str
+    arguments: dict[str, Any]
+
+
+class BrowserEnv(AbstractEnv):
+    def __init__(self, task_name: str, task: AbstractWebTask, backend: BrowserBackend, seed: int = 0):
+        self.task_name = task_name
+        self.task = task
+        self.seed = seed
+        self.backend = backend
+        self._turns = 0
+
+    def reset(self, seed: int):
+        self.seed = seed
+        setup_js = self.task.get_setup_js()
+        if setup_js:
+            js_result_str = self.backend.run_js(setup_js)
+            logger.info(f"Task reset result: {js_result_str}")
+
+    def step(self, action: BrowserAction) -> tuple[Observation, float, bool, bool, dict]:
+        logger.info(f"BrowserEnv.step() called with action {type(action)}")
+
+        action_exec_start = time.time()
+        finished = isinstance(action, StopStep)
+        if finished:
+            observation = Observation()  # empty observation
+        else:
+            observation = self._step(action)
+        action_exec_stop = time.time()
+        self._turns += 1
+
+        truncated = self._turns >= self.max_turns
+
+        if self.task.validate_per_step or finished or truncated:
+            reward = self.calculate_reward(action, observation)
+        else:
+            reward = None
+
+        env_info = {
+            "step_metadata": observation.metadata,
+            "action_exec_start": action_exec_start,
+            "action_exec_stop": action_exec_stop,
+            "action_exec_timeout": 0.0,
+        }
+        obs_view = observation.short_view() if isinstance(observation, Observation) else observation
+        logger.info(f"Action result in observation: {obs_view}")
+        return observation, reward, finished, truncated, env_info
+
+    def _step(self, action: Action) -> PageObservation:
+        tool_result = self.backend.call_tool(action.name, action.arguments)
+        return PageObservation(content=tool_result)
+
+    def calculate_reward(self, action: Action, observation: PageObservation) -> float:
+        validate_js = self.task.get_step_validate_js()
+        validate_result = self.backend.run_js(validate_js)
+        reward, other = self.task.parse_validation_result(validate_result)
+        return reward
+
+    def close(self):
+        teardown_js = self.task.get_teardown_js()
+        if teardown_js:
+            js_result_str = self.backend.run_js(teardown_js)
+            logger.info(f"Task teardown result: {js_result_str}")

src/agentlab/backends/browser/mcp_playwright.json

@@ -0,0 +1,17 @@
+{
+  "mcpServers": {
+    "playwright": {
+      "command": "npx",
+      "args": [
+        "@playwright/mcp@latest",
+        "--browser",
+        "chromium",
+        "--headless",
+        "--isolated"
+      ],
+      "env": {
+        "PLAYWRIGHT_BROWSERS_PATH": ""
+      }
+    }
+  }
+}

src/agentlab/backends/browser/mcp_playwright.py

@@ -0,0 +1,14 @@
+from agentlab.backends.browser.base import MCPBrowserBackend
+
+DEFAULT_CONFIG_PATH = "src/agentlab/backends/browser/mcp_playwright.json"
+
+class MCPPlaywright(MCPBrowserBackend):
+    def __init__(self, config_path: str | None = None):
+        super().__init__(config_path or DEFAULT_CONFIG_PATH)
+
+    def run_js(self, js: str):
+        raw_response = self.call_tool("browser_evaluate", {"function": js})
+        _, half_response = raw_response.split("### Result", maxsplit=1)
+        result_str, _ = half_response.split("\n### Ran", maxsplit=1)
+        result_str = result_str.strip()
+        return result_str

src/agentlab/benchmarks/miniwob/__init__.py

@@ -0,0 +1,4 @@
+from .benchmark import MiniWobBenchmark
+from .task import MiniWobTask
+
+__all__ = ["MiniWobBenchmark", "MiniWobTask"]

src/agentlab/benchmarks/miniwob/benchmark.py

@@ -0,0 +1,50 @@
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from pydantic import ConfigDict
+
+from agentlab.backends.browser.base import BrowserBackend
+from agentlab.backends.browser.env import BrowserEnv
+from agentlab.benchmarks.abstract_env import AbstractBenchmark, AbstractEnvArgs
+from agentlab.benchmarks.miniwob.task import MiniWobTask, get_miniwob_tasks
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class MiniwobArgs(AbstractEnvArgs):
+    task: MiniWobTask
+    task_seed: int
+    task_name: str
+    backend: BrowserBackend
+
+    def __init__(self, task_name: str, task: MiniWobTask, backend: BrowserBackend, task_seed: int = 0):
+        self.task_name = task_name
+        self.task = task
+        self.task_seed = task_seed
+        self.backend = backend
+
+    def make_env(self, exp_dir: Path, action_mapping=None) -> BrowserEnv:
+        env = BrowserEnv(task_name=self.task_name, task=self.task, backend=self.backend, seed=self.task_seed)
+        return env
+
+
+class MiniWobBenchmark(AbstractBenchmark):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    backend: BrowserBackend
+    name: str = "miniwob"
+    env_args_list: list[MiniwobArgs] = None  # type: ignore
+    dataset: list[MiniWobTask] = None  # type: ignore
+
+    def model_post_init(self, __context: Any) -> None:
+        self.env_args_list = []
+        if self.dataset is None:
+            self.dataset = get_miniwob_tasks()
+        for task in self.dataset:
+            name = f"miniwob.{task.task_id}"
+            env_args = MiniwobArgs(task_name=name, task=task, backend=self.backend)
+            self.env_args_list.append(env_args)
+        logger.info(f"Loaded {len(self.env_args_list)} miniwob tasks")