From aa47e47ca88a2b879259bcdcc36c81fa2a5da431 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Tue, 9 Jun 2026 11:46:31 +0000 Subject: [PATCH] a2a-source-edit: remove agent.py --- agent.py | 596 ------------------------------------------------------- 1 file changed, 596 deletions(-) delete mode 100644 agent.py diff --git a/agent.py b/agent.py deleted file mode 100644 index a7a828a..0000000 --- a/agent.py +++ /dev/null @@ -1,596 +0,0 @@ -from __future__ import annotations - -import csv -import json -import re -import time -from dataclasses import asdict, dataclass, field -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, - Pricing, - Resources, - RunContext, - WorkspaceAccess, - WorkspaceMode, - skill, -) - - -# ----------------------------- -# Public config/auth models -# ----------------------------- - - -class SeleniumbaseWebsiteScraperConfig(BaseModel): - # Reserved for future agent-level defaults - default_headless: bool = True - default_wait_seconds: float = 2.0 - - -# ----------------------------- -# Result models (for clarity) -# ----------------------------- - - -@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 = field(default_factory=ScrapeFiles) - - -# ----------------------------- -# Agent implementation -# ----------------------------- - - -class SeleniumbaseWebsiteScraper(A2AAgent[SeleniumbaseWebsiteScraperConfig, BaseModel]): - name = "seleniumbase-website-scraper" - description = ( - "Compliant website scraping via SeleniumBase/browser automation. " - "Respects robots.txt, same-origin pagination, and safety rails." - ) - version = "0.1.0" - - config_model = SeleniumbaseWebsiteScraperConfig - - # Deterministic/no-LLM agent: do not consume caller LLM credentials. - pricing = Pricing( - price_per_call_usd=0.0, - 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=200, - allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), - require_reason=False, - ) - - tools_used = ("seleniumbase", "beautifulsoup4") - - # --------------- - # 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, - *, - 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 - -# 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 - -items = [] -warnings = [] -pages_visited = 0 -blocked = False -block_reason = '' - -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) - -# 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 - -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) - -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)) -""" - return script