| """The OpenEnv environment: MCP tools over one shared episode. |
| |
| `reset(split=, index=, mode=, surface=)` picks a Crello template and how it is |
| played; the tools a model sees depend on the surface: |
| |
| tools add_image, add_text, add_shape, update_element, ... (structured edits) |
| html write_html, read_html (one page of markup) |
| gui screenshot, left_click, left_click_drag, type, key (a real editor) |
| |
| On the gui surface the editor is the built-in one in one of six layouts, or a real app (skin="penpot", "impress", |
| "inkscape-desktop", "gimp"; realapps/), which the server runs locally or in Daytona sandboxes it manages. |
| |
| and every surface shares get_task, view_reference and submit. `submit` ends the |
| episode and its reward is the score of the final document (see core/score.py). |
| Pure MCP clients, which have no `reset`, start with `start_task`. |
| |
| Tasks are served through OpenEnv's Task API (`list_splits`, `num_tasks`, |
| `get_task`, `get_task_range`), so a trainer can enumerate a split and then |
| reset onto any index. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import base64 |
| import json |
| import os |
| import random |
| from typing import Annotated, Any, Literal |
|
|
| from fastmcp import FastMCP |
| from fastmcp.exceptions import ToolError |
| from fastmcp.server.transforms.visibility import Visibility |
| from fastmcp.utilities.types import Image |
| from openenv.core.env_server.mcp_environment import MCPEnvironment |
| from openenv.core.env_server.mcp_types import CallToolAction |
| from openenv.core.env_server.types import Action, Observation |
| from pydantic import Field |
|
|
| try: |
| from ..core import ops |
| from ..core.contact import contact_sheet |
| from ..core.html_profile import HTMLProfileError, html_to_doc |
| from ..core.pack import TaskPack, load_pack |
| from ..core.prompts import PROMPT_VERSION, instructions |
| from ..models import DesignState |
| from .episodes import MODES, REAL_SKINS, SKINS, STORE, SURFACES, Episode |
| from .gui import VIEWPORT, GuiDriver, self_url |
| except ImportError: |
| from core import ops |
| from core.contact import contact_sheet |
| from core.html_profile import HTMLProfileError, html_to_doc |
| from core.pack import TaskPack, load_pack |
| from core.prompts import PROMPT_VERSION, instructions |
| from models import DesignState |
| from server.episodes import MODES, REAL_SKINS, SKINS, STORE, SURFACES, Episode |
| from server.gui import VIEWPORT, GuiDriver, self_url |
|
|
| EDIT_TOOLS = { |
| "add_image", "add_text", "add_shape", "update_element", "delete_element", |
| "duplicate_element", "reorder_element", "align_elements", "set_background", "undo", "redo", |
| "write_html", |
| } |
| DEFAULT_MAX_STEPS = {"tools": 60, "html": 20, "gui": 200} |
| |
| OBSERVATION_TOOLS = {"start_task", "get_task", "view_reference", "view_canvas", "view_assets", "list_fonts", |
| "list_elements", "read_html", "screenshot", "wait"} |
| OBSERVATION_FACTOR = 3 |
|
|
| Num = float | int |
| FILE_DIALOGS = ("mac", "windows", "linux") |
| COORDINATES = ("pixels", "normalized") |
| Coord = Annotated[list[int], Field(description="[x, y] on the screenshot: pixels, or a 0-1000 grid when " |
| "your instructions say so", min_length=2, max_length=2)] |
|
|
|
|
| def _key_sequence(keys) -> list[str]: |
| """'enter', ['ctrl+a', 'delete'] or the JSON string '["enter"]' -> key combos.""" |
| if isinstance(keys, str): |
| text = keys.strip() |
| if text.startswith("["): |
| try: |
| keys = json.loads(text) |
| except json.JSONDecodeError: |
| keys = [text.strip("[]\"' ")] |
| else: |
| return [text] if text else [] |
| return [str(k).strip() for k in keys if str(k).strip()] |
|
|
|
|
| def _json(obj: Any) -> str: |
| return json.dumps(obj, ensure_ascii=False, separators=(",", ":")) |
|
|
|
|
| class DesignCanvasEnvironment(MCPEnvironment): |
| """Recreate a Crello design on a real canvas, through tools, HTML or a GUI.""" |
|
|
| SUPPORTS_CONCURRENT_SESSIONS = True |
|
|
| def __init__(self, pack: TaskPack | None = None): |
| self.pack = pack or load_pack() |
| self._episode: Episode | None = None |
| self._state = DesignState() |
| self._reward_mode = "final" |
| self._last_total = 0.0 |
| mcp = FastMCP("design_canvas") |
| self._mcp = mcp |
| self._register_common(mcp) |
| self._register_tools_surface(mcp) |
| self._register_html_surface(mcp) |
| self._register_gui_surface(mcp) |
| super().__init__(mcp) |
| self._set_surface("tools") |
|
|
| |
|
|
| def list_splits(self): |
| return [{"name": s, "type": s, "num_tasks": self.pack.count(s)} |
| for s in self.pack.splits()] |
|
|
| def num_tasks(self, split): |
| return self.pack.count(split) |
|
|
| def get_task(self, split, index): |
| return self.pack.public(self.pack.at(split, int(index))) |
|
|
| def get_task_range(self, split, start=None, stop=None): |
| return self.pack.task_range(split, start, stop) |
|
|
| def list_tasks(self, split): |
| return self.get_task_range(split) |
|
|
| |
|
|
| def reset( |
| self, |
| seed: int | None = None, |
| episode_id: str | None = None, |
| split: str | None = None, |
| index: int | None = None, |
| task_id: str | None = None, |
| mode: str = "reference", |
| surface: str = "tools", |
| skin: str | None = None, |
| theme: str | None = None, |
| layout: str | None = None, |
| max_steps: int | None = None, |
| reward_mode: str = "final", |
| coordinates: str = "pixels", |
| file_dialog: str | None = None, |
| **kwargs: Any, |
| ) -> Observation: |
| if coordinates not in COORDINATES: |
| raise ValueError(f"coordinates must be one of {COORDINATES}") |
| if mode not in MODES: |
| raise ValueError(f"mode must be one of {MODES}") |
| if surface not in SURFACES: |
| raise ValueError(f"surface must be one of {SURFACES}") |
| if reward_mode not in ("final", "dense"): |
| raise ValueError("reward_mode must be final or dense") |
| if skin in REAL_SKINS and surface != "gui": |
| raise ValueError(f"skin {skin!r} is a real app: use surface='gui' (tools and html edit the canvas " |
| "directly, with no app)") |
| rng = random.Random(seed) |
| if task_id is not None: |
| task = self.pack.get(task_id) |
| else: |
| split = split or ("train" if "train" in self.pack.splits() else self.pack.splits()[0]) |
| count = self.pack.count(split) |
| index = rng.randrange(count) if index is None else int(index) |
| task = self.pack.at(split, index) |
| |
| |
| |
| skin = skin or (rng.choice(SKINS) if seed is not None else "figma") |
| theme = theme or (rng.choice(("light", "dark")) if seed is not None else "light") |
| layout = layout or "default" |
| if layout not in ("default", "swap"): |
| raise ValueError("layout must be default or swap") |
| |
| |
| file_dialog = file_dialog or (rng.choice(FILE_DIALOGS) if seed is not None else None) |
| if file_dialog is not None and file_dialog not in FILE_DIALOGS: |
| raise ValueError(f"file_dialog must be one of {FILE_DIALOGS}") |
| self._close_gui() |
| ep = Episode( |
| pack=self.pack, task=task, mode=mode, surface=surface, skin=skin, theme=theme, |
| max_steps=int(max_steps or DEFAULT_MAX_STEPS[surface]), |
| ) |
| if episode_id: |
| ep.episode_id = str(episode_id)[:64] |
| STORE.add(ep) |
| self._episode = ep |
| self._reward_mode = reward_mode |
| |
| |
| self._coordinates = coordinates |
| self._last_total = 0.0 |
| self._set_surface(surface) |
| editor = f"/editor/?episode={ep.episode_id}&skin={skin}&theme={theme}" + ( |
| "&layout=swap" if layout == "swap" else "") + (f"&os={file_dialog}" if file_dialog else "") + ( |
| |
| f"&fs={rng.randrange(1 << 30)}" if seed is not None else "") |
| if skin in REAL_SKINS: |
| from ..realapps import READY, available, get_backend |
|
|
| ok, why = available() |
| if not ok: |
| raise RuntimeError(f"{REAL_SKINS[skin][0]} is not available on this server: {why}") |
|
|
| backend = get_backend(skin, fonts_dir=self.pack.root / "fonts") |
| gt = self.pack.gt_doc(task) |
| if skin == "penpot": |
| session = backend.new_episode(f"{task['task_id']} {ep.episode_id}", gt.width, gt.height, ep.assets) |
| GuiDriver.get().open(ep.episode_id, session.url, cookies=backend.client.cookies(), |
| ready=READY[skin], headers=backend.client.headers or None).result(300) |
| else: |
| session = backend.new_episode(skin, gt.width, gt.height, ep.assets, fonts=self.pack.fonts) |
| GuiDriver.get().attach_remote(ep.episode_id, session.client) |
| ep.realapp = (backend, session) |
| |
| editor = f"/api/episodes/{ep.episode_id}/screen.png" |
| elif surface == "gui": |
| GuiDriver.get().open(ep.episode_id, self_url() + editor + "&agent=1").result(45) |
| self._state = DesignState( |
| episode_id=ep.episode_id, step_count=0, task_id=task["task_id"], split=task["split"], |
| mode=mode, surface=surface, skin=skin, editor_url=editor, |
| ) |
| public = os.environ.get("DESIGN_CANVAS_PUBLIC_URL", "").rstrip("/") |
| return self._observation(ep, public + editor) |
|
|
| def _observation(self, ep: Episode, editor: str) -> Observation: |
| info = ep.info() |
| meta = { |
| **info, |
| "instructions": instructions(ep.surface, ep.mode, info, |
| (VIEWPORT["width"], VIEWPORT["height"]), |
| coordinates=getattr(self, "_coordinates", "pixels"), skin=ep.skin), |
| "coordinates": getattr(self, "_coordinates", "pixels"), |
| "prompt_version": PROMPT_VERSION, |
| "skin": ep.skin, |
| "theme": ep.theme, |
| "layout": "swap" if "layout=swap" in editor else "default", |
| "file_dialog": next((d for d in FILE_DIALOGS if f"&os={d}" in editor), None), |
| "editor_url": editor, |
| "tools": sorted(self._surface_tools(ep.surface)), |
| } |
| if ep.skin in REAL_SKINS: |
| meta["app"] = {"skin": ep.skin, "name": REAL_SKINS[ep.skin][0], "real": True} |
| if ep.mode == "reference": |
| meta["reference_jpeg_b64"] = base64.b64encode( |
| self.pack.target_jpeg(ep.task, 768)).decode() |
| return Observation(done=False, reward=None, metadata=meta) |
|
|
| def _set_surface(self, surface: str) -> None: |
| |
| |
| self._mcp._transforms = [t for t in self._mcp._transforms |
| if not isinstance(t, Visibility)] |
| self._mcp.enable(tags={"common", surface}, only=True) |
|
|
| @staticmethod |
| def _surface_tools(surface: str) -> set[str]: |
| return {name for name, tags in _TOOL_TAGS.items() if tags & {"common", surface}} |
|
|
| def _close_gui(self): |
| if self._episode is not None and self._episode.surface == "gui": |
| try: |
| GuiDriver.get().close(self._episode.episode_id).result(10) |
| except Exception: |
| pass |
| if self._episode.realapp is not None and hasattr(self._episode.realapp[0], "release"): |
| GuiDriver.get().detach_remote(self._episode.episode_id) |
| backend, session = self._episode.realapp |
| self._episode.realapp = None |
| backend.release(session.url) |
|
|
| def close(self): |
| self._close_gui() |
| super().close() |
|
|
| @property |
| def state(self) -> DesignState: |
| ep = self._episode |
| if ep is not None: |
| self._state.step_count = ep.step_count |
| self._state.n_elements = len(ep.doc.elements) |
| self._state.done = ep.done |
| if ep.result is not None: |
| self._state.reward = ep.result.total |
| self._state.grade = ep.result.as_dict() |
| return self._state |
|
|
| |
|
|
| def _after_step(self, action: Action, obs: Observation) -> Observation: |
| ep = self._episode |
| if ep is None or not isinstance(action, CallToolAction): |
| return obs |
| tool = action.tool_name |
| |
| |
| |
| if tool in OBSERVATION_TOOLS: |
| ep.observation_count += 1 |
| if not ep.done and ep.observation_count >= OBSERVATION_FACTOR * ep.max_steps: |
| ep.submit("env", reason="observation limit") |
| else: |
| ep.step_count += 1 |
| if not ep.done and ep.step_count >= ep.max_steps: |
| ep.submit("env", reason="step limit") |
| if ep.done and ep.result is not None: |
| obs.done = True |
| final = ep.result.total |
| obs.reward = final - self._last_total if self._reward_mode == "dense" else final |
| self._last_total = final |
| obs.metadata = {**(obs.metadata or {}), "grade": ep.result.as_dict(), |
| "episode_id": ep.episode_id} |
| elif self._reward_mode == "dense" and tool in EDIT_TOOLS: |
| total = ep.score().total |
| obs.reward = total - self._last_total |
| self._last_total = total |
| else: |
| obs.reward = 0.0 |
| return obs |
|
|
| def step(self, action: Action, timeout_s: float | None = None, **kwargs: Any) -> Observation: |
| obs = super().step(action, timeout_s=timeout_s, **kwargs) |
| return self._after_step(action, obs) |
|
|
| async def step_async(self, action: Action, timeout_s: float | None = None, |
| **kwargs: Any) -> Observation: |
| obs = await super().step_async(action, timeout_s=timeout_s, **kwargs) |
| return self._after_step(action, obs) |
|
|
| def _step_impl(self, action: Action, timeout_s: float | None = None, **kwargs) -> Observation: |
| return Observation(done=False, reward=0.0, metadata={ |
| "error": f"unsupported action {type(action).__name__}; use CallToolAction"}) |
|
|
| |
|
|
| def _ep(self) -> Episode: |
| if self._episode is None: |
| raise ToolError("no task is loaded: call reset() or the start_task tool first") |
| return self._episode |
|
|
| def _edit(self, op: str, args: dict) -> str: |
| ep = self._ep() |
| if ep.done: |
| raise ToolError("the episode is finished; start a new task") |
| try: |
| result = ep.apply(op, {k: v for k, v in args.items() if v is not None}, "agent") |
| except ops.OpError as exc: |
| raise ToolError(str(exc)) from None |
| return _json(result) |
|
|
| |
|
|
| def _register_common(self, mcp: FastMCP) -> None: |
| env = self |
|
|
| @mcp.tool(tags={"common"}) |
| def start_task( |
| split: Annotated[str, Field(description="train, validation or test")] = "validation", |
| index: Annotated[int | None, Field(description="task index; random if omitted")] = None, |
| mode: Annotated[Literal["reference", "description"], Field( |
| description="reference: recreate a target image; description: build from a brief" |
| )] = "reference", |
| surface: Annotated[Literal["tools", "html"], Field( |
| description="tools: structured edit calls; html: write one page of markup" |
| )] = "tools", |
| ) -> str: |
| """Load a design task (for MCP clients without reset). Returns the instructions. |
| |
| After starting, open the returned editor_url in a browser to watch the canvas live. |
| """ |
| obs = env.reset(split=split, index=index, mode=mode, surface=surface) |
| meta = obs.metadata |
| return _json({k: meta[k] for k in ("episode_id", "task_id", "instructions", |
| "editor_url", "tools")}) |
|
|
| @mcp.tool(tags={"common"}) |
| def get_task() -> str: |
| """The current task: canvas size, assets (id, kind, aspect ratio), fonts used, and, |
| in description mode, the design brief.""" |
| return _json(env._ep().info()) |
|
|
| @mcp.tool(tags={"common"}) |
| def view_reference() -> Image: |
| """The target design you are recreating (reference mode only).""" |
| ep = env._ep() |
| if ep.mode != "reference": |
| raise ToolError("description mode has no target image; read the brief in get_task") |
| return Image(data=env.pack.target_jpeg(ep.task, 768), format="jpeg") |
|
|
| @mcp.tool(tags={"common"}) |
| def submit() -> str: |
| """Finish the episode and score the current canvas. Irreversible.""" |
| ep = env._ep() |
| grade = ep.submit("agent") |
| return grade.report() |
|
|
| @mcp.tool(tags={"tools", "html"}) |
| def view_canvas() -> Image: |
| """Render your current canvas, exactly as it will be scored.""" |
| ep = env._ep() |
| return Image(data=ep.render_jpeg(768), format="jpeg") |
|
|
| @mcp.tool(tags={"tools", "html"}) |
| def view_assets() -> Image: |
| """One image showing every asset of this task, labelled with its id, kind and |
| aspect ratio. Checkerboard = transparent.""" |
| ep = env._ep() |
| font = env.pack.fonts.file_path("Montserrat", "regular") |
| return Image(data=contact_sheet(ep.task["assets"], ep.assets, font), format="png") |
|
|
| @mcp.tool(tags={"tools", "html"}) |
| def list_fonts(query: Annotated[str | None, Field( |
| description="optional case-insensitive filter")] = None) -> str: |
| """The font families available. Any other family is rejected.""" |
| fams = env.pack.fonts.families() |
| if query: |
| fams = [f for f in fams if query.lower() in f.lower()] |
| return _json({"count": len(fams), "fonts": fams}) |
|
|
| def _register_tools_surface(self, mcp: FastMCP) -> None: |
| env = self |
| tags = {"tools"} |
|
|
| @mcp.tool(tags=tags) |
| def list_elements() -> str: |
| """Everything on the canvas, back to front, with ids, boxes and styles.""" |
| ep = env._ep() |
| return ep.doc.summary() + "\n\n" + _json(ep.doc.public()["elements"]) |
|
|
| @mcp.tool(tags=tags) |
| def add_image( |
| asset_id: Annotated[str, Field(description="asset id, e.g. a3")], |
| x: Num | None = None, y: Num | None = None, |
| width: Num | None = None, height: Num | None = None, |
| angle: Annotated[Num, Field(description="degrees clockwise")] = 0, |
| opacity: Annotated[Num, Field(ge=0, le=1)] = 1, |
| ) -> str: |
| """Place an asset on top of the canvas. Give width or height alone to keep the |
| asset's aspect ratio; omit both for a default size. Returns the new element id.""" |
| return env._edit("add_image", dict(asset_id=asset_id, x=x, y=y, width=width, |
| height=height, angle=angle, opacity=opacity)) |
|
|
| @mcp.tool(tags=tags) |
| def add_text( |
| text: Annotated[str, Field(description="the copy; use \\n for line breaks")], |
| x: Num | None = None, y: Num | None = None, |
| width: Annotated[Num | None, Field( |
| description="box width for alignment; omit to fit the text")] = None, |
| font_family: str = "Montserrat", |
| font_size: Annotated[Num, Field(description="px")] = 48, |
| color: Annotated[str, Field(description="#rrggbb")] = "#000000", |
| align: Literal["left", "center", "right"] = "left", |
| bold: bool = False, italic: bool = False, underline: bool = False, |
| uppercase: bool = False, |
| line_height: Annotated[Num, Field(description="multiple of font size")] = 1.0, |
| letter_spacing: Annotated[Num, Field(description="px added after each letter")] = 0, |
| angle: Num = 0, opacity: Annotated[Num, Field(ge=0, le=1)] = 1, |
| runs: Annotated[list[dict] | None, Field( |
| description="optional styled pieces [{text, color?, bold?, italic?}] that " |
| "concatenate to the text")] = None, |
| ) -> str: |
| """Add a text box on top of the canvas. Height follows the number of lines.""" |
| return env._edit("add_text", dict( |
| text=text, x=x, y=y, width=width, font_family=font_family, font_size=font_size, |
| color=color, align=align, bold=bold, italic=italic, underline=underline, |
| uppercase=uppercase, line_height=line_height, letter_spacing=letter_spacing, |
| angle=angle, opacity=opacity, runs=runs)) |
|
|
| @mcp.tool(tags=tags) |
| def add_shape( |
| shape: Literal["rect", "ellipse"] = "rect", |
| x: Num | None = None, y: Num | None = None, |
| width: Num | None = None, height: Num | None = None, |
| fill: Annotated[str, Field(description="#rrggbb or none")] = "#d9d9d9", |
| stroke: str | None = None, stroke_width: Num = 0, |
| radius: Annotated[Num, Field(description="corner radius (rect)")] = 0, |
| angle: Num = 0, opacity: Annotated[Num, Field(ge=0, le=1)] = 1, |
| ) -> str: |
| """Draw a rectangle or ellipse on top of the canvas.""" |
| return env._edit("add_shape", dict( |
| shape=shape, x=x, y=y, width=width, height=height, fill=fill, stroke=stroke, |
| stroke_width=stroke_width, radius=radius, angle=angle, opacity=opacity)) |
|
|
| @mcp.tool(tags=tags) |
| def update_element( |
| element_id: str, |
| properties: Annotated[dict, Field( |
| description="fields to change, e.g. {\"x\": 40, \"font_size\": 72, " |
| "\"color\": \"#ff0000\"}. Any field from add_image/add_text/" |
| "add_shape, plus visible, locked, flip_x, flip_y")], |
| ) -> str: |
| """Change an element's position, size, rotation or style.""" |
| if not isinstance(properties, dict): |
| raise ToolError("properties must be an object") |
| return env._edit("update_element", {"element_id": element_id, **properties}) |
|
|
| @mcp.tool(tags=tags) |
| def delete_element(element_id: str) -> str: |
| """Remove an element.""" |
| return env._edit("delete_element", {"element_id": element_id}) |
|
|
| @mcp.tool(tags=tags) |
| def duplicate_element(element_id: str, dx: Num = 20, dy: Num = 20) -> str: |
| """Copy an element just above itself, offset by (dx, dy).""" |
| return env._edit("duplicate_element", {"element_id": element_id, "dx": dx, "dy": dy}) |
|
|
| @mcp.tool(tags=tags) |
| def reorder_element( |
| element_id: str, |
| to: Annotated[str, Field( |
| description="front, back, forward, backward, or a layer index (0 = back)")], |
| ) -> str: |
| """Change an element's place in the stacking order.""" |
| return env._edit("reorder_element", {"element_id": element_id, "to": to}) |
|
|
| @mcp.tool(tags=tags) |
| def align_elements( |
| element_ids: list[str], |
| alignment: Literal["left", "center", "right", "top", "middle", "bottom"], |
| relative_to: Literal["canvas", "selection"] = "canvas", |
| ) -> str: |
| """Align elements to the canvas or to each other.""" |
| return env._edit("align_elements", {"element_ids": element_ids, |
| "alignment": alignment, |
| "relative_to": relative_to}) |
|
|
| @mcp.tool(tags=tags) |
| def set_background(color: Annotated[str, Field(description="#rrggbb")]) -> str: |
| """Set the canvas background colour.""" |
| return env._edit("set_background", {"color": color}) |
|
|
| @mcp.tool(tags=tags) |
| def undo() -> str: |
| """Undo the last change.""" |
| ok = env._ep().undo("agent") |
| return "undone" if ok else "nothing to undo" |
|
|
| @mcp.tool(tags=tags) |
| def redo() -> str: |
| """Redo the last undone change.""" |
| ok = env._ep().redo("agent") |
| return "redone" if ok else "nothing to redo" |
|
|
| def _register_html_surface(self, mcp: FastMCP) -> None: |
| env = self |
|
|
| @mcp.tool(tags={"html"}) |
| def write_html(html: Annotated[str, Field( |
| description='the full page: <div id="canvas" ...> with one child per element')], |
| ) -> str: |
| """Replace the canvas with this HTML. Returns what was parsed and any warnings.""" |
| ep = env._ep() |
| if ep.done: |
| raise ToolError("the episode is finished; start a new task") |
| try: |
| document, warnings = html_to_doc(html, ep.doc.width, ep.doc.height) |
| ep.apply("replace_document", {"document": document}, "agent") |
| except (HTMLProfileError, ops.OpError, ValueError) as exc: |
| raise ToolError(f"HTML rejected: {exc}") from None |
| out = {"elements": len(ep.doc.elements), "summary": ep.doc.summary()} |
| if warnings: |
| out["warnings"] = sorted(set(warnings)) |
| return _json(out) |
|
|
| @mcp.tool(tags={"html"}) |
| def read_html() -> str: |
| """The current canvas as HTML in the profile, ready to edit and write back.""" |
| return env._ep().html() |
|
|
| def _register_gui_surface(self, mcp: FastMCP) -> None: |
| env = self |
| tags = {"gui"} |
| driver = GuiDriver.get |
|
|
| def gui_ep() -> Episode: |
| ep = env._ep() |
| if ep.surface != "gui": |
| raise ToolError("GUI actions need reset(surface='gui')") |
| if ep.done: |
| raise ToolError("the episode is finished") |
| return ep |
|
|
| def mods(text: str | None) -> list[str]: |
| return [m for m in (text or "").split("+") if m] |
|
|
| def px(c) -> tuple[int, int]: |
| """The agent's [x, y] as screenshot pixels.""" |
| x, y = int(c[0]), int(c[1]) |
| if getattr(env, "_coordinates", "pixels") == "normalized": |
| x = round(x * (VIEWPORT["width"] - 1) / 1000) |
| y = round(y * (VIEWPORT["height"] - 1) / 1000) |
| return x, y |
|
|
| @mcp.tool(tags=tags) |
| async def screenshot() -> Image: |
| """Capture the editor window as the next observation.""" |
| ep = env._ep() |
| return Image(data=await driver().screenshot(ep.episode_id), format="png") |
|
|
| @mcp.tool(tags=tags) |
| async def left_click(coordinate: Coord, text: Annotated[str | None, Field( |
| description="modifier keys held during the click, e.g. shift or ctrl")] = None, |
| ) -> str: |
| """Left-click at [x, y].""" |
| ep = gui_ep() |
| await driver().click(ep.episode_id, *px(coordinate), modifiers=mods(text)) |
| return f"clicked {coordinate}" |
|
|
| @mcp.tool(tags=tags) |
| async def double_click(coordinate: Coord) -> str: |
| """Double-click at [x, y] (edits text, opens a field).""" |
| ep = gui_ep() |
| await driver().click(ep.episode_id, *px(coordinate), clicks=2) |
| return f"double-clicked {coordinate}" |
|
|
| @mcp.tool(tags=tags) |
| async def triple_click(coordinate: Coord) -> str: |
| """Triple-click at [x, y] (selects all text in a field).""" |
| ep = gui_ep() |
| await driver().click(ep.episode_id, *px(coordinate), clicks=3) |
| return f"triple-clicked {coordinate}" |
|
|
| @mcp.tool(tags=tags) |
| async def right_click(coordinate: Coord) -> str: |
| """Right-click at [x, y] (context menu).""" |
| ep = gui_ep() |
| await driver().click(ep.episode_id, *px(coordinate), button="right") |
| return f"right-clicked {coordinate}" |
|
|
| @mcp.tool(tags=tags) |
| async def mouse_move(coordinate: Coord) -> str: |
| """Move the pointer to [x, y] without clicking (shows hover states).""" |
| ep = gui_ep() |
| await driver().move(ep.episode_id, *px(coordinate)) |
| return f"moved to {coordinate}" |
|
|
| @mcp.tool(tags=tags) |
| async def left_click_drag(start_coordinate: Coord, coordinate: Coord, |
| text: str | None = None) -> str: |
| """Press at start_coordinate, drag to coordinate, release. Moves, resizes and |
| rotates elements via their handles, and drops assets onto the canvas.""" |
| ep = gui_ep() |
| await driver().drag(ep.episode_id, *px(start_coordinate), *px(coordinate), |
| modifiers=mods(text)) |
| return f"dragged {start_coordinate} -> {coordinate}" |
|
|
| @mcp.tool(tags=tags) |
| async def scroll(coordinate: Coord, |
| scroll_direction: Literal["up", "down", "left", "right"], |
| scroll_amount: Annotated[int, Field(ge=1, le=20)] = 3, |
| text: str | None = None) -> str: |
| """Scroll at [x, y]. Hold ctrl (text="ctrl") to zoom the canvas.""" |
| ep = gui_ep() |
| step = 100 * scroll_amount |
| dx = {"left": -step, "right": step}.get(scroll_direction, 0) |
| dy = {"up": -step, "down": step}.get(scroll_direction, 0) |
| await driver().scroll(ep.episode_id, *px(coordinate), dx, dy, modifiers=mods(text)) |
| return f"scrolled {scroll_direction} {scroll_amount}" |
|
|
| @mcp.tool(name="type", tags=tags) |
| async def type_text(text: str) -> str: |
| """Type text at the keyboard focus.""" |
| ep = gui_ep() |
| await driver().type_text(ep.episode_id, text) |
| return f"typed {len(text)} characters" |
|
|
| @mcp.tool(tags=tags) |
| async def key(keys: Annotated[str | list[str], Field( |
| description="a key or chord, e.g. enter, escape, delete, ctrl+z, shift+up; " |
| "a list presses each in turn")], |
| ) -> str: |
| """Press a key or key combination.""" |
| ep = gui_ep() |
| sequence = _key_sequence(keys) |
| if not sequence: |
| raise ToolError("keys is empty") |
| for combo in sequence: |
| await driver().key(ep.episode_id, combo) |
| return f"pressed {' then '.join(sequence)}" |
|
|
| @mcp.tool(tags=tags) |
| async def wait(duration: Annotated[Num, Field(ge=0, le=10)] = 1) -> str: |
| """Wait for the given seconds.""" |
| ep = gui_ep() |
| await driver().wait(ep.episode_id, float(duration)) |
| return f"waited {duration}s" |
|
|
|
|
| _TOOL_TAGS: dict[str, set[str]] = { |
| "start_task": {"common"}, "get_task": {"common"}, "view_reference": {"common"}, |
| "submit": {"common"}, |
| "view_canvas": {"tools", "html"}, "view_assets": {"tools", "html"}, |
| "list_fonts": {"tools", "html"}, |
| **{n: {"tools"} for n in ("list_elements", "add_image", "add_text", "add_shape", |
| "update_element", "delete_element", "duplicate_element", |
| "reorder_element", "align_elements", "set_background", "undo", |
| "redo")}, |
| "write_html": {"html"}, "read_html": {"html"}, |
| **{n: {"gui"} for n in ("screenshot", "left_click", "double_click", "triple_click", |
| "right_click", "mouse_move", "left_click_drag", "scroll", "type", |
| "key", "wait")}, |
| } |
|
|
