bobseares-2/seleniumbase-website-scraper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

deploy

Browse Source
This commit is contained in:
a2a-platform
2026-06-27 16:54:42 +00:00
parent 1801e85b16
commit 35739c9ab8
3 changed files with 305 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
.a2a/agent.dsl.json
View File

@@ -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"
}
}

2
a2a.yaml
View File

@@ -1,5 +1,5 @@
name: seleniumbase-website-scraper
version: 0.2.4
version: 0.2.5
entrypoint: agent:BrowserToApiAgent
expose:
public: true

288
agent.py
View File

@@ -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