apodex/frontier-agent-demo
14
1"""Web scraping tool with academic URL routing."""2 3from __future__ import annotations4 5import asyncio6import logging7 8import httpx9 10from frontier_agent.core.tool import tool11from frontier_agent.infra.config import FrontierAgentConfig, get_config12from frontier_agent.infra.summary_llm import summarize as _summary_llm_summarize13from frontier_agent.infra.usage_meter import record_api_request14from plugins.tools._academic_fetch import (15 biorxiv_to_pdf,16 extract_pmcid,17 fetch_pmc_fulltext,18 fetch_unpaywall_oa_url,19 is_garbage_content,20 pubmed_to_pmc,21 resolve_doi,22 route_url,23)24from plugins.tools._bounded_fetch import (25 MAX_REDIRECT_HOPS,26 RedirectRefused,27 binary_content_type,28 blocked_download_url,29 next_hop,30 non_public_url_error,31 pin_to_address,32 read_bounded,33 strip_cross_origin_credentials,34 vet_public_url,35)36from plugins.tools._render_check import unrendered_kind37from plugins.tools._scrape_cache import (38 ScrapeUnavailable,39 format_skip_message,40 scrape_result_cache,41)42from plugins.tools._scrape_cache import cache as _scrape_cache43 44logger = logging.getLogger(__name__)45 46_MAX_RETRIES = 347_SHORT_CONTENT_THRESHOLD = 500 # re-try via Unpaywall when content is this short48_PAYWALL_SHORT_THRESHOLD = 500 # tighter check used for known paywall domains49 50# Below this many chars the raw page is returned verbatim instead of being51# routed through the summary LLM, even when ``info_to_extract`` is set. The52# summary LLM exists purely to keep 50KB+ pages from blowing the agent's53# context; a short page (a policy paragraph, a financial snippet, a small54# table) costs nothing to carry whole, and paraphrasing it through a55# temperature=1.0 extractor risks drift.56#57# Threshold picked from an offline A/B on real pages: long policy/legal58# text (13-24K chars) summarised faithfully, but a short, number-dense59# page fabricated a derived percentage and back-filled figures that were60# never on the page. Drift tracks number density, not length, so the line61# sits high enough to route short number-dense pages to raw (a few K62# tokens whole — no real context cost) while long-form still gets63# compressed. Tune upward if headroom allows.64_SUMMARY_MIN_CHARS = 12_00065 66 67@tool68async def web_fetch(69 url: str | list[str],70 info_to_extract: str | list[str] = "",71) -> str:72 """Scrape and extract content from one or more web pages.73 74 Automatic backend selection based on URL domain: PMC / PubMed / bioRxiv /75 medRxiv URLs go to the corresponding OA API; known paywall domains are76 routed via Unpaywall; everything else goes through Jina Reader. Retry,77 negative cache, and arXiv PDF→HTML redirect are applied automatically.78 79 A non-empty ``info_to_extract`` routes the raw content through a cheap80 summary LLM that extracts only the information requested. With it81 omitted the raw extracted text is returned (subject to overflow trim).82 83 Args:84 url: A URL string, or a list of URLs for parallel fetch.85 info_to_extract: Optional focus for the summary LLM. A single86 string applies to every URL; a list pairs with ``url`` 1:1.87 88 Returns:89 For a single URL, the extracted content directly. For a list, a90 numbered block per URL: ``[i] URL: …\\n Info: …``.91 """92 urls, focuses = _normalise_inputs(url, info_to_extract)93 if not urls:94 return "Error: URL is required."95 96 if len(urls) == 1:97 return await _fetch_one(urls[0], focuses[0])98 99 results = await asyncio.gather(100 *(_fetch_one(u, f) for u, f in zip(urls, focuses, strict=False))101 )102 return "\n\n".join(103 f"[{i}] URL: {u}\n Info: {r}"104 for i, (u, r) in enumerate(zip(urls, results, strict=False), 1)105 )106 107 108def _normalise_inputs(109 url: str | list[str], info_to_extract: str | list[str],110) -> tuple[list[str], list[str]]:111 """Coerce the LangChain payload into paired URL + focus lists."""112 from plugins.tools._coerce import coerce_json_list113 urls = coerce_json_list(url) if isinstance(url, str) else url114 if isinstance(urls, str):115 urls = [urls]116 elif not isinstance(urls, list):117 urls = []118 urls = [u.strip() for u in urls if isinstance(u, str) and u.strip()]119 120 focuses = (121 coerce_json_list(info_to_extract) if isinstance(info_to_extract, str)122 else info_to_extract123 )124 if isinstance(focuses, str):125 focuses = [focuses] * len(urls)126 elif isinstance(focuses, list):127 focuses = [str(f) for f in focuses]128 if len(focuses) < len(urls):129 focuses = focuses + [""] * (len(urls) - len(focuses))130 else:131 focuses = [""] * len(urls)132 return urls, focuses133 134 135async def _fetch_one(url: str, info_to_extract: str) -> str:136 """Run the full fetch pipeline for a single URL."""137 if not url or not url.strip():138 return "Error: URL is required."139 140 url = url.strip()141 if not url.startswith(("http://", "https://")):142 url = f"https://{url}"143 144 # Vet the target before ANYTHING leaves the process — including the scrape145 # provider's request, which would otherwise be handed an internal URL.146 non_public = await non_public_url_error(url)147 if non_public:148 return (149 f"[BLOCKED] {non_public}. Only public http(s) endpoints may be "150 "fetched. Use a public source."151 )152 153 # Operator hard-block (WEB_DOMAIN_BLACKLIST_EXTRA) — same list that154 # filters web_search results, enforced here so a direct URL from page155 # content cannot bypass it.156 from plugins.tools.web_search import is_domain_blocked157 if is_domain_blocked(url):158 return (159 f"[BLOCKED] The domain of {url} is on the operator blocklist "160 f"and must not be accessed. Use a different source."161 )162 163 blocked_ext = blocked_download_url(url)164 if blocked_ext:165 return (166 f"[BLOCKED] This URL is a dataset/archive download ({blocked_ext}), "167 "not a web page. Do not download data files — read the dataset's "168 "documentation/landing page instead, or use aggregate API queries."169 )170 171 # arXiv PDFs consistently fail extraction — redirect to HTML abstract page.172 if "arxiv.org/pdf/" in url:173 html_url = url.replace("/pdf/", "/abs/").split(".pdf")[0]174 logger.info("Redirecting arxiv PDF → HTML: %s", html_url)175 url = html_url176 elif "arxiv.org/pdf" in url:177 html_url = url.replace("/pdf", "/abs")178 logger.info("Redirecting arxiv PDF → HTML: %s", html_url)179 url = html_url180 181 # Negative-cache check: skip URLs that recently returned 403/422/429.182 cached = _scrape_cache.check(url)183 if cached is not None:184 msg = format_skip_message(url, cached)185 logger.info("web_fetch skipped (cached %d): %s", cached.status, url[:60])186 return msg187 188 config = get_config()189 route = route_url(url)190 191 # A fresh scrape is classified in the scrape function, the cache predicate,192 # and the caller-facing warning path. Keep the verdicts for this one fetch193 # so HTML parsing happens once per distinct response body.194 render_verdicts: list[tuple[str, str | None]] = []195 196 def _render_kind(content: str) -> str | None:197 for seen, verdict in render_verdicts:198 if content is seen or content == seen:199 return verdict200 verdict = unrendered_kind(content)201 render_verdicts.append((content, verdict))202 return verdict203 204 # Single-flight cross-run cache: sibling agents fetching the same URL share205 # one round-trip. Only validated, non-garbage content is cached; the empty206 # / garbage branches raise so the failure is neither stored nor shared.207 async def _scrape() -> str:208 content = await _fetch_via_route(url, route, config)209 # Post-fetch quality check: Jina can return a CAPTCHA/login page as a210 # 200, and some un-listed paywalls slip past the route table. Try one211 # Unpaywall-driven retry (cheap — fails fast when no DOI resolves).212 content = await _maybe_recover_via_unpaywall(url, route, content, config)213 if not content:214 raise ScrapeUnavailable("empty")215 # Final garbage check — Jina can return a CAPTCHA/login page as success.216 if is_garbage_content(content):217 raise ScrapeUnavailable("garbage")218 # A pre-hydration DOM survived even the browser-engine escalation219 # above. Only the high-confidence ``shell`` verdict fails here: a220 # merely short body ("empty") is legitimate content often enough that221 # erroring on it would lose real pages.222 if _render_kind(content) == "shell":223 raise ScrapeUnavailable("unrendered")224 return content225 226 try:227 content = await scrape_result_cache.get_or_scrape(228 url,229 _scrape,230 # A low-confidence short page is still useful enough to return,231 # but it may be a pre-hydration race. Never let it become the232 # process-wide answer for every later fetch of this URL.233 should_cache=lambda scraped: _render_kind(scraped) is None,234 )235 except ScrapeUnavailable as exc:236 if str(exc) == "unrendered":237 return (238 f"[NOT RENDERED] The page at {url} is a JavaScript app that "239 f"served no content even with browser rendering. Fetching it "240 f"again the same way will not help: look for the same material "241 f"at another source (the site's own API/JSON endpoint, a PDF, "242 f"or an archive copy), or search for the page title instead."243 )244 if str(exc) == "garbage":245 return (246 f"[ACCESS BLOCKED] The page at {url} is behind a paywall or "247 f"anti-bot protection. Please try searching for an open-access "248 f"version (arxiv.org, PMC, institutional repositories)."249 )250 return f"Could not extract content from {url}"251 252 _scrape_cache.record_success(url)253 254 # LLM extraction if requested — uses dedicated SUMMARY_LLM_* config and255 # gracefully returns truncated raw content when unconfigured. Skipped for256 # short pages (< _SUMMARY_MIN_CHARS): they carry whole at no context cost,257 # and returning them verbatim avoids paraphrase drift on the exact numbers258 # / scope qualifiers the agent asked for. The focus is still served — the259 # agent reads ``info_to_extract`` straight from the raw page.260 if (261 info_to_extract262 and info_to_extract.strip()263 and len(content) >= _SUMMARY_MIN_CHARS264 ):265 output = await _summary_llm_summarize(content, info_to_extract)266 else:267 from plugins.tools._overflow import maybe_overflow268 output = maybe_overflow("web_fetch", content)269 270 if _render_kind(content) == "empty":271 return (272 f"[POSSIBLY NOT RENDERED] The page at {url} remained suspiciously "273 "short after the browser-render retry. This result was deliberately "274 "not cached. Use it if it answers the question; otherwise switch to "275 "another source instead of repeatedly fetching this URL.\n\n"276 f"{output}"277 )278 return output279 280 281# ── Routing ───────────────────────────────────────────────────────────────282 283async def _fetch_via_route(url: str, route: str, config: FrontierAgentConfig) -> str:284 """Dispatch to the domain-specific backend; always falls back to Jina."""285 if route == "pmc":286 return await _fetch_pmc(url, config)287 if route == "pubmed":288 return await _fetch_pubmed(url, config)289 if route == "biorxiv":290 return await _fetch_biorxiv(url, config)291 if route == "paywall":292 return await _fetch_paywall(url, config)293 294 # Generic route — Jina first, then trafilatura fallback.295 content = ""296 if config.jina_api_key:297 content = await _jina_scrape(url, config.jina_api_key, config.jina_base_url)298 if content.startswith("[UPSTREAM_EXHAUSTED"):299 direct = await _direct_scrape(url)300 return direct or f"Error: {content}"301 if not content:302 content = await _direct_scrape(url)303 return content304 305 306async def _fetch_pmc(url: str, config: FrontierAgentConfig) -> str:307 pmcid = extract_pmcid(url)308 if pmcid:309 text = await fetch_pmc_fulltext(pmcid)310 if text:311 return text312 return await _jina_or_empty(url, config)313 314 315async def _fetch_pubmed(url: str, config: FrontierAgentConfig) -> str:316 pmcid = await pubmed_to_pmc(url)317 if pmcid:318 text = await fetch_pmc_fulltext(pmcid)319 if text:320 return text321 return await _jina_or_empty(url, config)322 323 324async def _fetch_biorxiv(url: str, config: FrontierAgentConfig) -> str:325 pdf_url = biorxiv_to_pdf(url)326 text = ""327 if pdf_url:328 logger.info("[bioRxiv] Auto PDF: %s", pdf_url)329 text = await _jina_or_empty(pdf_url, config)330 if not text or len(text) < _SHORT_CONTENT_THRESHOLD:331 fallback = await _jina_or_empty(url, config)332 if fallback and len(fallback) > len(text):333 text = fallback334 return text335 336 337async def _fetch_paywall(url: str, config: FrontierAgentConfig) -> str:338 doi = await resolve_doi(url)339 text = ""340 if doi:341 oa_url = await fetch_unpaywall_oa_url(doi)342 if oa_url:343 logger.info("[Paywall bypass] %s → OA PDF: %s", url[:60], oa_url)344 text = await _jina_or_empty(oa_url, config)345 if not text or len(text) < _PAYWALL_SHORT_THRESHOLD:346 fallback = await _jina_or_empty(url, config)347 if fallback and len(fallback) > len(text):348 text = fallback349 return text350 351 352async def _jina_or_empty(url: str, config: FrontierAgentConfig) -> str:353 """Run Jina with the existing retry/cache logic; empty string on failure."""354 if not config.jina_api_key:355 return ""356 try:357 return await _jina_scrape(url, config.jina_api_key, config.jina_base_url)358 except Exception as exc:359 logger.warning("Jina fetch failed for %s: %s", url[:60], exc)360 return ""361 362 363async def _maybe_recover_via_unpaywall(364 url: str, route: str, content: str, config: FrontierAgentConfig,365) -> str:366 """If the routed fetch returned garbage or was suspiciously short on a367 paywall domain, attempt one Unpaywall-driven retry.368 369 Skipped for ``pmc``/``pubmed`` — if the BioC API couldn't resolve the370 article there's no DOI detour worth trying that Jina wouldn't already hit.371 """372 if route in ("pmc", "pubmed"):373 return content374 375 garbage = is_garbage_content(content)376 short_on_paywall = (377 route == "paywall"378 and content379 and len(content) < _PAYWALL_SHORT_THRESHOLD380 )381 if not garbage and not short_on_paywall:382 return content383 384 reason = "garbage" if garbage else "short"385 logger.warning(386 "[Quality] %s content detected for %s — trying Unpaywall fallback",387 reason, url[:60],388 )389 doi = await resolve_doi(url)390 if not doi:391 return content392 oa_url = await fetch_unpaywall_oa_url(doi)393 if not oa_url:394 return content395 396 alt = await _jina_or_empty(oa_url, config)397 if alt and not is_garbage_content(alt) and len(alt) > len(content):398 logger.info("[Quality] Unpaywall fallback succeeded: %d chars", len(alt))399 return alt400 return content401 402 403def _extract_jina_body(raw: str) -> str:404 """Return the usable content from a Jina response body.405 406 Jina's POST endpoint returns JSON wrapped as ``{"code":200,"status":...,407 "data":{"title":...,"content":"markdown..."}}`` when ``Accept: application/json``408 is set. Some proxies may return raw markdown instead. Handle both shapes409 and also detect balance errors.410 """411 if not raw:412 return ""413 stripped = raw.lstrip()414 if not stripped.startswith("{"):415 return raw416 try:417 import json as _json418 data = _json.loads(raw)419 except (ValueError, TypeError):420 return raw421 422 if isinstance(data, dict):423 # Balance-error sentinel.424 if data.get("name") == "InsufficientBalanceError":425 logger.error("Jina API: Insufficient balance")426 return ""427 # Wrapped success payload.428 inner = data.get("data")429 if isinstance(inner, dict) and isinstance(inner.get("content"), str):430 return inner["content"]431 return raw432 433 434async def _jina_request(435 url: str,436 api_key: str,437 base_url: str,438 *,439 browser: bool,440) -> tuple[int, str]:441 """Single Jina call. Returns (status_code, body_text). Status -1 on transport error.442 443 ``browser=True`` switches the Jina extraction engine to ``browser`` (real444 page render), which recovers many origins that block the direct fetch445 engine with 403/422. Uses a longer ``X-Timeout`` so dynamic content has446 time to load.447 """448 headers = {449 "Authorization": f"Bearer {api_key}",450 "Content-Type": "application/json",451 "Accept": "application/json",452 "X-Timeout": "30" if browser else "20",453 }454 if browser:455 headers["X-Engine"] = "browser"456 # The escalation exists because the first attempt produced nothing457 # usable; Jina caches its own responses, so without this the retry can458 # be answered from that same result.459 headers["X-No-Cache"] = "true"460 461 try:462 async with httpx.AsyncClient(timeout=60) as client:463 async with client.stream(464 "POST",465 base_url,466 headers=headers,467 json={"url": url},468 ) as resp:469 # Bounded read — a Jina conversion of a huge document470 # must not buffer past the byte cap.471 body, _ = await read_bounded(resp)472 # One answered Jina request is billable; non-2xx473 # statuses still consumed a reader call upstream, so count474 # them as requests too and tag the error separately.475 record_api_request(476 "jina", errors=0 if resp.status_code < 400 else 1,477 )478 # Jina returns UTF-8 JSON but omits ``charset`` in Content-Type,479 # which makes httpx fall back to chardet sniffing — chardet480 # mis-classifies CJK-heavy bodies with mostly-ASCII headers as481 # Windows-1252, producing mojibake like ``千と千尋`` →482 # ``åã¨åå°``. Force UTF-8 to skip the sniff entirely.483 return resp.status_code, body.decode("utf-8", errors="replace")484 except httpx.TimeoutException:485 record_api_request("jina", requests=0, errors=1)486 return -1, "timeout"487 except Exception as e:488 logger.error("Jina transport error for %s: %s", url[:60], e)489 record_api_request("jina", requests=0, errors=1)490 return -1, str(e)491 492 493async def _jina_scrape(url: str, api_key: str, base_url: str) -> str:494 """Scrape via Jina Reader with escalation retry.495 496 Strategy:497 1. Default (fast) engine first.498 2. On 403/422, escalate once to ``X-Engine: browser`` — covers SPAs and499 origins that block the direct fetch engine. Only caches the URL as500 banned if the browser-engine attempt also fails.501 3. Also escalate on a 200 whose body came back un-rendered: an SPA that had502 not hydrated when the reader captured it answers 200 with a503 navigation-only DOM, so a status-only rule never noticed (see504 ``_render_check``). The retry additionally bypasses Jina's own response505 cache, which would otherwise hand back the same shell.506 4. Retry 429/5xx with exponential backoff (unchanged).507 """508 escalated = False509 # Only a 429 means Jina capped us for volume; exhausting the retries against510 # 5xx is an outage. The two must not reach the visitor as the same message.511 rate_limited = False512 513 for attempt in range(_MAX_RETRIES):514 status, body = await _jina_request(url, api_key, base_url, browser=escalated)515 516 if status == 200:517 content = _extract_jina_body(body)518 kind = unrendered_kind(content)519 if kind is not None and not escalated:520 logger.info(521 "Jina returned an un-rendered page for %s (%s, %d chars) "522 "— escalating to browser engine", url[:60], kind, len(content),523 )524 escalated = True525 continue526 return content527 528 if status == 403:529 if not escalated:530 logger.info("Jina 403 for %s — escalating to browser engine", url[:60])531 escalated = True532 continue533 _scrape_cache.record_failure(url, 403)534 logger.warning("Jina 403 for %s after browser retry — cached (1h)", url[:60])535 return ""536 537 if status == 422:538 if not escalated:539 logger.info("Jina 422 for %s — escalating to browser engine", url[:60])540 escalated = True541 continue542 _scrape_cache.record_failure(url, 422)543 logger.info("Jina 422 for %s after browser retry — recorded", url[:60])544 return ""545 546 if status == 429:547 rate_limited = True548 wait = 2 ** attempt549 logger.warning("Jina 429 rate limited for %s, retrying in %ds", url[:60], wait)550 await asyncio.sleep(wait)551 continue552 553 if status >= 500:554 wait = 2 ** attempt555 logger.warning("Jina %d server error for %s, retrying in %ds", status, url[:60], wait)556 await asyncio.sleep(wait)557 continue558 559 if status == -1:560 # Transport error / timeout — treat like a soft retry.561 logger.warning("Jina transport failure for %s: %s (attempt %d)", url[:60], body[:60], attempt + 1)562 if attempt < _MAX_RETRIES - 1:563 await asyncio.sleep(1)564 continue565 return ""566 567 # Any other status — log and abort without caching.568 logger.error("Jina HTTP %d for %s (body: %s)", status, url[:60], body[:200])569 if status == 402:570 return "[UPSTREAM_EXHAUSTED provider=jina status=402]"571 return ""572 573 # Retries exhausted on 429/5xx — record as rate-limited so we back off.574 _scrape_cache.record_failure(url, 429)575 logger.error("Jina scrape exhausted retries for %s — cached 429 (5min)", url[:60])576 return "[UPSTREAM_EXHAUSTED provider=jina status=429]" if rate_limited else ""577 578 579async def _direct_scrape(url: str) -> str:580 """Direct scrape with httpx + trafilatura extraction."""581 try:582 import trafilatura583 except ImportError:584 return ""585 586 try:587 headers = {588 # Sent to every site this scrapes. Identify your deployment589 # here if you want operators to be able to contact you — the590 # default names the project, not any account.591 "User-Agent": (592 "Mozilla/5.0 (compatible; FrontierAgent/1.0; "593 "+https://github.com/ApodexAI/FrontierAgent)"594 ),595 }596 # Redirects are walked by hand so every hop is vetted: following them597 # automatically would let a vetted public URL hand us a 302 to598 # localhost or a metadata endpoint (see ``next_hop``).599 async with httpx.AsyncClient(timeout=60, follow_redirects=False) as client:600 hop_url, hop_headers = url, headers601 for _ in range(MAX_REDIRECT_HOPS):602 refusal, addresses = await vet_public_url(hop_url)603 if refusal:604 raise RedirectRefused(refusal)605 dial_url, dial_headers, extensions = pin_to_address(606 hop_url, addresses, hop_headers,607 )608 async with client.stream(609 "GET", dial_url, headers=dial_headers, extensions=extensions,610 ) as resp:611 target = await next_hop(resp, hop_url)612 if target is not None:613 hop_headers = strip_cross_origin_credentials(614 hop_headers, hop_url, target,615 )616 hop_url = target617 continue618 resp.raise_for_status()619 # A data blob (dataset zip, media) has no page text for620 # trafilatura; skip the download instead of buffering it.621 if binary_content_type(resp.headers.get("content-type")):622 logger.warning(623 "Direct scrape skipped binary content for %s", url[:60],624 )625 return ""626 body, _ = await read_bounded(resp)627 break628 else:629 logger.warning("Direct scrape exceeded redirect limit for %s", url[:60])630 return ""631 except RedirectRefused as exc:632 logger.warning("Direct scrape refused a redirect for %s: %s", url[:60], exc)633 return ""634 except httpx.TimeoutException:635 logger.warning("Direct scrape timeout for %s", url[:60])636 return ""637 except Exception as e:638 logger.warning("Direct scrape failed for %s: %s", url[:60], e)639 return ""640 641 # Pass raw bytes so trafilatura reads the HTML ``<meta charset>``642 # itself instead of trusting httpx's chardet sniff — pages that mix643 # heavy ASCII boilerplate with a small CJK payload get mis-classified644 # as Windows-1252 by chardet (UTF-8 → mojibake).645 return trafilatura.extract(body) or ""646