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

deploy

Browse Source
This commit is contained in:
a2a-platform
2026-06-27 01:23:07 +00:00
parent 6a6f5122b7
commit 46b4132727
1 changed files with 72 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
agent.py
View File

@@ -11,6 +11,7 @@ import asyncio
import hashlib
import html
import json
import os
import re
import time
from dataclasses import asdict, dataclass, field
@@ -22,7 +23,7 @@ from a2a_pack import A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, ski
from pydantic import BaseModel, Field
DEFAULT_OUTPUT_DIR = Path("outputs/browser-to-api")
DEFAULT_OUTPUT_DIR = Path(os.environ.get("BROWSER_TO_API_OUTPUT_DIR", "/tmp/browser-to-api"))
JSONISH_TYPES = (
"application/json",
"application/problem+json",
@@ -166,6 +167,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
if not url or not str(url).strip().startswith(("http://", "https://")):
return {"error": "url must start with http:// or https://"}
try:
await ctx.emit_progress("launching browser and capturing network traffic")
samples, capture_log = await asyncio.to_thread(
capture_browser_traffic,
@@ -188,22 +190,9 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
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"],
}
return _result_payload(url=url, samples=samples, bundle=bundle, output_dir=output_dir, capture_log=capture_log)
except Exception as exc: # noqa: BLE001
return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"}
@skill(
name="discover_api_from_trace",
@@ -226,6 +215,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
min_samples: int = 1,
redact: list[str] | None = None,
) -> dict[str, Any]:
try:
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(
@@ -239,20 +229,9 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
)
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"],
}
return _result_payload(samples=samples, bundle=bundle, output_dir=output_dir)
except Exception as exc: # noqa: BLE001
return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"}
def capture_browser_traffic(
@@ -842,6 +821,35 @@ def write_bundle(bundle: dict[str, Any], output_dir: Path) -> Path:
return output_dir
def _result_payload(
*,
samples: list[TrafficSample],
bundle: dict[str, Any],
output_dir: Path,
url: str | None = None,
capture_log: list[str] | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"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"],
}
if url is not None:
payload["url"] = url
if capture_log is not None:
payload["capture_log"] = capture_log
return payload
def infer_schema(values: list[Any]) -> dict[str, Any]:
values = [value for value in values if value is not None]
if not values: