From 6a6f5122b767f54024b995381ade1bb1e3dd2141 Mon Sep 17 00:00:00 2001 From: a2a-platform Date: Sat, 27 Jun 2026 01:15:21 +0000 Subject: [PATCH] deploy --- .a2a-builder-state.json | 12 - .a2a/agent.dsl.json | 351 ++++ README.md | 157 +- a2a.yaml | 48 +- agent.py | 1743 +++++++++++++------- requirements.txt | 4 +- skills/compliant-browser-scraping/SKILL.md | 45 - skills/llm-iterative-scrape/SKILL.md | 31 - smoke_tests/quotes_toscrape.json | 17 - smoke_tests/test_iterative_quotes_plan.py | 46 - smoke_tests/test_render_scraper_script.py | 59 - 11 files changed, 1578 insertions(+), 935 deletions(-) delete mode 100644 .a2a-builder-state.json create mode 100644 .a2a/agent.dsl.json delete mode 100644 skills/compliant-browser-scraping/SKILL.md delete mode 100644 skills/llm-iterative-scrape/SKILL.md delete mode 100644 smoke_tests/quotes_toscrape.json delete mode 100644 smoke_tests/test_iterative_quotes_plan.py delete mode 100644 smoke_tests/test_render_scraper_script.py diff --git a/.a2a-builder-state.json b/.a2a-builder-state.json deleted file mode 100644 index 39a5c35..0000000 --- a/.a2a-builder-state.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "agent": "seleniumbase-website-scraper", - "file_count": 9, - "last_sandbox": { - "exit_code": 0, - "source_hash": "800286217c3eb6945d8f9b09404a6da74147243f3a05ed4fb46f11b6c0ff4cf1", - "updated_at": 1781005877 - }, - "repo_head_sha": "59e28467b7e98c3ef0abe4da070be833ccacb7dd", - "source": "repo-sync", - "updated_at": 1781005822 -} \ No newline at end of file diff --git a/.a2a/agent.dsl.json b/.a2a/agent.dsl.json new file mode 100644 index 0000000..cf64474 --- /dev/null +++ b/.a2a/agent.dsl.json @@ -0,0 +1,351 @@ +{ + "schema_version": "2026-06-04", + "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.0", + "entrypoint": { + "module": "agent", + "class_name": "BrowserToApiAgent", + "function": null, + "command": [] + }, + "skills": [ + { + "name": "discover_api_from_browser", + "description": "Open a website in a browser, capture observable HTTP traffic, infer an OpenAPI 3.1 spec, and write a browsable API discovery report.", + "handler": "discover_api_from_browser", + "tags": [ + "browser", + "seleniumbase", + "openapi", + "api-discovery" + ], + "scopes": [], + "stream": false, + "policy": { + "timeout_seconds": 600.0, + "idempotent": false, + "max_retries": 0, + "cost_class": null, + "allow_scope_expansion": false, + "grant_mode": null, + "grant_allow_patterns": [], + "grant_deny_patterns": [], + "grant_outputs_prefix": null, + "grant_write_prefixes": [], + "grant_ttl_seconds": null, + "grant_run_timeout_seconds": null, + "grant_approval_timeout_seconds": null, + "grant_scope_approval_timeout_seconds": null + }, + "input_schema": { + "type": "object", + "properties": { + "url": { + "type": "string" + }, + "title": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "origins": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] + }, + "include": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] + }, + "exclude": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] + }, + "wait_seconds": { + "type": "integer" + }, + "max_clicks": { + "type": "integer" + }, + "search_term": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "min_samples": { + "type": "integer" + }, + "redact": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "url" + ], + "additionalProperties": false + }, + "output_schema": { + "additionalProperties": true, + "type": "object" + } + }, + { + "name": "discover_api_from_trace", + "description": "Replay request/response JSONL traffic into the same OpenAPI 3.1 discovery pipeline used by live browser capture.", + "handler": "discover_api_from_trace", + "tags": [ + "trace", + "openapi", + "api-discovery" + ], + "scopes": [], + "stream": false, + "policy": { + "timeout_seconds": 240.0, + "idempotent": false, + "max_retries": 0, + "cost_class": null, + "allow_scope_expansion": false, + "grant_mode": null, + "grant_allow_patterns": [], + "grant_deny_patterns": [], + "grant_outputs_prefix": null, + "grant_write_prefixes": [], + "grant_ttl_seconds": null, + "grant_run_timeout_seconds": null, + "grant_approval_timeout_seconds": null, + "grant_scope_approval_timeout_seconds": null + }, + "input_schema": { + "type": "object", + "properties": { + "requests_jsonl": { + "type": "string" + }, + "responses_jsonl": { + "type": "string" + }, + "title": { + "type": "string" + }, + "origins": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] + }, + "include": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] + }, + "exclude": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] + }, + "min_samples": { + "type": "integer" + }, + "redact": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "requests_jsonl" + ], + "additionalProperties": false + }, + "output_schema": { + "additionalProperties": true, + "type": "object" + } + } + ], + "capabilities": { + "browser_to_api": { + "capture": "Launches a browser, records XHR/fetch/document traffic with response bodies, then infers OpenAPI.", + "offline": "Can replay JSONL request/response captures into the same OpenAPI emitter.", + "outputs": [ + "openapi.json", + "openapi.yaml", + "report.md", + "index.html", + "client.mjs", + "confidence.json", + "samples/*.json" + ], + "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." + } + }, + "input_modes": [ + "application/json" + ], + "output_modes": [ + "application/json" + ], + "required_secrets": [], + "required_env": [], + "consumer_setup": { + "fields": [] + }, + "runtime": { + "lifecycle": "ephemeral", + "availability": "on_demand", + "state": "none", + "sandbox": "microsandbox", + "resources": { + "cpu": "2000m", + "memory": "4Gi", + "gpu": 0, + "max_runtime_seconds": 900 + }, + "concurrency": 1, + "egress": { + "allow_hosts": [ + "*" + ], + "allow_internal_services": [], + "deny_internet_by_default": false + }, + "tools_used": [ + "seleniumbase", + "playwright", + "browser-to-api", + "openapi" + ], + "llm_provisioning": "platform", + "pricing": { + "price_per_call_usd": 0.0, + "caller_pays_llm": false, + "notes": "" + }, + "wants_cp_jwt": false, + "platform_resources": { + "memory": null, + "databases": [] + }, + "endpoints": [], + "apt_packages": [] + }, + "template_lineage": null, + "meta_agent_manifest": null, + "state_schema": null, + "workspace_access": { + "enabled": false, + "max_files": 0, + "allowed_modes": [], + "require_reason": true, + "deny_patterns": [], + "require_human_approval": false, + "max_total_size_bytes": 104857600 + }, + "config_schema": { + "properties": {}, + "title": "BrowserToApiConfig", + "type": "object" + }, + "auth": { + "model": "a2a_pack.auth.NoAuth", + "strategy": "public", + "principal_schema": { + "additionalProperties": false, + "description": "Public agent: no caller identity required.", + "properties": {}, + "title": "NoAuth", + "type": "object" + }, + "resolver": null, + "required": false + }, + "metadata": { + "source": "python-a2a-pack", + "project_manifest": { + "name": "seleniumbase-website-scraper", + "version": "0.2.0", + "entrypoint": "agent:BrowserToApiAgent" + } + } +} diff --git a/README.md b/README.md index 2fb8845..4160329 100644 --- a/README.md +++ b/README.md @@ -1,145 +1,40 @@ # seleniumbase-website-scraper -Private A2A agent for compliant, authorized website scraping with SeleniumBase/browser automation where available. +Browser-to-API discovery agent. -## Safety policy +This upgrades the original SeleniumBase website scraper idea into a hosted API +discovery workflow: -Use this agent only for authorized scraping of public or user-owned sites. The agent must not bypass, solve, defeat, or evade CAPTCHAs, anti-bot protections, login gates, paywalls, `robots.txt`, ToS restrictions, or access controls. If a CAPTCHA, anti-bot challenge, login wall, paywall, or access-control indicator is detected, the run stops and returns a manual-intervention/blocked status. +- opens a site in a browser +- records observable XHR/fetch/document traffic +- captures JSON response bodies when the browser can read them +- filters analytics/static noise +- templates paths and query parameters +- splits GraphQL/JSON-RPC style dispatches +- infers request and response JSON schemas +- emits `openapi.json`, `openapi.yaml`, `report.md`, `index.html`, + `client.mjs`, `confidence.json`, and redacted samples -The agent does not use stealth/undetected-browser modes, CAPTCHA solvers, proxy rotation, credential injection, or access-control circumvention. +## Run Locally -## Skills - -### scrape_website - -Inputs: -- url — absolute http or https URL. -- selectors — mapping of output field names to CSS selectors. -- max_pages — maximum same-origin pages to visit. -- wait_seconds — wait after page open before capture/extraction. -- output_format — json or csv. -- headless — toggle browser headless mode. -- user_agent — optional custom single-line user agent. -- respect_robots — when true, checks robots.txt before fetch and pagination. -- rate_limit_seconds — polite delay between pagination requests. -- screenshot — save PNG screenshots for visited pages. - -Outputs: -- status — ok, invalid_input, blocked_by_robots_txt, manual_intervention_required, browser_runtime_unavailable, or error. -- items — extracted records. -- saved_paths — output files under outputs/seleniumbase-website-scraper/. -- warnings — safety, robots, selector, browser, or runtime warnings. -- screenshots — screenshot file paths when requested. -- pages_visited and html_captures. - -### iterate_scrape_goal (new) - -Parameters: -- start_url, allowed_domains (default same-origin), seed_selectors, max_steps, max_pages, - max_runtime_seconds, wait_seconds, rate_limit_seconds, headless, user_agent, - respect_robots (default true), output_format (json/csv), screenshot. - -Returns fields include: {status, items, pages_visited, html_captures, screenshots, saved_paths, warnings, steps, goal_summary}. - -An LLM-powered planner that iterates scrapes until a goal is satisfied or a safety/stop condition occurs. -Inputs: -- goal — natural-language target (e.g., "Collect quotes text and authors from the first 2 pages"). -- start_url — start page (absolute http/https). Same-origin only for all iterations unless allowed_domains is provided. -- seed_selectors — optional initial CSS selector map (defaults to {"title": "title"}). -- max_steps — planner steps (hard-capped at 3 improvements). -- max_pages — pages per scrape pass (default 1, capped by agent config). -- max_runtime_seconds — total runtime budget across steps. -- wait_seconds, output_format, headless, user_agent, respect_robots, rate_limit_seconds, screenshot — mirror scrape_website. -- allowed_domains — optional whitelist; else restricted to same-origin. - -Outputs: -- status — ok | no_items | blocked_by_robots_txt | manual_intervention_required | browser_runtime_unavailable | error -- items — accumulated unique items across passes -- pages_visited — union of pages visited across passes -- html_captures — captures from passes -- screenshots — when requested -- saved_paths — includes an iter-plan manifest under outputs/seleniumbase-website-scraper/ -- warnings — including planner parsing issues, safety stops, or budget stops -- iterations — number of planner decisions taken -- decisions — array of {iteration, action, reason, confidence, url?, selectors?} - -Behavior and guarantees: -- Uses platform-provided ctx.llm with strict system prompt that forbids any evasion of CAPTCHAs, anti-bot, logins, paywalls, or robots.txt. -- Planner can only propose same-origin URLs; cross-origin plans are rejected and end the loop. -- Each pass runs the deterministic scrape_website skill; stops on any challenge or robots block. -- Produces a compact decisions trail and saves a small manifest under outputs/seleniumbase-website-scraper/. - -## Selector syntax - -Default extraction returns visible text: - -```json -{"title": "h1", "summary": ".article-summary"} +```bash +python -m pip install -r requirements.txt +python -m playwright install chromium +python agent.py https://example.com --wait-seconds 8 --max-clicks 4 ``` -Supported suffixes: -- ::text — visible text. -- ::html — inner HTML. -- ::attr(name) — attribute extraction, e.g. a.card::attr(href). +## Run as A2A -Special pagination selector field names are not emitted as data fields and are used to find the next page: -- next -- _next -- __next__ -- pagination_next -- next_page - -Pagination is restricted to the same scheme and host as the start URL. - -## Example payloads - -Simple one-pass scrape: -```json -{ - "url": "https://example.com/articles", - "selectors": { - "headline": "h2.article-title", - "links": "a.article-link::attr(href)", - "next": "a[rel='next']::attr(href)" - }, - "max_pages": 3, - "wait_seconds": 2, - "output_format": "json", - "headless": true, - "respect_robots": true -} +```bash +a2a dev +a2a test --invoke --skill discover_api_from_browser --args-json '{"url":"https://example.com","wait_seconds":8,"max_clicks":2}' ``` -Iterative planner: +Use `origins` to focus on the API host: + ```json -{ - "goal": "collect at least 30 product tiles with name, price, link", - "url": "https://example.com/store", - "selectors": {"name": ".tile .name", "price": ".tile .price", "href": ".tile a::attr(href)"}, - "min_items": 30, - "max_iterations": 3, - "per_pass_max_pages": 1, - "respect_robots": true -} +{"url":"https://app.example.com","origins":["api.example.com"],"min_samples":2} ``` -## Smoke test: quotes.toscrape.com - -Goal: "Collect quotes text and authors from the first 2 pages" -- Expected: status=ok, >= 1 item, <= 2 iterations (often 1–2), a saved JSON manifest, no screenshots when disabled. -- Example call (schematic): -```json -{ - "goal": "Collect quotes text and authors from the first 2 pages", - "url": "https://quotes.toscrape.com/", - "selectors": {"text": ".quote .text", "author": ".quote .author", "pagination_next": "li.next a::attr(href)"}, - "min_items": 2, - "max_iterations": 2, - "per_pass_max_pages": 2, - "screenshot": false -} -``` - -## Deployment notes - -The agent declares a larger runtime budget because browser work can be CPU/memory intensive. Actual browser execution is performed through the workspace sandbox helper using the SeleniumBase image, so files written under /workspace/outputs/seleniumbase-website-scraper/ are durable workspace outputs. +The replay skill accepts JSONL request/response captures and runs the same +OpenAPI emitter without opening a browser. diff --git a/a2a.yaml b/a2a.yaml index 2d58a37..dad8a46 100644 --- a/a2a.yaml +++ b/a2a.yaml @@ -1,14 +1,42 @@ -# Project identity for `a2a deploy`. Most metadata (resources, scopes, -# secrets, workspace, etc.) lives on the Python class — this file only -# tells the CLI how to find it. name: seleniumbase-website-scraper -version: 0.1.2 -entrypoint: agent:SeleniumbaseWebsiteScraper +version: 0.2.0 +entrypoint: agent:BrowserToApiAgent expose: - public: false + public: true + runtime: resources: - cpu: "2" - memory: 2Gi - max_runtime_seconds: 600 - + cpu: 2000m + memory: 4Gi + max_runtime_seconds: 900 + egress: + allow_hosts: + - "*" + deny_internet_by_default: false + apt_packages: + - ca-certificates + - fonts-liberation + - libasound2 + - libatk-bridge2.0-0 + - libatk1.0-0 + - libcairo2 + - libcups2 + - libdbus-1-3 + - libdrm2 + - libgbm1 + - libglib2.0-0 + - libgtk-3-0 + - libnspr4 + - libnss3 + - libpango-1.0-0 + - libx11-6 + - libxcb1 + - libxcomposite1 + - libxdamage1 + - libxext6 + - libxfixes3 + - libxkbcommon0 + - libxrandr2 + - xdg-utils + install_commands: + - python -m playwright install chromium diff --git a/agent.py b/agent.py index 9d97294..7d194b2 100644 --- a/agent.py +++ b/agent.py @@ -1,596 +1,1175 @@ +"""Browser-to-API discovery agent. + +The Browserbase browser-to-api skill is a strong offline pipeline. This agent +turns the same idea into a hosted A2A workflow: drive a browser, capture useful +HTTP traffic, infer schemas, and emit a deployable OpenAPI bundle. +""" from __future__ import annotations -import csv +import argparse +import asyncio +import hashlib +import html 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 pathlib import Path +from typing import Any +from urllib.parse import parse_qsl, urlparse -from pydantic import BaseModel +from a2a_pack import A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill +from pydantic import BaseModel, Field -from a2a_pack import ( - A2AAgent, - Pricing, - Resources, - RunContext, - WorkspaceAccess, - WorkspaceMode, - skill, + +DEFAULT_OUTPUT_DIR = Path("outputs/browser-to-api") +JSONISH_TYPES = ( + "application/json", + "application/problem+json", + "application/vnd.api+json", + "text/json", ) +TEXT_TYPES = ("text/plain", "application/x-ndjson") +DEFAULT_REDACT_KEYS = { + "authorization", + "cookie", + "set-cookie", + "x-csrf-token", + "x-xsrf-token", + "x-api-key", + "proxy-authorization", + "password", + "token", + "secret", + "api_key", + "apikey", + "access_token", + "accesstoken", + "refresh_token", + "refreshtoken", + "creditcard", + "ssn", +} +NOISE_URL_PATTERNS = [ + r"google-analytics\.com", + r"googletagmanager\.com", + r"doubleclick\.net", + r"facebook\.com/tr", + r"segment\.(io|com)", + r"mixpanel\.com", + r"amplitude\.com", + r"fullstory\.com", + r"hotjar\.com", + r"intercom\.io", + r"clarity\.ms", + r"cloudflareinsights\.com", + r"sentry\.io", + r"datadog(hq)?\.com", + r"/(pixel|beacon|track|pageview|impression)(/|$|\?)", + r"\.(png|jpe?g|gif|svg|webp|ico|woff2?|ttf|eot|otf|css|map|mp4|webm|mp3)(\?|$)", + r"/(sw|service-worker)\.js(\?|$)", + r"/manifest\.json(\?|$)", + r"/robots\.txt(\?|$)", + r"/favicon\.ico(\?|$)", +] -# ----------------------------- -# 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.2" - - 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: +class BrowserToApiConfig(BaseModel): 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)) + +@dataclass +class TrafficSample: + request_id: str + method: str + url: str + status: int | None = None + resource_type: str | None = None + request_headers: dict[str, Any] = field(default_factory=dict) + response_headers: dict[str, Any] = field(default_factory=dict) + request_body: Any = None + response_body: Any = None + response_body_truncated: bool = False + content_type: str | None = None + timestamp: float = field(default_factory=time.time) + error: str | None = None + + +@dataclass +class EndpointGroup: + key: str + origin: str + method: str + path: str + source_path: str + dispatch: str | None + samples: list[TrafficSample] = field(default_factory=list) + path_params: list[dict[str, Any]] = field(default_factory=list) + query_params: dict[str, list[str]] = field(default_factory=dict) + flags: list[str] = field(default_factory=list) + + +class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): + 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.0" + + config_model = BrowserToApiConfig + auth_model = NoAuth + llm_provisioning = LLMProvisioning.PLATFORM + pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False) + tools_used = ("seleniumbase", "playwright", "browser-to-api", "openapi") + capabilities = { + "browser_to_api": { + "capture": "Launches a browser, records XHR/fetch/document traffic with response bodies, then infers OpenAPI.", + "offline": "Can replay JSONL request/response captures into the same OpenAPI emitter.", + "outputs": [ + "openapi.json", + "openapi.yaml", + "report.md", + "index.html", + "client.mjs", + "confidence.json", + "samples/*.json", + ], + "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.", + } + } + + @skill( + name="discover_api_from_browser", + description=( + "Open a website in a browser, capture observable HTTP traffic, infer " + "an OpenAPI 3.1 spec, and write a browsable API discovery report." + ), + tags=("browser", "seleniumbase", "openapi", "api-discovery"), + timeout_seconds=600, + ) + async def discover_api_from_browser( + self, + ctx: RunContext[NoAuth], + url: str, + title: str | None = None, + origins: list[str] | None = None, + include: list[str] | None = None, + exclude: list[str] | None = None, + wait_seconds: int = 8, + max_clicks: int = 4, + search_term: str | None = None, + min_samples: int = 1, + redact: list[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://"} + + await ctx.emit_progress("launching browser and capturing network traffic") + samples, capture_log = await asyncio.to_thread( + capture_browser_traffic, + url=str(url).strip(), + wait_seconds=max(1, min(int(wait_seconds), 60)), + max_clicks=max(0, min(int(max_clicks), 20)), + search_term=search_term, + ) + + await ctx.emit_progress(f"captured {len(samples)} network samples; inferring API surface") + bundle = build_openapi_bundle( + samples, + title=title or _title_from_url(url), + origins=origins, + include=include, + exclude=exclude, + min_samples=max(1, int(min_samples)), + redact=redact, + ) + output_dir = write_bundle(bundle, DEFAULT_OUTPUT_DIR / safe_run_id(url)) + + await ctx.emit_progress(f"wrote OpenAPI bundle to {output_dir}") + return { + "url": url, + "captured_samples": len(samples), + "included_samples": bundle["summary"]["included_samples"], + "endpoints": bundle["summary"]["endpoints"], + "origins": bundle["summary"]["origins"], + "output_dir": str(output_dir), + "openapi_json": str(output_dir / "openapi.json"), + "openapi_yaml": str(output_dir / "openapi.yaml"), + "html_report": str(output_dir / "index.html"), + "markdown_report": str(output_dir / "report.md"), + "client": str(output_dir / "client.mjs"), + "confidence": str(output_dir / "confidence.json"), + "capture_log": capture_log, + "top_endpoints": bundle["summary"]["top_endpoints"], + } + + @skill( + name="discover_api_from_trace", + description=( + "Replay request/response JSONL traffic into the same OpenAPI 3.1 " + "discovery pipeline used by live browser capture." + ), + tags=("trace", "openapi", "api-discovery"), + timeout_seconds=240, + ) + async def discover_api_from_trace( + self, + ctx: RunContext[NoAuth], + requests_jsonl: str, + responses_jsonl: str = "", + title: str = "Discovered Website API", + origins: list[str] | None = None, + include: list[str] | None = None, + exclude: list[str] | None = None, + min_samples: int = 1, + redact: list[str] | None = None, + ) -> dict[str, Any]: + await ctx.emit_progress("pairing replayed request/response trace") + samples = parse_trace_jsonl(requests_jsonl=requests_jsonl, responses_jsonl=responses_jsonl) + bundle = build_openapi_bundle( + samples, + title=title, + origins=origins, + include=include, + exclude=exclude, + min_samples=max(1, int(min_samples)), + redact=redact, + ) + output_dir = write_bundle(bundle, DEFAULT_OUTPUT_DIR / safe_run_id(title)) + await ctx.emit_progress(f"wrote replayed OpenAPI bundle to {output_dir}") + return { + "captured_samples": len(samples), + "included_samples": bundle["summary"]["included_samples"], + "endpoints": bundle["summary"]["endpoints"], + "origins": bundle["summary"]["origins"], + "output_dir": str(output_dir), + "openapi_json": str(output_dir / "openapi.json"), + "openapi_yaml": str(output_dir / "openapi.yaml"), + "html_report": str(output_dir / "index.html"), + "markdown_report": str(output_dir / "report.md"), + "client": str(output_dir / "client.mjs"), + "confidence": str(output_dir / "confidence.json"), + "top_endpoints": bundle["summary"]["top_endpoints"], + } + + +def capture_browser_traffic( + *, + url: str, + wait_seconds: int = 8, + max_clicks: int = 4, + search_term: str | None = None, +) -> tuple[list[TrafficSample], list[str]]: + """Capture useful browser traffic with Playwright. + + SeleniumBase remains part of the packaged agent because it is valuable for + anti-bot/UC flows, but Playwright's response body APIs make it the pragmatic + capture engine for schema inference. + """ + from playwright.sync_api import TimeoutError as PlaywrightTimeoutError + from playwright.sync_api import sync_playwright + + log: list[str] = [] + samples_by_request: dict[int, TrafficSample] = {} + + with sync_playwright() as p: + browser = p.chromium.launch( + headless=True, + args=[ + "--disable-dev-shm-usage", + "--disable-blink-features=AutomationControlled", + "--no-sandbox", + ], + ) + context = browser.new_context( + viewport={"width": 1440, "height": 1000}, + ignore_https_errors=True, + user_agent=( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/126.0 Safari/537.36" + ), + ) + page = context.new_page() + + def on_request(request: Any) -> None: + parsed_body = _parse_maybe_json(request.post_data) + samples_by_request[id(request)] = TrafficSample( + request_id=str(id(request)), + method=request.method.upper(), + url=request.url, + resource_type=request.resource_type, + request_headers=dict(request.headers or {}), + request_body=parsed_body, + timestamp=time.time(), + ) + + def on_response(response: Any) -> None: + request = response.request + sample = samples_by_request.get(id(request)) + if not sample: + sample = TrafficSample( + request_id=str(id(request)), + method=request.method.upper(), + url=request.url, + resource_type=request.resource_type, + request_headers=dict(request.headers or {}), + request_body=_parse_maybe_json(request.post_data), + timestamp=time.time(), + ) + samples_by_request[id(request)] = sample + sample.status = response.status + sample.response_headers = dict(response.headers or {}) + sample.content_type = sample.response_headers.get("content-type") + try: + body = response.body() + sample.response_body_truncated = len(body) > 1_000_000 + if body and _is_body_worth_capturing(sample.content_type): + body = body[:1_000_000] + text = body.decode("utf-8", errors="replace") + sample.response_body = _parse_maybe_json(text) + except Exception as exc: # noqa: BLE001 + sample.error = f"{type(exc).__name__}: {exc}" + + page.on("request", on_request) + page.on("response", on_response) + + try: + page.goto(url, wait_until="domcontentloaded", timeout=45_000) + log.append(f"open: {url}") + except PlaywrightTimeoutError: + log.append("open warning: initial navigation timed out after domcontentloaded wait") + except Exception as exc: # noqa: BLE001 + 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) + page.wait_for_timeout(1500) + context.close() + browser.close() + + return list(samples_by_request.values()), log + + +def _fill_search_inputs(page: Any, search_term: str | None, log: list[str]) -> None: + if not search_term: + return + selectors = [ + "input[type='search']", + "input[name*='search' i]", + "input[placeholder*='search' i]", + "input[type='text']", + ] + for selector in selectors: + try: + locator = page.locator(selector).first + if locator.count() > 0 and locator.is_visible(timeout=1000): + locator.fill(search_term, timeout=2000) + locator.press("Enter", timeout=2000) + page.wait_for_timeout(2000) + log.append(f"interaction: filled {selector}") + return + except Exception: + continue + + +def _auto_scroll(page: Any, log: list[str]) -> None: + try: + for _ in range(3): + page.mouse.wheel(0, 900) + page.wait_for_timeout(700) + page.mouse.wheel(0, -400) + log.append("interaction: auto-scroll") + except Exception as exc: # noqa: BLE001 + log.append(f"scroll warning: {type(exc).__name__}: {exc}") + + +def _click_interesting_elements(page: Any, *, max_clicks: int, log: list[str]) -> None: + if max_clicks <= 0: + return + selectors = [ + "button:visible", + "[role='button']:visible", + "a[href]:visible", + ] + clicked = 0 + for selector in selectors: + if clicked >= max_clicks: + break + try: + count = min(page.locator(selector).count(), max_clicks * 3) + except Exception: + continue + for index in range(count): + if clicked >= max_clicks: + break + try: + item = page.locator(selector).nth(index) + text = (item.inner_text(timeout=800) or "").strip().lower() + href = item.get_attribute("href", timeout=800) if selector.startswith("a") else None + if href and href.startswith(("mailto:", "tel:", "#")): + continue + if href and urlparse(href).netloc and urlparse(href).netloc != urlparse(page.url).netloc: + continue + if _looks_destructive(text): + continue + item.click(timeout=2000) + page.wait_for_timeout(1600) + clicked += 1 + log.append(f"interaction: clicked {selector} #{index + 1}") + except Exception: + continue + + +def _looks_destructive(text: str) -> bool: + return any(word in text for word in ("delete", "remove", "logout", "sign out", "purchase", "checkout", "pay")) + + +def parse_trace_jsonl(*, requests_jsonl: str, responses_jsonl: str = "") -> list[TrafficSample]: + request_rows = [_json_loads(line) for line in requests_jsonl.splitlines() if line.strip()] + response_rows = [_json_loads(line) for line in responses_jsonl.splitlines() if line.strip()] + response_by_id = { + str(row.get("requestId") or row.get("id") or row.get("request_id")): row + for row in response_rows + if isinstance(row, dict) + } + samples: list[TrafficSample] = [] + for index, row in enumerate(row for row in request_rows if isinstance(row, dict)): + params = row.get("params") if isinstance(row.get("params"), dict) else row + req = params.get("request") if isinstance(params.get("request"), dict) else params + request_id = str(params.get("requestId") or row.get("requestId") or row.get("id") or index) + response = response_by_id.get(request_id, {}) + response_params = response.get("params") if isinstance(response.get("params"), dict) else response + response_obj = response_params.get("response") if isinstance(response_params.get("response"), dict) else response_params + samples.append( + TrafficSample( + request_id=request_id, + method=str(req.get("method") or "GET").upper(), + url=str(req.get("url") or params.get("url") or ""), + status=_int_or_none(response_obj.get("status")), + resource_type=params.get("type") or params.get("resourceType"), + request_headers=_dict_or_empty(req.get("headers")), + response_headers=_dict_or_empty(response_obj.get("headers")), + request_body=_parse_maybe_json(req.get("postData") or req.get("body")), + response_body=_parse_maybe_json(response_obj.get("body") or response_params.get("body")), + content_type=_content_type(response_obj.get("headers")), + timestamp=float(params.get("timestamp") or time.time()), + ) + ) + return [sample for sample in samples if sample.url] + + +def build_openapi_bundle( + samples: list[TrafficSample], + *, + title: str, + origins: list[str] | None = None, + include: list[str] | None = None, + exclude: list[str] | None = None, + min_samples: int = 1, + redact: list[str] | None = None, +) -> dict[str, Any]: + redactions = set(DEFAULT_REDACT_KEYS) + redactions.update(_normalize_redact_key(key) for key in (redact or [])) + filtered = [ + _redact_sample(sample, redactions) + for sample in samples + if _include_sample(sample, origins=origins, include=include, exclude=exclude) + ] + groups = group_samples(filtered) + openapi = emit_openapi(title=title, groups=groups, min_samples=min_samples) + confidence = emit_confidence(groups, min_samples=min_samples) + report = emit_markdown_report(title=title, groups=groups, confidence=confidence, min_samples=min_samples) + client = emit_client(openapi) + html_report = emit_html_report(title=title, markdown=report, confidence=confidence) + summary = summarize(groups, filtered, min_samples=min_samples) + return { + "openapi": openapi, + "confidence": confidence, + "report": report, + "html": html_report, + "client": client, + "samples": emit_samples(groups, min_samples=min_samples), + "summary": summary, + } + + +def _include_sample( + sample: TrafficSample, + *, + origins: list[str] | None, + include: list[str] | None, + exclude: list[str] | None, +) -> bool: + if not sample.url.startswith(("http://", "https://")): + return False + parsed = urlparse(sample.url) + origin = _origin(parsed) + if origins and not any(_origin_matches(origin, candidate) for candidate in origins): + return False + resource_type = (sample.resource_type or "").lower() + if resource_type and resource_type not in {"xhr", "fetch", "document"} and not _looks_like_api_url(sample.url): + return False + all_excludes = NOISE_URL_PATTERNS + list(exclude or []) + if any(_regex_search(pattern, sample.url) for pattern in all_excludes): + if not include or not any(_regex_search(pattern, sample.url) for pattern in include): + return False + if include and not any(_regex_search(pattern, sample.url) for pattern in include): + return False + content_type = sample.content_type or _content_type(sample.response_headers) + if sample.method == "GET" and content_type and "text/html" in content_type and not _looks_like_api_url(sample.url): + return False + return True + + +def group_samples(samples: list[TrafficSample]) -> list[EndpointGroup]: + groups: dict[str, EndpointGroup] = {} + for sample in samples: + parsed = urlparse(sample.url) + origin = _origin(parsed) + path, path_params = template_path(parsed.path or "/") + dispatch = _dispatch_name(sample, parsed) + display_path = path if not dispatch else f"{path}:{safe_operation_token(dispatch)}" + key = f"{sample.method} {origin}{display_path}" + group = groups.get(key) + if not group: + group = EndpointGroup( + key=key, + origin=origin, + method=sample.method.upper(), + path=display_path, + source_path=path, + dispatch=dispatch, + path_params=path_params, + ) + groups[key] = group + group.samples.append(sample) + query = dict(parse_qsl(parsed.query, keep_blank_values=True)) + for name, value in query.items(): + group.query_params.setdefault(name, []).append(value) + for group in groups.values(): + if len(group.samples) == 1: + group.flags.append("single-sample") + statuses = {sample.status for sample in group.samples if sample.status} + if len(statuses) <= 1: + group.flags.append("single-status") + content_types = {_content_type(sample.response_headers) for sample in group.samples if _content_type(sample.response_headers)} + if len(content_types) > 1: + group.flags.append("mixed-content-types") + if group.dispatch: + group.flags.append("dispatch-split") + return sorted(groups.values(), key=lambda group: (-len(group.samples), group.key)) + + +def emit_openapi(*, title: str, groups: list[EndpointGroup], min_samples: int) -> dict[str, Any]: + servers = sorted({group.origin for group in groups if len(group.samples) >= min_samples}) + spec: dict[str, Any] = { + "openapi": "3.1.0", + "info": { + "title": title, + "version": "0.1.0-discovered", + "description": "Generated from observed browser traffic. Coverage is bounded by captured flows.", + }, + "servers": [{"url": server} for server in servers], + "paths": {}, + "components": {"schemas": {}}, + } + schema_components: dict[str, Any] = {} + seen_schema_hashes: dict[str, str] = {} + + for group in groups: + if len(group.samples) < min_samples: + continue + method = group.method.lower() + path_item = spec["paths"].setdefault(group.path, {}) + request_bodies = [sample.request_body for sample in group.samples if sample.request_body is not None] + response_bodies = [sample.response_body for sample in group.samples if sample.response_body is not None] + operation_id = operation_id_for(group) + parameters = list(group.path_params) + parameters.extend(query_parameters_for(group)) + responses: dict[str, Any] = {} + for status in sorted({sample.status or 0 for sample in group.samples}): + status_key = str(status or "default") + matching = [sample.response_body for sample in group.samples if (sample.status or 0) == status and sample.response_body is not None] + response_schema = _schema_ref(infer_schema(matching or response_bodies), schema_components, seen_schema_hashes, operation_id + "Response") + responses[status_key] = { + "description": f"Observed {status_key} response", + "content": {"application/json": {"schema": response_schema}} if matching or response_bodies else {}, + } + if not responses: + responses["default"] = {"description": "Observed response", "content": {}} + operation: dict[str, Any] = { + "operationId": operation_id, + "summary": f"{group.method} {group.source_path}", + "parameters": parameters, + "responses": responses, + "x-origin": group.origin, + "x-source-path": group.source_path, + "x-sample-count": len(group.samples), + "x-confidence": confidence_for(group), + } + auth_headers = observed_auth_headers(group) + if auth_headers: + operation["x-observed-auth"] = auth_headers + if group.dispatch: + operation["x-dispatch-key"] = group.dispatch + if request_bodies: + request_schema = _schema_ref(infer_schema(request_bodies), schema_components, seen_schema_hashes, operation_id + "Request") + operation["requestBody"] = { + "required": True, + "content": {"application/json": {"schema": request_schema}}, + } + path_item[method] = operation + + spec["components"]["schemas"] = schema_components + return spec + + +def query_parameters_for(group: EndpointGroup) -> list[dict[str, Any]]: + total = max(len(group.samples), 1) + params = [] + for name, values in sorted(group.query_params.items()): + params.append( + { + "name": name, + "in": "query", + "required": len(values) == total, + "schema": infer_schema(values), + } + ) + return params + + +def emit_confidence(groups: list[EndpointGroup], *, min_samples: int) -> dict[str, Any]: + endpoints = [] + for group in groups: + response_body_known = any(sample.response_body is not None for sample in group.samples) + request_body_known = any(sample.request_body is not None for sample in group.samples) + statuses = sorted({sample.status for sample in group.samples if sample.status}) + endpoints.append( + { + "key": group.key, + "method": group.method, + "origin": group.origin, + "path": group.path, + "sourcePath": group.source_path, + "dispatch": group.dispatch, + "samples": len(group.samples), + "included": len(group.samples) >= min_samples, + "statusCodes": statuses, + "requestBodyKnown": request_body_known, + "responseBodyKnown": response_body_known, + "normalizationFlags": group.flags, + "confidence": confidence_for(group)["level"], + } + ) + return {"endpoints": endpoints} + + +def emit_markdown_report( + *, + title: str, + groups: list[EndpointGroup], + confidence: dict[str, Any], + min_samples: int, +) -> str: + lines = [ + f"# {title}", + "", + "Generated from observed browser traffic. This is an inductive API map, not a vendor contract.", + "", + f"- Observed endpoints: {len(groups)}", + f"- Included endpoints: {sum(1 for group in groups if len(group.samples) >= min_samples)}", + f"- Minimum samples: {min_samples}", + "", + "## Endpoints", + "", + ] + confidence_by_key = {entry["key"]: entry for entry in confidence["endpoints"]} + for group in groups: + entry = confidence_by_key[group.key] + included = "included" if entry["included"] else "below min_samples" + lines.extend( + [ + f"### `{group.method} {group.path}`", + "", + f"- Origin: `{group.origin}`", + f"- Samples: {len(group.samples)} ({included})", + f"- Confidence: `{entry['confidence']}`", + f"- Statuses: `{', '.join(map(str, entry['statusCodes'])) or 'unknown'}`", + f"- Flags: `{', '.join(group.flags) or 'none'}`", + "", + "```bash", + curl_example(group), + "```", + "", + ] + ) + lines.extend( + [ + "## Coverage Notes", + "", + "- Exercise more logged-in flows to reveal endpoints behind auth-gated UI.", + "- Run with `origins` restricted to the API host when marketing/vendor noise dominates.", + "- Raise `min_samples` to keep only endpoints seen repeatedly.", + "- Response schemas are strongest when the browser can read JSON response bodies.", + "", + ] + ) + return "\n".join(lines) + + +def emit_html_report(*, title: str, markdown: str, confidence: dict[str, Any]) -> str: + rows = [] + for endpoint in confidence["endpoints"]: + rows.append( + "" + f"{html.escape(endpoint['method'])}" + f"{html.escape(endpoint['path'])}" + f"{html.escape(endpoint['confidence'])}" + f"{endpoint['samples']}" + f"{'yes' if endpoint['responseBodyKnown'] else 'no'}" + f"{html.escape(', '.join(endpoint['normalizationFlags']) or 'none')}" + "" + ) + return f""" + + + + + {html.escape(title)} API discovery + + + +

