Computer Use avanzado: patrones para escenarios de automatización complejos

Claude recibe la tarea
→ Pide una captura (screenshot)
→ Analiza lo que ve
→ Decide qué acción ejecutar
→ Ejecuta la acción (click, type, key)
→ Toma otra captura: REVISA el resultado
→ Repite hasta terminar la tarea

import anthropic
import base64
import time
from pathlib import Path

client = anthropic.Anthropic()

def take_screenshot() -> str:
    """
    En producción: toma una captura con scrot/PIL/mss y la codifica en base64.
    Aquí es un ejemplo; en la práctica, reemplázalo con tu implementación.
    """
    # 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:
    """
    Ejecuta una tarea de Computer Use con lógica de reintento en cada paso.
    
    Patrón de recuperación:
    - Si Claude reporta un error → tomamos una captura y le pasamos el contexto
    - Si no encuentra el elemento → probamos desplazarnos o esperar a que cargue
    - Si los errores se acumulan → escalamos (detenemos y registramos)
    """
    
    messages = []
    # Conjunto de herramientas actual: sin encabezado beta y sin tamaño de pantalla.
    # Las capturas que devolvemos deben respetar por sí mismas los límites de tamaño del modelo.
    tools = [{"type": "computer_toolset_20260801"}]
    
    # Captura inicial: "mira lo que hay ahora en la pantalla"
    initial_screenshot = take_screenshot()
    
    messages.append({
        "role": "user",
        "content": [
            image_block(initial_screenshot),
            {
                "type": "text",
                "text": f"""Este es el estado actual de la pantalla. Ejecuta la siguiente tarea:

{task}

REGLAS IMPORTANTES:
1. Después de cada clic o de escribir texto, toma una captura para revisar
2. Si el elemento no se ve, primero desplázate y luego búscalo
3. Si algo salió mal, describe el problema a detalle antes del siguiente intento
4. Al terminar la tarea, escribe TASK_COMPLETE y describe en breve lo que hiciste""",
            },
        ],
    })
    
    step_count = 0
    retry_context = []
    
    while step_count < max_steps:
        step_count += 1
        
        try:
            response = client.messages.create(
                model="claude-opus-5-5",   # modelos actuales: página "Lo vigente"
                max_tokens=4096,
                tools=tools,
                messages=messages,
            )
            
            # Revisamos si terminó
            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}
                # Terminó sin nuestro marcador: también está bien
                return {"status": "complete", "steps": step_count, "summary": final_text}
            
            # Procesamos los tool calls. El modelo puede mandar varias acciones
            # seguidas: las ejecutamos en orden y nos detenemos en la primera falla.
            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 es la acción misma: screenshot, left_click, type, key, scroll...
                            # En producción aquí va tu código para controlar el mouse y el teclado
                            execute_computer_action(block.name, block.input)
                            time.sleep(pause_between_steps)
                            if block.name in ("screenshot", "zoom"):
                                # PATRÓN CLAVE: captura fresca cuando el modelo la pide.
                                # La regla "después de una acción, captura" está en el prompt de arriba.
                                # Para zoom devuelve la zona recortada; aquí, por brevedad, la pantalla completa.
                                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)
            
            # Lo agregamos al historial y seguimos
            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,
                }
            # Agregamos el contexto del error y volvemos a intentar
            error_screenshot = take_screenshot()
            messages.append({
                "role": "user",
                "content": [
                    image_block(error_screenshot),
                    {"type": "text", "text": f"Ocurrió un error: {str(e)}. Esta es la pantalla actual. Prueba otro enfoque."},
                ],
            })
    
    return {"status": "max_steps_reached", "steps": step_count}


# Los nombres de teclas del modelo (Return, Escape) son distintos a los de pyautogui (enter, esc)
KEY_MAP = {"return": "enter", "escape": "esc", "page_down": "pagedown", "page_up": "pageup"}


def execute_computer_action(name: str, params: dict) -> None:
    """
    Ejemplo: en la práctica aquí va PyAutoGUI, xdotool o un cliente VNC nativo.
    Los nombres de las acciones y los campos salen de la documentación de la herramienta computer.
    """
    # pip install pyautogui
    import pyautogui
    
    if name in ("screenshot", "zoom"):
        return  # la captura la tomamos por separado
    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"Acción no soportada: {name}")

