Computer Use Advanced — паттерны для сложных сценариев автоматизации

Claude получает задачу
→ Запрашивает скриншот (screenshot)
→ Анализирует что видит
→ Решает какое действие выполнить
→ Выполняет действие (click, type, key)
→ Снова берёт скриншот — ПРОВЕРЯЕТ результат
→ Повторяет пока задача не выполнена

import anthropic
import base64
import time
from pathlib import Path

client = anthropic.Anthropic()

def take_screenshot() -> str:
    """
    В продакшне: берёт скриншот через scrot/PIL/mss и кодирует в base64.
    Здесь — заглушка, в реальности замените на вашу реализацию.
    """
    # pip install mss Pillow
    import mss
    import io
    from PIL import Image
    
    with mss.mss() as sct:
        monitor = {"top": 0, "left": 0, "width": 1280, "height": 800}
        screenshot = sct.grab(monitor)
        img = Image.frombytes("RGB", screenshot.size, screenshot.bgra, "raw", "BGRX")
        buffer = io.BytesIO()
        img.save(buffer, format="PNG")
        return base64.standard_b64encode(buffer.getvalue()).decode("utf-8")


def image_block(b64: str) -> dict:
    return {
        "type": "image",
        "source": {"type": "base64", "media_type": "image/png", "data": b64},
    }


def computer_use_with_retry(
    task: str,
    max_steps: int = 30,
    max_retries_per_step: int = 3,
    pause_between_steps: float = 1.0,
) -> dict:
    """
    Запускает Computer Use задачу с retry-логикой на каждом шаге.
    
    Паттерн восстановления:
    - Если Claude сообщает об ошибке → делаем скриншот, передаём контекст
    - Если элемент не найден → пробуем прокрутку или ждём загрузки
    - Если стек ошибок растёт → эскалируем (останавливаем, логируем)
    """
    
    messages = []
    # Современный набор инструментов: без beta-заголовка и без размеров экрана.
    # Скриншоты, которые мы возвращаем, должны сами укладываться в лимиты модели по размеру.
    tools = [{"type": "computer_toolset_20260801"}]
    
    # Начальный скриншот — "посмотри что сейчас на экране"
    initial_screenshot = take_screenshot()
    
    messages.append({
        "role": "user",
        "content": [
            image_block(initial_screenshot),
            {
                "type": "text",
                "text": f"""Вот текущее состояние экрана. Выполни следующую задачу:

{task}

ВАЖНЫЕ ПРАВИЛА:
1. После каждого клика или ввода текста — сделай скриншот для проверки
2. Если элемент не виден — сначала прокрути, потом ищи
3. Если что-то пошло не так — опиши проблему в деталях перед следующей попыткой
4. Завершая задачу — сообщи TASK_COMPLETE и кратко опиши что было сделано""",
            },
        ],
    })
    
    step_count = 0
    retry_context = []
    
    while step_count < max_steps:
        step_count += 1
        
        try:
            response = client.messages.create(
                model="claude-opus-5-5",   # актуальные модели: страница «Актуальное сейчас»
                max_tokens=4096,
                tools=tools,
                messages=messages,
            )
            
            # Проверяем завершение
            if response.stop_reason == "end_turn":
                final_text = " ".join(
                    block.text for block in response.content 
                    if block.type == "text"
                )
                if "TASK_COMPLETE" in final_text:
                    return {"status": "success", "steps": step_count, "summary": final_text}
                # Завершился без нашего маркера — тоже ок
                return {"status": "complete", "steps": step_count, "summary": final_text}
            
            # Обрабатываем tool calls. Модель может прислать несколько действий
            # подряд: выполняем по порядку и останавливаемся на первом сбое.
            tool_results = []
            failed = False
            for block in response.content:
                if block.type == "tool_use" and getattr(block, "toolset_name", None) == "computer":
                    result = {
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "toolset_name": "computer",
                    }
                    if failed:
                        result["is_error"] = True
                        result["content"] = "Not executed: an earlier computer action in this turn failed."
                    else:
                        try:
                            # block.name — это само действие: screenshot, left_click, type, key, scroll...
                            # В продакшне здесь ваш код управления мышью/клавиатурой
                            execute_computer_action(block.name, block.input)
                            time.sleep(pause_between_steps)
                            if block.name in ("screenshot", "zoom"):
                                # КЛЮЧЕВОЙ ПАТТЕРН: свежий скриншот по запросу модели.
                                # Правило «после действия — скриншот» задано в промпте выше.
                                # Для zoom верните вырезанную область, здесь для краткости — весь экран.
                                result["content"] = [image_block(take_screenshot())]
                            else:
                                result["content"] = [{"type": "text", "text": "OK"}]
                        except Exception as action_error:
                            failed = True
                            result["is_error"] = True
                            result["content"] = str(action_error)
                    tool_results.append(result)
            
            # Добавляем в историю и продолжаем
            messages.append({"role": "assistant", "content": response.content})
            if tool_results:
                messages.append({"role": "user", "content": tool_results})
                
        except Exception as e:
            retry_context.append(str(e))
            if len(retry_context) >= max_retries_per_step:
                return {
                    "status": "error",
                    "steps": step_count,
                    "errors": retry_context,
                }
            # Добавляем контекст ошибки и пробуем снова
            error_screenshot = take_screenshot()
            messages.append({
                "role": "user",
                "content": [
                    image_block(error_screenshot),
                    {"type": "text", "text": f"Произошла ошибка: {str(e)}. Вот текущий экран. Попробуй другой подход."},
                ],
            })
    
    return {"status": "max_steps_reached", "steps": step_count}


