apodex/frontier-agent-demo
14
1"""Jina-backed web fetch, byte-compatible with the reference agent's tool.2 3Extraction prompt, retry schedule and output formatting are reproduced4exactly: they shape what the agent sees, and the model was trained on this5form.6"""7 8from __future__ import annotations9 10import asyncio11import json12import logging13import os14from typing import Any15 16import httpx17 18from frontier_agent.core.tool import tool19from frontier_agent.infra.summary_llm import (20 FALLBACK_TRUNCATE,21 summary_llm_candidates,22)23from frontier_agent.infra.usage_meter import record_api_request, record_llm_usage24from plugins.tools._bounded_fetch import (25 MAX_REDIRECT_HOPS,26 RedirectRefused,27 binary_content_type,28 blocked_download_url,29 decode_body,30 next_hop,31 non_public_url_error,32 pin_to_address,33 read_bounded,34 strip_cross_origin_credentials,35 vet_public_url,36)37from plugins.tools._render_check import (38 MIN_RENDERED_BODY_CHARS,39 reader_body,40 unrendered_kind,41)42from plugins.tools._scrape_cache import ScrapeUnavailable, scrape_result_cache43 44logger = logging.getLogger(__name__)45 46 47# ── Env (read at call time, not import time — see web_search_aligned.py) ──48 49 50def _jina_api_key() -> str:51 return os.getenv("JINA_API_KEY", "")52 53 54def _jina_base_url() -> str:55 # Reference default: ``https://r.jina.ai`` (direct). The .env can56 # override to a proxy without changing this fallback.57 return os.getenv("JINA_BASE_URL", "https://r.jina.ai")58 59 60def _summary_llm_base_url() -> str | None:61 return os.environ.get("SUMMARY_LLM_BASE_URL")62 63 64def _summary_llm_model_name() -> str | None:65 return os.environ.get("SUMMARY_LLM_MODEL_NAME")66 67 68def _summary_llm_api_key() -> str | None:69 return os.environ.get("SUMMARY_LLM_API_KEY")70 71 72# Matches the reference tool's list. Just two patterns —73# Twitter/Reddit/etc are intentionally fetchable.74_BANNED_URL_PATTERNS: tuple[str, ...] = (75 "huggingface.co/datasets",76 "huggingface.co/spaces",77)78 79 80def _ensure_list(val: Any) -> Any:81 """Unwrap doubly-serialised JSON list payloads (mirrors web_search)."""82 if isinstance(val, str) and val.startswith("["):83 try:84 parsed = json.loads(val)85 if isinstance(parsed, list):86 return parsed87 except (json.JSONDecodeError, ValueError):88 pass89 return val90 91 92# ── Un-rendered ("app shell") handling ────────────────────────────────93#94# Detection lives in ``_render_check`` (shared with ``web_fetch.py``). Aliased95# to the module-private names this file uses elsewhere.96_reader_body = reader_body97_unrendered_kind = unrendered_kind98_MIN_RENDERED_BODY_CHARS = MIN_RENDERED_BODY_CHARS99 100# Render budget handed to Jina. The reader's own default gives an SPA too101# little time and returns the pre-hydration DOM; ``web_fetch.py`` (the102# non-aligned implementation) has always sent 20-30s. Verified on103# ``icmconjectures.com/1983-prob-8`` (2026-07-29): without this header the URL104# yields a 285-byte shell, with it 4945+ bytes of content.105_JINA_TIMEOUT_S = 30106_JINA_RETRY_TIMEOUT_S = 40107 108 109# ── Jina scraping ─────────────────────────────────────────────────────110 111 112async def _scrape_url_with_jina(113 url: str,114 custom_headers: dict[str, str] | None = None,115 max_chars: int = 102400 * 4,116 *,117 engine: str | None = None,118 no_cache: bool = False,119 render_timeout_s: int = _JINA_TIMEOUT_S,120) -> dict[str, Any]:121 """Scrape via Jina reader API. Mirrors the reference implementation exactly.122 123 Retries on connect/read timeouts and 5xx/408/409/425/429 with the124 fixed schedule ``[1, 2, 4, 8]`` seconds. Detects Jina's125 ``InsufficientBalanceError`` JSON body and returns it as a structured126 failure rather than treating the body as content.127 128 ``engine`` / ``no_cache`` / ``render_timeout_s`` drive the reader's129 rendering: the escalation retry in :func:`_fetch_single` re-requests a130 known-empty page with the browser engine, a longer render budget and131 Jina's own response cache bypassed.132 """133 api_key = _jina_api_key()134 if not api_key:135 return {"success": False, "content": "", "error": "JINA_API_KEY not set"}136 137 # Avoid duplicate Jina URL prefix — if the user already passed a138 # ``https://r.jina.ai/<inner>`` URL, strip the outer prefix so we139 # don't double-wrap.140 if url.startswith("https://r.jina.ai/") and url.count("http") >= 2:141 url = url[len("https://r.jina.ai/") :]142 143 jina_url = f"{_jina_base_url()}/{url}"144 headers = {145 "Authorization": f"Bearer {api_key}",146 # Render budget (see _JINA_TIMEOUT_S). Sent on every call, not just the147 # retry — the default budget is what loses the race in the first place.148 "x-timeout": str(render_timeout_s),149 }150 if engine:151 headers["x-engine"] = engine152 if no_cache:153 # Jina caches its own responses, so a shell it captured once is served154 # back for every later attempt. Bypass it when re-trying.155 headers["x-no-cache"] = "true"156 if custom_headers:157 headers.update(custom_headers)158 159 retry_delays = [1, 2, 4, 8]160 response: httpx.Response | None = None161 162 body = b""163 for attempt, delay in enumerate(retry_delays, 1):164 try:165 async with httpx.AsyncClient() as client, client.stream(166 "GET",167 jina_url,168 headers=headers,169 timeout=httpx.Timeout(None, connect=20, read=60),170 follow_redirects=True,171 ) as response:172 response.raise_for_status()173 # Bounded read: stop at the byte cap instead of174 # buffering a whole (possibly multi-GB) body that175 # ``[:max_chars]`` would throw away anyway.176 body, _ = await read_bounded(response)177 # Count one answered, billable Jina reader request.178 # The direct-httpx fallback (``_scrape_url_with_python``)179 # intentionally records nothing — it doesn't bill Jina.180 record_api_request("jina")181 break182 except (httpx.ConnectTimeout, httpx.ConnectError, httpx.ReadTimeout) as e:183 record_api_request("jina", requests=0, errors=1)184 if attempt < len(retry_delays):185 await asyncio.sleep(delay)186 continue187 return {"success": False, "content": "", "error": str(e)}188 except httpx.HTTPStatusError as e:189 sc = e.response.status_code190 record_api_request("jina", requests=0, errors=1)191 if (sc >= 500 or sc in [408, 409, 425, 429]) and attempt < len(192 retry_delays,193 ):194 await asyncio.sleep(delay)195 continue196 return {"success": False, "content": "", "error": str(e)}197 except Exception as e:198 record_api_request("jina", requests=0, errors=1)199 return {"success": False, "content": "", "error": str(e)}200 201 if response is None:202 return {"success": False, "content": "", "error": "No response received"}203 204 content = decode_body(response, body)205 if not content:206 return {"success": False, "content": "", "error": "Empty response from Jina"}207 208 # Detect Jina balance exhaustion — the body is JSON like209 # ``{"name": "InsufficientBalanceError", ...}`` rather than the page.210 try:211 maybe_err = json.loads(content)212 if (213 isinstance(maybe_err, dict)214 and maybe_err.get("name") == "InsufficientBalanceError"215 ):216 return {217 "success": False,218 "content": "",219 "error": "Jina insufficient balance",220 }221 except json.JSONDecodeError:222 pass223 224 return {"success": True, "content": content[:max_chars], "error": ""}225 226 227async def _scrape_url_with_python(228 url: str,229 custom_headers: dict[str, str] | None = None,230 max_chars: int = 102400 * 4,231) -> dict[str, Any]:232 """Direct httpx GET fallback when Jina fails. Same retry policy."""233 headers = {234 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",235 }236 if custom_headers:237 headers.update(custom_headers)238 239 retry_delays = [1, 2, 4]240 241 for attempt, delay in enumerate(retry_delays, 1):242 try:243 async with httpx.AsyncClient() as client:244 # Redirects are walked by hand so every hop is vetted the way245 # the initial URL was; following them automatically lets a246 # vetted public URL answer 302 → localhost (see ``next_hop``).247 # Each hop is also pinned to the address that passed the check,248 # so the name cannot resolve to something else at connect time,249 # and credentials are dropped when a hop changes origin.250 hop_url, hop_headers = url, headers251 for _ in range(MAX_REDIRECT_HOPS):252 refusal, addresses = await vet_public_url(hop_url)253 if refusal:254 raise RedirectRefused(refusal)255 dial_url, dial_headers, extensions = pin_to_address(256 hop_url, addresses, hop_headers,257 )258 async with client.stream(259 "GET",260 dial_url,261 headers=dial_headers,262 timeout=httpx.Timeout(None, connect=20, read=60),263 follow_redirects=False,264 extensions=extensions,265 ) as response:266 target = await next_hop(response, hop_url)267 if target is not None:268 hop_headers = strip_cross_origin_credentials(269 hop_headers, hop_url, target,270 )271 hop_url = target272 continue273 response.raise_for_status()274 # Direct fetch sees raw origin bytes (no Jina text275 # conversion), so a data-blob content-type means a276 # dataset/archive download — tell the agent plainly277 # instead of feeding mojibake to the SUMMARY_LLM.278 blob_type = binary_content_type(279 response.headers.get("content-type"),280 )281 if blob_type:282 declared = response.headers.get("content-length", "?")283 return {284 "success": False,285 "content": "",286 "error": (287 f"binary content ({blob_type}, "288 f"{declared} bytes) — a data file, not a "289 "web page; do not fetch it as text"290 ),291 }292 body, _ = await read_bounded(response)293 break294 else:295 return {296 "success": False, "content": "",297 "error": "too many redirects",298 }299 content = decode_body(response, body)300 if not content:301 return {"success": False, "content": "", "error": "Empty response"}302 return {"success": True, "content": content[:max_chars], "error": ""}303 except RedirectRefused as e:304 # A refused hop is a policy decision, not a transport fault: do not305 # burn the remaining retries re-requesting the same chain.306 return {"success": False, "content": "", "error": str(e)}307 except (httpx.ConnectTimeout, httpx.ConnectError, httpx.ReadTimeout) as e:308 if attempt < len(retry_delays):309 await asyncio.sleep(delay)310 continue311 return {"success": False, "content": "", "error": str(e)}312 except httpx.HTTPStatusError as e:313 sc = e.response.status_code314 if (sc >= 500 or sc in [408, 409, 425, 429]) and attempt < len(315 retry_delays,316 ):317 await asyncio.sleep(delay)318 continue319 return {"success": False, "content": "", "error": str(e)}320 except Exception as e:321 return {"success": False, "content": "", "error": str(e)}322 323 return {"success": False, "content": "", "error": "All retries exhausted"}324 325 326# ── LLM extraction ───────────────────────────────────────────────────327 328# Copied from the reference implementation. The wording matters: it shapes329# the LLM's extraction style and ultimately the content the agent sees.330_EXTRACT_INFO_PROMPT = """You are given a piece of content and the requirement of information to extract. Your task is to extract the information specifically requested. Be precise and focus exclusively on the requested information.331 332INFORMATION TO EXTRACT:333{}334 335INSTRUCTIONS:3361. Extract the information relevant to the focus above.3372. If the exact information is not found, extract the most closely related details.3383. Be specific and include exact details when available.3394. Clearly organize the extracted information for easy understanding.3405. Do not include general summaries or unrelated content.341 342CONTENT TO ANALYZE:343{}344 345EXTRACTED INFORMATION:"""346 347 348def _truncate_fallback(content: str) -> str:349 """Raw-content path used when no summary LLM is reachable."""350 if len(content) > FALLBACK_TRUNCATE:351 return content[:FALLBACK_TRUNCATE] + "\n\n[Content truncated...]"352 return content353 354 355async def _extract_info_with_llm(356 content: str,357 info_to_extract: str,358 truncate_last_num_chars: int = -1,359) -> dict[str, Any]:360 """Call the cheap SUMMARY_LLM to focus-extract from the scraped page.361 362 Resolution + fallback (2026-06-03): candidates come from363 ``frontier_agent.infra.summary_llm.summary_llm_candidates()`` — profile364 ``summary_llm:`` block (contextvar override, installed by the agent_team365 ``main_agent_node``) first, then its ``fallback:`` sub-block, then366 env ``SUMMARY_LLM_*`` / ``SUMMARY_LLM_FALLBACK_*``. Each candidate367 keeps the reference retry policy; a candidate that368 exhausts retries (or returns empty content — runaway reasoning)369 falls through to the next.370 """371 if not content or not content.strip():372 return {"success": False, "extracted_info": "", "error": "Empty content"}373 374 # No focus — nothing to extract. Hand back the raw page, matching375 # ``web_fetch``, instead of prompting the LLM with an empty focus.376 if not info_to_extract or not info_to_extract.strip():377 return {378 "success": True,379 "extracted_info": _truncate_fallback(content),380 "error": "",381 }382 383 candidates = summary_llm_candidates()384 if not candidates:385 # Legacy quirk preserved: BASE_URL set without MODEL_NAME used to386 # run with model="default" — keep that working.387 legacy_base = _summary_llm_base_url()388 if legacy_base:389 candidates = [{390 "endpoint": legacy_base,391 "model": _summary_llm_model_name() or "default",392 "api_key": _summary_llm_api_key() or "",393 "provider": "summary_llm",394 }]395 else:396 # No summary LLM anywhere — not even the primary-model last397 # resort ``summary_llm_candidates()`` appends. Degrade to398 # truncated raw content like ``web_fetch`` does instead of399 # failing every fetch: a page the agent can read beats an400 # error string.401 logger.warning(402 "Summary LLM not configured — returning truncated raw content",403 )404 return {405 "success": True,406 "extracted_info": _truncate_fallback(content),407 "error": "",408 }409 410 text = content411 if truncate_last_num_chars > 0:412 text = content[:-truncate_last_num_chars] + "[...truncated]"413 414 last: dict[str, Any] = {415 "success": False, "extracted_info": "", "error": "No response",416 }417 for cand in candidates:418 last = await _extract_with_candidate(cand, content, text, info_to_extract)419 if last["success"] and last["extracted_info"]:420 return last421 return last422 423 424async def _extract_with_candidate(425 cand: dict[str, str],426 content: str,427 text: str,428 info_to_extract: str,429) -> dict[str, Any]:430 """One candidate's extraction attempt (reference retry policy)."""431 endpoint = cand["endpoint"]432 model = cand["model"]433 prompt = _EXTRACT_INFO_PROMPT.format(info_to_extract, text)434 435 payload: dict[str, Any] = {436 "model": model,437 "max_tokens": 8192,438 "messages": [{"role": "user", "content": prompt}],439 "temperature": 1.0,440 }441 # GPT-5/4-style models reject ``max_tokens`` and need the new key.442 if "gpt" in model:443 payload["max_completion_tokens"] = payload.pop("max_tokens")444 if "gpt-5" in model.lower() or "gpt5" in model.lower():445 payload["service_tier"] = "flex"446 payload["reasoning_effort"] = "minimal"447 # Self-hosted reasoning models behind SGLang: extraction is an448 # auxiliary call — disable thinking so the reasoning prefix doesn't449 # eat the whole budget and return content=None. These keys match the450 # *model name*, so they are a contract with the serving stack — do451 # not rename them cosmetically or the branch stops firing.452 if any(k in model.lower() for k in ("qwen", "apodex", "sglang", "397b")):453 payload["chat_template_kwargs"] = {"enable_thinking": False}454 455 headers: dict[str, str] = {"Content-Type": "application/json"}456 api_key = cand.get("api_key") or ""457 if api_key:458 headers["Authorization"] = f"Bearer {api_key}"459 460 retry_delays = [1, 2, 4, 8]461 response: httpx.Response | None = None462 463 for attempt, delay in enumerate(retry_delays, 1):464 try:465 async with httpx.AsyncClient() as client:466 response = await client.post(467 endpoint,468 headers=headers,469 json=payload,470 timeout=httpx.Timeout(None, connect=30, read=300),471 )472 473 # Context-overflow recovery: chop a chunk off the tail and474 # retry. Each attempt cuts a larger slice (40K, 80K, 120K…).475 if response and (476 "exceeds the model's maximum context length" in response.text477 or "longer than the model's context length" in response.text478 ):479 payload["messages"][0]["content"] = _EXTRACT_INFO_PROMPT.format(480 info_to_extract,481 content[: -(40960 * attempt)] + "[...truncated]",482 )483 continue484 485 response.raise_for_status()486 break487 except httpx.HTTPError as e:488 # GPT-5 sometimes rejects ``service_tier`` — drop and retry.489 if (490 "gpt-5" in model.lower() or "gpt5" in model.lower()491 ) and "service_tier" in payload:492 payload.pop("service_tier", None)493 # Retry every transient HTTP/network error within the budget.494 # The SUMMARY_LLM proxy returns 401 under bursty parallel load495 # even with a valid key — a follow-up request almost always496 # succeeds.497 if attempt < len(retry_delays):498 await asyncio.sleep(delay)499 continue500 return {"success": False, "extracted_info": "", "error": str(e)}501 except Exception as e:502 return {"success": False, "extracted_info": "", "error": str(e)}503 504 if response is None:505 return {"success": False, "extracted_info": "", "error": "No response"}506 507 try:508 data = response.json()509 except json.JSONDecodeError as e:510 return {511 "success": False,512 "extracted_info": "",513 "error": f"JSON parse error: {e}",514 }515 516 if data.get("choices"):517 try:518 extracted = data["choices"][0]["message"]["content"]519 except (KeyError, IndexError) as e:520 return {"success": False, "extracted_info": "", "error": str(e)}521 # This raw-httpx LLM call bypasses the middleware522 # chain entirely — without this forward its tokens appear523 # nowhere (not even per-run). Lands in the top-level524 # ``usage_summary.llm["{model}@summary_llm"]`` slot.525 usage = data.get("usage") or {}526 if usage:527 record_llm_usage(528 model=model,529 provider=cand.get("provider") or "summary_llm",530 prompt_tokens=int(usage.get("prompt_tokens", 0) or 0),531 completion_tokens=int(usage.get("completion_tokens", 0) or 0),532 cache_read_tokens=int(533 (usage.get("prompt_tokens_details") or {}).get(534 "cached_tokens", 0,535 ) or 0,536 ),537 )538 return {"success": True, "extracted_info": extracted, "error": ""}539 540 return {541 "success": False,542 "extracted_info": "",543 "error": f"Unexpected response: {data}",544 }545 546 547# ── Single-URL fetch + extract ────────────────────────────────────────548 549 550async def _fetch_single(551 url: str,552 info_to_extract: str,553 custom_headers: dict[str, str] | None = None,554) -> str:555 """Scrape one URL (Jina → fallback to direct) then run LLM extraction.556 557 The scrape is served through ``scrape_result_cache`` (single-flight,558 cross-run) so sibling agents fetching the same URL share one Jina559 round-trip. Only the SCRAPE is cached — the SUMMARY_LLM extraction below560 still runs per call so each run keeps its own ``info_to_extract`` focus.561 """562 # Vet the target before ANYTHING leaves the process — before the scrape563 # cache and before Jina, which would otherwise be handed an internal URL.564 # This implementation is the one the shipped react profile selects565 # (``web_fetch_impl: aligned``), so the guard has to live here too and not566 # only in ``plugins/tools/web_fetch.py``.567 non_public = await non_public_url_error(url)568 if non_public:569 return (570 f"Blocked: {non_public}. Only public http(s) endpoints may be "571 "fetched. Use a public source."572 )573 574 if any(pat in url for pat in _BANNED_URL_PATTERNS):575 return "Blocked: scraping Hugging Face datasets/spaces is not allowed."576 577 blocked_ext = blocked_download_url(url)578 if blocked_ext:579 return (580 f"Blocked: this URL is a dataset/archive download ({blocked_ext}), "581 "not a web page. Do not download data files — read the dataset's "582 "documentation/landing page instead, or use aggregate API queries."583 )584 585 # The initial body and browser retry each flow through several decisions586 # (retry, selection, cache admission, and the final warning). Reuse each587 # verdict within this fetch instead of reparsing the same HTML every time.588 render_verdicts: list[tuple[str, str | None]] = []589 590 def _render_kind(content: str) -> str | None:591 for seen, verdict in render_verdicts:592 if content is seen or content == seen:593 return verdict594 verdict = _unrendered_kind(content)595 render_verdicts.append((content, verdict))596 return verdict597 598 async def _scrape() -> str:599 scrape = await _scrape_url_with_jina(url, custom_headers)600 if scrape["success"]:601 kind = _render_kind(scrape["content"])602 if kind is not None:603 # HTTP 200 with a navigation-only body: the page had not604 # hydrated when the reader captured it. Escalate ONCE — browser605 # engine, longer render budget, Jina's cache bypassed — before606 # believing the emptiness.607 logger.info(608 "Jina returned an un-rendered page for %s (%s, %d chars) "609 "— retrying with the browser engine",610 url, kind, len(scrape["content"]),611 )612 retry = await _scrape_url_with_jina(613 url,614 custom_headers,615 engine="browser",616 no_cache=True,617 render_timeout_s=_JINA_RETRY_TIMEOUT_S,618 )619 if retry["success"] and _render_kind(retry["content"]) is None:620 return retry["content"]621 # Still nothing. Keep whichever attempt carried more body — a622 # legitimately tiny page reads as "empty" too, and turning that623 # into an error would lose real content.624 if retry["success"] and len(625 _reader_body(retry["content"]),626 ) > len(_reader_body(scrape["content"])):627 scrape = retry628 if _render_kind(scrape["content"]) == "shell":629 raise ScrapeUnavailable(630 "the page is a JavaScript app shell — it returned no "631 "content even with browser rendering and no cache. "632 "Fetching it again the same way will not help: look for "633 "the same material at another source (the site's own "634 "API/JSON endpoint, a PDF or an archive copy), or search "635 "for the page title instead"636 )637 return scrape["content"]638 639 logger.warning(640 "Jina failed for %s: %s, trying direct", url, scrape["error"],641 )642 # Plain httpx executes no JavaScript, so for a shell page this fallback643 # can only ever confirm the emptiness — say so instead of handing the644 # extractor a navigation-only DOM to describe.645 scrape = await _scrape_url_with_python(url, custom_headers)646 if not scrape["success"]:647 raise ScrapeUnavailable(scrape["error"])648 if _render_kind(scrape["content"]) == "shell":649 raise ScrapeUnavailable(650 "the reader was unavailable and the raw HTML is a JavaScript "651 "app shell with no content in it (a direct fetch runs no "652 "JavaScript). Try another source for the same material rather "653 "than re-fetching this URL"654 )655 return scrape["content"]656 657 try:658 # Custom headers can change what the origin returns, so a header-bearing659 # request bypasses the URL-keyed shared cache to avoid serving content660 # fetched under different headers.661 if custom_headers:662 content = await _scrape()663 else:664 content = await scrape_result_cache.get_or_scrape(665 url,666 _scrape,667 # Return a low-confidence short page to this caller, but do not668 # publish it as the process-wide answer. A later fetch must be669 # free to win the render race.670 should_cache=lambda scraped: _render_kind(scraped) is None,671 )672 except ScrapeUnavailable as exc:673 return f"[ERROR]: Scraping failed: {exc}"674 675 result = await _extract_info_with_llm(content, info_to_extract)676 if not result["success"]:677 return f"[ERROR]: Extraction failed: {result['error']}"678 extracted = result["extracted_info"]679 if _render_kind(content) == "empty":680 return (681 f"[POSSIBLY NOT RENDERED] The page at {url} remained suspiciously "682 "short after the browser-render retry. This result was deliberately "683 "not cached. Use it if it answers the question; otherwise switch to "684 "another source instead of repeatedly fetching this URL.\n\n"685 f"{extracted}"686 )687 return extracted688 689 690# ── Tool ──────────────────────────────────────────────────────────────691 692 693@tool(name="web_fetch")694async def web_fetch_aligned(695 url: str | list[str],696 info_to_extract: str | list[str] = "",697 custom_headers: dict[str, str] | None = None,698) -> str:699 """Fetch content from a URL and extract specific types of information.700 701 Args:702 url: The URL to fetch, or a list of URLs to fetch in parallel703 info_to_extract: The specific types of information to extract (usually a question), or a list of extraction prompts (one per URL). Omit to get the raw page back704 custom_headers (Dict[str, str]): Additional headers to include in the request705 706 Returns:707 Extracted information as plain text. For multiple URLs, results are numbered708 """709 # The reference tool accepts JSON-encoded lists for both fields — mirror that.710 url = _ensure_list(url)711 info_to_extract = _ensure_list(info_to_extract)712 713 urls = url if isinstance(url, list) else [url]714 urls = [u for u in urls if u and u.strip()]715 if not urls:716 return "[ERROR]: url is required and cannot be empty."717 718 # ``info_to_extract`` broadcast rules, matching the reference tool:719 # - len matches urls → pair up 1:1720 # - exactly one entry → broadcast to every URL721 # - mismatched (>=2 != N) → join into one prompt and broadcast722 # - plain string → broadcast723 if isinstance(info_to_extract, list):724 if len(info_to_extract) == len(urls):725 infos = info_to_extract726 elif len(info_to_extract) == 1:727 infos = info_to_extract * len(urls)728 else:729 infos = [" ".join(info_to_extract)] * len(urls)730 else:731 infos = [info_to_extract] * len(urls)732 733 # Dedup identical (url, info) pairs so a multi-URL call doesn't burn734 # SUMMARY_LLM tokens on duplicates.735 seen: set[tuple[str, str]] = set()736 deduped_urls: list[str] = []737 deduped_infos: list[str] = []738 for u, info in zip(urls, infos, strict=False):739 key = (u, info)740 if key in seen:741 continue742 seen.add(key)743 deduped_urls.append(u)744 deduped_infos.append(info)745 urls = deduped_urls746 infos = deduped_infos747 748 try:749 results = await asyncio.gather(750 *[751 _fetch_single(u, info, custom_headers)752 for u, info in zip(urls, infos, strict=False)753 ],754 )755 756 # The reference tool formats single and multi-URL identically:757 # ``[N] URL: <u>\n Info: <text>``.758 lines: list[str] = []759 for i, (u, text) in enumerate(zip(urls, results, strict=False), 1):760 lines.append(f"[{i}] URL: {u}")761 lines.append(f" Info: {text}")762 return "\n".join(lines)763 764 except Exception as e:765 return f"[ERROR]: Unexpected error: {e!s}"766 767 768__all__ = ["web_fetch_aligned"]769