Gemini 2.5 Computer Use登場：Googleが実現した「自律操作AI」が人間の代わりにPCを動かす時代へ

from typing import List, Tuple
import time
from playwright.sync_api import sync_playwright
from google import genai
from google.genai import types
from google.genai.types import Content, Part

# ===============================
# 設定
# ===============================

API_KEY = ""  # ← ここに自分のGemini APIキーをハードコード
MODEL_NAME = "gemini-2.5-computer-use-preview-10-2025"

SCREEN_WIDTH = 1440
SCREEN_HEIGHT = 900
TURN_LIMIT = 8

# ===============================
# 座標補正関数
# ===============================

def denormalize_x(x: int, screen_width: int) -> int:
    return int(x / 1000 * screen_width)

def denormalize_y(y: int, screen_height: int) -> int:
    return int(y / 1000 * screen_height)

# ===============================
# モデル出力（関数呼び出し）を実行
# ===============================

def execute_function_calls(candidate, page, screen_width, screen_height):
    results = []
    for part in candidate.content.parts:
        if not part.function_call:
            continue

        fc = part.function_call
        fname = fc.name
        args = fc.args or {}

        if "safety_decision" in args:
            print("⚠ Safety check:", args["safety_decision"].get("explanation", ""))
            user = input("Proceed? [y/N]: ").strip().lower()
            if user != "y":
                results.append((fname, {"error": "user_denied"}))
                continue

        try:
            if fname == "open_web_browser":
                pass
            elif fname == "navigate":
                page.goto(args["url"])
            elif fname == "click_at":
                x = denormalize_x(args["x"], screen_width)
                y = denormalize_y(args["y"], screen_height)
                page.mouse.click(x, y)
            elif fname == "type_text_at":
                x = denormalize_x(args["x"], screen_width)
                y = denormalize_y(args["y"], screen_height)
                page.mouse.click(x, y)
                page.keyboard.press("Meta+A")
                page.keyboard.press("Backspace")
                page.keyboard.type(args["text"])
                if args.get("press_enter", True):
                    page.keyboard.press("Enter")
            elif fname == "scroll_document":
                direction = args.get("direction", "down")
                delta = 1200 if direction == "down" else -1200
                page.mouse.wheel(0, delta)
            else:
                print(f"Unhandled function: {fname}")

            page.wait_for_load_state(timeout=5000)
            time.sleep(0.8)
            results.append((fname, {"status": "ok"}))

        except Exception as e:
            results.append((fname, {"error": str(e)}))
            print(f"Error executing {fname}: {e}")
    return results

# ===============================
# 実行後のスクショ＋URLを返す
# ===============================

def get_function_responses(page, results):
    screenshot_bytes = page.screenshot(type="png")
    current_url = page.url
    frs = []
    for name, result in results:
        payload = {"url": current_url}
        payload.update(result)
        frs.append(
            types.FunctionResponse(
                name=name,
                response=payload,
                parts=[types.FunctionResponsePart(
                    inline_data=types.FunctionResponseBlob(
                        mime_type="image/png", data=screenshot_bytes
                    )
                )]
            )
        )
    return frs

# ===============================
# メイン処理
# ===============================