# Названия клавиш у модели (Return, Escape) отличаются от названий в pyautogui (enter, esc)
KEY_MAP = {"return": "enter", "escape": "esc", "page_down": "pagedown", "page_up": "pageup"}


def execute_computer_action(name: str, params: dict) -> None:
    """
    Заглушка — в реальности здесь PyAutoGUI, xdotool или нативный VNC-клиент.
    Имена действий и поля берём из документации инструмента computer.
    """
    # pip install pyautogui
    import pyautogui
    
    if name in ("screenshot", "zoom"):
        return  # скриншот мы берём отдельно
    elif name == "left_click":
        x, y = params["coordinate"]
        pyautogui.click(x, y)
    elif name == "double_click":
        x, y = params["coordinate"]
        pyautogui.doubleClick(x, y)
    elif name == "type":
        pyautogui.write(params["text"], interval=0.05)
    elif name == "key":
        keys = [KEY_MAP.get(k.lower(), k.lower()) for k in params["text"].split("+")]
        pyautogui.hotkey(*keys)
    elif name == "scroll":
        direction = params.get("scroll_direction", "down")
        amount = params.get("scroll_amount", 3)
        x, y = params.get("coordinate") or pyautogui.position()
        pyautogui.scroll(amount if direction == "up" else -amount, x=x, y=y)
    elif name == "wait":
        time.sleep(params.get("duration", 1))
    else:
        raise ValueError(f"Действие не поддерживается: {name}")

# Всегда фиксируйте виртуальное разрешение для Computer Use
VIRTUAL_WIDTH = 1280
VIRTUAL_HEIGHT = 800

# При съёмке реального экрана — масштабируйте вниз
# При передаче координат от Claude обратно — масштабируйте вверх

def normalize_coordinates(x: int, y: int, real_width: int, real_height: int) -> tuple:
    """Переводим виртуальные координаты Claude в реальные."""
    real_x = int(x * real_width / VIRTUAL_WIDTH)
    real_y = int(y * real_height / VIRTUAL_HEIGHT)
    return real_x, real_y

# Запуск headless с возможностью VNC-наблюдения
docker run -d \
  -e DISPLAY=:1 \
  -p 5900:5900 \
  --name cu-sandbox \
  my-computer-use-image

# Внутри контейнера
Xvfb :1 -screen 0 1280x800x24 &
x11vnc -display :1 -nopw -listen 0.0.0.0 -forever &

# Правильная архитектура для продакшна
# 1. Отдельный Docker-контейнер с VNC
# 2. В контейнере — чистый пользователь без доступа к production данным
# 3. Только нужные приложения установлены
# 4. Результаты работы передаются через volume mount, не через буфер обмена

from playwright.async_api import async_playwright

async def hybrid_automation():
    # Шаг 1: Computer Use для сложного логина (SSO + 2FA)
    login_result = computer_use_with_retry(
        task="Открой сайт example.com, нажми 'Войти через корпоративный аккаунт', "
             "введи логин [email protected], дождись 2FA SMS и введи код",
        max_steps=20
    )
    
    # Шаг 2: Playwright подхватывает уже аутентифицированную сессию
    # (передаём cookies или storage state из браузера)
    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp("http://localhost:9222")
        page = browser.contexts[0].pages[0]  # берём уже открытую страницу
        
        # Теперь быстрая структурированная работа через селекторы
        rows = await page.query_selector_all("table.reports tr")
        data = []
        for row in rows:
            cells = await row.query_selector_all("td")
            data.append([await cell.inner_text() for cell in cells])
        
        return data

# Устанавливаем зависимости
pip install anthropic mss Pillow pyautogui

# Для Linux/Docker: устанавливаем Xvfb + x11vnc
# sudo apt-get install xvfb x11vnc

# Запускаем виртуальный дисплей (только для Linux без монитора)
export DISPLAY=:1
Xvfb :1 -screen 0 1280x800x24 &

import mss
import io
import base64
from PIL import Image

VIRTUAL_WIDTH = 1280
VIRTUAL_HEIGHT = 800

def capture_screen(region=None) -> str:
    """Снимает скриншот и возвращает base64."""
    with mss.mss() as sct:
        monitor = region or {"top": 0, "left": 0, "width": VIRTUAL_WIDTH, "height": VIRTUAL_HEIGHT}
        screenshot = sct.grab(monitor)
        img = Image.frombytes("RGB", screenshot.size, screenshot.bgra, "raw", "BGRX")
        img = img.resize((VIRTUAL_WIDTH, VIRTUAL_HEIGHT))
        buffer = io.BytesIO()
        img.save(buffer, format="PNG", optimize=True)
        return base64.standard_b64encode(buffer.getvalue()).decode("utf-8")

