Gemini Computer Use

pip install "orgo" "google-genai>=2.7.0" pillow python-dotenv

ORGO_API_KEY=your_orgo_api_key
GEMINI_API_KEY=your_gemini_api_key

export ORGO_API_KEY=your_orgo_api_key
export GEMINI_API_KEY=your_gemini_api_key

import os
import io
import json
import time
import base64
from google import genai
from orgo import Computer
from PIL import Image
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Initialize Gemini client
client = genai.Client(api_key=os.environ.get('GEMINI_API_KEY'))

# Connect to your Orgo computer
# Get your computer_id from https://orgo.ai/workspaces
computer = Computer(computer_id="your-computer-id")

MODEL = 'gemini-3.7-flash'

TOOLS = [{
    "type": "computer_use",
    "environment": "desktop",
    "enable_prompt_injection_detection": True,
}]

# Screen resolution
SCREEN_WIDTH = 1280
SCREEN_HEIGHT = 720

# System prompt with Ubuntu-specific instructions
SYSTEM_PROMPT = f"""You are controlling an Ubuntu Linux virtual machine with a display resolution of {SCREEN_WIDTH}x{SCREEN_HEIGHT}.

<UBUNTU_DESKTOP_GUIDELINES>
* CRITICAL: When opening applications or files on the Ubuntu desktop, you MUST USE DOUBLE-CLICK, not single-click
* Single-click only selects desktop icons but DOES NOT open them
* Desktop interactions:
  - Desktop icons (apps/folders): DOUBLE-CLICK to open
  - Menu items: SINGLE-CLICK to select
  - Taskbar/launcher icons: SINGLE-CLICK to open
  - Window buttons (close/minimize/maximize): SINGLE-CLICK
  - File browser items: DOUBLE-CLICK to open
* When you need to submit or confirm, use the 'Enter' key
</UBUNTU_DESKTOP_GUIDELINES>

<IMPORTANT_NOTES>
* Wait for applications to load before acting on them
* Batch multiple actions together when possible before checking the result
</IMPORTANT_NOTES>"""


def denormalize_x(x: int) -> int:
    """Convert a normalized x coordinate (0-999) to an actual pixel."""
    return int(x / 1000 * SCREEN_WIDTH)


def denormalize_y(y: int) -> int:
    """Convert a normalized y coordinate (0-999) to an actual pixel."""
    return int(y / 1000 * SCREEN_HEIGHT)


def get_screenshot_png() -> str:
    """Get a base64 PNG screenshot. Orgo returns JPEG, Gemini requires PNG."""
    jpeg_data = base64.b64decode(computer.screenshot_base64())
    image = Image.open(io.BytesIO(jpeg_data))
    png_buffer = io.BytesIO()
    image.save(png_buffer, format='PNG')
    return base64.b64encode(png_buffer.getvalue()).decode('utf-8')


def execute_action(name: str, args: dict) -> dict:
    """Run one Gemini action on the computer. Returns extra result fields."""
    if name in ("click", "double_click", "right_click"):
        x, y = denormalize_x(args["x"]), denormalize_y(args["y"])
        {
            "click": computer.left_click,
            "double_click": computer.double_click,
            "right_click": computer.right_click,
        }[name](x, y)

    elif name == "type":
        computer.type(args["text"])
        if args.get("press_enter", False):
            computer.key("Return")

    elif name == "press_key":
        computer.key(args["key"])

    elif name == "hotkey":
        computer.key("+".join(args["keys"]).lower())

    elif name == "scroll":
        # Orgo scrolls in clicks, so treat the pixel magnitude as a hint.
        amount = max(1, args.get("magnitude_in_pixels", 300) // 100)
        computer.scroll(args["direction"], amount)

    elif name == "wait":
        computer.wait(args.get("seconds", 2))

    elif name == "take_screenshot":
        pass  # Every result already carries a fresh screenshot

    else:
        return {"error": f"Unsupported action: {name}"}

    return {}


def execute_function_calls(interaction) -> list:
    """Run every function call in the turn, in order."""
    results = []

    for step in interaction.steps:
        if step.type != "function_call":
            continue

        print(f"  -> {step.name}: {step.arguments.get('intent', '')}")

        try:
            result = execute_action(step.name, step.arguments)
        except Exception as error:
            print(f"    Error: {error}")
            result = {"error": str(error)}

        time.sleep(1)  # Wait for the UI to update
        results.append((step.name, step.call_id, result))

    return results


def get_function_results(results: list) -> list:
    """Return one result per call, each carrying the new screen."""
    screenshot_png = get_screenshot_png()

    return [
        {
            "type": "function_result",
            "name": name,
            "call_id": call_id,
            "result": [
                {"type": "text", "text": json.dumps({"status": "completed", **result})},
                {"type": "image", "data": screenshot_png, "mime_type": "image/png"},
            ],
        }
        for name, call_id, result in results
    ]


# Define the task
task = "Open Chrome and search for 'gemini ai'"
print(f"Task: {task}\n")

interaction = client.interactions.create(
    model=MODEL,
    system_instruction=SYSTEM_PROMPT,
    input=[
        {"type": "text", "text": task},
        {"type": "image", "data": get_screenshot_png(), "mime_type": "image/png"},
    ],
    tools=TOOLS,
)

# Agent loop
for turn in range(20):
    print(f"\n--- Turn {turn + 1} ---")

    has_function_calls = any(
        step.type == "function_call" for step in interaction.steps
    )

    if not has_function_calls:
        text = " ".join(
            block.text
            for step in interaction.steps if step.type == "model_output"
            for block in step.content if block.type == "text"
        )
        print(f"Agent finished: {text}")
        break

    results = execute_function_calls(interaction)

    interaction = client.interactions.create(
        model=MODEL,
        previous_interaction_id=interaction.id,
        input=get_function_results(results),
        tools=TOOLS,
    )

# The computer keeps running. Call computer.destroy() to tear it down.