{html.escape(title)} API discovery

+

OpenAPI, client, samples, and confidence files were generated next to this report.

+ + + {''.join(rows)} +
MethodPathConfidenceSamplesResponse bodyFlags
+

Markdown Report

+ 
{html.escape(markdown)}
+ + """ - return script + + +def emit_client(openapi: dict[str, Any]) -> str: + functions = [] + for path, item in sorted(openapi.get("paths", {}).items()): + for method, operation in sorted(item.items()): + name = operation.get("operationId") or safe_operation_token(f"{method}_{path}") + functions.append( + f"""export async function {name}(baseUrl, options = {{}}) {{ + const params = options.params || {{}}; + let path = {json.dumps(path)}; + for (const [key, value] of Object.entries(params.path || {{}})) {{ + path = path.replace(`{{${{key}}}}`, encodeURIComponent(String(value))); + }} + const query = new URLSearchParams(params.query || {{}}).toString(); + const response = await fetch(`${{baseUrl}}${{path}}${{query ? `?${{query}}` : ""}}`, {{ + method: {json.dumps(method.upper())}, + headers: {{ "content-type": "application/json", ...(options.headers || {{}}) }}, + body: options.body === undefined ? undefined : JSON.stringify(options.body), + }}); + const text = await response.text(); + return text ? JSON.parse(text) : null; +}}""" + ) + return "\n\n".join(functions) + "\n" + + +def emit_samples(groups: list[EndpointGroup], *, min_samples: int) -> dict[str, Any]: + output = {} + for group in groups: + if len(group.samples) < min_samples: + continue + sample = next((item for item in reversed(group.samples) if item.status and 200 <= item.status < 300), group.samples[-1]) + output[operation_id_for(group)] = { + "method": sample.method, + "url": sample.url, + "status": sample.status, + "requestHeaders": sample.request_headers, + "requestBody": sample.request_body, + "responseHeaders": sample.response_headers, + "responseBody": sample.response_body, + } + return output + + +def summarize(groups: list[EndpointGroup], samples: list[TrafficSample], *, min_samples: int) -> dict[str, Any]: + included = [group for group in groups if len(group.samples) >= min_samples] + return { + "included_samples": len(samples), + "endpoints": len(included), + "origins": sorted({group.origin for group in included}), + "top_endpoints": [ + { + "method": group.method, + "path": group.path, + "origin": group.origin, + "samples": len(group.samples), + "confidence": confidence_for(group)["level"], + } + for group in included[:10] + ], + } + + +def write_bundle(bundle: dict[str, Any], output_dir: Path) -> Path: + output_dir.mkdir(parents=True, exist_ok=True) + (output_dir / "samples").mkdir(exist_ok=True) + (output_dir / "openapi.json").write_text(json.dumps(bundle["openapi"], indent=2, sort_keys=True), encoding="utf-8") + (output_dir / "openapi.yaml").write_text(to_yaml(bundle["openapi"]), encoding="utf-8") + (output_dir / "confidence.json").write_text(json.dumps(bundle["confidence"], indent=2, sort_keys=True), encoding="utf-8") + (output_dir / "report.md").write_text(bundle["report"], encoding="utf-8") + (output_dir / "index.html").write_text(bundle["html"], encoding="utf-8") + (output_dir / "client.mjs").write_text(bundle["client"], encoding="utf-8") + for name, sample in bundle["samples"].items(): + (output_dir / "samples" / f"{name}.json").write_text(json.dumps(sample, indent=2, sort_keys=True), encoding="utf-8") + return output_dir + + +def infer_schema(values: list[Any]) -> dict[str, Any]: + values = [value for value in values if value is not None] + if not values: + return {} + if all(isinstance(value, dict) for value in values): + keys = sorted({key for value in values for key in value}) + required = [key for key in keys if all(key in value for value in values)] + properties = {key: infer_schema([value.get(key) for value in values if key in value]) for key in keys} + schema: dict[str, Any] = {"type": "object", "properties": properties} + if required: + schema["required"] = required + return schema + if all(isinstance(value, list) for value in values): + items = [item for value in values for item in value] + return {"type": "array", "items": infer_schema(items)} + type_names = sorted({_json_type(value) for value in values}) + schema = {"type": type_names[0] if len(type_names) == 1 else type_names} + if type_names == ["string"]: + formats = {_format_hint(str(value)) for value in values} + formats.discard(None) + if len(formats) == 1: + schema["format"] = formats.pop() + distinct = sorted({str(value) for value in values}) + if 1 < len(distinct) <= 8 and len(values) >= 5: + schema["enum"] = distinct + return schema + + +def _schema_ref(schema: dict[str, Any], components: dict[str, Any], seen: dict[str, str], name_hint: str) -> dict[str, Any]: + if not schema: + return {} + body = json.dumps(schema, sort_keys=True) + digest = hashlib.sha1(body.encode("utf-8")).hexdigest()[:12] + if digest in seen: + return {"$ref": f"#/components/schemas/{seen[digest]}"} + name = safe_operation_token(name_hint) + if not name[:1].isalpha(): + name = f"Schema{name}" + candidate = name + suffix = 2 + while candidate in components: + candidate = f"{name}{suffix}" + suffix += 1 + components[candidate] = schema + seen[digest] = candidate + return {"$ref": f"#/components/schemas/{candidate}"} + + +def template_path(path: str) -> tuple[str, list[dict[str, Any]]]: + params: list[dict[str, Any]] = [] + counters: dict[str, int] = {} + output = [] + for segment in path.split("/"): + if not segment: + continue + replacement, schema = _segment_template(segment) + if replacement: + counters[replacement] = counters.get(replacement, 0) + 1 + name = replacement if counters[replacement] == 1 else f"{replacement}{counters[replacement]}" + output.append(f"{{{name}}}") + params.append({"name": name, "in": "path", "required": True, "schema": schema}) + else: + output.append(segment) + return "/" + "/".join(output), params + + +def _segment_template(segment: str) -> tuple[str | None, dict[str, Any]]: + if re.fullmatch(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}", segment): + return "id", {"type": "string", "format": "uuid"} + if re.fullmatch(r"\d+", segment): + return "id", {"type": "integer"} + if re.fullmatch(r"[A-Za-z0-9_-]{12,}", segment): + return "id", {"type": "string"} + return None, {} + + +def _dispatch_name(sample: TrafficSample, parsed: Any) -> str | None: + body = sample.request_body + if isinstance(body, dict): + for key in ("operationName", "method", "action", "op", "opname"): + value = body.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + query = dict(parse_qsl(parsed.query, keep_blank_values=True)) + for key in ("operationName", "method", "action", "op", "opname"): + value = query.get(key) + if value: + return value + return None + + +def operation_id_for(group: EndpointGroup) -> str: + raw = f"{group.method}_{group.path}".replace("{", "by_").replace("}", "") + return safe_operation_token(raw) + + +def safe_operation_token(value: str) -> str: + token = re.sub(r"[^A-Za-z0-9_]+", "_", value).strip("_") + if not token: + token = "operation" + if token[0].isdigit(): + token = f"op_{token}" + return token[:96] + + +def confidence_for(group: EndpointGroup) -> dict[str, Any]: + samples = len(group.samples) + statuses = sorted({sample.status for sample in group.samples if sample.status}) + if samples >= 10 and len(statuses) > 1 and not group.flags: + level = "high" + elif samples >= 3 and not any(flag in group.flags for flag in ("mixed-content-types",)): + level = "medium" + else: + level = "low" + return {"level": level, "samples": samples, "statusCodes": statuses, "normalizationFlags": group.flags} + + +def observed_auth_headers(group: EndpointGroup) -> list[str]: + found = set() + for sample in group.samples: + for name in sample.request_headers: + clean = name.lower() + if clean in DEFAULT_REDACT_KEYS or "token" in clean or "secret" in clean or "signature" in clean: + found.add(clean) + return sorted(found) + + +def curl_example(group: EndpointGroup) -> str: + sample = next((item for item in reversed(group.samples) if item.status and 200 <= item.status < 300), group.samples[-1]) + parts = ["curl", "-X", sample.method, json.dumps(sample.url)] + if sample.request_body is not None: + parts.extend(["-H", json.dumps("content-type: application/json"), "--data", json.dumps(json.dumps(sample.request_body))]) + return " ".join(parts) + + +def to_yaml(value: Any, indent: int = 0) -> str: + spaces = " " * indent + if isinstance(value, dict): + lines = [] + for key, item in value.items(): + if isinstance(item, (dict, list)): + lines.append(f"{spaces}{key}:") + lines.append(to_yaml(item, indent + 2).rstrip()) + else: + lines.append(f"{spaces}{key}: {yaml_scalar(item)}") + return "\n".join(lines) + "\n" + if isinstance(value, list): + if not value: + return f"{spaces}[]\n" + lines = [] + for item in value: + if isinstance(item, (dict, list)): + lines.append(f"{spaces}-") + lines.append(to_yaml(item, indent + 2).rstrip()) + else: + lines.append(f"{spaces}- {yaml_scalar(item)}") + return "\n".join(lines) + "\n" + return f"{spaces}{yaml_scalar(value)}\n" + + +def yaml_scalar(value: Any) -> str: + if value is None: + return "null" + if value is True: + return "true" + if value is False: + return "false" + if isinstance(value, (int, float)): + return str(value) + text = str(value) + if not text or any(ch in text for ch in ":{}[]#,&*?|<>=!%@`\\n") or text.strip() != text: + return json.dumps(text) + return text + + +def _redact_sample(sample: TrafficSample, redactions: set[str]) -> TrafficSample: + data = asdict(sample) + data["request_headers"] = _redact_value(data["request_headers"], redactions) + data["response_headers"] = _redact_value(data["response_headers"], redactions) + data["request_body"] = _redact_value(data["request_body"], redactions) + data["response_body"] = _redact_value(data["response_body"], redactions) + return TrafficSample(**data) + + +def _redact_value(value: Any, redactions: set[str]) -> Any: + if isinstance(value, dict): + output = {} + for key, item in value.items(): + clean = _normalize_redact_key(str(key)) + output[key] = "" if clean in redactions or "token" in clean or "secret" in clean or "signature" in clean else _redact_value(item, redactions) + return output + if isinstance(value, list): + return [_redact_value(item, redactions) for item in value] + if isinstance(value, str): + if re.fullmatch(r"eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+", value): + return "" + if re.search(r"[^@\s]+@[^@\s]+\.[^@\s]+", value): + return "" + return value + + +def _parse_maybe_json(value: Any) -> Any: + if value is None or isinstance(value, (dict, list, int, float, bool)): + return value + text = str(value) + if not text: + return None + try: + return json.loads(text) + except Exception: + return text[:100_000] + + +def _json_loads(line: str) -> Any: + try: + return json.loads(line) + except Exception: + return None + + +def _json_type(value: Any) -> str: + if isinstance(value, bool): + return "boolean" + if isinstance(value, int) and not isinstance(value, bool): + return "integer" + if isinstance(value, float): + return "number" + if isinstance(value, str): + return "string" + if value is None: + return "null" + if isinstance(value, list): + return "array" + if isinstance(value, dict): + return "object" + return "string" + + +def _format_hint(value: str) -> str | None: + if re.fullmatch(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}", value): + return "uuid" + if re.fullmatch(r"\d{4}-\d{2}-\d{2}T.+", value): + return "date-time" + if value.startswith(("http://", "https://")): + return "uri" + if re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", value): + return "email" + return None + + +def _is_body_worth_capturing(content_type: str | None) -> bool: + if not content_type: + return False + clean = content_type.lower() + return any(kind in clean for kind in JSONISH_TYPES + TEXT_TYPES) + + +def _content_type(headers: Any) -> str | None: + if not isinstance(headers, dict): + return None + for key, value in headers.items(): + if str(key).lower() == "content-type": + return str(value).split(";")[0].strip().lower() + return None + + +def _dict_or_empty(value: Any) -> dict[str, Any]: + return value if isinstance(value, dict) else {} + + +def _int_or_none(value: Any) -> int | None: + try: + return int(value) + except Exception: + return None + + +def _origin(parsed: Any) -> str: + return f"{parsed.scheme}://{parsed.netloc}" + + +def _origin_matches(origin: str, candidate: str) -> bool: + if candidate.startswith(("http://", "https://")): + return origin == candidate.rstrip("/") + return urlparse(origin).netloc == candidate.strip().strip("/") + + +def _looks_like_api_url(url: str) -> bool: + parsed = urlparse(url) + path = parsed.path.lower() + return any(marker in path for marker in ("/api/", "/graphql", "/gql", "/rpc", "/v1/", "/v2/", "/v3/")) or "json" in path + + +def _regex_search(pattern: str, value: str) -> bool: + try: + return re.search(pattern, value, re.IGNORECASE) is not None + except re.error: + return False + + +def _normalize_redact_key(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "_", value.lower()).strip("_") + + +def _title_from_url(url: str) -> str: + host = urlparse(url).netloc or "Website" + return f"{host} Discovered API" + + +def safe_run_id(value: str) -> str: + stamp = time.strftime("%Y%m%d-%H%M%S") + token = re.sub(r"[^A-Za-z0-9]+", "-", value.lower()).strip("-")[:48] or "run" + return f"{stamp}-{token}" + + +def main() -> None: + parser = argparse.ArgumentParser(description="Discover an API from browser traffic.") + parser.add_argument("url") + parser.add_argument("--title") + parser.add_argument("--origin", action="append", dest="origins") + parser.add_argument("--wait-seconds", type=int, default=8) + parser.add_argument("--max-clicks", type=int, default=4) + args = parser.parse_args() + samples, log = capture_browser_traffic(url=args.url, wait_seconds=args.wait_seconds, max_clicks=args.max_clicks) + bundle = build_openapi_bundle(samples, title=args.title or _title_from_url(args.url), origins=args.origins) + output_dir = write_bundle(bundle, DEFAULT_OUTPUT_DIR / safe_run_id(args.url)) + print(json.dumps({"output_dir": str(output_dir), "summary": bundle["summary"], "capture_log": log}, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt index 2c9db06..b79f708 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ # a2a-pack is auto-installed by the deploy build. -# This agent installs SeleniumBase at runtime inside the sandbox script. -# Leave this file empty to avoid CLI dependency conflicts during card build. +playwright>=1.44 +seleniumbase==4.49.10 diff --git a/skills/compliant-browser-scraping/SKILL.md b/skills/compliant-browser-scraping/SKILL.md deleted file mode 100644 index 26b464b..0000000 --- a/skills/compliant-browser-scraping/SKILL.md +++ /dev/null @@ -1,45 +0,0 @@ ---- -name: compliant-browser-scraping -description: "Safety and workflow guidance for authorized public or user-owned browser scraping with SeleniumBase. Use for website scraping runs, selector extraction, pagination, robots.txt checks, challenge detection, and artifact reporting." ---- -# compliant-browser-scraping - -# Compliant Browser Scraping - -This agent is for authorized scraping of public or user-owned websites only. - -## Mandatory safety rules - -- Do not bypass, solve, defeat, evade, or work around CAPTCHAs, anti-bot protections, login gates, paywalls, robots.txt restrictions, ToS restrictions, IP blocks, access controls, or technical protection measures. -- If a CAPTCHA, anti-bot challenge, login wall, paywall, access denial, or equivalent challenge is detected, stop the run and return a clear `manual_intervention_required` or `blocked` status with warnings. -- Do not use stealth/undetected-browser modes, CAPTCHA solvers, proxy rotation, credential stuffing, cookie injection, or other circumvention techniques. -- Respect robots.txt when requested by the caller. -- Paginate only within the same scheme+host as the starting URL unless the caller starts a new scrape explicitly for another URL. -- Apply polite rate limiting between page visits. - -## Selector mapping conventions - -The public `selectors` input is a mapping from output field names to CSS selectors. Examples: - -```json -{ - "title": "h1", - "price": ".price", - "links": "a.result::attr(href)", - "next": "a[rel='next']::attr(href)" -} -``` - -Special selector names `next`, `_next`, `__next__`, `pagination_next`, or `next_page` are interpreted as pagination selectors and are not emitted as data fields. - -Selector suffixes: - -- `::text` extracts visible text (default behavior for element selectors). -- `::html` extracts innerHTML. -- `::attr(name)` extracts an attribute. - -When multiple elements match a field, return a list. When one element matches, return a string. When nothing matches, return null. - -## Output artifacts - -Save machine-readable output under `outputs/seleniumbase-website-scraper/` as JSON or CSV. If screenshots are requested, save PNG screenshots in the same directory. Include warnings and status in the response. diff --git a/skills/llm-iterative-scrape/SKILL.md b/skills/llm-iterative-scrape/SKILL.md deleted file mode 100644 index ffcad0a..0000000 --- a/skills/llm-iterative-scrape/SKILL.md +++ /dev/null @@ -1,31 +0,0 @@ ---- -name: llm-iterative-scrape -description: "Iterative compliant scraping workflow: plan -> bounded browser step -> evaluate -> refine until goal satisfied or safety/budget stops." ---- -# llm-iterative-scrape - -Use this skill for iterative, goal-driven scraping on authorized public or user-owned sites. - -Loop: -1) Translate the natural-language goal into a minimal set of CSS selectors and a short plan for ONE bounded browser step. -2) Execute exactly one small browser step with a polite wait/rate limit via the host agent's primitives. -3) Evaluate observations vs. the goal. Refine minimally. Stop when satisfied or when any safety/budget condition triggers. - -Hard safety constraints: -- Never solve or route around CAPTCHAs/anti-bot. -- Never log in or pass paywalls. -- Restrict pagination to same-origin as the start URL by default; also respect provided allowed_domains. -- When `respect_robots` is true, treat any robots.txt disallow or failure-to-check as a stop condition. - -Selectors guidance: -- Use compact CSS selectors. Suffix conventions: `::text`, `::html`, `::attr(name)`. -- Use one pagination field: `pagination_next` or `next` for the next-page `href`. -- Prefer domain-agnostic patterns like `a[rel='next']::attr(href)` or `li.next a::attr(href)`. - -Planning output contract: -Return strictly compact JSON only: {"plan_summary": str, "selectors": {str:str}, "continue": bool, "notes": str}. Keep under ~600 chars when possible. - -Evaluation guidance: -- If extracted data matches the requested fields and count threshold, set continue=false and summarize. -- If pagination is needed for more pages and a next link exists, keep continue=true. -- On any safety trigger (captcha/login/paywall/robots), set continue=false and summarize the stop reason. diff --git a/smoke_tests/quotes_toscrape.json b/smoke_tests/quotes_toscrape.json deleted file mode 100644 index a3398a7..0000000 --- a/smoke_tests/quotes_toscrape.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "url": "https://quotes.toscrape.com/", - "selectors": { - "quotes": ".quote", - "text": ".text", - "author": ".author", - "tags": ".tag" - }, - "max_pages": 1, - "wait_seconds": 1, - "output_format": "json", - "headless": true, - "user_agent": "", - "respect_robots": true, - "rate_limit_seconds": 1, - "screenshot": false -} diff --git a/smoke_tests/test_iterative_quotes_plan.py b/smoke_tests/test_iterative_quotes_plan.py deleted file mode 100644 index 7e49b8f..0000000 --- a/smoke_tests/test_iterative_quotes_plan.py +++ /dev/null @@ -1,46 +0,0 @@ -from __future__ import annotations - -import inspect -from pathlib import Path - - -def _load_agent_class(): - import importlib.util - - agent_path = Path(__file__).resolve().parents[1] / "agent.py" - spec = importlib.util.spec_from_file_location("agent", agent_path) - assert spec is not None and spec.loader is not None - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return getattr(module, "SeleniumbaseWebsiteScraper") - - -def test_iterative_signature_has_required_params() -> None: - Agent = _load_agent_class() - fn = getattr(Agent, "iterative_scrape_goal") - sig = inspect.signature(fn) - params = set(sig.parameters.keys()) - # skill method signature includes (self, ctx, ...) - required = { - "ctx", - "goal", - "start_url", - "allowed_domains", - "seed_selectors", - "max_steps", - "max_pages", - "max_runtime_seconds", - "wait_seconds", - "output_format", - "headless", - "user_agent", - "respect_robots", - "rate_limit_seconds", - "screenshot", - } - assert required.issubset(params) - - -if __name__ == "__main__": - test_iterative_signature_has_required_params() - print("iterative signature smoke ok") diff --git a/smoke_tests/test_render_scraper_script.py b/smoke_tests/test_render_scraper_script.py deleted file mode 100644 index 16e0a14..0000000 --- a/smoke_tests/test_render_scraper_script.py +++ /dev/null @@ -1,59 +0,0 @@ -from __future__ import annotations - -import ast -import importlib.util -import re -from pathlib import Path - - -def _load_agent_module(): - agent_path = Path(__file__).resolve().parents[1] / "agent.py" - spec = importlib.util.spec_from_file_location("agent", agent_path) - assert spec is not None and spec.loader is not None - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module - - -def _render_script() -> str: - module = _load_agent_module() - return module._render_scraper_script( - { - "url": "https://example.com/", - "selectors": {"title": "title"}, - "max_pages": 1, - "wait_seconds": 0, - "output_format": "json", - "headless": True, - "user_agent": "SmokeTest", - "respect_robots": True, - "rate_limit_seconds": 0, - "screenshot": False, - "output_dir": "/workspace/outputs/seleniumbase-website-scraper", - } - ) - - -def test_render_scraper_script_parses_as_python_35() -> None: - script = _render_script() - ast.parse(script, feature_version=(3, 5)) - - -def test_render_scraper_script_has_no_forbidden_modern_syntax_markers() -> None: - script = _render_script() - - numeric_underscore_literal = re.compile(r"(?