Claude computer use

pip install anthropic playwright httpx

export SPECSTREAM_API_KEY="YOUR_SPECSTREAM_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_API_KEY"

import base64
import os
import time

import anthropic
import httpx
from playwright.sync_api import sync_playwright

API = "https://api.specstream.ai/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['SPECSTREAM_API_KEY']}"}
MODEL = "claude-opus-5-5"
TASK = "Open https://news.ycombinator.com and tell me the title of the top story."

TOOLS = [
    # The session's screenshots are already full size, so zoom is switched off.
    {"type": "computer_toolset_20260801", "configs": {"zoom": {"enabled": False}}},
    {
        "name": "goto",
        "description": "Open a URL in the browser. The screenshots show the page only, with no address bar.",
        "input_schema": {
            "type": "object",
            "properties": {"url": {"type": "string"}},
            "required": ["url"],
        },
    },
]

KEYS = {
    "return": "Enter", "enter": "Enter", "kp_enter": "Enter", "escape": "Escape", "esc": "Escape",
    "tab": "Tab", "space": "Space", "backspace": "Backspace", "delete": "Delete",
    "home": "Home", "end": "End", "page_up": "PageUp", "prior": "PageUp",
    "page_down": "PageDown", "next": "PageDown", "up": "ArrowUp", "down": "ArrowDown",
    "left": "ArrowLeft", "right": "ArrowRight", "ctrl": "Control", "control": "Control",
    "alt": "Alt", "shift": "Shift", "super": "Meta", "cmd": "Meta", "meta": "Meta",
}
CLICKS = {
    "left_click": ("left", 1), "right_click": ("right", 1), "middle_click": ("middle", 1),
    "double_click": ("left", 2), "triple_click": ("left", 3),
}
SCROLL = {"up": (0, -100), "down": (0, 100), "left": (-100, 0), "right": (100, 0)}


def keys(text):
    return [KEYS.get(part.lower(), part) for part in text.split("+")]


def screenshot(page):
    data = base64.b64encode(page.screenshot(type="png")).decode()
    return [{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": data}}]


def act(page, name, args, pos):
    """Run one member of the computer toolset on the page and return its result."""
    if name == "screenshot":
        return screenshot(page)
    if name == "cursor_position":
        return f"X={pos[0]}, Y={pos[1]}"
    if name == "type":
        page.keyboard.type(args["text"])
    elif name == "key":
        for _ in range(args.get("repeat", 1)):
            page.keyboard.press("+".join(keys(args["text"])))
    elif name == "hold_key":
        held = keys(args["text"])
        for key in held:
            page.keyboard.down(key)
        time.sleep(args["duration"])
        for key in reversed(held):
            page.keyboard.up(key)
    elif name == "wait":
        time.sleep(args["duration"])
    elif name == "left_mouse_down":
        page.mouse.down()
    elif name == "left_mouse_up":
        page.mouse.up()
    else:
        # The mouse members: text holds the modifier keys for this one action.
        held = keys(args["text"]) if args.get("text") else []
        for key in held:
            page.keyboard.down(key)
        try:
            if name == "left_click_drag":
                page.mouse.move(*args["start_coordinate"])
                page.mouse.down()
                page.mouse.move(*args["coordinate"], steps=10)
                page.mouse.up()
                pos[:] = args["coordinate"]
            else:
                if args.get("coordinate"):
                    pos[:] = args["coordinate"]
                page.mouse.move(*pos)
                if name in CLICKS:
                    button, count = CLICKS[name]
                    page.mouse.click(*pos, button=button, click_count=count)
                elif name == "scroll":
                    dx, dy = SCROLL[args["scroll_direction"]]
                    amount = args.get("scroll_amount", 3)
                    page.mouse.wheel(dx * amount, dy * amount)
                elif name != "mouse_move":
                    raise ValueError(f"{name} is not implemented")
        finally:
            for key in reversed(held):
                page.keyboard.up(key)
    return "OK"


def run_calls(page, response, pos):
    """Run every tool_use block in order and answer each one."""
    results, failed = [], False
    for block in response.content:
        if block.type != "tool_use":
            continue
        computer = getattr(block, "toolset_name", None) == "computer"
        result = {"type": "tool_result", "tool_use_id": block.id}
        if computer:
            result["toolset_name"] = "computer"
        if failed and computer:
            result["is_error"] = True
            result["content"] = "Not executed: an earlier computer action in this turn failed."
        else:
            try:
                if computer:
                    result["content"] = act(page, block.name, block.input, pos)
                else:
                    page.goto(block.input["url"])
                    result["content"] = "OK"
            except Exception as error:
                failed = True
                result["is_error"] = True
                result["content"] = str(error)
        results.append(result)
    return results


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 = anthropic.Anthropic()
    messages = [{"role": "user", "content": TASK}]
    pos = [0, 0]
    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()
            for _ in range(30):
                response = client.messages.create(
                    model=MODEL, max_tokens=16000, tools=TOOLS, messages=messages
                )
                messages.append({"role": "assistant", "content": response.content})
                results = run_calls(page, response, pos)
                if not results:
                    print("".join(b.text for b in response.content if b.type == "text"))
                    break
                messages.append({"role": "user", "content": results})
    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