Gemini computer use

pip install mandala-computer google-genai

import base64
import json
import sys

import mandala_computer
from google import genai
from mandala_computer import Client

MODEL = "gemini-3.8-flash"
TOOLS = [{"type": "computer_use", "environment": "desktop"}]
CLICKS = {"click": "click", "double_click": "double_click", "triple_click": "triple_click",
          "middle_click": "middle_click", "right_click": "right_click"}


class Desktop:
    """Runs Gemini's desktop actions on a Mandala computer."""

    def __init__(self, computer):
        self.c = computer
        self.width, self.height = computer.screen
        self.held = []  # keys pressed with key_down and not yet released

    def xy(self, args, x="x", y="y"):
        # Gemini answers on a 0-999 grid; the SDK wants screen pixels.
        return args[x] * self.width // 1000, args[y] * self.height // 1000

    def do(self, name, args):
        c = self.c
        if name in CLICKS:
            getattr(c, CLICKS[name])(*self.xy(args), *self.held)
        elif name == "mouse_down":
            c.mouse_down(*self.xy(args))
        elif name == "mouse_up":
            c.mouse_up(*self.xy(args))
        elif name == "move":
            c.move(*self.xy(args))
        elif name == "type":
            text = args["text"]
            for i in range(0, len(text), 400):  # at most 400 characters per call
                c.type(text[i : i + 400])
            if args.get("press_enter"):
                c.key("Return")
        elif name == "drag_and_drop":
            to_x, to_y = self.xy(args, "end_x", "end_y")
            from_x, from_y = self.xy(args, "start_x", "start_y")
            c.drag(to_x, to_y, from_x=from_x, from_y=from_y)
        elif name == "scroll":
            notches = max(1, round(args.get("magnitude_in_pixels", 300) / 100))
            c.scroll(*self.xy(args), direction=args["direction"], amount=notches, modifiers=tuple(self.held))
        elif name == "press_key":
            c.key(*self.held, args["key"])
        elif name == "hotkey":
            c.key(*self.held, *args["keys"])
        elif name == "key_down":
            self.held.append(args["key"])
        elif name == "key_up":
            self.held = [k for k in self.held if k != args["key"]]
        elif name == "wait":
            c.wait(min(args.get("seconds", 1), 30))
        elif name != "take_screenshot":
            raise ValueError(f"unsupported action: {name}")

    def screenshot(self):
        png = self.c.screenshot(fresh=True)
        return {"type": "image", "mime_type": "image/png", "data": base64.b64encode(png).decode()}


def confirm(decision):
    """Ask a person before an action Gemini's safety system flagged."""
    print(f"Gemini wants confirmation: {decision.get('explanation', '')}")
    return input("Continue? [y/N] ").strip().lower() == "y"


def run(computer, task, max_turns=40):
    client = genai.Client()  # reads GEMINI_API_KEY
    desktop = Desktop(computer)
    interaction = client.interactions.create(
        model=MODEL, tools=TOOLS, input=[{"type": "text", "text": task}, desktop.screenshot()]
    )
    for _ in range(max_turns):
        calls = [step for step in interaction.steps if step.type == "function_call"]
        if not calls:
            return " ".join(
                block.text
                for step in interaction.steps
                if step.type == "model_output"
                for block in step.content
                if block.type == "text"
            )

        # Run the calls in order, and stop at the first failure.
        outcomes, failed = [], False
        for call in calls:
            args = dict(call.arguments)
            decision = args.pop("safety_decision", None)
            if failed:
                outcomes.append((call, {"error": "Not executed: an earlier action failed."}, True))
                continue
            if decision and decision.get("decision") == "require_confirmation" and not confirm(decision):
                return "Stopped at a safety confirmation."
            outcome = {"safety_acknowledgement": True} if decision else {}
            try:
                desktop.do(call.name, args)
                outcomes.append((call, outcome, False))
            except (mandala_computer.MandalaError, ValueError, KeyError) as e:
                failed = True
                outcomes.append((call, {"error": f"{type(e).__name__}: {e}", **outcome}, True))

        shot = desktop.screenshot()
        results = [
            {"type": "function_result", "name": call.name, "call_id": call.id, "is_error": error,
             "result": [{"type": "text", "text": json.dumps(outcome)}, shot]}
            for call, outcome, error in outcomes
        ]
        interaction = client.interactions.create(
            model=MODEL, tools=TOOLS, previous_interaction_id=interaction.id, input=results
        )
    return "Stopped after max_turns."


if __name__ == "__main__":
    task = " ".join(sys.argv[1:]) or "Open a terminal and show today's date."
    client = Client()  # reads MANDALA_API_KEY
    with client.computers.ephemeral(template="base", name="gemini-computer-use") as computer:
        computer.wait_for_guest()
        print(run(computer, task))