def main():
    print("🚀 Starting Gemini 2.5 Computer Use demo (NVDA stock)...")

    # Playwright起動（headless=False で操作を可視化）
    pw = sync_playwright().start()
    browser = pw.chromium.launch(headless=False)
    context = browser.new_context(viewport={"width": SCREEN_WIDTH, "height": SCREEN_HEIGHT})
    page = context.new_page()
    page.goto("https://www.google.com")

    # Gemini APIクライアント
    client = genai.Client(api_key=API_KEY)

    # Computer Useツール設定
    config = types.GenerateContentConfig(
        tools=[types.Tool(
            computer_use=types.ComputerUse(
                environment=types.Environment.ENVIRONMENT_BROWSER
            )
        )]
    )

    # 目標を指定
    goal = (
        "Open Google Finance and read the current stock price and price change for NVIDIA (ticker NVDA). "
        "You can go directly to https://www.google.com/finance/quote/NVDA:NASDAQ. "
        "When you find the data, stop and reply with: 'NVIDIA stock: $<price> (<change>)'. "
        "Do not perform purchases or sign in."
    )

    # 初回入力
    contents: List[Content] = [
        Content(role="user", parts=[
            Part(text=goal),
            Part.from_bytes(data=page.screenshot(type="png"), mime_type="image/png"),
        ])
    ]

    # ループ実行
    for turn in range(TURN_LIMIT):
        print(f"\n--- Turn {turn + 1} ---")
        response = client.models.generate_content(
            model=MODEL_NAME,
            contents=contents,
            config=config,
        )
        candidate = response.candidates[0]
        contents.append(candidate.content)

        has_actions = any(p.function_call for p in candidate.content.parts)
        if not has_actions:
            final_text = " ".join([p.text for p in candidate.content.parts if getattr(p, "text", None)])
            print("\n✅ Result:", final_text.strip())
            break

        results = execute_function_calls(candidate, page, SCREEN_WIDTH, SCREEN_HEIGHT)
        frs = get_function_responses(page, results)
        contents.append(Content(role="user", parts=[Part(function_response=fr) for fr in frs]))

    # 終了
    browser.close()
    pw.stop()
    print("✅ Completed.")

if __name__ == "__main__":
    main()

import json
import random
import time
import subprocess
from typing import List, Dict, Any
import anthropic

API_KEY = ""  # ← ClaudeのAPIキーをここに記入
MODEL = "claude-3-7-sonnet-20250219"
BETA_FLAG = "computer-use-2025-01-24"

client = anthropic.Anthropic(api_key=API_KEY)

def run_bash(cmd: str, retries: int = 5, base_delay: float = 0.8) -> Dict[str, Any]:
    last = {"exit_code": -1, "stdout": "", "stderr": "", "http_status": None}
    for attempt in range(1, retries + 1):
        try:
            completed = subprocess.run(
                cmd, shell=True, capture_output=True, text=True, timeout=20
            )
            out = completed.stdout
            http_status = None
            if "HTTP_STATUS:" in out:
                *body_lines, status_line = out.strip().splitlines()
                if status_line.startswith("HTTP_STATUS:"):
                    http_status = int(status_line.split(":")[1])
                    out = "\n".join(body_lines)
            last = {
                "exit_code": completed.returncode,
                "stdout": out[:200000],
                "stderr": completed.stderr[:200000],
                "http_status": http_status,
            }
            if http_status and 200 <= http_status < 300:
                return last
            if http_status in (429, 500, 502, 503, 504):
                delay = base_delay * (2 ** (attempt - 1)) + random.uniform(0, 0.4)
                print(f"⚠ HTTP {http_status}. retry {attempt}/{retries} after {delay:.1f}s")
                time.sleep(delay)
                continue
            return last
        except Exception as e:
            last = {"exit_code": -1, "stdout": "", "stderr": str(e), "http_status": None}
            time.sleep(0.3)
    return last

