deploy

agent.py

@@ -24,7 +24,7 @@ from pathlib import Path
 from typing import Any
 from urllib.parse import parse_qsl, urljoin, urlparse
 
-from a2a_pack import A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill
+from a2a_pack import A2AAgent, EgressPolicy, LLMProvisioning, NoAuth, Pricing, RunContext, skill
 from pydantic import BaseModel, Field
 
 
@@ -84,6 +84,14 @@ NOISE_URL_PATTERNS = [
     r"/favicon\.ico(\?|$)",
 ]
 DEFAULT_CAPTCHA_SOLVE_DOMAINS = ("a2acloud.io", ".a2acloud.io")
+ALLOWED_TARGET_DOMAINS = ("a2acloud.io", ".a2acloud.io")
+ALLOWED_EGRESS_HOSTS = (
+    "a2acloud.io",
+    "www.a2acloud.io",
+    "blog.a2acloud.io",
+    "docs.a2acloud.io",
+    "api.a2acloud.io",
+)
 CAPTCHA_POLICIES = {"detect_only", "solve_if_allowed", "never"}
 BROWSER_MODES = {"seleniumbase_stealth", "playwright"}
 
@@ -135,6 +143,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
     auth_model = NoAuth
     llm_provisioning = LLMProvisioning.PLATFORM
     pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False)
+    egress = EgressPolicy(allow_hosts=ALLOWED_EGRESS_HOSTS, deny_internet_by_default=True)
     tools_used = ("seleniumbase", "playwright", "browser-to-api", "openapi")
     capabilities = {
         "browser_to_api": {
@@ -151,7 +160,8 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
             ],
             "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.",
-            "captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains.",
+            "target_scope": "Public browser capture is restricted to a2acloud.io and subdomains.",
+            "captcha_policy": "SeleniumBase solve_captcha is fixed to a2acloud.io and subdomains.",
             "llm_exploration": "Uses ctx.llm to choose bounded, non-destructive page actions after the page loads.",
         }
     }
@@ -180,13 +190,18 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
         redact: list[str] | None = None,
         browser_mode: str = "seleniumbase_stealth",
         captcha_policy: str = "solve_if_allowed",
-        allowed_captcha_domains: list[str] | None = None,
         ai_explore: bool = True,
         ai_steps: int = 10,
         ai_goal: str | None = None,
     ) -> dict[str, Any]:
-        if not url or not str(url).strip().startswith(("http://", "https://")):
+        clean_url = str(url).strip()
+        if not clean_url or not clean_url.startswith(("http://", "https://")):
             return {"error": "url must start with http:// or https://"}
+        if not _host_allowed_for_target(clean_url):
+            return {
+                "error": "url host is outside this public agent's allowed target scope",
+                "allowed_domains": list(ALLOWED_TARGET_DOMAINS),
+            }
         if browser_mode not in BROWSER_MODES:
             return {"error": f"browser_mode must be one of: {', '.join(sorted(BROWSER_MODES))}"}
         if captcha_policy not in CAPTCHA_POLICIES:
@@ -196,13 +211,12 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
             await ctx.emit_progress("launching browser and capturing network traffic")
             samples, capture_log = await asyncio.to_thread(
                 capture_browser_traffic,
-                url=str(url).strip(),
+                url=clean_url,
                 wait_seconds=max(1, min(int(wait_seconds), 60)),
                 max_clicks=max(0, min(int(max_clicks), 20)),
                 search_term=search_term,
                 browser_mode=browser_mode,
                 captcha_policy=captcha_policy,
-                allowed_captcha_domains=allowed_captcha_domains,
                 ai_explore=ai_explore,
                 ai_steps=max(0, min(int(ai_steps), 20)),
                 ai_goal=ai_goal,
@@ -223,14 +237,14 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
 
             await ctx.emit_progress(f"wrote OpenAPI bundle to {output_dir}")
             return _result_payload(
-                url=url,
+                url=clean_url,
                 samples=samples,
                 bundle=bundle,
                 output_dir=output_dir,
                 capture_log=capture_log,
                 browser_mode=browser_mode,
                 captcha_policy=captcha_policy,
-                allowed_captcha_domains=_effective_captcha_domains(allowed_captcha_domains),
+                allowed_captcha_domains=_effective_captcha_domains(None),
                 ai_explore=ai_explore,
                 ai_steps=max(0, min(int(ai_steps), 20)),
             )
@@ -994,14 +1008,18 @@ def _seleniumbase_sleep(sb: Any, seconds: int) -> None:
 
 
 def _effective_captcha_domains(allowed_captcha_domains: list[str] | None) -> list[str]:
-    domains = [str(item).strip().lower() for item in (allowed_captcha_domains or []) if str(item).strip()]
-    for domain in DEFAULT_CAPTCHA_SOLVE_DOMAINS:
-        if domain not in domains:
-            domains.append(domain)
-    return domains
+    return list(DEFAULT_CAPTCHA_SOLVE_DOMAINS)
+
+
+def _host_allowed_for_target(url: str) -> bool:
+    return _host_allowed_for_domains(url, list(ALLOWED_TARGET_DOMAINS))
 
 
 def _host_allowed_for_captcha(url: str, allowed_domains: list[str]) -> bool:
+    return _host_allowed_for_domains(url, allowed_domains)
+
+
+def _host_allowed_for_domains(url: str, allowed_domains: list[str]) -> bool:
     host = urlparse(url).hostname or ""
     host = host.lower().rstrip(".")
     for raw_domain in allowed_domains: