OpenAI Computer Use

import os
import time
import requests
from openai import OpenAI
from orgo import Computer
from dotenv import load_dotenv

load_dotenv()

API = "https://www.orgo.ai/api"
HEADERS = {"Authorization": f"Bearer {os.environ['ORGO_API_KEY']}"}

MODEL = "gpt-5.6"
TOOLS = [{"type": "computer"}]

INSTRUCTIONS = """You are controlling a desktop computer.
- Always double-click desktop icons to open applications.
- Use keyboard shortcuts as single commands (e.g. 'ctrl+c', not separate keys)."""


def run_computer_task(task, computer_id=None, max_turns=20):
    """Execute a task using the OpenAI computer tool with Orgo."""

    client = OpenAI()
    # Pass a size for a new computer: SDKs before 0.0.47 default to 2 GB, which the API rejects.
    computer = Computer(computer_id=computer_id, ram=4, cpu=1)
    print(f"Computer ID: {computer.computer_id}")

    response = client.responses.create(
        model=MODEL,
        tools=TOOLS,
        instructions=INSTRUCTIONS,
        input=task,
        reasoning={"summary": "concise"},
    )

    for _ in range(max_turns):
        # Display progress
        for item in response.output:
            if item.type == "reasoning":
                for summary in getattr(item, "summary", []):
                    print(f"[thinking] {getattr(summary, 'text', '')}")

        call = next(
            (item for item in response.output if item.type == "computer_call"),
            None,
        )

        # No computer call means the task is done.
        if call is None:
            print("Task completed")
            break

        # Run the whole batch, in order.
        for action in call.actions:
            print(f"-> {action.type}")
            execute_action(computer, action)
        time.sleep(1)  # Allow the UI to settle

        screenshot = screenshot_base64(computer)

        response = client.responses.create(
            model=MODEL,
            tools=TOOLS,
            previous_response_id=response.id,
            input=[{
                "type": "computer_call_output",
                "call_id": call.call_id,
                "output": {
                    "type": "computer_screenshot",
                    "image_url": f"data:image/png;base64,{screenshot}",
                    "detail": "original",
                },
            }],
            reasoning={"summary": "concise"},
        )

    return computer


def screenshot_base64(computer):
    """The screen as base64 PNG, straight from the API."""
    response = requests.get(
        f"{API}/computers/{computer.computer_id}/screenshot",
        params={"response_format": "base64"},
        headers=HEADERS,
    )
    response.raise_for_status()
    return response.json()["image"]


def scroll(computer, x, y, direction, amount):
    """Scroll at (x, y). Orgo scrolls up or down only."""
    if direction not in ("up", "down"):
        raise ValueError(f"Unsupported scroll direction: {direction}")
    response = requests.post(
        f"{API}/computers/{computer.computer_id}/scroll",
        json={"x": x, "y": y, "direction": direction, "amount": amount},
        headers=HEADERS,
    )
    response.raise_for_status()


# Orgo presses keys with xdotool, which expects X keysym names.
XDOTOOL_KEYS = {
    "enter": "Return", "return": "Return", "esc": "Escape", "escape": "Escape",
    "backspace": "BackSpace", "tab": "Tab", "space": "space", "delete": "Delete",
    "arrowup": "Up", "arrowdown": "Down", "arrowleft": "Left", "arrowright": "Right",
    "pageup": "Page_Up", "pagedown": "Page_Down", "home": "Home", "end": "End",
}


def to_xdotool(key):
    return XDOTOOL_KEYS.get(key.lower(), key.lower())


def execute_action(computer, action):
    """Execute one computer action using Orgo."""

    match action.type:
        case "click":
            if getattr(action, "button", "left") == "right":
                computer.right_click(action.x, action.y)
            else:
                computer.left_click(action.x, action.y)

        case "double_click":
            computer.double_click(action.x, action.y)

        case "type":
            computer.type(action.text)

        case "keypress":
            keys = getattr(action, "keys", [])
            if keys:
                # One combo, e.g. ["CTRL", "C"] becomes "ctrl+c"
                computer.key("+".join(to_xdotool(k) for k in keys))

        case "scroll":
            scroll_y = getattr(action, "scroll_y", 0)
            direction = "down" if scroll_y > 0 else "up"
            scroll(computer, action.x, action.y, direction, max(1, abs(scroll_y) // 100))

        case "wait":
            computer.wait(2)

        case "screenshot":
            pass  # The loop screenshots after every batch


if __name__ == "__main__":
    computer = run_computer_task("Open a terminal and list files")

    # Always clean up
    computer.destroy()