#!/usr/bin/env python3 """browser_agent.py — Vision agent: screenshot → Claude Vision → azione → loop. Non richiede conoscenza del DOM. Vede la pagina come un umano e agisce. Usa OpenRouter + claude-sonnet per la visione. Uso: python3 browser_agent.py "cerca minipc N150 su amazon.it e dimmi i primi 3 risultati con prezzo" python3 browser_agent.py "vai su booking.com e cerca hotel a Milano il 20 luglio 1 notte" """ import base64 import json import os import signal import subprocess import sys import time import urllib.request from pathlib import Path import httpx from dotenv import load_dotenv from playwright.sync_api import sync_playwright load_dotenv(Path(__file__).parent.parent / ".env") OPENROUTER_KEY = os.getenv("OPENROUTER_API_KEY", "") MODEL = "anthropic/claude-sonnet-4-5" MAX_STEPS = 20 SCREENSHOT_W = 1280 SCREENSHOT_H = 800 # Profilo Chrome dedicato al browser agent — completamente separato da Elon AGENT_PROFILE = Path.home() / ".config/chrome-debug-agent" AGENT_CDP_PORT = 9223 # porta diversa da Elon (9222) AGENT_CDP_URL = f"http://localhost:{AGENT_CDP_PORT}" AGENT_CHROME_CMD = [ "google-chrome-stable", f"--remote-debugging-port={AGENT_CDP_PORT}", f"--user-data-dir={AGENT_PROFILE}", "--no-first-run", "--profile-directory=Default", "--start-maximized", ] SYSTEM_PROMPT = """Sei un agente che naviga il web tramite screenshot. Ad ogni step ricevi uno screenshot della pagina corrente e il tuo obiettivo. Rispondi SOLO con un JSON valido, nessun altro testo. Azioni disponibili: - {"action": "click", "x": N, "y": N, "reason": "..."} - {"action": "type", "text": "...", "reason": "..."} - {"action": "key", "key": "Enter", "reason": "..."} - {"action": "scroll", "y": N, "reason": "..."} (N positivo = giù, negativo = su) - {"action": "navigate", "url": "...", "reason": "..."} - {"action": "press_hold", "x": N, "y": N, "seconds": N, "reason": "..."} (tieni premuto il mouse — per captcha 'press and hold') - {"action": "wait", "seconds": N, "reason": "..."} (aspetta N secondi che la pagina carichi) - {"action": "done", "result": "...", "reason": "..."} (quando hai la risposta) - {"action": "fail", "reason": "..."} (se impossibile procedere) Regole CRITICHE: - Non cliccare mai su "Acquista", "Aggiungi al carrello", "Prenota", "Paga" o simili - Solo ricerca e lettura, mai azioni irreversibili - Coordinate X,Y sono in pixel rispetto all'angolo in alto a sinistra dello screenshot - Se la pagina ha un banner cookie/privacy, chiudilo prima di procedere - Dopo aver cliccato "Cerca" o "Vai" usa {"action": "wait", "seconds": 4} per aspettare il caricamento - IMPORTANTE: appena vedi risultati (lista voli, hotel, prodotti con prezzi), leggili SUBITO e rispondi con "done". Non continuare a scrollare se hai già i dati visibili. - Se vedi un pulsante "PREMI E TIENI PREMUTO" o "press and hold", usa {"action": "press_hold", "x": N, "y": N, "seconds": 3} al centro del pulsante — NON usare click semplice. - CRITICO: usa "done" SOLO se i dati richiesti sono VISIBILI nello screenshot corrente. MAI inventare o dedurre dati non visibili. Se non vedi i dati, usa "fail". - Se dopo press_hold il captcha è ancora presente, usa "fail" — non continuare a tentare. - Se hai già scrollato 3+ volte senza trovare nuovi dati, usa "fail" invece di inventare. """ def _start_agent_chrome() -> subprocess.Popen: """Avvia Chrome sul profilo agent (porta 9223). Ritorna il processo.""" env = {**os.environ, "DISPLAY": ":0"} proc = subprocess.Popen( AGENT_CHROME_CMD, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, # crea process group separato → kill group funziona ) # Attendi CDP pronto for _ in range(20): time.sleep(1) try: urllib.request.urlopen(AGENT_CDP_URL + "/json", timeout=2) print(f"[agent] Chrome avviato (pid {proc.pid}, porta {AGENT_CDP_PORT})") return proc except Exception: pass print("[agent] WARNING: CDP non risponde dopo 20s") return proc def _stop_agent_chrome(proc: subprocess.Popen) -> None: """Chiude il Chrome agent e tutti i suoi figli.""" try: # Termina il gruppo di processi (Chrome spawna child processes) os.killpg(os.getpgid(proc.pid), signal.SIGTERM) except Exception: try: proc.terminate() except Exception: pass time.sleep(1) # Forza kill se ancora in piedi result = subprocess.run(["pgrep", "-f", "chrome-debug-agent"], capture_output=True, text=True) for pid in result.stdout.split(): try: os.kill(int(pid), signal.SIGKILL) except Exception: pass print("[agent] Chrome chiuso.") def screenshot_b64(page) -> str: img = page.screenshot(clip={"x": 0, "y": 0, "width": SCREENSHOT_W, "height": SCREENSHOT_H}) return base64.standard_b64encode(img).decode() def ask_vision(goal: str, screenshot: str, history: list[dict]) -> dict: messages = [] for h in history[-6:]: # ultimi 6 step di history (testo) messages.append(h) messages.append({ "role": "user", "content": [ { "type": "image", "source": {"type": "base64", "media_type": "image/png", "data": screenshot} }, { "type": "text", "text": ( f"Obiettivo: {goal}\n\n" "Questo è lo screenshot ATTUALE della pagina. " "Ricevi sempre uno screenshot ad ogni step. " "Cosa faccio adesso? Rispondi solo con JSON valido." ) } ] }) r = httpx.post( "https://openrouter.ai/api/v1/chat/completions", headers={ "Authorization": f"Bearer {OPENROUTER_KEY}", "HTTP-Referer": "https://adrian.privcloud.dev", }, json={ "model": MODEL, "messages": [{"role": "system", "content": SYSTEM_PROMPT}] + messages, "max_tokens": 300, "temperature": 0, }, timeout=30, ) r.raise_for_status() content = r.json()["choices"][0]["message"]["content"].strip() # Estrai JSON anche se il modello aggiunge testo start = content.find("{") end = content.rfind("}") + 1 return json.loads(content[start:end]) def execute_action(page, action: dict) -> str: a = action["action"] if a == "click": page.mouse.click(action["x"], action["y"]) time.sleep(1.5) return f"click ({action['x']}, {action['y']})" elif a == "type": page.keyboard.type(action["text"], delay=50) time.sleep(0.5) return f"type '{action['text']}'" elif a == "key": page.keyboard.press(action["key"]) time.sleep(1.5) return f"key {action['key']}" elif a == "scroll": page.mouse.wheel(0, action["y"]) time.sleep(0.8) return f"scroll {action['y']}" elif a == "press_hold": secs = min(float(action.get("seconds", 2)), 5) page.mouse.move(action["x"], action["y"]) page.mouse.down() time.sleep(secs) page.mouse.up() time.sleep(1.5) return f"press_hold ({action['x']}, {action['y']}) per {secs}s" elif a == "wait": secs = min(int(action.get("seconds", 3)), 10) time.sleep(secs) return f"wait {secs}s" elif a == "navigate": page.goto(action["url"], wait_until="domcontentloaded", timeout=30000) page.bring_to_front() time.sleep(3) return f"navigate {action['url']}" elif a in ("done", "fail"): return f"STOP: {action.get('result', action.get('reason', ''))}" return f"unknown action: {a}" def run(goal: str) -> str: print(f"\n[agent] Obiettivo: {goal}") print(f"[agent] Modello: {MODEL}, max {MAX_STEPS} step") print(f"[agent] Avvio Chrome dedicato (porta {AGENT_CDP_PORT})...\n") chrome_proc = _start_agent_chrome() with sync_playwright() as p: browser = p.chromium.connect_over_cdp(AGENT_CDP_URL) context = browser.contexts[0] page = context.new_page() page.set_viewport_size({"width": SCREENSHOT_W, "height": SCREENSHOT_H}) page.bring_to_front() history = [] result = "" debug_dir = Path("/tmp/browser_agent_debug") debug_dir.mkdir(exist_ok=True) for step in range(1, MAX_STEPS + 1): print(f"[step {step}/{MAX_STEPS}] Screenshot...", end=" ", flush=True) raw_img = page.screenshot(clip={"x": 0, "y": 0, "width": SCREENSHOT_W, "height": SCREENSHOT_H}) ss = base64.standard_b64encode(raw_img).decode() (debug_dir / f"step_{step:02d}.png").write_bytes(raw_img) print("Vision...", end=" ", flush=True) action = None for attempt in range(3): try: action = ask_vision(goal, ss, history) break except Exception as e: print(f"(retry {attempt+1}: {e})", end=" ", flush=True) time.sleep(2) if action is None: print("Errore vision persistente, stop.") break reason = action.get("reason", "") print(f"→ {action['action']} | {reason}") if action["action"] in ("done", "fail"): result = action.get("result", action.get("reason", "")) break executed = execute_action(page, action) # Aggiorna history (solo testo, non immagini — troppo pesante) history.append({ "role": "assistant", "content": json.dumps(action, ensure_ascii=False) }) history.append({ "role": "user", "content": f"Eseguito: {executed}" }) browser.close() _stop_agent_chrome(chrome_proc) return result if __name__ == "__main__": if len(sys.argv) < 2: print("Uso: python3 browser_agent.py 'obiettivo in linguaggio naturale'") sys.exit(1) goal = sys.argv[1] result = run(goal) print(f"\n{'='*60}") print(f"RISULTATO:\n{result}") print('='*60)