From 69cc656afbb147b54df753dbf1ce98e1319fecf9 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Mon, 8 Jun 2026 22:41:15 +0000 Subject: [PATCH] edit agent.py --- agent.py | 391 ++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 387 insertions(+), 4 deletions(-) diff --git a/agent.py b/agent.py index fbf41ea..74cc48d 100644 --- a/agent.py +++ b/agent.py @@ -5,6 +5,12 @@ selector mapping, pagination limits, browser options, robots.txt behavior, and output format. It never attempts to bypass CAPTCHAs, anti-bot challenges, access controls, paywalls, login gates, robots.txt restrictions, or terms of service restrictions. + +This module now also exposes an LLM-powered iterative skill that can plan and +refine safe scraping runs using the base browser skill as an execution step. +It uses a platform-provided LLM grant (ctx.llm) with strict safety rails and +never attempts to evade protections. The deep agent proposes selector/URL +adjustments limited to the same origin and stops on any challenge. """ from __future__ import annotations @@ -14,6 +20,7 @@ import textwrap import urllib.parse import urllib.request import urllib.robotparser +from datetime import datetime, timezone from pathlib import Path from typing import Any, Literal @@ -35,6 +42,7 @@ from a2a_pack import ( OUTPUT_DIR = "outputs/seleniumbase-website-scraper" RUNTIME_SKILLS_DIR = "seleniumbase-website-scraper/.deepagents/skills/" +DEEPAGENTS_RECURSION_LIMIT = 400 class SeleniumbaseWebsiteScraperConfig(BaseModel): @@ -49,7 +57,7 @@ class SeleniumbaseWebsiteScraper(A2AAgent[SeleniumbaseWebsiteScraperConfig, NoAu "automation with robots.txt checks, same-domain pagination safeguards, " "screenshots, HTML capture, JSON/CSV output, retries, and polite rate limiting." ) - version = "1.0.3" + version = "1.1.0" config_model = SeleniumbaseWebsiteScraperConfig auth_model = NoAuth @@ -59,13 +67,13 @@ class SeleniumbaseWebsiteScraper(A2AAgent[SeleniumbaseWebsiteScraperConfig, NoAu price_per_call_usd=0.0, caller_pays_llm=True, notes=( - "Browser scraping is deterministic; any future LLM use must read the " - "caller-provided ctx.llm credential. The caller pays any forwarded LLM cost." + "Browser scraping is deterministic; any LLM use reads the caller-provided ctx.llm credential. " + "The caller pays any forwarded LLM cost." ), ) resources = Resources(cpu="2", memory="2Gi", max_runtime_seconds=900) egress = EgressPolicy(deny_internet_by_default=False) - tools_used = ("seleniumbase", "selenium", "chromium", "browser-automation") + tools_used = ("seleniumbase", "selenium", "chromium", "browser-automation", "deepagents", "langchain") workspace_access = WorkspaceAccess.dynamic( max_files=256, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), @@ -227,6 +235,293 @@ class SeleniumbaseWebsiteScraper(A2AAgent[SeleniumbaseWebsiteScraperConfig, NoAu "robots": {"checked": bool(respect_robots)}, } + @skill( + description=( + "Iteratively plan and run compliant scraping passes until a goal is met, " + "using a platform-provided LLM with strict safety rails. The deep agent may " + "refine CSS selectors and pagination hints within the same origin only, and will stop " + "on any CAPTCHA/anti-bot/login/paywall/access-control indicator." + ), + timeout_seconds=900, + cost_class="llm-medium", + grant_outputs_prefix=OUTPUT_DIR, + grant_write_prefixes=(OUTPUT_DIR,), + grant_run_timeout_seconds=900, + ) + async def iterate_scrape_goal( + self, + ctx: RunContext[NoAuth], + goal: str, + url: str, + selectors: dict[str, str] | None = None, + min_items: int = 10, + max_iterations: int = 5, + per_pass_max_pages: int = 1, + wait_seconds: float = 3.0, + output_format: Literal["json", "csv"] = "json", + headless: bool = True, + user_agent: str = "", + respect_robots: bool = True, + rate_limit_seconds: float = 1.0, + screenshot: bool = False, + allow_selector_updates: bool = True, + ) -> dict[str, Any]: + """Use an LLM planner to refine selectors/pagination until the goal is met or a stop condition occurs.""" + start_url = str(url or "").strip() + if not start_url: + return {"status": "invalid_input", "items": [], "saved_paths": [], "warnings": ["url is required"], "screenshots": []} + + # Initial validation (if selectors are provided) + if selectors: + v = _validate_inputs( + url=start_url, + selectors=selectors, + max_pages=per_pass_max_pages, + max_pages_limit=self.config.max_pages_limit, + wait_seconds=wait_seconds, + output_format=output_format, + user_agent=user_agent, + rate_limit_seconds=rate_limit_seconds, + ) + if v["errors"]: + return {"status": "invalid_input", "items": [], "saved_paths": [], "warnings": v["errors"], "screenshots": []} + + creds = ctx.llm + same_origin_safeguard = _same_origin + await ctx.emit_progress("starting deep-agent iterative scrape planner") + + # Planner system prompt with strict safety policy + system_prompt = ( + "You are a cautious scraping planner for a compliant browser scraper. " + "Your actions are limited to proposing: (a) CSS selectors to extract fields; " + "(b) a same-origin URL (relative or absolute) for the next pass; (c) a reason to stop.\n" + "Hard safety rules:\n" + "- Never suggest evading CAPTCHAs, anti-bot, or access controls.\n" + "- Never suggest login, credential use, or paywall bypass.\n" + "- Always respect robots.txt and same-origin restriction with the initial URL's scheme+host.\n" + "- Do NOT use stealth, proxies, or automation-evasion.\n" + "Output only compact JSON with keys: action ('scrape'|'stop'), reason, selectors (object, optional), url (string, optional), confidence (0..1)." + ) + + decisions: list[dict[str, Any]] = [] + all_items: list[dict[str, Any]] = [] + all_pages: list[str] = [] + all_htmls: list[str] = [] + all_screens: list[str] = [] + warnings: list[str] = [] + saved_paths: list[str] = [] + + if not creds.api_key: + # Deterministic fallback: run one pass with provided selectors (or trivial title extraction) + await ctx.emit_progress("LLM creds unavailable; running deterministic single-pass fallback") + sel = selectors or {"title": "title"} + one = await self.scrape_website( + ctx=ctx, + url=start_url, + selectors=sel, + max_pages=per_pass_max_pages, + wait_seconds=wait_seconds, + output_format=output_format, + headless=headless, + user_agent=user_agent, + respect_robots=respect_robots, + rate_limit_seconds=rate_limit_seconds, + screenshot=screenshot, + ) + return { + "status": one.get("status") or "ok", + "items": one.get("items") or [], + "saved_paths": one.get("saved_paths") or [], + "warnings": one.get("warnings") or [], + "screenshots": one.get("screenshots") or [], + "pages_visited": one.get("pages_visited") or [], + "html_captures": one.get("html_captures") or [], + "iterations": 1, + "decisions": [], + } + + # Build deep agent planner + planner = self._build_deep_agent(ctx=ctx, creds=creds, system_prompt=system_prompt) + current_url = start_url + current_selectors = selectors or {"title": "title"} + + for iteration in range(1, int(max_iterations) + 1): + # Execute one browser pass + run = await self.scrape_website( + ctx=ctx, + url=current_url, + selectors=current_selectors, + max_pages=per_pass_max_pages, + wait_seconds=wait_seconds, + output_format=output_format, + headless=headless, + user_agent=user_agent, + respect_robots=respect_robots, + rate_limit_seconds=rate_limit_seconds, + screenshot=screenshot, + ) + status = str(run.get("status") or "ok") + items = list(run.get("items") or []) + all_items.extend(items) + all_pages.extend(run.get("pages_visited") or []) + all_htmls.extend(run.get("html_captures") or []) + all_screens.extend(run.get("screenshots") or []) + warnings.extend(run.get("warnings") or []) + saved_paths.extend(run.get("saved_paths") or []) + + if status in {"challenge_detected", "manual_intervention_required", "blocked", "blocked_by_robots_txt", "browser_runtime_unavailable"}: + warnings.append(f"iteration {iteration} stopped due to status={status}") + break + if len(all_items) >= int(min_items): + decisions.append({"iteration": iteration, "action": "stop", "reason": "min_items reached", "confidence": 1.0}) + break + + # Ask planner what to do next + summary = { + "iteration": iteration, + "goal": str(goal or "").strip(), + "start_url": start_url, + "current_url": current_url, + "same_origin": same_origin_safeguard(start_url, current_url), + "items_this_pass": len(items), + "items_total": len(all_items), + "example_item": items[0] if items else None, + "current_selectors": current_selectors, + "status": status, + } + prompt = json.dumps( + { + "instruction": "Propose next step using only same-origin URL and safe CSS selectors, or stop.", + "schema": { + "action": "scrape|stop", + "reason": "string", + "selectors": {"field": "CSS"}, + "url": "relative or absolute same-origin URL", + "confidence": 0.0, + }, + "context": summary, + }, + ensure_ascii=False, + ) + state = await planner.ainvoke( + {"messages": [{"role": "user", "content": prompt}]}, + config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}, + ) + raw = _last_message_text(state) + try: + decision = _extract_json_object(raw) + except Exception as exc: # noqa: BLE001 + warnings.append("planner response was not JSON; stopping: " + _friendly_error(exc)) + break + + action = str(decision.get("action") or "").strip().lower() + reason = str(decision.get("reason") or "").strip() + next_url = str(decision.get("url") or current_url).strip() or current_url + proposed_selectors = decision.get("selectors") or {} + confidence = float(decision.get("confidence") or 0.0) + + # Enforce safety: same-origin only, and only allow selector updates if permitted + if not same_origin_safeguard(start_url, next_url): + decisions.append({ + "iteration": iteration, + "action": "stop", + "reason": "planner suggested cross-origin URL; stopping for safety", + "confidence": confidence, + }) + break + + if action == "stop": + decisions.append({"iteration": iteration, "action": "stop", "reason": reason, "confidence": confidence}) + break + + if action != "scrape": + decisions.append({ + "iteration": iteration, + "action": "stop", + "reason": "planner returned unknown action", + "confidence": confidence, + }) + break + + # Adopt next plan + current_url = next_url + if allow_selector_updates and isinstance(proposed_selectors, dict) and proposed_selectors: + # ensure keys/values are strings + safe_map: dict[str, str] = {} + for k, v in proposed_selectors.items(): + k2 = str(k).strip() + v2 = str(v).strip() + if k2 and v2: + safe_map[k2] = v2 + if safe_map: + current_selectors = safe_map + decisions.append({ + "iteration": iteration, + "action": "scrape", + "reason": reason or "refining selectors/pagination", + "confidence": confidence, + "url": current_url, + "selectors": current_selectors, + }) + + # Deduplicate items by URL+page_index if present + seen_keys: set[tuple[Any, Any]] = set() + unique_items: list[dict[str, Any]] = [] + for it in all_items: + key = (it.get("url"), it.get("page_index")) + if key in seen_keys: + continue + seen_keys.add(key) + unique_items.append(it) + + # Persist a small run manifest for traceability + manifest_path = await _write_iter_manifest(ctx, { + "goal": goal, + "start_url": start_url, + "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "iterations": len(decisions) or min(int(max_iterations), 1), + "decisions": decisions, + "items_total": len(unique_items), + }) + if manifest_path: + saved_paths.append(manifest_path) + + return { + "status": "ok" if unique_items else "no_items", + "items": unique_items, + "saved_paths": saved_paths, + "warnings": warnings, + "screenshots": all_screens, + "pages_visited": all_pages, + "html_captures": all_htmls, + "iterations": len(decisions) or 0, + "decisions": decisions, + } + + def _build_deep_agent(self, *, ctx: RunContext[NoAuth], creds: Any, system_prompt: str): + from a2a_pack.deepagents import create_a2a_deep_agent + from langchain.agents.middleware import wrap_model_call + + @wrap_model_call + async def safe_model_call(request: Any, handler: Any) -> Any: + messages = request.state.get("messages", []) + print( + "[seleniumbase-website-scraper] planner model_call " + f"model={creds.model} source={creds.source} messages={len(messages)}" + ) + return await handler(request) + + backend = ctx.workspace_backend() + skill_sources = _seed_runtime_skills(backend, ctx) + return create_a2a_deep_agent( + ctx, + creds=creds, + backend=backend, + skills=skill_sources or None, + middleware=[safe_model_call], + system_prompt=system_prompt, + ) + def _validate_inputs( *, @@ -556,6 +851,11 @@ def _parse_last_json(stdout: str) -> dict[str, Any] | None: return None +def _same_origin(a: str, b: str) -> bool: + pa, pb = urllib.parse.urlparse(a), urllib.parse.urlparse(b) + return pa.scheme == pb.scheme and pa.netloc.lower() == pb.netloc.lower() + + async def _emit_workspace_artifact_if_available(ctx: RunContext[Any], path: str) -> None: try: reader = getattr(ctx.workspace, "read_bytes", None) @@ -575,3 +875,86 @@ async def _emit_workspace_artifact_if_available(ctx: RunContext[Any], path: str) await ctx.emit_artifact(ref) except Exception: # noqa: BLE001 return + + +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}" + + +def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: + # Seed any packaged runtime skills if present + root = Path(__file__).parent / "skills" + if not root.exists(): + return [] + runtime_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_root + rel, path.read_bytes())) + if uploads: + backend.upload_files(uploads) + return [runtime_root] + return [] + + +def _extract_json_object(raw: str) -> dict[str, Any]: + text = str(raw or "").strip() + if text.startswith("```"): + text = re.sub(r"^```(?:json)?\s*", "", text) + text = re.sub(r"\s*```$", "", text) + try: + parsed = json.loads(text) + except json.JSONDecodeError: + start = text.find("{") + end = text.rfind("}") + if start < 0 or end <= start: + raise + parsed = json.loads(text[start : end + 1]) + if not isinstance(parsed, dict): + raise ValueError("planner result was not a JSON object") + return parsed + + +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) + return str(content or messages[-1]) + + +def _friendly_error(exc: Exception) -> str: + message = str(exc) or exc.__class__.__name__ + message = re.sub(r"(?i)(password|passwd|pwd|secret|token|key)=\S+", r"\1=", message) + message = re.sub(r"(?i)(auth(?:entication)? failed).*", r"\1", message) + return message[:400] + + +async def _write_iter_manifest(ctx: RunContext[Any], manifest: dict[str, Any]) -> str | "": + try: + out_dir = OUTPUT_DIR + name = f"iter-plan-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ".json" + path = f"{out_dir}/{name}" + await ctx.workspace.write(path, json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8")) + return path + except Exception: # noqa: BLE001 + return ""