localailb/assistant
0
1"""playwright_search_tool.py
2
3Playwright-driven web search tools for Open Deep Research / rdtii, used ALONGSIDE the
4existing paid SerpAPI/Serper `GoogleSearchTool` and `DuckDuckGoSearchTool` (not a
5replacement for either).
6
7Two search tools are provided:
8 - PlaywrightDuckDuckGoSearchTool (RECOMMENDED as primary): scrapes DuckDuckGo's
9 server-rendered HTML results page. No JavaScript wait, no CAPTCHA/consent wall in
10 normal use, and stable markup -- the most scrape-friendly of the two.
11 - PlaywrightGoogleSearchTool (fallback): scrapes Google's JS-rendered results page.
12 Broader index (useful for some non-English/regional government sites DuckDuckGo
13 misses), but far more bot-detection friction -- occasional CAPTCHA/consent walls
14 and markup that shifts to resist scraping.
15
16Why either exists at all: SerpAPI/Serper cost money per query and DuckDuckGo (via the
17`ddgs` library) gets rate-limited under heavy/rapid use. These instead launch a real
18(headless) Chromium browser via Playwright and scrape results directly -- no paid API,
19no third-party search library, no per-query cost or account quota.
20
21They're drop-in-shaped tools: single `query` input, and a similar "numbered list of
22title/link/snippet" output shape as smolagents' GoogleSearchTool. Once wired in
23alongside the other search tools, the search_agent keeps using its existing tools
24(VisitTool, PageUpTool, PageDownTool, FinderTool, FindNextTool, ArchiveSearchTool,
25TextInspectorTool) to open, crawl, and read whatever pages these tools find -- only the
26initial "search" step differs.
27
28Requirements (not in requirements.txt by default -- add both):
29 pip install playwright
30 playwright install chromium
31
32Usage:
33 from playwright_search_tool import PlaywrightDuckDuckGoSearchTool, PlaywrightGoogleSearchTool
34 WEB_TOOLS = [GoogleSearchTool(...), DuckDuckGoSearchTool(), PlaywrightDuckDuckGoSearchTool(), PlaywrightGoogleSearchTool(), ...]
35
36---------------------------------------------------------------------------
37Why PlaywrightGoogleSearchTool's extraction logic is NOT based on Google's CSS class
38names (div.g, div.MjjYud, etc.):
39
40Google's result-page markup (a) changes its class names frequently and without
41notice, and (b) is frequently served *differently* to traffic it suspects is
42automated -- a plain, unauthenticated, headless-looking browser can get a
43stripped-down or reflowed results page where those old class names simply don't
44exist, even though a person opening the same query in a normal browser sees full
45results. Relying on class names is why an earlier version of this tool could report
46"No Google results found" for a query that clearly has results.
47
48Instead, extraction here is structural: Google's results almost always still put each
49organic result's title inside an <h3>, wrapped (directly or via an ancestor) in an
50<a href="..."> to the real destination. Walking up from every <h3> to its nearest
51linked ancestor is far more resistant to markup/class churn than any fixed selector
52list. A handful of anti-bot-detection tweaks (a realistic user agent, a normal
53viewport, hiding the `navigator.webdriver` flag, and waiting for actual content
54instead of a fixed sleep) also reduce how often Google serves the stripped-down page
55in the first place.
56
57DuckDuckGo's HTML endpoint below needs none of this -- it's server-rendered with no
58client-side JS to fight, which is exactly why it's the recommended primary tool.
59---------------------------------------------------------------------------
60"""
61from __future__ import annotations
62
63import json
64import os
65import tempfile
66import urllib.parse
67
68from smolagents import Tool
69
70_USER_AGENT = (
71 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
72 "(KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36"
73)
74
75_CONSENT_BUTTON_LABELS = ("Accept all", "I agree", "Accept", "Reject all", "Alle akzeptieren")
76
77# Hosts that are Google's own chrome (nav bars, "Sign in", "Settings", cached-page
78# links, translate links, etc.) rather than an actual search result -- filtered out
79# of both the h3-based pass and the broad fallback pass below.
80_IGNORED_HOST_FRAGMENTS = (
81 "google.com/search", "google.com/preferences", "google.com/advanced_search",
82 "google.com/intl", "accounts.google.", "support.google.", "policies.google.",
83 "maps.google.", "webcache.googleusercontent.", "translate.google.",
84 "/url?q=/search", "consent.google.",
85)
86
87_ANTI_DETECTION_INIT_SCRIPT = """
88Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
89Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
90Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
91window.chrome = window.chrome || { runtime: {} };
92"""
93
94
95class PlaywrightDuckDuckGoSearchTool(Tool):
96 """Searches DuckDuckGo's plain HTML endpoint (html.duckduckgo.com/html/) with
97 Playwright. This is the PREFERRED Playwright search tool for automated use, ahead
98 of PlaywrightGoogleSearchTool below:
99
100 - The page is fully server-rendered HTML with no JavaScript required, so there's
101 no client-side rendering to wait for and no anti-bot-detection script needed.
102 - No CAPTCHA/consent wall in normal use (unlike Google, which frequently serves
103 one to automated-looking traffic).
104 - Stable, simple markup (a.result__a / .result__snippet) that changes far less
105 often than Google's, which is deliberately obfuscated/rotated against scrapers.
106
107 Trade-off: DuckDuckGo's index is thinner than Google's for some non-English or
108 regional government sites, which is why PlaywrightGoogleSearchTool is kept as an
109 explicit fallback rather than removed.
110 """
111
112 name = "playwright_duckduckgo_search"
113 description = (
114 "Performs a DuckDuckGo web search for your query and returns the top results (title, "
115 "link, short snippet) as a numbered list. Uses a real browser (Playwright) against "
116 "DuckDuckGo's plain HTML results page -- no JavaScript rendering wait, no CAPTCHA/consent "
117 "wall in normal use, and more stable markup than Google's. Use this as your PRIMARY search "
118 "tool for almost every search -- call it FIRST, before any other search tool. Only fall "
119 "back to the 'playwright_google_search' tool if this one comes back with no results, or a "
120 "topic needs Google's broader index (e.g. some non-English regional government sites). "
121 "Keep queries short and simple -- the plain topic name plus a country/keyword or two -- "
122 "rather than long boolean or quoted-phrase chains."
123 )
124 inputs = {
125 "query": {"type": "string", "description": "The search query to perform. Keep it short and simple."},
126 }
127 output_type = "string"
128
129 def __init__(self, max_results: int = 8, headless: bool = False, timeout_ms: int = 20000):
130 super().__init__()
131 self.max_results = max_results
132 self.headless = headless
133 self.timeout_ms = timeout_ms
134
135 def forward(self, query: str) -> str:
136 try:
137 from playwright.sync_api import sync_playwright
138 except ImportError as e:
139 raise ImportError(
140 "PlaywrightDuckDuckGoSearchTool needs the `playwright` package and its "
141 "browser binaries. Install with:\n"
142 " pip install playwright\n"
143 " playwright install chromium"
144 ) from e
145
146 query = (query or "").strip()
147 if not query:
148 return "Error: empty search query."
149
150 url = "https://html.duckduckgo.com/html/?" + urllib.parse.urlencode({"q": query})
151
152 with sync_playwright() as p:
153 browser = p.chromium.launch(
154 headless=self.headless,
155 args=["--disable-blink-features=AutomationControlled"],
156 )
157 try:
158 context = browser.new_context(
159 user_agent=_USER_AGENT,
160 locale="en-US",
161 viewport={"width": 1366, "height": 900},
162 extra_http_headers={"Accept-Language": "en-US,en;q=0.9"},
163 )
164 page = context.new_page()
165 # Server-rendered HTML -- domcontentloaded is enough, no networkidle wait
166 # needed the way the JS-heavy Google results page requires.
167 page.goto(url, timeout=self.timeout_ms, wait_until="domcontentloaded")
168
169 extracted = page.evaluate(
170 """
171 () => {
172 const results = [];
173 document.querySelectorAll('.result, .web-result').forEach(block => {
174 const a = block.querySelector('a.result__a');
175 if (!a || !a.href) return;
176 const snippetEl = block.querySelector('.result__snippet');
177 results.push({
178 title: (a.innerText || '').trim(),
179 link: a.href,
180 snippet: snippetEl ? (snippetEl.innerText || '').trim() : '',
181 });
182 });
183 return { results, pageTitle: document.title };
184 }
185 """
186 )
187 results = extracted.get("results", [])
188 finally:
189 browser.close()
190
191 if not results:
192 return (
193 f"No DuckDuckGo results found for '{query}'. Try a shorter/simpler query with "
194 f"playwright_duckduckgo_search again, OR call the fallback Google tool like this:\n"
195 f"```python\n"
196 f'result = playwright_google_search(query="{query}")\n'
197 f"print(result)\n"
198 f"```"
199 )
200
201 lines = [f"## DuckDuckGo Search Results for '{query}'\n"]
202 n = 0
203 for r in results:
204 link = (r.get("link") or "").strip()
205 if not link:
206 continue
207 n += 1
208 title = (r.get("title") or link).strip()
209 snippet = (r.get("snippet") or "").strip()
210 lines.append(f"{n}. [{title}]({link})\n{snippet}\n")
211 if n >= self.max_results:
212 break
213
214 return "\n".join(lines)
215
216
217class PlaywrightGoogleSearchTool(Tool):
218 """Searches Google by driving a real headless browser with Playwright, as an
219 additional, free alternative alongside the existing paid/rate-limited search
220 tools -- not a replacement for them.
221
222 A fresh, isolated browser is launched per call (rather than one shared browser
223 kept alive across calls) -- slightly slower per search, but avoids any
224 thread-safety concerns from Gradio potentially handling requests on different
225 threads, and keeps this tool's lifecycle dead simple (nothing to explicitly shut
226 down at app exit).
227 """
228
229 name = "playwright_google_search"
230 description = (
231 "Performs a Google web search for your query and returns the top results "
232 "(title, link, short snippet) as a numbered list. Uses a real browser "
233 "(Playwright) that types the query into google.com directly -- no paid "
234 "search API, no API key, and no third-party search library involved. This "
235 "is the FALLBACK search tool -- try 'playwright_duckduckgo_search' FIRST for "
236 "almost every search, and only call this one if that comes back with no "
237 "results or the topic needs Google's broader index. Keep queries short and "
238 "simple -- the plain topic name plus a country/keyword or two -- rather than "
239 "long boolean or quoted-phrase chains."
240 )
241 inputs = {
242 "query": {"type": "string", "description": "The search query to perform. Keep it short and simple."},
243 "filter_year": {
244 "type": "string",
245 "description": "(Optional) restrict results to this year, e.g. '2023'.",
246 "nullable": True,
247 },
248 }
249 output_type = "string"
250
251 def __init__(self, max_results: int = 8, headless: bool = False, timeout_ms: int = 20000, debug: bool = False):
252 super().__init__()
253 self.max_results = max_results
254 self.headless = headless
255 self.timeout_ms = timeout_ms
256 self.debug = debug
257
258 def forward(self, query: str, filter_year: str | None = None) -> str:
259 try:
260 from playwright.sync_api import sync_playwright
261 except ImportError as e:
262 raise ImportError(
263 "PlaywrightGoogleSearchTool needs the `playwright` package and its "
264 "browser binaries. Install with:\n"
265 " pip install playwright\n"
266 " playwright install chromium"
267 ) from e
268
269 query = (query or "").strip()
270 if not query:
271 return "Error: empty search query."
272
273 params = {"q": query, "num": str(max(self.max_results, 10)), "hl": "en", "gl": "us", "pws": "0"}
274 if filter_year:
275 params["tbs"] = f"cdr:1,cd_min:1/1/{filter_year},cd_max:12/31/{filter_year}"
276 url = "https://www.google.com/search?" + urllib.parse.urlencode(params)
277
278 page_info = ""
279 with sync_playwright() as p:
280 browser = p.chromium.launch(
281 headless=self.headless,
282 args=["--disable-blink-features=AutomationControlled"],
283 )
284 try:
285 context = browser.new_context(
286 user_agent=_USER_AGENT,
287 locale="en-US",
288 viewport={"width": 1366, "height": 900},
289 extra_http_headers={"Accept-Language": "en-US,en;q=0.9"},
290 )
291 context.add_init_script(_ANTI_DETECTION_INIT_SCRIPT)
292 page = context.new_page()
293 page.goto(url, timeout=self.timeout_ms, wait_until="domcontentloaded")
294
295 # Best-effort consent-dialog dismissal (Google's EU/UK cookie wall) --
296 # harmless no-op if no such dialog appears for this session/region.
297 for label in _CONSENT_BUTTON_LABELS:
298 try:
299 page.get_by_role("button", name=label).click(timeout=1200)
300 page.wait_for_load_state("domcontentloaded", timeout=self.timeout_ms)
301 break
302 except Exception:
303 continue
304
305 # Wait for actual result content rather than a fixed sleep -- far more
306 # reliable across Google's varying render times. If nothing shows up
307 # in time, keep going anyway and let the extraction below report
308 # honestly (this also surfaces a captcha/consent page as "no results"
309 # rather than hanging).
310 try:
311 page.wait_for_selector("h3, #rso, #search", timeout=6000)
312 except Exception:
313 pass
314 page.wait_for_timeout(400)
315
316 extracted = page.evaluate(
317 """
318 ([ignoredFragments]) => {
319 const isIgnored = (href) => ignoredFragments.some(f => href.includes(f));
320
321 const results = [];
322 const seenLinks = new Set();
323
324 // Primary pass: every <h3> (the title Google puts on each organic
325 // result), walking up to its nearest linked ancestor.
326 document.querySelectorAll('h3').forEach(h3 => {
327 const a = h3.closest('a[href]') || h3.parentElement?.querySelector('a[href]');
328 if (!a || !a.href || !a.href.startsWith('http')) return;
329 if (isIgnored(a.href) || seenLinks.has(a.href)) return;
330
331 // Snippet: look at the result's outer block (a few levels up
332 // from the link) and take its text, minus the title itself.
333 let block = a.closest('div[data-hveid], div.g, div') || a.parentElement;
334 let text = block ? block.innerText : '';
335 const title = h3.innerText || '';
336 if (title && text.startsWith(title)) text = text.slice(title.length);
337 text = text.split('\\n').map(s => s.trim()).filter(Boolean).slice(0, 2).join(' ');
338
339 seenLinks.add(a.href);
340 results.push({ title: title || a.href, link: a.href, snippet: text });
341 });
342
343 // Fallback pass: if the h3-based pass found nothing (e.g. an
344 // unusual layout variant), broadly scan external links with
345 // non-trivial link text instead.
346 if (results.length === 0) {
347 document.querySelectorAll('a[href^="http"]').forEach(a => {
348 const href = a.href;
349 if (isIgnored(href) || seenLinks.has(href)) return;
350 const text = (a.innerText || '').trim();
351 if (text.length < 8) return;
352 seenLinks.add(href);
353 results.push({ title: text, link: href, snippet: '' });
354 });
355 }
356
357 return {
358 results: results.slice(0, 30),
359 pageTitle: document.title,
360 bodyLen: document.body ? document.body.innerText.length : 0,
361 };
362 }
363 """,
364 [_IGNORED_HOST_FRAGMENTS],
365 )
366 results = extracted.get("results", [])
367 if self.debug:
368 page_info = (
369 f" [debug: page title='{extracted.get('pageTitle')}', "
370 f"body chars={extracted.get('bodyLen')}]"
371 )
372 finally:
373 browser.close()
374
375 if not results:
376 year_note = f" (filtered to {filter_year})" if filter_year else ""
377 return (
378 f"No Google results found for '{query}'{year_note}.{page_info} This can happen if Google "
379 "served a consent/CAPTCHA page instead of results. Try playwright_duckduckgo_search "
380 "instead, like this:\n"
381 f"```python\n"
382 f'result = playwright_duckduckgo_search(query="{query}")\n'
383 f"print(result)\n"
384 f"```"
385 )
386
387 lines = [f"## Search Results for '{query}'{page_info}\n"]
388 n = 0
389 for r in results:
390 link = (r.get("link") or "").strip()
391 if not link:
392 continue
393 n += 1
394 title = (r.get("title") or link).strip()
395 snippet = (r.get("snippet") or "").strip()
396 lines.append(f"{n}. [{title}]({link})\n{snippet}\n")
397 if n >= self.max_results:
398 break
399
400 return "\n".join(lines)
401
402
403# ---------------------------------------------------------------------------
404# Extensions/keywords that make a link look like the actual downloadable document
405# file rather than a landing/details page about it. Deliberately broad (covers the
406# common legal-document file types plus generic "get me the file" wording) -- a
407# false positive here just means the agent double-checks a link that turns out to
408# be another HTML page, which playwright_visit_page reports honestly either way.
409# ---------------------------------------------------------------------------
410_DOWNLOAD_EXT_RE = r"\.(pdf|docx?|rtf|txt)(\?|#|$)"
411_DOWNLOAD_KEYWORD_RE = r"download|full[\s_-]?text|view[\s_-]?(the[\s_-])?(authoris|full)|attachment|/file/"
412
413
414class PlaywrightVisitPageTool(Tool):
415 """Opens a URL with a real headless Chromium browser (Playwright) and returns its
416 visible text plus every link on the page, with links that look like an actual
417 downloadable document (by file extension or surrounding wording) flagged and
418 listed first. This is the "read/browse a page" half of a Playwright-only
419 search-and-browse loop -- paired with PlaywrightGoogleSearchTool for the
420 "search" half -- so neither step needs the `requests` library, a paid search
421 API, or the existing requests-based SimpleTextBrowser.
422
423 If the URL itself resolves directly to a non-HTML file (a PDF, DOCX, etc.),
424 that's reported immediately as "this URL IS the document" instead of trying to
425 scrape it as a web page -- the agent doesn't need to visit it a second time.
426 """
427
428 name = "playwright_visit_page"
429 description = (
430 "Opens a URL with a real browser (Playwright) and returns the page's visible text plus "
431 "every link found on it, with links that look like an actual downloadable document file "
432 "(PDF/DOC/DOCX/TXT, or wording like 'Download'/'Full text'/'View authorised version') "
433 "flagged with a leading '⬇️ DOWNLOAD?' marker and listed first. Use this to read a search "
434 "result page, then call it again on any flagged link found on that page to go one level "
435 "deeper (e.g. from a government 'Details' page to the actual PDF it links to). If the URL "
436 "you pass in is itself already a direct file (not an HTML page), this tool reports that "
437 "immediately instead of trying to scrape it as a web page."
438 )
439 inputs = {
440 "url": {"type": "string", "description": "The exact URL to open (from a search result or a link found on a previously visited page)."},
441 }
442 output_type = "string"
443
444 def __init__(
445 self,
446 headless: bool = False,
447 timeout_ms: int = 20000,
448 max_text_chars: int = 5000,
449 max_links: int = 40,
450 ):
451 super().__init__()
452 self.headless = headless
453 self.timeout_ms = timeout_ms
454 self.max_text_chars = max_text_chars
455 self.max_links = max_links
456
457 def forward(self, url: str) -> str:
458 try:
459 from playwright.sync_api import sync_playwright
460 except ImportError as e:
461 raise ImportError(
462 "PlaywrightVisitPageTool needs the `playwright` package and its browser "
463 "binaries. Install with:\n"
464 " pip install playwright\n"
465 " playwright install chromium"
466 ) from e
467
468 url = (url or "").strip()
469 if not url:
470 return "Error: empty URL."
471 if not (url.startswith("http://") or url.startswith("https://")):
472 return f"Error: '{url}' is not an http(s) URL."
473
474 with sync_playwright() as p:
475 browser = p.chromium.launch(
476 headless=self.headless,
477 args=["--disable-blink-features=AutomationControlled"],
478 )
479 try:
480 context = browser.new_context(
481 user_agent=_USER_AGENT,
482 locale="en-US",
483 viewport={"width": 1366, "height": 900},
484 extra_http_headers={"Accept-Language": "en-US,en;q=0.9"},
485 accept_downloads=True,
486 )
487 context.add_init_script(_ANTI_DETECTION_INIT_SCRIPT)
488 page = context.new_page()
489
490 try:
491 response = page.goto(url, timeout=self.timeout_ms, wait_until="domcontentloaded")
492 except Exception as e:
493 # A direct file link (e.g. a PDF) sometimes triggers Playwright's own
494 # download handling rather than a normal navigation, which surfaces as
495 # an exception here rather than a response -- treat that as a strong
496 # signal this URL IS a downloadable file, not a real navigation failure.
497 if "Download is starting" in str(e):
498 return (
499 f"'{url}' triggered a file download when opened -- this URL IS the "
500 "document itself (not an HTML landing page). Use it directly as the "
501 "document's URL; no further browsing of this link is needed."
502 )
503 return f"Error: could not load '{url}': {e}"
504
505 status = response.status if response else None
506 if status and status >= 400:
507 return f"Error: '{url}' returned HTTP {status}."
508
509 content_type = (response.headers.get("content-type", "") if response else "").lower()
510 if content_type and "html" not in content_type and "text/plain" not in content_type:
511 cl = response.headers.get("content-length") if response else None
512 size_note = f", {cl} bytes" if cl and cl.isdigit() else ""
513 return (
514 f"'{url}' is itself a direct downloadable file (Content-Type: "
515 f"{content_type}{size_note}). This URL IS the document -- no further "
516 "browsing of this link is needed."
517 )
518
519 # Best-effort settle for JS-rendered content -- keep going even if the page
520 # never truly goes idle (some sites keep long-poll/analytics connections open).
521 try:
522 page.wait_for_load_state("networkidle", timeout=min(self.timeout_ms, 8000))
523 except Exception:
524 pass
525
526 data = page.evaluate(
527 """
528 ([downloadExtRe, downloadKeywordRe]) => {
529 const extRe = new RegExp(downloadExtRe, 'i');
530 const kwRe = new RegExp(downloadKeywordRe, 'i');
531 const seen = new Set();
532 const links = [];
533 document.querySelectorAll('a[href^="http"]').forEach(a => {
534 const href = a.href;
535 if (!href || seen.has(href)) return;
536 const text = (a.innerText || a.textContent || '').trim().replace(/\\s+/g, ' ');
537 if (!text && !extRe.test(href)) return;
538 seen.add(href);
539 const likelyDownload = extRe.test(href) || kwRe.test(href) || kwRe.test(text);
540 links.push({ href, text: text.slice(0, 120), likelyDownload });
541 });
542 return {
543 title: document.title || '',
544 text: document.body ? document.body.innerText : '',
545 links,
546 };
547 }
548 """,
549 [_DOWNLOAD_EXT_RE, _DOWNLOAD_KEYWORD_RE],
550 )
551 finally:
552 browser.close()
553
554 title = data.get("title", "") or url
555 text = (data.get("text", "") or "").strip()
556 if len(text) > self.max_text_chars:
557 text = text[: self.max_text_chars] + "\n...[truncated]"
558
559 links = data.get("links", [])
560 flagged = [ln for ln in links if ln.get("likelyDownload")]
561 other = [ln for ln in links if not ln.get("likelyDownload")]
562 shown = (flagged + other)[: self.max_links]
563
564 lines = [f"## Page: {title}\nURL: {url}\n", "### Content\n", text or "(no visible text extracted)", ""]
565 lines.append(f"### Links found on this page ({len(links)} total, showing {len(shown)})")
566 if flagged:
567 lines.append(f"({len(flagged)} flagged as possible document downloads -- listed first)")
568 for i, ln in enumerate(shown, start=1):
569 marker = "⬇️ DOWNLOAD? " if ln.get("likelyDownload") else ""
570 label = ln.get("text") or ln.get("href")
571 lines.append(f"{i}. {marker}[{label}]({ln.get('href')})")
572
573 _save_page_cache(url, title, text)
574
575 return "\n".join(lines)
576
577
578# ---------------------------------------------------------------------------
579# PlaywrightExtractLegalDocumentLinksTool
580#
581# Purpose: a dedicated "extract every link, then score/rank it against my
582# target topic" step -- run BEFORE opening individual candidate pages one by
583# one. Many government/legal-register sites (results listings, regulator
584# publication pages, gazette indexes) are JavaScript-rendered, so a plain
585# `requests`-based link scrape misses most or all of the real links; this
586# reuses the same Playwright browser the rest of this module already drives.
587#
588# Deliberately a SEPARATE tool from PlaywrightVisitPageTool rather than a mode
589# flag on it: PlaywrightVisitPageTool's job is "read this one page's content +
590# its links"; this tool's job is "given a page, tell me which of its links are
591# actually worth opening for a SPECIFIC target topic" -- it takes a
592# topic_keywords argument and returns a ranked, scored list, not raw content.
593# Named specifically (not "playwright_extract_links") so it's unambiguous next
594# to playwright_visit_page / playwright_duckduckgo_search / playwright_google_search
595# in a tool list, and so the agent doesn't confuse it with a generic scraper.
596# ---------------------------------------------------------------------------
597class PlaywrightExtractLegalDocumentLinksTool(Tool):
598 """Opens a URL with a real browser (Playwright -- required for JavaScript-
599 rendered listing/index pages a plain HTTP request can't see the links on),
600 extracts every link on the page, scores each one for relevance against a
601 target legal topic (keyword overlap with the link's visible text/URL, plus
602 a bonus for links that look like an actual downloadable document file or
603 carry download-style wording), and returns a ranked list -- most relevant
604 first -- instead of a flat, unordered dump of every link on the page.
605
606 Use this BEFORE opening individual candidate pages one by one: run it on a
607 search-result page or a regulator's publications/listing page, look at the
608 top-ranked links, and only then call `playwright_visit_page` on the ones
609 that actually look relevant. If it reports zero links at all (a page with
610 no outbound links, or a direct file URL), fall back to reading the page's
611 own content with `playwright_visit_page` instead.
612 """
613
614 name = "playwright_extract_legal_document_links"
615 description = (
616 "Opens a URL with a real browser (Playwright) and extracts every link on the page, then "
617 "scores and ranks each link by how relevant it looks to a target legal topic you specify "
618 "(keyword overlap with the link's visible text/URL, plus a bonus for links that look like "
619 "an actual downloadable document file -- PDF/DOC/DOCX/TXT -- or carry wording like "
620 "'Download'/'Full text'/'View authorised version'). Returns the top-ranked links first, "
621 "each with its relevance score, so you can decide which ones are actually worth opening "
622 "before visiting them one by one -- use this BEFORE playwright_visit_page on a search-"
623 "result page, a regulator's publications/listing page, or any other page likely to link "
624 "out to several documents. Needs a real browser because many such pages are JavaScript-"
625 "rendered and a plain HTTP request can't see their links at all. If this reports zero "
626 "links (e.g. the URL is itself a direct file, or a page with no outbound links), fall "
627 "back to reading the page's own content with playwright_visit_page instead."
628 )
629 inputs = {
630 "url": {"type": "string", "description": "The exact URL of the page whose links you want to extract and rank."},
631 "topic_keywords": {
632 "type": "string",
633 "description": (
634 "Short, plain keywords describing the legal document/topic you're looking for, e.g. "
635 "'cross-border data policy Cambodia' or 'data protection act Philippines'. Used to "
636 "score each link's relevance -- keep it short, the same way you'd phrase a search query."
637 ),
638 },
639 }
640 output_type = "string"
641
642 def __init__(
643 self,
644 headless: bool = False,
645 timeout_ms: int = 20000,
646 max_links_extracted: int = 200,
647 max_links_shown: int = 25,
648 ):
649 super().__init__()
650 self.headless = headless
651 self.timeout_ms = timeout_ms
652 self.max_links_extracted = max_links_extracted
653 self.max_links_shown = max_links_shown
654
655 @staticmethod
656 def _score_link(text: str, href: str, keywords: list[str]) -> int:
657 """Cheap, deterministic relevance score -- no LLM call. Counts keyword
658 hits in the link's visible text and URL (text hits weighted higher,
659 since a link's own wording is a stronger relevance signal than
660 incidental words in its URL path), plus a flat bonus for anything that
661 already looks like a direct document download (by extension or
662 surrounding wording, reusing this module's own download-detection
663 regexes so the signal is consistent with playwright_visit_page)."""
664 import re as _re
665
666 text_l = (text or "").lower()
667 href_l = (href or "").lower()
668 score = 0
669 for kw in keywords:
670 kw = kw.lower().strip()
671 if not kw:
672 continue
673 if kw in text_l:
674 score += 3
675 if kw in href_l:
676 score += 1
677 if _re.search(_DOWNLOAD_EXT_RE, href_l):
678 score += 5
679 if _re.search(_DOWNLOAD_KEYWORD_RE, href_l) or _re.search(_DOWNLOAD_KEYWORD_RE, text_l):
680 score += 3
681 return score
682
683 def forward(self, url: str, topic_keywords: str) -> str:
684 try:
685 from playwright.sync_api import sync_playwright
686 except ImportError as e:
687 raise ImportError(
688 "PlaywrightExtractLegalDocumentLinksTool needs the `playwright` package and its "
689 "browser binaries. Install with:\n"
690 " pip install playwright\n"
691 " playwright install chromium"
692 ) from e
693
694 url = (url or "").strip()
695 if not url:
696 return "Error: empty URL."
697 if not (url.startswith("http://") or url.startswith("https://")):
698 return f"Error: '{url}' is not an http(s) URL."
699
700 keywords = [w for w in (topic_keywords or "").replace(",", " ").split() if len(w) > 2]
701
702 with sync_playwright() as p:
703 browser = p.chromium.launch(
704 headless=self.headless,
705 args=["--disable-blink-features=AutomationControlled"],
706 )
707 try:
708 context = browser.new_context(
709 user_agent=_USER_AGENT,
710 locale="en-US",
711 viewport={"width": 1366, "height": 900},
712 extra_http_headers={"Accept-Language": "en-US,en;q=0.9"},
713 accept_downloads=True,
714 )
715 context.add_init_script(_ANTI_DETECTION_INIT_SCRIPT)
716 page = context.new_page()
717
718 try:
719 response = page.goto(url, timeout=self.timeout_ms, wait_until="domcontentloaded")
720 except Exception as e:
721 if "Download is starting" in str(e):
722 return (
723 f"'{url}' triggered a file download when opened -- this URL IS a document "
724 "itself, not a listing/index page with links to extract. Use it directly as "
725 "a candidate document; there are no links to rank here."
726 )
727 return f"Error: could not load '{url}': {e}"
728
729 status = response.status if response else None
730 if status and status >= 400:
731 return f"Error: '{url}' returned HTTP {status}."
732
733 content_type = (response.headers.get("content-type", "") if response else "").lower()
734 if content_type and "html" not in content_type and "text/plain" not in content_type:
735 return (
736 f"'{url}' is itself a direct downloadable file (Content-Type: {content_type}). "
737 "This URL IS a document -- there are no links to extract/rank here."
738 )
739
740 try:
741 page.wait_for_load_state("networkidle", timeout=min(self.timeout_ms, 8000))
742 except Exception:
743 pass
744
745 data = page.evaluate(
746 """
747 () => {
748 const seen = new Set();
749 const links = [];
750 document.querySelectorAll('a[href^="http"]').forEach(a => {
751 const href = a.href;
752 if (!href || seen.has(href)) return;
753 const text = (a.innerText || a.textContent || '').trim().replace(/\\s+/g, ' ');
754 seen.add(href);
755 links.push({ href, text: text.slice(0, 160) });
756 });
757 return { title: document.title || '', links };
758 }
759 """
760 )
761 finally:
762 browser.close()
763
764 title = data.get("title", "") or url
765 links = data.get("links", [])[: self.max_links_extracted]
766
767 if not links:
768 return (
769 f"'{url}' (page: {title}) has no outbound http(s) links to extract. Fall back to "
770 "reading this page's own content with playwright_visit_page instead."
771 )
772
773 scored = []
774 for ln in links:
775 href, text = ln.get("href", ""), ln.get("text", "")
776 scored.append((self._score_link(text, href, keywords), href, text))
777 scored.sort(key=lambda t: t[0], reverse=True)
778
779 shown = scored[: self.max_links_shown]
780 n_relevant = sum(1 for s, _, _ in scored if s > 0)
781
782 lines = [
783 f"## Links extracted from: {title}\nURL: {url}\n",
784 f"Keywords used for scoring: {', '.join(keywords) if keywords else '(none given)'}\n",
785 f"### Ranked links ({len(links)} total found, {n_relevant} scored relevant, showing top {len(shown)})\n",
786 "Open the highest-scoring links first with playwright_visit_page. A score of 0 means no "
787 "keyword or download signal matched at all -- usually site navigation/boilerplate, not a "
788 "candidate document.\n",
789 ]
790 for i, (score, href, text) in enumerate(shown, start=1):
791 label = text or href
792 lines.append(f"{i}. [score={score}] [{label}]({href})")
793
794 if n_relevant == 0:
795 lines.append(
796 "\nNote: nothing on this page scored above 0 for the given keywords -- consider a "
797 "different topic_keywords phrasing, or fall back to reading the page's own content "
798 "with playwright_visit_page in case the relevant text (not a link) is on this page "
799 "itself."
800 )
801
802 return "\n".join(lines)
803
804
805class PlaywrightReadEmbeddedPdfTool(Tool):
806 """Opens a URL using Playwright, scans the page DOM to identify any embedded PDF files
807 (via iframe, embed, or object tags, or direct PDF download links), downloads the PDF,
808 and extracts its text content using pdfplumber to return to the agent.
809 """
810
811 name = "playwright_read_embedded_pdf"
812 description = (
813 "Opens a webpage URL using Playwright, scans the DOM to identify any embedded PDF files "
814 "(in iframes, embed/object tags, or direct PDF links), downloads the PDF, and extracts "
815 "its text content using pdfplumber. Use this tool if you suspect a page contains an "
816 "embedded PDF viewer displaying document text that isn't captured by normal page reading."
817 )
818 inputs = {
819 "url": {"type": "string", "description": "The URL of the webpage that contains the embedded PDF."},
820 }
821 output_type = "string"
822
823 def __init__(
824 self,
825 headless: bool = False,
826 timeout_ms: int = 20000,
827 max_pdf_chars: int = 25000,
828 ):
829 super().__init__()
830 self.headless = headless
831 self.timeout_ms = timeout_ms
832 self.max_pdf_chars = max_pdf_chars
833
834 def forward(self, url: str) -> str:
835 try:
836 from playwright.sync_api import sync_playwright
837 except ImportError as e:
838 raise ImportError(
839 "PlaywrightReadEmbeddedPdfTool needs the `playwright` package. Install with:\n"
840 " pip install playwright\n"
841 " playwright install chromium"
842 ) from e
843
844 import requests
845 import tempfile
846 import os
847 import pdfplumber # MIT — replaced PyMuPDF (AGPL) for Store distribution
848 import urllib.parse
849
850 url = (url or "").strip()
851 if not url:
852 return "Error: empty URL."
853
854 # If the input URL is already a PDF, process it directly
855 if url.lower().endswith(".pdf"):
856 pdf_urls = [url]
857 page_title = "Direct PDF Link"
858 else:
859 if not (url.startswith("http://") or url.startswith("https://")):
860 return f"Error: '{url}' is not an http(s) URL."
861
862 pdf_urls = []
863 page_title = url
864 with sync_playwright() as p:
865 browser = p.chromium.launch(
866 headless=self.headless,
867 args=["--disable-blink-features=AutomationControlled"],
868 )
869 try:
870 context = browser.new_context(
871 user_agent=_USER_AGENT,
872 locale="en-US",
873 viewport={"width": 1366, "height": 900},
874 extra_http_headers={"Accept-Language": "en-US,en;q=0.9"},
875 accept_downloads=True,
876 )
877 context.add_init_script(_ANTI_DETECTION_INIT_SCRIPT)
878 page = context.new_page()
879
880 try:
881 response = page.goto(url, timeout=self.timeout_ms, wait_until="domcontentloaded")
882 page_title = page.title() or url
883 except Exception as e:
884 if "Download is starting" in str(e):
885 # The page URL itself triggered a direct download
886 pdf_urls = [url]
887 else:
888 return f"Error: could not load '{url}': {e}"
889
890 if not pdf_urls:
891 # Scan the DOM for embedded PDFs
892 detected = page.evaluate(
893 """
894 () => {
895 const urls = new Set();
896
897 // Check iframes (common for PDF.js or direct PDF embed)
898 document.querySelectorAll('iframe').forEach(el => {
899 const src = el.src || '';
900 if (!src) return;
901 if (src.toLowerCase().includes('.pdf')) {
902 urls.add(src);
903 }
904 // Try extracting from common PDF viewer query params like viewer.html?file=URL
905 const fileParam = src.match(/[?&]file=([^&]+)/);
906 if (fileParam) {
907 try { urls.add(decodeURIComponent(fileParam[1])); } catch(err) {}
908 }
909 });
910
911 // Check embed tags
912 document.querySelectorAll('embed').forEach(el => {
913 const src = el.src || '';
914 if (src && (src.toLowerCase().includes('.pdf') || (el.type || '').toLowerCase() === 'application/pdf')) {
915 urls.add(src);
916 }
917 });
918
919 // Check object tags
920 document.querySelectorAll('object').forEach(el => {
921 const data = el.data || '';
922 if (data && (data.toLowerCase().includes('.pdf') || (el.type || '').toLowerCase() === 'application/pdf')) {
923 urls.add(data);
924 }
925 });
926
927 // Check links that look like PDFs
928 document.querySelectorAll('a[href]').forEach(el => {
929 const href = el.href || '';
930 const text = (el.innerText || el.textContent || '').trim().toLowerCase();
931 if (href.toLowerCase().includes('.pdf') || text.includes('view pdf') || text.includes('download pdf')) {
932 urls.add(href);
933 }
934 });
935
936 return Array.from(urls);
937 }
938 """
939 )
940 # Resolve absolute URLs
941 for u in detected:
942 resolved = urllib.parse.urljoin(url, u)
943 if resolved not in pdf_urls:
944 pdf_urls.append(resolved)
945 finally:
946 browser.close()
947
948 if not pdf_urls:
949 return f"No embedded PDFs or direct PDF links found on webpage: '{url}'."
950
951 # Fetch and extract text from the detected PDFs (up to 3 to avoid excessive time/text limit issues)
952 lines = [f"## Webpage Analyzed: {page_title}", f"URL: {url}", f"Detected {len(pdf_urls)} embedded PDF(s). Processing..."]
953
954 for idx, pdf_url in enumerate(pdf_urls[:3], 1):
955 lines.append(f"\n--- PDF #{idx}: {pdf_url} ---")
956 temp_path = None
957 try:
958 headers = {"User-Agent": _USER_AGENT}
959 res = requests.get(pdf_url, headers=headers, timeout=20, stream=True)
960 if res.status_code >= 400:
961 lines.append(f"Error downloading PDF: HTTP Status {res.status_code}")
962 continue
963
964 # Write to temp file
965 with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tf:
966 for chunk in res.iter_content(chunk_size=8192):
967 tf.write(chunk)
968 temp_path = tf.name
969
970 # Extract text using pdfplumber (MIT)
971 pdf_text_parts = []
972 with pdfplumber.open(temp_path) as pdf:
973 for page in pdf.pages:
974 page_text = page.extract_text() or ""
975 pdf_text_parts.append(page_text)
976 if sum(len(x) for x in pdf_text_parts) > self.max_pdf_chars:
977 pdf_text_parts.append("\n...[truncated due to length limit]...")
978 break
979
980 extracted_text = "\n".join(pdf_text_parts).strip()
981 if extracted_text:
982 lines.append(extracted_text)
983 else:
984 lines.append("(No text could be extracted from this PDF -- it might be scanned image pages)")
985 except Exception as pdf_err:
986 lines.append(f"Failed to process PDF: {pdf_err}")
987 finally:
988 if temp_path and os.path.exists(temp_path):
989 try:
990 os.remove(temp_path)
991 except:
992 pass
993
994 return "\n".join(lines)
995
996
997# ── Shared page state for navigation tools ──────────────────────────
998# PlaywrightVisitPageTool caches the last-visited page here so
999# PageUp/PageDown/FindOnPage/FindNext can navigate within its content
1000# without re-launching a browser.
1001_page_cache = {
1002 "url": None,
1003 "title": None,
1004 "full_text": "",
1005 "viewport_pos": 0,
1006 "viewport_size": 2000,
1007 "find_query": None,
1008 "find_last_viewport": None,
1009}
1010
1011
1012def _save_page_cache(url, title, text):
1013 _page_cache["url"] = url
1014 _page_cache["title"] = title
1015 _page_cache["full_text"] = text or ""
1016 _page_cache["viewport_pos"] = 0
1017 _page_cache["find_query"] = None
1018 _page_cache["find_last_viewport"] = None
1019
1020
1021def _format_viewport(cache):
1022 total = len(cache["full_text"])
1023 vsize = cache["viewport_size"]
1024 pos = cache["viewport_pos"]
1025 total_pages = max(1, (total + vsize - 1) // vsize)
1026 current_page = pos // vsize + 1
1027 content = cache["full_text"][pos:pos + vsize]
1028 return (
1029 f"Address: {cache['url']}\n"
1030 f"Title: {cache['title']}\n"
1031 f"Viewport position: Showing page {current_page} of {total_pages}.\n"
1032 f"=======================\n"
1033 f"{content}"
1034 )
1035
1036
1037class PlaywrightPageDownTool(Tool):
1038 name = "playwright_page_down"
1039 description = "Scroll the viewport DOWN one page-length in the currently visited page and return the new viewport content."
1040 inputs = {}
1041 output_type = "string"
1042
1043 def forward(self) -> str:
1044 cache = _page_cache
1045 if not cache["url"]:
1046 return "No page has been visited yet. Use playwright_visit_page first."
1047 total = len(cache["full_text"])
1048 cache["viewport_pos"] = min(cache["viewport_pos"] + cache["viewport_size"], max(0, total - 1))
1049 return _format_viewport(cache)
1050
1051
1052class PlaywrightPageUpTool(Tool):
1053 name = "playwright_page_up"
1054 description = "Scroll the viewport UP one page-length in the currently visited page and return the new viewport content."
1055 inputs = {}
1056 output_type = "string"
1057
1058 def forward(self) -> str:
1059 cache = _page_cache
1060 if not cache["url"]:
1061 return "No page has been visited yet. Use playwright_visit_page first."
1062 cache["viewport_pos"] = max(cache["viewport_pos"] - cache["viewport_size"], 0)
1063 return _format_viewport(cache)
1064
1065
1066class PlaywrightFindOnPageTool(Tool):
1067 name = "playwright_find_on_page"
1068 description = "Scroll the viewport to the first occurrence of the search string. This is equivalent to Ctrl+F on the currently visited page."
1069 inputs = {
1070 "search_string": {
1071 "type": "string",
1072 "description": "The string to search for on the page. Supports wildcards like '*'.",
1073 }
1074 }
1075 output_type = "string"
1076
1077 def forward(self, search_string: str) -> str:
1078 cache = _page_cache
1079 if not cache["url"]:
1080 return "No page has been visited yet. Use playwright_visit_page first."
1081
1082 import re
1083 query = re.sub(r"\*", ".*", re.escape(search_string))
1084 full = cache["full_text"]
1085
1086 match = re.search(query, full, re.IGNORECASE)
1087 if not match:
1088 cache["find_query"] = search_string
1089 cache["find_last_viewport"] = None
1090 return (
1091 f"Address: {cache['url']}\n"
1092 f"Title: {cache['title']}\n"
1093 f"=======================\n"
1094 f"The search string '{search_string}' was not found on this page."
1095 )
1096
1097 pos = match.start()
1098 vsize = cache["viewport_size"]
1099 cache["viewport_pos"] = max(0, pos - vsize // 4)
1100 cache["find_query"] = search_string
1101 cache["find_last_viewport"] = cache["viewport_pos"]
1102 return _format_viewport(cache)
1103
1104
1105class PlaywrightFindNextTool(Tool):
1106 name = "playwright_find_next"
1107 description = "Scroll the viewport to the next occurrence of the search string. Use after playwright_find_on_page."
1108 inputs = {}
1109 output_type = "string"
1110
1111 def forward(self) -> str:
1112 cache = _page_cache
1113 if not cache["url"]:
1114 return "No page has been visited yet. Use playwright_visit_page first."
1115 if not cache["find_query"]:
1116 return "No active search. Use playwright_find_on_page first."
1117
1118 import re
1119 query = re.sub(r"\*", ".*", re.escape(cache["find_query"]))
1120 full = cache["full_text"]
1121 start = cache["viewport_pos"] + cache["viewport_size"]
1122 if cache["find_last_viewport"] is not None and cache["find_last_viewport"] >= start:
1123 start = cache["find_last_viewport"] + 1
1124
1125 if start >= len(full):
1126 start = 0
1127
1128 match = re.search(query, full[start:], re.IGNORECASE)
1129 if not match:
1130 cache["find_last_viewport"] = None
1131 return (
1132 f"Address: {cache['url']}\n"
1133 f"Title: {cache['title']}\n"
1134 f"=======================\n"
1135 f"No more occurrences of '{cache['find_query']}' found."
1136 )
1137
1138 pos = start + match.start()
1139 vsize = cache["viewport_size"]
1140 cache["viewport_pos"] = max(0, pos - vsize // 4)
1141 cache["find_last_viewport"] = cache["viewport_pos"]
1142 return _format_viewport(cache)
1143
1144
1145class PlaywrightArchiveSearchTool(Tool):
1146 name = "playwright_find_archived_url"
1147 description = "Given a url, searches the Wayback Machine and returns the archived version of the url closest to the desired date. Use this when a page is dead, blocked, or has changed since the date you care about."
1148 inputs = {
1149 "url": {
1150 "type": "string",
1151 "description": "The url you need the archive for.",
1152 },
1153 "date": {
1154 "type": "string",
1155 "description": "The date to find the archive for, in 'YYYYMMDD' format (e.g. '27 June 2008' → '20080627').",
1156 "nullable": True,
1157 },
1158 }
1159 output_type = "string"
1160
1161 def forward(self, url: str, date: str | None = None) -> str:
1162 import requests
1163
1164 no_timestamp_url = f"https://archive.org/wayback/available?url={urllib.parse.quote(url)}"
1165 archive_url = no_timestamp_url + (f"×tamp={date}" if date else "")
1166 try:
1167 response = requests.get(archive_url, timeout=30).json()
1168 except Exception as e:
1169 return f"Error querying Wayback Machine: {e}"
1170
1171 if "archived_snapshots" in response and "closest" in response["archived_snapshots"]:
1172 closest = response["archived_snapshots"]["closest"]
1173 else:
1174 try:
1175 response2 = requests.get(no_timestamp_url, timeout=30).json()
1176 if "archived_snapshots" in response2 and "closest" in response2["archived_snapshots"]:
1177 closest = response2["archived_snapshots"]["closest"]
1178 else:
1179 return f"No archive found for '{url}' on Wayback Machine."
1180 except Exception as e:
1181 return f"Error querying Wayback Machine: {e}"
1182
1183 target_url = closest["url"]
1184 snapshot_date = closest["timestamp"][:8] if "timestamp" in closest else "unknown"
1185 return (
1186 f"Web archive for url {url}, snapshot taken at date {snapshot_date}:\n"
1187 f"Archived URL: {target_url}\n\n"
1188 f"To view this archived page, call playwright_visit_page with the archived URL above."
1189 )
1190