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

a2a-source-edit: write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-06-09 11:40:29 +00:00
parent fc4ba965b8
commit 74e1b1a69e
1 changed files with 547 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

693
agent.py
View File

@@ -1,189 +1,596 @@
"""seleniumbase-website-scraper agent.
Starter stack:
- DeepAgents for tool-calling orchestration
- Caller-provided LLM credentials via ctx.llm
- A tiny model-call middleware hook you can replace with tracing,
routing, rate limits, or policy checks
"""
from __future__ import annotations
import csv
import json
from pathlib import Path
from typing import Any
import re
import time
from dataclasses import asdict, dataclass
from typing import Any, Dict, List, Literal, Optional, Tuple
from urllib.parse import urljoin, urlparse
from urllib import robotparser
from pydantic import BaseModel
from a2a_pack import (
A2AAgent,
LLMProvisioning,
{{ auth_type }},
Pricing,
Resources,
RunContext,
WorkspaceAccess,
WorkspaceMode,
skill,
)
from a2a_pack.context import LLMCreds
# -----------------------------
# Public config/auth models
# -----------------------------
class SeleniumbaseWebsiteScraperConfig(BaseModel):
pass
# Reserved for future agent-level defaults
default_headless: bool = True
default_wait_seconds: float = 2.0
SYSTEM_PROMPT = """\
You are a compact tool-calling agent.
Use the text_stats tool when the user asks about text, counts, summaries,
or anything where exact length/word numbers would help. Mention tool results
briefly instead of dumping raw JSON.
"""
RUNTIME_SKILLS_DIR = "seleniumbase-website-scraper/.deepagents/skills/"
DEEPAGENTS_RECURSION_LIMIT = 500
# -----------------------------
# Result models (for clarity)
# -----------------------------
class SeleniumbaseWebsiteScraper(A2AAgent[SeleniumbaseWebsiteScraperConfig, {{ auth_type }}]):
@dataclass
class ScrapeFiles:
json_path: Optional[str] = None
csv_path: Optional[str] = None
screenshots: Tuple[str, ...] = ()
@dataclass
class ScrapeResult:
ok: bool
items: List[Dict[str, Any]]
pages_visited: int
robots_respected: bool
blocked: bool = False
block_reason: str = ""
warnings: Tuple[str, ...] = ()
files: ScrapeFiles = ScrapeFiles()
# -----------------------------
# Agent implementation
# -----------------------------
class SeleniumbaseWebsiteScraper(A2AAgent[SeleniumbaseWebsiteScraperConfig, BaseModel]):
name = "seleniumbase-website-scraper"
description = "A compliant website scraping agent built on SeleniumBase for structured extraction with robots.txt respect, pagination, optional screenshots, and CSV/JSON output."
description = (
"Compliant website scraping via SeleniumBase/browser automation. "
"Respects robots.txt, same-origin pagination, and safety rails."
)
version = "0.1.0"
config_model = SeleniumbaseWebsiteScraperConfig
auth_model = {{ auth_type }}
# Hosted generated agents read the caller's saved LLM credential through
# ctx.llm. The platform may proxy that credential through LiteLLM, but agent
# code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
# directly.
llm_provisioning = LLMProvisioning.PLATFORM
# Deterministic/no-LLM agent: do not consume caller LLM credentials.
pricing = Pricing(
price_per_call_usd=0.0,
caller_pays_llm=True,
notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
caller_pays_llm=False,
notes=(
"Deterministic SeleniumBase scraper. No LLM usage; caller never billed for LLM."
),
)
# Browser automation benefits from more resources than the tiny default.
resources = Resources(cpu="2", memory="2Gi", max_runtime_seconds=600)
# Allow writing outputs (JSON/CSV/screenshots) into the caller workspace.
workspace_access = WorkspaceAccess.dynamic(
max_files=64,
max_files=200,
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
require_reason=False,
)
tools_used = ("deepagents", "langchain")
@skill(description="Ask the starter DeepAgent to answer with tool calls when useful")
async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
creds = ctx.llm
await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
if not creds.api_key:
return (
"LLM key required. Add an LLM credential in Settings > LLM "
"credentials before running this agent; for local --invoke "
"runs set AGENT_LLM_KEY."
)
graph = self._build_deep_agent(ctx=ctx, creds=creds)
state = await graph.ainvoke(
{"messages": [{"role": "user", "content": prompt}]},
config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
)
await ctx.emit_progress("deepagent finished")
return _last_message_text(state)
tools_used = ("seleniumbase", "beautifulsoup4")
def _build_deep_agent(
# ---------------
# Helper methods
# ---------------
@staticmethod
def _robots_allows(url: str, user_agent: Optional[str]) -> Tuple[bool, List[str]]:
ua = user_agent or "*"
parsed = urlparse(url)
if not parsed.scheme or not parsed.netloc:
return False, ["invalid_url"]
robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"
warnings: List[str] = []
rp = robotparser.RobotFileParser()
try:
rp.set_url(robots_url)
rp.read()
except Exception as exc: # noqa: BLE001
warnings.append(f"robots_fetch_error:{type(exc).__name__}")
# If robots cannot be fetched, be conservative: allow but record warning
return True, warnings
try:
allowed = bool(rp.can_fetch(ua, url))
except Exception as exc: # noqa: BLE001
warnings.append(f"robots_parse_error:{type(exc).__name__}")
allowed = True
crawl_delay = getattr(rp, "crawl_delay", None)
if callable(crawl_delay):
try:
delay = crawl_delay(ua) # type: ignore[call-arg]
if delay:
warnings.append(f"robots_crawl_delay:{delay}")
except Exception:
pass
return allowed, warnings
@staticmethod
def _blocked_by_content(page_html: str) -> Tuple[bool, str]:
hay = page_html.lower()
patterns = {
"captcha": r"captcha|are you human|i am not a robot|recaptcha",
"access_denied": r"access denied|error 403|forbidden|blocked",
"login_required": r"login required|please log in|sign in to continue",
"paywall": r"subscribe now|membership required|paywall|premium only",
"bot_detected": r"unusual traffic|automated queries|bot detected",
}
for reason, rx in patterns.items():
if re.search(rx, hay, re.IGNORECASE):
return True, reason
return False, ""
@staticmethod
def _same_origin(u1: str, u2: str) -> bool:
p1, p2 = urlparse(u1), urlparse(u2)
return (p1.scheme, p1.netloc) == (p2.scheme, p2.netloc)
# ----------------------
# Public skills
# ----------------------
@skill(
description=(
"Scrape a website with SeleniumBase given a URL and CSS selectors. "
"Honors robots.txt, supports same-origin pagination, optional screenshots, and JSON/CSV outputs."
)
)
async def scrape_website(
self,
ctx: RunContext[BaseModel],
url: str,
selectors: Dict[str, str],
pagination_selector: Optional[str] = None,
item_selector: Optional[str] = None,
max_pages: int = 1,
headless: bool = True,
wait_seconds: float = 2.0,
user_agent: Optional[str] = None,
rate_limit_seconds: float = 1.0,
same_origin_only: bool = True,
output_format: Literal["json", "csv"] = "json",
screenshot: bool = False,
out_basename: str = "scrape",
) -> Dict[str, Any]:
# Robots.txt check up front
allowed, robots_warnings = self._robots_allows(url, user_agent)
if not allowed:
result = ScrapeResult(
ok=False,
blocked=True,
block_reason="robots_disallow",
items=[],
pages_visited=0,
robots_respected=True,
warnings=tuple(robots_warnings),
files=ScrapeFiles(),
)
return asdict(result)
await ctx.emit_progress("starting SeleniumBase scrape")
# Prepare and run scraper inside the workspace sandbox so outputs persist
script = self._render_seleniumbase_script(
start_url=url,
selectors=selectors,
pagination_selector=pagination_selector,
item_selector=item_selector,
max_pages=max_pages,
headless=headless,
wait_seconds=wait_seconds,
user_agent=user_agent,
rate_limit_seconds=rate_limit_seconds,
same_origin_only=same_origin_only,
output_format=output_format,
screenshot=screenshot,
out_basename=out_basename,
)
exec_result = await ctx.workspace_python(
script,
image="python:3.11-slim",
timeout_seconds=min(max(60.0, 10.0 * max_pages), 540.0),
memory_mib=1536,
cpus=2,
)
warnings: List[str] = []
if exec_result.exit_code != 0:
warnings.append(
f"scraper_exit_nonzero:{exec_result.exit_code}: {exec_result.output[:300]}"
)
try:
payload = json.loads(exec_result.output.strip().splitlines()[-1])
except Exception:
payload = {
"ok": False,
"blocked": False,
"block_reason": "",
"items": [],
"pages_visited": 0,
"robots_respected": True,
"warnings": warnings + ["no_json_payload"],
"files": asdict(ScrapeFiles()),
}
else:
# merge sandbox warnings with ours
if isinstance(payload.get("warnings"), list):
payload["warnings"] = list(set(payload["warnings"] + warnings))
else:
payload["warnings"] = warnings
return payload
@skill(
description=(
"Iteratively attempt to scrape toward a goal without LLMs. "
"Heuristically refines selectors and pagination (max 2 iterations) within same-origin and robots rules."
)
)
async def iterate_scrape_goal(
self,
ctx: RunContext[BaseModel],
goal: str,
start_url: str,
min_items: int = 25,
max_steps: int = 2,
headless: bool = True,
wait_seconds: float = 2.0,
user_agent: Optional[str] = None,
rate_limit_seconds: float = 1.0,
output_format: Literal["json", "csv"] = "json",
screenshot: bool = False,
out_basename: str = "iter",
) -> Dict[str, Any]:
allowed, robots_warnings = self._robots_allows(start_url, user_agent)
if not allowed:
return asdict(
ScrapeResult(
ok=False,
blocked=True,
block_reason="robots_disallow",
items=[],
pages_visited=0,
robots_respected=True,
warnings=tuple(robots_warnings),
)
)
await ctx.emit_progress("heuristic discovery of selectors")
# Minimal heuristic candidates for item container and fields
container_candidates = [
"article",
"li",
"div.card",
"div.post",
"div.result",
"div.item",
"div.entry",
"div.product",
]
field_candidates = {
"title": ["h1", "h2", "h3", "a[title]", "a"],
"link@href": ["a[href]"],
"description": ["p", ".description", ".summary"],
}
steps = max(1, min(max_steps, 2))
best_payload: Dict[str, Any] = {}
best_count = 0
for step in range(steps):
item_selector = container_candidates[min(step, len(container_candidates) - 1)]
selectors: Dict[str, str] = {}
for key, opts in field_candidates.items():
selectors[key] = opts[min(step, len(opts) - 1)]
payload = await self.scrape_website(
ctx,
url=start_url,
selectors=selectors,
pagination_selector='a[rel="next"], a.next, a[aria-label="Next"], a:contains("Next"), a:contains("›")',
item_selector=item_selector,
max_pages=10,
headless=headless,
wait_seconds=wait_seconds,
user_agent=user_agent,
rate_limit_seconds=rate_limit_seconds,
same_origin_only=True,
output_format=output_format,
screenshot=screenshot,
out_basename=f"{out_basename}_s{step+1}",
)
items = payload.get("items") or []
if len(items) >= min_items:
payload.setdefault("warnings", []).append(
f"goal_reached_in_step:{step+1}"
)
return payload
if len(items) > best_count:
best_payload, best_count = payload, len(items)
# return best effort
if best_payload:
best_payload.setdefault("warnings", []).append(
"min_items_not_reached_best_effort"
)
return best_payload
# fallback empty struct
return asdict(
ScrapeResult(
ok=False,
items=[],
pages_visited=0,
robots_respected=True,
blocked=False,
warnings=("no_items_found", *tuple(robots_warnings)),
)
)
# ----------------------
# Sandbox script factory
# ----------------------
def _render_seleniumbase_script(
self,
*,
ctx: RunContext[{{ auth_type }}],
creds: LLMCreds,
) -> Any:
# Lazy imports keep `a2a card` usable before local dependencies are
# installed. `a2a deploy` installs requirements.txt during the build.
from a2a_pack.deepagents import create_a2a_deep_agent
from langchain.agents.middleware import wrap_model_call
from langchain_core.tools import tool
start_url: str,
selectors: Dict[str, str],
pagination_selector: Optional[str],
item_selector: Optional[str],
max_pages: int,
headless: bool,
wait_seconds: float,
user_agent: Optional[str],
rate_limit_seconds: float,
same_origin_only: bool,
output_format: str,
screenshot: bool,
out_basename: str,
) -> str:
# Inline Python to run inside the sandbox with the workspace mounted
# at /workspace. Keep imports local to avoid affecting agent card load.
sel_json = json.dumps(selectors)
pag_sel = json.dumps(pagination_selector) if pagination_selector else "None"
item_sel = json.dumps(item_selector) if item_selector else "None"
ua = json.dumps(user_agent) if user_agent else "None"
script = f"""
import json, os, sys, time, csv, re
from urllib.parse import urlparse, urljoin
from urllib import robotparser
@tool
def text_stats(text: str) -> str:
"""Return exact word, character, and line counts for text."""
words = [part for part in text.split() if part.strip()]
return json.dumps(
{
"characters": len(text),
"words": len(words),
"lines": len(text.splitlines()) or 1,
}
)
# Install runtime deps lightweightly if missing
try:
import seleniumbase # type: ignore
except Exception:
import subprocess, sys
subprocess.check_call([sys.executable, '-m', 'pip', 'install', '--no-cache-dir', 'seleniumbase==4.28.4'])
import seleniumbase # type: ignore
@wrap_model_call
async def log_model_call(request: Any, handler: Any) -> Any:
messages = request.state.get("messages", [])
print(
"[middleware] model_call "
f"model={creds.model} source={creds.source} messages={len(messages)}"
)
return await handler(request)
items = []
warnings = []
pages_visited = 0
blocked = False
block_reason = ''
backend = ctx.workspace_backend()
skill_sources = _seed_runtime_skills(backend, ctx)
# create_a2a_deep_agent resolves provider:model strings with
# langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
# provider-specific extra body, and runtime model overrides.
return create_a2a_deep_agent(
ctx,
creds=creds,
backend=backend,
skills=skill_sources or None,
tools=[text_stats],
middleware=[log_model_call],
system_prompt=SYSTEM_PROMPT,
)
start_url = {json.dumps(start_url)}
selectors = {sel_json}
pagination_selector = {pag_sel}
item_selector = {item_sel}
max_pages = int({max_pages})
headless = bool({str(headless)})
wait_seconds = float({wait_seconds})
user_agent = {ua}
rate_limit_seconds = float({rate_limit_seconds})
same_origin_only = bool({str(same_origin_only)})
output_format = {json.dumps(output_format)}
screenshot = bool({str(screenshot)})
out_basename = {json.dumps(out_basename)}
origin = '{{}}://{{}}'.format(urlparse(start_url).scheme, urlparse(start_url).netloc)
def _runtime_skills_root(ctx: RunContext[Any]) -> str:
workspace = getattr(ctx, "_workspace", None)
prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
if not prefixes:
outputs_prefix = getattr(workspace, "outputs_prefix", None)
prefixes = (outputs_prefix or "outputs/",)
prefix = str(prefixes[0]).strip("/")
return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
# Robots.txt re-check in sandbox (best effort)
rp = robotparser.RobotFileParser()
try:
rp.set_url(origin + '/robots.txt')
rp.read()
if not rp.can_fetch(user_agent or '*', start_url):
print(json.dumps({{'ok': False, 'blocked': True, 'block_reason': 'robots_disallow', 'items': [], 'pages_visited': 0, 'robots_respected': True, 'warnings': ['sandbox_robots_disallow'], 'files': {{'json_path': None, 'csv_path': None, 'screenshots': []}}}}))
sys.exit(0)
except Exception as e:
warnings.append('robots_sandbox_warning')
# Start SeleniumBase driver (undetected-chromedriver helps with bot walls)
from seleniumbase import Driver # type: ignore
def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
"""Copy packaged DeepAgents skills into the invocation workspace.
try:
driver = Driver(uc=True, headless=headless, agent=user_agent)
except Exception as e:
warnings.append('driver_init_error:' + e.__class__.__name__)
print(json.dumps({{'ok': False, 'blocked': False, 'block_reason': '', 'items': [], 'pages_visited': 0, 'robots_respected': True, 'warnings': warnings, 'files': {{'json_path': None, 'csv_path': None, 'screenshots': []}}}}))
sys.exit(0)
DeepAgents loads skills from its backend, while source-controlled
``skills/`` folders live in the image. This bridge lets generated agents
ship reusable SKILL.md bundles without giving up durable A2A workspace
files.
current_url = start_url
try:
for page in range(max_pages):
driver.get(current_url)
time.sleep(max(0.0, wait_seconds))
pages_visited += 1
html = driver.page_source or ''
low = html.lower()
# Blocker heuristics
for key, rx in {{
'captcha': r'captcha|are you human|i am not a robot|recaptcha',
'access_denied': r'access denied|error 403|forbidden|blocked',
'login_required': r'login required|please log in|sign in to continue',
'paywall': r'subscribe now|membership required|paywall|premium only',
'bot_detected': r'unusual traffic|automated queries|bot detected',
}}.items():
if re.search(rx, low, re.IGNORECASE):
blocked = True
block_reason = key
break
if blocked:
break
def pick_attr(selector: str) -> tuple[str, str | None]:
# Support 'selector@attr' to pick element attribute
if '@' in selector:
base, attr = selector.split('@', 1)
return base.strip(), attr.strip() or None
return selector, None
# Extraction
page_items = []
try:
from selenium.webdriver.common.by import By # type: ignore
if item_selector:
containers = driver.find_elements(By.CSS_SELECTOR, item_selector)[:200]
for c in containers:
row: dict[str, str] = {{}}
for key, sel in selectors.items():
base_sel, attr = pick_attr(sel)
try:
el = c.find_element(By.CSS_SELECTOR, base_sel)
val = (el.get_attribute(attr) if attr else el.text).strip()
except Exception:
val = ''
row[key] = val
# compact empty rows
if any(v for v in row.values()):
page_items.append(row)
else:
# No item container: single snapshot using first match of each selector
row: dict[str, str] = {{}}
for key, sel in selectors.items():
base_sel, attr = pick_attr(sel)
try:
el = driver.find_element(By.CSS_SELECTOR, base_sel)
val = (el.get_attribute(attr) if attr else el.text).strip()
except Exception:
val = ''
row[key] = val
if any(v for v in row.values()):
page_items.append(row)
except Exception as e:
warnings.append('extract_error:' + e.__class__.__name__)
items.extend(page_items)
# Screenshot
if screenshot:
try:
os.makedirs('outputs', exist_ok=True)
shot_path = f"outputs/{{out_basename}}_p{{page+1}}.png"
driver.save_screenshot(shot_path)
except Exception as e:
warnings.append('screenshot_error')
# Pagination
if pagination_selector:
try:
from selenium.webdriver.common.by import By # type: ignore
next_candidates = driver.find_elements(By.CSS_SELECTOR, pagination_selector)
next_href = None
for el in next_candidates:
href = el.get_attribute('href')
if href:
if same_origin_only:
if urlparse(href).netloc and urlparse(href).netloc != urlparse(start_url).netloc:
continue
next_href = href
break
if not next_href and next_candidates:
try:
next_candidates[0].click()
time.sleep(max(0.0, wait_seconds))
current_url = driver.current_url
except Exception:
break
else:
if not next_href:
break
if same_origin_only and urlparse(next_href).netloc and urlparse(next_href).netloc != urlparse(start_url).netloc:
break
current_url = urljoin(current_url, next_href)
time.sleep(max(0.0, rate_limit_seconds))
continue
except Exception:
break
else:
break
finally:
try:
driver.quit()
except Exception:
pass
# Persist outputs
os.makedirs('outputs', exist_ok=True)
json_path = f"outputs/{{out_basename}}.json"
csv_path = f"outputs/{{out_basename}}.csv"
try:
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(items, f, ensure_ascii=False, indent=2)
except Exception as e:
warnings.append('json_write_error')
json_path = None
try:
if items:
keys = sorted({{k for row in items for k in row.keys()}})
with open(csv_path, 'w', encoding='utf-8', newline='') as f:
w = csv.DictWriter(f, fieldnames=keys)
w.writeheader()
for row in items:
w.writerow(row)
else:
csv_path = None
except Exception:
warnings.append('csv_write_error')
csv_path = None
# Collect screenshot paths
shots = []
try:
for name in os.listdir('outputs'):
if name.startswith(out_basename + '_p') and name.endswith('.png'):
shots.append('outputs/' + name)
except Exception:
pass
payload = {{
'ok': (not blocked),
'items': items,
'pages_visited': pages_visited,
'robots_respected': True,
'blocked': blocked,
'block_reason': block_reason,
'warnings': warnings,
'files': {{'json_path': json_path, 'csv_path': csv_path, 'screenshots': shots}},
}}
print(json.dumps(payload))
"""
root = Path(__file__).parent / "skills"
if not root.exists():
return []
runtime_skills_root = _runtime_skills_root(ctx)
uploads: list[tuple[str, bytes]] = []
for path in root.rglob("*"):
if path.is_file():
rel = path.relative_to(root).as_posix()
uploads.append((runtime_skills_root + rel, path.read_bytes()))
if uploads:
backend.upload_files(uploads)
return [runtime_skills_root]
return []
def _last_message_text(state: dict[str, Any]) -> str:
messages = state.get("messages") or []
if not messages:
return json.dumps(state, default=str)
content = getattr(messages[-1], "content", None)
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for item in content:
if isinstance(item, dict):
text = item.get("text") or item.get("content")
if text:
parts.append(str(text))
elif item:
parts.append(str(item))
return "\n".join(parts) if parts else json.dumps(content, default=str)
return str(content or messages[-1])
return script