OpenAI computer use

pip install openai playwright httpx

export SPECSTREAM_API_KEY="YOUR_SPECSTREAM_API_KEY"
export OPENAI_API_KEY="YOUR_OPENAI_API_KEY"

import base64
import os
import time

import httpx
from openai import OpenAI
from playwright.sync_api import sync_playwright

API = "https://api.specstream.ai/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['SPECSTREAM_API_KEY']}"}
MODEL = "gpt-6.1-sol"
TOOLS = [{"type": "computer"}]
START_URL = "https://news.ycombinator.com"
TASK = "Tell me the title of the top story on this page. Use the computer tool for UI interaction."

KEYS = {
    "ENTER": "Enter", "RETURN": "Enter", "ESC": "Escape", "ESCAPE": "Escape", "TAB": "Tab",
    "SPACE": "Space", "BACKSPACE": "Backspace", "DELETE": "Delete", "DEL": "Delete",
    "HOME": "Home", "END": "End", "PAGEUP": "PageUp", "PAGEDOWN": "PageDown",
    "UP": "ArrowUp", "DOWN": "ArrowDown", "LEFT": "ArrowLeft", "RIGHT": "ArrowRight",
    "ARROWUP": "ArrowUp", "ARROWDOWN": "ArrowDown", "ARROWLEFT": "ArrowLeft",
    "ARROWRIGHT": "ArrowRight", "CTRL": "Control", "CONTROL": "Control", "SHIFT": "Shift",
    "OPTION": "Alt", "ALT": "Alt", "META": "Meta", "CMD": "Meta", "COMMAND": "Meta",
}
BUTTONS = {"left": "left", "right": "right", "wheel": "middle"}


def key(name):
    return KEYS.get(name.upper(), name)


def handle(page, actions):
    """Run the actions of one computer_call on the page, in order."""
    for action in actions:
        kind = action.type
        if kind == "screenshot":
            continue  # one is taken after every batch
        if kind == "keypress":
            page.keyboard.press("+".join(key(k) for k in action.keys))
            continue
        if kind == "type":
            page.keyboard.type(action.text)
            continue
        if kind == "wait":
            time.sleep(2)
            continue
        # The mouse actions: keys holds the modifiers for this one action.
        held = [key(k) for k in (getattr(action, "keys", None) or [])]
        for k in held:
            page.keyboard.down(k)
        try:
            if kind == "click":
                button = BUTTONS[getattr(action, "button", "left")]
                page.mouse.click(action.x, action.y, button=button)
            elif kind == "double_click":
                page.mouse.dblclick(action.x, action.y)
            elif kind == "move":
                page.mouse.move(action.x, action.y)
            elif kind == "scroll":
                page.mouse.move(action.x, action.y)
                page.mouse.wheel(action.scroll_x, action.scroll_y)
            elif kind == "drag":
                path = [(point.x, point.y) for point in action.path]
                page.mouse.move(*path[0])
                page.mouse.down()
                for x, y in path[1:]:
                    page.mouse.move(x, y)
                page.mouse.up()
            else:
                raise ValueError(f"Unsupported action: {kind}")
        finally:
            for k in reversed(held):
                page.keyboard.up(k)


def main():
    session = httpx.post(
        f"{API}/browser/sessions", headers=HEADERS, json={"timeout_s": 600}, timeout=60
    )
    session.raise_for_status()
    session = session.json()["data"]
    print("Watch it:", session["live_url"])

    client = OpenAI()
    try:
        with sync_playwright() as p:
            browser = p.chromium.connect_over_cdp(session["connect_url"])
            context = browser.contexts[0]
            page = context.pages[0] if context.pages else context.new_page()
            page.goto(START_URL)
            response = client.responses.create(model=MODEL, tools=TOOLS, input=TASK)
            for _ in range(30):
                call = next((item for item in response.output if item.type == "computer_call"), None)
                if call is None:
                    print(response.output_text)
                    break
                handle(page, call.actions)
                image = base64.b64encode(page.screenshot(type="png")).decode()
                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,{image}",
                                "detail": "original",
                            },
                        }
                    ],
                )
    finally:
        ended = httpx.post(
            f"{API}/browser/sessions/{session['id']}/release", headers=HEADERS, timeout=60
        ).json()["data"]
        print(f"{ended['seconds']} seconds, ${ended['charged_usd']:.4f}")


if __name__ == "__main__":
    main()

python agent.py