# Fija siempre una resolución virtual para Computer Use
VIRTUAL_WIDTH = 1280
VIRTUAL_HEIGHT = 800

# Al capturar la pantalla real, reescala hacia abajo
# Al devolver las coordenadas de Claude, reescala hacia arriba

def normalize_coordinates(x: int, y: int, real_width: int, real_height: int) -> tuple:
    """Convertimos las coordenadas virtuales de Claude en coordenadas reales."""
    real_x = int(x * real_width / VIRTUAL_WIDTH)
    real_y = int(y * real_height / VIRTUAL_HEIGHT)
    return real_x, real_y

# Arranque headless con opción de observar por VNC
docker run -d \
  -e DISPLAY=:1 \
  -p 5900:5900 \
  --name cu-sandbox \
  my-computer-use-image

# Dentro del contenedor
Xvfb :1 -screen 0 1280x800x24 &
x11vnc -display :1 -nopw -listen 0.0.0.0 -forever &

# Arquitectura correcta para producción
# 1. Un contenedor Docker aparte con VNC
# 2. Dentro del contenedor, un usuario limpio sin acceso a datos de producción
# 3. Solo las aplicaciones necesarias instaladas
# 4. Los resultados se pasan por un volume mount, no por el portapapeles

from playwright.async_api import async_playwright

async def hybrid_automation():
    # Paso 1: Computer Use para el login complicado (SSO + 2FA)
    login_result = computer_use_with_retry(
        task="Abre el sitio example.com, haz clic en 'Iniciar sesión con cuenta corporativa', "
             "escribe el usuario [email protected], espera el SMS de 2FA y escribe el código",
        max_steps=20
    )
    
    # Paso 2: Playwright toma la sesión ya autenticada
    # (pasamos las cookies o el storage state del navegador)
    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp("http://localhost:9222")
        page = browser.contexts[0].pages[0]  # tomamos la página que ya está abierta
        
        # Ahora, trabajo rápido y estructurado con selectores
        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

# Instalamos dependencias
pip install anthropic mss Pillow pyautogui

# Para Linux/Docker: instalamos Xvfb + x11vnc
# sudo apt-get install xvfb x11vnc

# Arrancamos una pantalla virtual (solo para Linux sin monitor)
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:
    """Toma una captura y la devuelve en 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  # Mouse a una esquina = detener

# Los nombres de teclas del modelo (Return, Escape) son distintos a los de pyautogui (enter, esc)
KEY_MAP = {"return": "enter", "escape": "esc"}

def execute_action(name: str, params: dict) -> None:
    """Ejecuta una acción del modelo. Si falla, lanza una excepción: el código que llama devolverá is_error."""
    if name in ("screenshot", "zoom"):
        return  # la captura la toma el código que llama
    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)  # Pequeña pausa antes de escribir
        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"Acción no soportada: {name}")
    time.sleep(0.5)  # Esperamos a que responda la interfaz

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()
    
    # Sin encabezado beta y sin tamaño de pantalla (conjunto de herramientas actual)
    tools = [{"type": "computer_toolset_20260801"}]
    
    system_prompt = f"""Estás automatizando una tarea en la aplicación {app_name}.
    
REGLAS:
- Después de cada acción toma una captura para revisar
- Si ves una ventana modal o una notificación, ciérrala
- Si no encuentras el elemento, desplázate por la página, espera 2 segundos e intenta de nuevo
- Cuando termines la tarea, escribe TASK_COMPLETE y describe lo que hiciste
- Si la tarea es imposible, escribe TASK_FAILED y explica el motivo"""
    
    messages = [{
        "role": "user",
        "content": [
            screenshot_block(),
            {"type": "text", "text": f"Ejecuta la tarea: {task_description}"},
        ],
    }]
    
    for step in range(40):  # Máximo 40 pasos
        response = client.messages.create(
            model="claude-opus-5-5",   # modelos actuales: página "Lo vigente"
            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"Terminado en {step+1} pasos: {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"Error al ejecutar la acción {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("Se superó el número máximo de pasos")
    return False

# Uso
run_desktop_task(
    task_description="Abre el menú Archivo → Reportes → Diario. Elige la fecha de ayer. Haz clic en Exportar → CSV. Guárdalo en la carpeta /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")