From 6042ced1430b379e61bdbad1d02ad051ec613b89 Mon Sep 17 00:00:00 2001 From: a2a-platform Date: Sat, 27 Jun 2026 23:43:30 +0000 Subject: [PATCH] deploy --- .a2a/agent.dsl.json | 4 +- a2a.yaml | 2 +- agent.py | 274 +++++++++++++++++++++++++++++++++++++++----- 3 files changed, 247 insertions(+), 33 deletions(-) diff --git a/.a2a/agent.dsl.json b/.a2a/agent.dsl.json index 7ce838b..a9b3f7a 100644 --- a/.a2a/agent.dsl.json +++ b/.a2a/agent.dsl.json @@ -3,7 +3,7 @@ "language": "python", "name": "seleniumbase-website-scraper", "description": "SeleniumBase-style browser agent that turns observed website traffic into OpenAPI specs, coverage reports, samples, and a replay client.", - "version": "0.2.9", + "version": "0.3.0", "entrypoint": { "module": "agent", "class_name": "BrowserToApiAgent", @@ -381,7 +381,7 @@ "source": "python-a2a-pack", "project_manifest": { "name": "seleniumbase-website-scraper", - "version": "0.2.9", + "version": "0.3.0", "entrypoint": "agent:BrowserToApiAgent" } } diff --git a/a2a.yaml b/a2a.yaml index fb001a5..764caed 100644 --- a/a2a.yaml +++ b/a2a.yaml @@ -1,5 +1,5 @@ name: seleniumbase-website-scraper -version: 0.2.9 +version: 0.3.0 entrypoint: agent:BrowserToApiAgent expose: public: true diff --git a/agent.py b/agent.py index bb6e2a6..891ae33 100644 --- a/agent.py +++ b/agent.py @@ -22,7 +22,7 @@ import urllib.request from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any -from urllib.parse import parse_qsl, urlparse +from urllib.parse import parse_qsl, urljoin, urlparse from a2a_pack import A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill from pydantic import BaseModel, Field @@ -129,7 +129,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): "SeleniumBase-style browser agent that turns observed website traffic " "into OpenAPI specs, coverage reports, samples, and a replay client." ) - version = "0.2.9" + version = "0.3.0" config_model = BrowserToApiConfig auth_model = NoAuth @@ -182,7 +182,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): captcha_policy: str = "solve_if_allowed", allowed_captcha_domains: list[str] | None = None, ai_explore: bool = True, - ai_steps: int = 4, + ai_steps: int = 10, ai_goal: str | None = None, ) -> dict[str, Any]: if not url or not str(url).strip().startswith(("http://", "https://")): @@ -204,7 +204,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): captcha_policy=captcha_policy, allowed_captcha_domains=allowed_captcha_domains, ai_explore=ai_explore, - ai_steps=max(0, min(int(ai_steps), 8)), + ai_steps=max(0, min(int(ai_steps), 20)), ai_goal=ai_goal, llm_config=_llm_config_from_ctx(ctx), ) @@ -232,7 +232,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): captcha_policy=captcha_policy, allowed_captcha_domains=_effective_captcha_domains(allowed_captcha_domains), ai_explore=ai_explore, - ai_steps=max(0, min(int(ai_steps), 8)), + ai_steps=max(0, min(int(ai_steps), 20)), ) except Exception as exc: # noqa: BLE001 return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"} @@ -287,7 +287,7 @@ def capture_browser_traffic( captcha_policy: str = "solve_if_allowed", allowed_captcha_domains: list[str] | None = None, ai_explore: bool = True, - ai_steps: int = 4, + ai_steps: int = 10, ai_goal: str | None = None, llm_config: dict[str, Any] | None = None, ) -> tuple[list[TrafficSample], list[str]]: @@ -370,8 +370,8 @@ def capture_browser_traffic( except Exception as exc: # noqa: BLE001 sample.error = f"{type(exc).__name__}: {exc}" - page.on("request", on_request) - page.on("response", on_response) + context.on("request", on_request) + context.on("response", on_response) try: page.goto(url, wait_until="domcontentloaded", timeout=45_000) @@ -391,7 +391,7 @@ def capture_browser_traffic( _ai_explore_page( page, llm_config=llm_config, - steps=min(ai_steps, max_clicks if max_clicks > 0 else ai_steps), + steps=ai_steps, goal=ai_goal or _default_ai_goal(url), search_term=search_term, log=log, @@ -446,27 +446,63 @@ def _ai_explore_page( search_term: str | None, log: list[str], ) -> None: + start_url = page.url + visited_urls = [_normalize_url(start_url)] + history: list[dict[str, Any]] = [] + min_steps = min(max(steps, 0), 6) + log.append(f"ai: exploration budget {steps} steps") for step in range(1, max(steps, 0) + 1): - snapshot = _page_action_snapshot(page) + snapshot = _page_action_snapshot(page, start_url=start_url) action = _llm_choose_action( llm_config, snapshot=snapshot, goal=goal, step=step, + total_steps=steps, search_term=search_term, + history=history, + visited_urls=visited_urls, log=log, ) if not action: log.append(f"ai: step {step}: no valid action returned") _auto_scroll(page, log) + history.append({"step": step, "action": "fallback_scroll", "url": _safe_page_url(page)}) continue - applied = _apply_ai_action(page, action, snapshot.get("candidates", []), search_term, log, step) + if str(action.get("action") or "").lower().strip() == "stop" and step < min_steps: + log.append(f"ai: step {step}: ignored early stop; continuing exploration") + action = {"action": "scroll", "reason": "continue exploration before stopping"} + applied = _apply_ai_action( + page, + action, + snapshot.get("candidates", []), + search_term, + start_url, + log, + step, + ) + if applied and action.get("action") != "stop": + _wait_for_action_settle(page) + _recover_if_external(page, start_url, log, step) + current_url = _normalize_url(_safe_page_url(page)) + if current_url and current_url not in visited_urls: + visited_urls.append(current_url) + log.append(f"ai: visited {current_url}") + history.append( + { + "step": step, + "action": str(action.get("action") or ""), + "index": action.get("index"), + "url": current_url, + "reason": str(action.get("reason") or "")[:180], + } + ) if not applied or action.get("action") == "stop": break - page.wait_for_timeout(1800) + log.append(f"ai: explored {len(visited_urls)} page url(s)") -def _page_action_snapshot(page: Any) -> dict[str, Any]: +def _page_action_snapshot(page: Any, *, start_url: str) -> dict[str, Any]: try: candidates = page.evaluate( """ @@ -485,33 +521,81 @@ def _page_action_snapshot(page: Any) -> dict[str, Any]: const tag = node.tagName.toLowerCase(); const text = (node.innerText || node.value || node.getAttribute('aria-label') || node.getAttribute('title') || '').trim(); const href = node.getAttribute('href') || ''; + const absoluteHref = href ? new URL(href, window.location.href).href : ''; const placeholder = node.getAttribute('placeholder') || ''; const role = node.getAttribute('role') || ''; const type = node.getAttribute('type') || ''; + const disabled = Boolean(node.disabled || node.getAttribute('aria-disabled') === 'true'); const id = `ai-discovery-${index}`; node.setAttribute('data-ai-discovery-id', id); - out.push({ index, selector: `[data-ai-discovery-id="${id}"]`, tag, text: text.slice(0, 140), href, placeholder, role, type }); + out.push({ + index, + selector: `[data-ai-discovery-id="${id}"]`, + tag, + text: text.slice(0, 180), + href, + absolute_href: absoluteHref, + placeholder, + role, + type, + disabled + }); index += 1; - if (out.length >= 45) break; + if (out.length >= 90) break; } - return out; + return { + candidates: out, + scroll: { + x: window.scrollX, + y: window.scrollY, + height: document.documentElement.scrollHeight, + viewportHeight: window.innerHeight + } + }; } """ ) except Exception: - candidates = [] + candidates = {"candidates": [], "scroll": {}} + raw_candidates = candidates.get("candidates") if isinstance(candidates, dict) else candidates + annotated_candidates = _annotate_action_candidates(raw_candidates, start_url=start_url, current_url=page.url) try: - visible_text = page.locator("body").inner_text(timeout=1500)[:4000] + visible_text = page.locator("body").inner_text(timeout=1500)[:6000] except Exception: visible_text = "" return { "url": page.url, "title": _safe_page_title(page), "visible_text": visible_text, - "candidates": candidates if isinstance(candidates, list) else [], + "scroll": candidates.get("scroll", {}) if isinstance(candidates, dict) else {}, + "candidates": annotated_candidates, } +def _annotate_action_candidates(raw_candidates: Any, *, start_url: str, current_url: str) -> list[dict[str, Any]]: + if not isinstance(raw_candidates, list): + return [] + base_url = current_url or start_url + annotated: list[dict[str, Any]] = [] + for item in raw_candidates: + if not isinstance(item, dict): + continue + candidate = dict(item) + href = str(candidate.get("href") or "") + absolute_href = str(candidate.get("absolute_href") or "") + if href and not absolute_href: + absolute_href = urljoin(base_url, href) + candidate["absolute_href"] = absolute_href + if href.startswith(("mailto:", "tel:", "javascript:")): + candidate["same_site"] = False + elif absolute_href: + candidate["same_site"] = _is_same_site_url(absolute_href, start_url) + else: + candidate["same_site"] = True + annotated.append(candidate) + return annotated + + def _safe_page_title(page: Any) -> str: try: return str(page.title() or "") @@ -519,37 +603,132 @@ def _safe_page_title(page: Any) -> str: return "" +def _safe_page_url(page: Any) -> str: + try: + return str(page.url or "") + except Exception: + return "" + + +def _normalize_url(url: str) -> str: + if not url: + return "" + parsed = urlparse(str(url)) + if not parsed.scheme: + return str(url).rstrip("/") + path = (parsed.path or "/").rstrip("/") or "/" + return parsed._replace(path=path, fragment="").geturl() + + +def _is_same_site_url(candidate_url: str, start_url: str) -> bool: + if not candidate_url: + return True + resolved = urljoin(start_url, candidate_url) + parsed = urlparse(resolved) + if parsed.scheme not in {"http", "https"}: + return False + candidate_site = _site_key(parsed.hostname or "") + start_site = _site_key(urlparse(start_url).hostname or "") + return bool(candidate_site and start_site and candidate_site == start_site) + + +def _site_key(host: str) -> str: + parts = [part for part in host.lower().rstrip(".").split(".") if part] + if len(parts) <= 2: + return ".".join(parts) + if len(parts[-1]) == 2 and len(parts[-2]) <= 3 and len(parts) >= 3: + return ".".join(parts[-3:]) + return ".".join(parts[-2:]) + + +def _recover_if_external(page: Any, start_url: str, log: list[str], step: int) -> None: + current_url = _safe_page_url(page) + if not current_url or _is_same_site_url(current_url, start_url): + return + log.append(f"ai warning: step {step}: landed off-site; returning to start site") + try: + page.go_back(wait_until="domcontentloaded", timeout=6000) + except Exception: + try: + page.goto(start_url, wait_until="domcontentloaded", timeout=10_000) + except Exception as exc: # noqa: BLE001 + log.append(f"ai warning: step {step}: recovery failed: {type(exc).__name__}: {exc}") + + +def _wait_for_action_settle(page: Any) -> None: + page.wait_for_timeout(900) + try: + page.wait_for_load_state("domcontentloaded", timeout=2500) + except Exception: + pass + try: + page.wait_for_load_state("networkidle", timeout=3000) + except Exception: + pass + page.wait_for_timeout(600) + + def _llm_choose_action( llm_config: dict[str, Any], *, snapshot: dict[str, Any], goal: str, step: int, + total_steps: int, search_term: str | None, + history: list[dict[str, Any]], + visited_urls: list[str], log: list[str], ) -> dict[str, Any] | None: candidates = snapshot.get("candidates", []) + visited_set = {_normalize_url(item) for item in visited_urls if item} compact_candidates = [ - {key: item.get(key) for key in ("index", "tag", "text", "href", "placeholder", "role", "type")} - for item in candidates[:45] + { + **{ + key: item.get(key) + for key in ( + "index", + "tag", + "text", + "href", + "absolute_href", + "placeholder", + "role", + "type", + "same_site", + "disabled", + ) + }, + "visited": _normalize_url(str(item.get("absolute_href") or "")) in visited_set, + } + for item in candidates[:90] if isinstance(item, dict) ] system = ( "You choose one safe browser action for API discovery. Return only JSON. " - "Allowed actions: click, fill, scroll, wait, stop. Prefer actions likely " - "to trigger XHR/fetch/API traffic. Avoid login, signup, checkout, delete, " - "logout, payment, account settings, destructive operations, and external domains." + "Allowed actions: click, fill, scroll, wait, back, stop. Use the available " + "budget to explore deeply: prefer unvisited same-site links, list/detail " + "pages, search boxes, filters, load-more controls, navigation, docs, feed " + "or content links that are likely to trigger XHR/fetch/API traffic. Avoid " + "external domains, login, signup, checkout, delete, logout, payment, account " + "settings, and destructive operations. Do not stop early unless there are no " + "safe same-site actions left." ) user = { "goal": goal, "step": step, + "total_steps": total_steps, + "remaining_steps": max(total_steps - step, 0), "current_url": snapshot.get("url"), "title": snapshot.get("title"), - "visible_text": str(snapshot.get("visible_text") or "")[:2200], + "scroll": snapshot.get("scroll"), + "visited_urls": visited_urls[-14:], + "recent_history": history[-8:], + "visible_text": str(snapshot.get("visible_text") or "")[:3000], "search_term": search_term, "candidates": compact_candidates, "response_schema": { - "action": "click|fill|scroll|wait|stop", + "action": "click|fill|scroll|wait|back|stop", "index": "candidate index for click/fill, optional", "text": "text to fill, optional", "reason": "short reason", @@ -646,6 +825,7 @@ def _apply_ai_action( action: dict[str, Any], candidates: list[dict[str, Any]], search_term: str | None, + start_url: str, log: list[str], step: int, ) -> bool: @@ -662,18 +842,29 @@ def _apply_ai_action( log.append(f"ai: step {step}: scroll ({reason})") _auto_scroll(page, log) return True + if action_name == "back": + try: + page.go_back(wait_until="domcontentloaded", timeout=6000) + log.append(f"ai: step {step}: back ({reason})") + except Exception as exc: # noqa: BLE001 + log.append(f"ai warning: step {step}: back failed: {type(exc).__name__}: {exc}") + return True index = _int_or_none(action.get("index")) candidate = _candidate_by_index(candidates, index) if not candidate: log.append(f"ai warning: step {step}: missing candidate for action {action_name}") - return False + return True text_bits = " ".join(str(candidate.get(key) or "") for key in ("text", "href", "placeholder")).lower() if _looks_destructive(text_bits): log.append(f"ai: step {step}: skipped destructive-looking candidate {index}") return True + href = str(candidate.get("absolute_href") or candidate.get("href") or "") + if href and not _is_same_site_url(href, start_url): + log.append(f"ai: step {step}: skipped external candidate {index}") + return True selector = str(candidate.get("selector") or "") if not selector: - return False + return True try: locator = page.locator(selector).first if action_name == "fill": @@ -683,14 +874,19 @@ def _apply_ai_action( log.append(f"ai: step {step}: fill candidate {index} ({reason})") return True if action_name == "click": + try: + locator.evaluate("(node) => node.removeAttribute('target')", timeout=1000) + except Exception: + pass locator.click(timeout=4000) log.append(f"ai: step {step}: click candidate {index} ({reason})") + _recover_if_external(page, start_url, log, step) return True except Exception as exc: # noqa: BLE001 log.append(f"ai warning: step {step}: {action_name} failed: {type(exc).__name__}: {exc}") return True log.append(f"ai warning: step {step}: unsupported action {action_name}") - return False + return True def _candidate_by_index(candidates: list[dict[str, Any]], index: int | None) -> dict[str, Any] | None: @@ -967,7 +1163,25 @@ def _click_interesting_elements(page: Any, *, max_clicks: int, log: list[str]) - def _looks_destructive(text: str) -> bool: - return any(word in text for word in ("delete", "remove", "logout", "sign out", "purchase", "checkout", "pay")) + blocked_words = ( + "delete", + "remove", + "logout", + "log out", + "sign out", + "purchase", + "checkout", + "pay", + "billing", + "login", + "log in", + "sign in", + "signup", + "sign up", + "account", + "settings", + ) + return any(word in text for word in blocked_words) def parse_trace_jsonl(*, requests_jsonl: str, responses_jsonl: str = "") -> list[TrafficSample]: