edit agent.py

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=<redacted>", 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 ""