Gemini Computer Use

pip install "orgo>=0.0.47" "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
import requests
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
# Use the computer UUID (the `id` returned when you create a computer)
computer = Computer(computer_id="your-computer-id")

MODEL = 'gemini-3.8-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, re-encoded so it always matches image/png."""
    # Ask the API for inline base64. SDK versions before 0.0.47 cannot read
    # the stored-image path the API returns by default.
    response = requests.get(
        f"https://www.orgo.ai/api/computers/{computer.computer_id}/screenshot",
        params={"response_format": "base64"},
        headers={"Authorization": f"Bearer {os.environ['ORGO_API_KEY']}"},
    )
    response.raise_for_status()
    image_data = base64.b64decode(response.json()["image"])
    image = Image.open(io.BytesIO(image_data))
    png_buffer = io.BytesIO()
    image.save(png_buffer, format='PNG')
    return base64.b64encode(png_buffer.getvalue()).decode('utf-8')


# Orgo presses keys with xdotool, which takes case-sensitive X keysym names
# such as Return and BackSpace. Map the names that differ, and lowercase
# single letters so ctrl+C does not also press Shift.
KEYSYMS = {
    "enter": "Return", "return": "Return", "backspace": "BackSpace",
    "escape": "Escape", "esc": "Escape", "tab": "Tab", "space": "space",
    "delete": "Delete", "control": "ctrl", "ctrl": "ctrl", "alt": "alt",
    "shift": "shift",
}


def keysym(name: str) -> str:
    if len(name) == 1:
        return name.lower()
    return KEYSYMS.get(name.lower(), name)


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(keysym(args["key"]))

    elif name == "hotkey":
        # keys is a list of key names. Orgo takes one string joined with +, such as ctrl+c.
        computer.key("+".join(keysym(key) for key in args["keys"]))

    elif name == "scroll":
        # Orgo scrolls up or down in clicks, so treat the pixel magnitude as a hint.
        if args["direction"] not in ("up", "down"):
            return {"error": f"Unsupported scroll direction: {args['direction']}"}
        amount = max(1, args.get("magnitude_in_pixels", 300) // 100)
        # Call the API with coordinates: scroll() before SDK 0.0.47 takes none.
        response = requests.post(
            f"https://www.orgo.ai/api/computers/{computer.computer_id}/scroll",
            json={
                "x": denormalize_x(args["x"]),
                "y": denormalize_y(args["y"]),
                "direction": args["direction"],
                "amount": amount,
            },
            headers={"Authorization": f"Bearer {os.environ['ORGO_API_KEY']}"},
        )
        response.raise_for_status()

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

    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.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)

    # system_instruction and tools apply to one interaction only, so send both every time.
    interaction = client.interactions.create(
        model=MODEL,
        system_instruction=SYSTEM_PROMPT,
        previous_interaction_id=interaction.id,
        input=get_function_results(results),
        tools=TOOLS,
    )

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