def main():
    user_goal = (
        "Use ONE bash call to fetch raw JSON from "
        "https://query1.finance.yahoo.com/v7/finance/quote?symbols=NVDA . "
        "I will parse JSON and include a parsed object (brief) in tool_result. "
        "AFTER receiving tool_result, DO NOT call any tool again and DO NOT retry. "
        "If brief contains price data, output exactly one line: "
        "'NVDA price: $<price> (<change>, <changePercent>%)'. "
        "If brief has an 'error' field, output exactly one line: "
        "'Temporarily rate-limited. Please try again later.'"
    )

    tools = [
        {
            "type": "computer_20250124",
            "name": "computer",
            "display_width_px": 1024,
            "display_height_px": 768,
            "display_number": 1,
        },
        {
            "type": "text_editor_20250124",
            "name": "str_replace_editor",  # ← ここが正解！
        },
        {
            "type": "bash_20250124",
            "name": "bash",
        },
    ]

    messages: List[Dict[str, Any]] = [{"role": "user", "content": user_goal}]

    for _ in range(4):
        resp = client.beta.messages.create(
            model=MODEL,
            max_tokens=800,
            messages=messages,
            tools=tools,
            betas=[BETA_FLAG],
        )
        content = resp.content
        messages.append({"role": "assistant", "content": content})
        tool_calls = [b for b in content if getattr(b, "type", None) == "tool_use"]
        if not tool_calls:

            final_texts = [b.text for b in content if getattr(b, "type", None) == "text"]
            print("✅ Result:", " ".join(final_texts).strip())
            break

        tool_results_blocks: List[Dict[str, Any]] = []
        for call in tool_calls:
            name = call.name
            tool_input = call.input or {}

            if name == "bash":
                # ---- 固定curlコマンド（ヘッダ付き・HTTPコード出力）----
                url = "https://query1.finance.yahoo.com/v7/finance/quote?symbols=NVDA"
                ua = (
                    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) "
                    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0 Safari/537.36"
                )
                cmd = (
                    "curl -s --compressed "
                    f"-H 'User-Agent: {ua}' "
                    "-H 'Accept: application/json' "
                    "-H 'Accept-Language: en-US,en;q=0.9' "
                    "-H 'Connection: keep-alive' "
                    f"'{url}' "
                    r"-w '\nHTTP_STATUS:%{http_code}\n'"
                )
                result = run_bash(cmd)

                # ---- HTTPステータス確認 ----
                status = result.get("http_status")
                body = result.get("stdout", "")

                if not status or status < 200 or status >= 300:
                    brief = {"error": "upstream_error", "http_status": status, "note": "rate-limited or blocked"}
                else:
                    try:
                        data = json.loads(body)
                        res = (data.get("quoteResponse") or {}).get("result") or []
                        if res:
                            q = res[0]
                            brief = {
                                "symbol": q.get("symbol"),
                                "price": q.get("regularMarketPrice"),
                                "change": q.get("regularMarketChange"),
                                "changePercent": q.get("regularMarketChangePercent"),
                                "currency": q.get("currency"),
                                "marketState": q.get("marketState"),
                            }
                        else:
                            brief = {"error": "empty_result", "http_status": status}
                    except Exception as e:
                        brief = {"error": f"parse_failed: {e}", "http_status": status}

                # Claudeへのtool_result（briefのみ）
                tool_results_blocks.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": [{"type": "text", "text": json.dumps(brief)}],
                })

        # ---- Claudeへtool_resultを返して最終回答を得る ----
        messages.append({"role": "user", "content": tool_results_blocks})
        resp_final = client.beta.messages.create(
            model=MODEL,
            max_tokens=800,
            messages=messages,
            tools=tools,
            betas=[BETA_FLAG],
        )

        final_texts = [b.text for b in resp_final.content if getattr(b, "type", None) == "text"]
        final_text = " ".join(final_texts).strip()

        if final_text:
            print("✅ Result:", final_text)
        else:
            # Claudeが応答しなかった場合のフォールバック
            try:
                last_brief = json.loads(tool_results_blocks[-1]["content"][0]["text"])
                if last_brief.get("price") is not None:
                    p = last_brief["price"]
                    ch = last_brief.get("change")
                    chp = last_brief.get("changePercent")
                    print(f"✅ Fallback: NVDA price: ${p} ({ch}, {chp}%)")
                else:
                    print("⚠ Claude returned no text and brief had no price.")
            except Exception as e:
                print(f"⚠ Claude returned no text. Fallback failed: {e}")
        break

if __name__ == "__main__":
    main()

⚠ HTTP 429. retry 1/5 after 1.1s
⚠ HTTP 429. retry 2/5 after 1.7s
⚠ HTTP 429. retry 3/5 after 3.5s
⚠ HTTP 429. retry 4/5 after 6.6s
⚠ HTTP 429. retry 5/5 after 13.2s
✅ Result: Temporarily rate-limited. Please try again later.