diff --git a/.a2a/agent.dsl.json b/.a2a/agent.dsl.json index 1294f28..3c4ca75 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.4", + "version": "0.2.5", "entrypoint": { "module": "agent", "class_name": "BrowserToApiAgent", @@ -144,6 +144,22 @@ "type": "null" } ] + }, + "ai_explore": { + "type": "boolean" + }, + "ai_steps": { + "type": "integer" + }, + "ai_goal": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] } }, "required": [ @@ -277,7 +293,8 @@ ], "noise_filtering": "Drops analytics, static assets, service workers, and obvious bot-defense plumbing by default.", "schema_inference": "Infers request/response JSON schemas, query params, path params, GraphQL dispatches, and auth-shaped headers.", - "captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains." + "captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains.", + "llm_exploration": "Uses ctx.llm to choose bounded, non-destructive page actions after the page loads." } }, "input_modes": [ @@ -364,7 +381,7 @@ "source": "python-a2a-pack", "project_manifest": { "name": "seleniumbase-website-scraper", - "version": "0.2.4", + "version": "0.2.5", "entrypoint": "agent:BrowserToApiAgent" } } diff --git a/a2a.yaml b/a2a.yaml index 4eb51a7..1cd1156 100644 --- a/a2a.yaml +++ b/a2a.yaml @@ -1,5 +1,5 @@ name: seleniumbase-website-scraper -version: 0.2.4 +version: 0.2.5 entrypoint: agent:BrowserToApiAgent expose: public: true diff --git a/agent.py b/agent.py index 2046d1f..f72dac9 100644 --- a/agent.py +++ b/agent.py @@ -17,6 +17,8 @@ import shutil import subprocess import sys import time +import urllib.error +import urllib.request from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any @@ -127,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.4" + version = "0.2.5" config_model = BrowserToApiConfig auth_model = NoAuth @@ -150,6 +152,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): "noise_filtering": "Drops analytics, static assets, service workers, and obvious bot-defense plumbing by default.", "schema_inference": "Infers request/response JSON schemas, query params, path params, GraphQL dispatches, and auth-shaped headers.", "captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains.", + "llm_exploration": "Uses ctx.llm to choose bounded, non-destructive page actions after the page loads.", } } @@ -178,6 +181,9 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): browser_mode: str = "seleniumbase_stealth", captcha_policy: str = "solve_if_allowed", allowed_captcha_domains: list[str] | None = None, + ai_explore: bool = True, + ai_steps: int = 4, + ai_goal: str | None = None, ) -> dict[str, Any]: if not url or not str(url).strip().startswith(("http://", "https://")): return {"error": "url must start with http:// or https://"} @@ -197,6 +203,10 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): browser_mode=browser_mode, captcha_policy=captcha_policy, allowed_captcha_domains=allowed_captcha_domains, + ai_explore=ai_explore, + ai_steps=max(0, min(int(ai_steps), 8)), + ai_goal=ai_goal, + llm_config=_llm_config_from_ctx(ctx), ) await ctx.emit_progress(f"captured {len(samples)} network samples; inferring API surface") @@ -221,6 +231,8 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): browser_mode=browser_mode, 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)), ) except Exception as exc: # noqa: BLE001 return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"} @@ -274,6 +286,10 @@ def capture_browser_traffic( browser_mode: str = "seleniumbase_stealth", captcha_policy: str = "solve_if_allowed", allowed_captcha_domains: list[str] | None = None, + ai_explore: bool = True, + ai_steps: int = 4, + ai_goal: str | None = None, + llm_config: dict[str, Any] | None = None, ) -> tuple[list[TrafficSample], list[str]]: """Capture useful browser traffic with Playwright. @@ -366,9 +382,22 @@ def capture_browser_traffic( log.append(f"open warning: {type(exc).__name__}: {exc}") page.wait_for_timeout(wait_seconds * 1000) - _fill_search_inputs(page, search_term, log) - _auto_scroll(page, log) - _click_interesting_elements(page, max_clicks=max_clicks, log=log) + if ai_explore and ai_steps > 0 and llm_config: + _ai_explore_page( + page, + llm_config=llm_config, + steps=min(ai_steps, max_clicks if max_clicks > 0 else ai_steps), + goal=ai_goal or _default_ai_goal(url), + search_term=search_term, + log=log, + ) + elif ai_explore and ai_steps > 0: + log.append("ai: ctx.llm unavailable; falling back to passive scroll only") + _auto_scroll(page, log) + else: + _fill_search_inputs(page, search_term, log) + _auto_scroll(page, log) + _click_interesting_elements(page, max_clicks=max_clicks, log=log) page.wait_for_timeout(1500) context.close() browser.close() @@ -376,6 +405,251 @@ def capture_browser_traffic( return list(samples_by_request.values()), log +def _llm_config_from_ctx(ctx: RunContext[NoAuth]) -> dict[str, Any] | None: + creds = getattr(ctx, "llm", None) + if creds is None: + return None + api_key = str(getattr(creds, "api_key", "") or "") + base_url = str(getattr(creds, "base_url", "") or "").rstrip("/") + model = str(getattr(creds, "model", "") or "") + if not (api_key and base_url and model): + return None + return {"api_key": api_key, "base_url": base_url, "model": model} + + +def _default_ai_goal(url: str) -> str: + host = urlparse(url).netloc or "this site" + return ( + f"Explore {host} to reveal useful same-site API, JSON, XHR, fetch, search, " + "navigation, post listing, filter, load-more, or content endpoints. Avoid " + "login, signup, checkout, destructive, payment, account, or external actions." + ) + + +def _ai_explore_page( + page: Any, + *, + llm_config: dict[str, Any], + steps: int, + goal: str, + search_term: str | None, + log: list[str], +) -> None: + for step in range(1, max(steps, 0) + 1): + snapshot = _page_action_snapshot(page) + action = _llm_choose_action( + llm_config, + snapshot=snapshot, + goal=goal, + step=step, + search_term=search_term, + log=log, + ) + if not action: + log.append(f"ai: step {step}: no valid action returned") + _auto_scroll(page, log) + continue + applied = _apply_ai_action(page, action, snapshot.get("candidates", []), search_term, log, step) + if not applied or action.get("action") == "stop": + break + page.wait_for_timeout(1800) + + +def _page_action_snapshot(page: Any) -> dict[str, Any]: + try: + candidates = page.evaluate( + """ + () => { + const nodes = Array.from(document.querySelectorAll( + 'a[href], button, [role="button"], input, textarea, select' + )); + const out = []; + let index = 0; + for (const node of nodes) { + const rect = node.getBoundingClientRect(); + const style = window.getComputedStyle(node); + if (rect.width < 4 || rect.height < 4 || style.visibility === 'hidden' || style.display === 'none') { + continue; + } + 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 placeholder = node.getAttribute('placeholder') || ''; + const role = node.getAttribute('role') || ''; + const type = node.getAttribute('type') || ''; + 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 }); + index += 1; + if (out.length >= 45) break; + } + return out; + } + """ + ) + except Exception: + candidates = [] + try: + visible_text = page.locator("body").inner_text(timeout=1500)[:4000] + except Exception: + visible_text = "" + return { + "url": page.url, + "title": _safe_page_title(page), + "visible_text": visible_text, + "candidates": candidates if isinstance(candidates, list) else [], + } + + +def _safe_page_title(page: Any) -> str: + try: + return str(page.title() or "") + except Exception: + return "" + + +def _llm_choose_action( + llm_config: dict[str, Any], + *, + snapshot: dict[str, Any], + goal: str, + step: int, + search_term: str | None, + log: list[str], +) -> dict[str, Any] | None: + candidates = snapshot.get("candidates", []) + compact_candidates = [ + {key: item.get(key) for key in ("index", "tag", "text", "href", "placeholder", "role", "type")} + for item in candidates[:45] + 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." + ) + user = { + "goal": goal, + "step": step, + "current_url": snapshot.get("url"), + "title": snapshot.get("title"), + "visible_text": str(snapshot.get("visible_text") or "")[:2200], + "search_term": search_term, + "candidates": compact_candidates, + "response_schema": { + "action": "click|fill|scroll|wait|stop", + "index": "candidate index for click/fill, optional", + "text": "text to fill, optional", + "reason": "short reason", + }, + } + try: + response = _chat_completion_json(llm_config, system=system, user=json.dumps(user, ensure_ascii=False)) + except Exception as exc: # noqa: BLE001 + log.append(f"ai warning: llm action selection failed: {type(exc).__name__}: {exc}") + return None + action = _parse_json_object(response) + if not isinstance(action, dict): + log.append(f"ai warning: non-json action: {_truncate_text(response, 240)}") + return None + return action + + +def _chat_completion_json(llm_config: dict[str, Any], *, system: str, user: str) -> str: + url = f"{llm_config['base_url'].rstrip('/')}/chat/completions" + payload = { + "model": llm_config["model"], + "messages": [ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], + "max_tokens": 320, + "response_format": {"type": "json_object"}, + } + req = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={ + "authorization": f"Bearer {llm_config['api_key']}", + "content-type": "application/json", + }, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=45) as response: + data = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace")[:1000] + raise RuntimeError(f"LLM HTTP {exc.code}: {detail}") from exc + choices = data.get("choices") if isinstance(data, dict) else None + if not choices: + return "" + message = choices[0].get("message") if isinstance(choices[0], dict) else {} + return str(message.get("content") or "") + + +def _apply_ai_action( + page: Any, + action: dict[str, Any], + candidates: list[dict[str, Any]], + search_term: str | None, + log: list[str], + step: int, +) -> bool: + action_name = str(action.get("action") or "").lower().strip() + reason = str(action.get("reason") or "").strip() + if action_name == "stop": + log.append(f"ai: step {step}: stop ({reason})") + return False + if action_name == "wait": + log.append(f"ai: step {step}: wait ({reason})") + page.wait_for_timeout(2000) + return True + if action_name == "scroll": + log.append(f"ai: step {step}: scroll ({reason})") + _auto_scroll(page, log) + 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 + 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 + selector = str(candidate.get("selector") or "") + if not selector: + return False + try: + locator = page.locator(selector).first + if action_name == "fill": + fill_text = str(action.get("text") or search_term or "api").strip()[:120] + locator.fill(fill_text, timeout=3000) + locator.press("Enter", timeout=3000) + log.append(f"ai: step {step}: fill candidate {index} ({reason})") + return True + if action_name == "click": + locator.click(timeout=4000) + log.append(f"ai: step {step}: click candidate {index} ({reason})") + 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 + + +def _candidate_by_index(candidates: list[dict[str, Any]], index: int | None) -> dict[str, Any] | None: + if index is None: + return None + for item in candidates: + if isinstance(item, dict) and item.get("index") == index: + return item + return None + + def seleniumbase_stealth_prep( *, url: str, @@ -1067,6 +1341,8 @@ def _result_payload( browser_mode: str | None = None, captcha_policy: str | None = None, allowed_captcha_domains: list[str] | None = None, + ai_explore: bool | None = None, + ai_steps: int | None = None, ) -> dict[str, Any]: bot_signals = _bot_defense_signals(samples) payload: dict[str, Any] = { @@ -1093,6 +1369,10 @@ def _result_payload( payload["captcha_policy"] = captcha_policy if allowed_captcha_domains is not None: payload["allowed_captcha_domains"] = allowed_captcha_domains + if ai_explore is not None: + payload["ai_explore"] = ai_explore + if ai_steps is not None: + payload["ai_steps"] = ai_steps if capture_log is not None: payload["capture_log"] = capture_log return payload