import pyautogui
import time

pyautogui.FAILSAFE = True  # Мышь в угол = остановка

# Названия клавиш у модели (Return, Escape) отличаются от названий в pyautogui (enter, esc)
KEY_MAP = {"return": "enter", "escape": "esc"}

def execute_action(name: str, params: dict) -> None:
    """Выполняет одно действие модели. При сбое бросает исключение: вызывающий код вернёт is_error."""
    if name in ("screenshot", "zoom"):
        return  # скриншот берёт вызывающий код
    if name == "left_click":
        x, y = params["coordinate"]
        pyautogui.click(x, y)
    elif name == "double_click":
        x, y = params["coordinate"]
        pyautogui.doubleClick(x, y)
    elif name == "type":
        time.sleep(0.2)  # Небольшая пауза перед вводом
        pyautogui.write(params["text"], interval=0.03)
    elif name == "key":
        keys = [KEY_MAP.get(k.lower(), k.lower()) for k in params["text"].split("+")]
        pyautogui.hotkey(*keys)
    elif name == "scroll":
        direction = params.get("scroll_direction", "down")
        amount = params.get("scroll_amount", 3)
        x, y = params.get("coordinate") or pyautogui.position()
        pyautogui.scroll(-amount if direction == "down" else amount, x=x, y=y)
    elif name == "wait":
        time.sleep(params.get("duration", 1))
    else:
        raise ValueError(f"Действие не поддерживается: {name}")
    time.sleep(0.5)  # Ждём отклика UI

import anthropic
import time
from cu_screenshot import capture_screen
from executor import execute_action

def screenshot_block() -> dict:
    return {
        "type": "image",
        "source": {"type": "base64", "media_type": "image/png", "data": capture_screen()},
    }

def run_desktop_task(task_description: str, app_name: str):
    client = anthropic.Anthropic()
    
    # Без beta-заголовка и без размеров экрана (современный набор инструментов)
    tools = [{"type": "computer_toolset_20260801"}]
    
    system_prompt = f"""Ты автоматизируешь задачу в приложении {app_name}.
    
ПРАВИЛА:
- После каждого действия делай скриншот для проверки
- Если видишь модальное окно или уведомление — закрой его
- Если элемент не найден — прокрути страницу, подожди 2 секунды, попробуй ещё раз
- Когда задача выполнена — напиши TASK_COMPLETE и опиши что сделал
- Если задача невозможна — напиши TASK_FAILED и объясни причину"""
    
    messages = [{
        "role": "user",
        "content": [
            screenshot_block(),
            {"type": "text", "text": f"Выполни задачу: {task_description}"},
        ],
    }]
    
    for step in range(40):  # Максимум 40 шагов
        response = client.messages.create(
            model="claude-opus-5-5",   # актуальные модели: страница «Актуальное сейчас»
            max_tokens=4096,
            system=system_prompt,
            tools=tools,
            messages=messages,
        )
        
        if response.stop_reason == "end_turn":
            final = " ".join(b.text for b in response.content if b.type == "text")
            print(f"Завершено за {step+1} шагов: {final}")
            return "TASK_COMPLETE" in final
        
        tool_results = []
        failed = False
        for block in response.content:
            if block.type == "tool_use" and getattr(block, "toolset_name", None) == "computer":
                result = {"type": "tool_result", "tool_use_id": block.id, "toolset_name": "computer"}
                if failed:
                    result["is_error"] = True
                    result["content"] = "Not executed: an earlier computer action in this turn failed."
                else:
                    try:
                        execute_action(block.name, block.input)
                        time.sleep(1.0)
                        if block.name in ("screenshot", "zoom"):
                            result["content"] = [screenshot_block()]
                        else:
                            result["content"] = [{"type": "text", "text": "OK"}]
                    except Exception as e:
                        failed = True
                        print(f"Ошибка выполнения действия {block.name}: {e}")
                        result["is_error"] = True
                        result["content"] = str(e)
                tool_results.append(result)
        
        messages.append({"role": "assistant", "content": response.content})
        if tool_results:
            messages.append({"role": "user", "content": tool_results})
    
    print("Превышено максимальное количество шагов")
    return False

# Использование
run_desktop_task(
    task_description="Открой меню Файл → Отчёты → Ежедневный. Выбери вчерашнюю дату. Нажми Экспорт → CSV. Сохрани в папку /tmp/reports/",
    app_name="LegacyAccountingApp"
)

import json
from datetime import datetime
from pathlib import Path

def log_session(task: str, success: bool, steps: int, errors: list):
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "task": task[:100],
        "success": success,
        "steps": steps,
        "errors": errors,
    }
    log_path = Path("logs/computer_use.jsonl")
    log_path.parent.mkdir(exist_ok=True)
    with open(log_path, "a") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")