125 lines
4.6 KiB
Python
125 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import ipaddress
|
|
import json
|
|
import socket
|
|
from collections.abc import Sequence
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
from micro_agents import Tool
|
|
|
|
|
|
class WebPageTools:
|
|
def __init__(self, *, timeout_seconds: float = 45, max_chars: int = 60_000):
|
|
if timeout_seconds <= 0:
|
|
raise ValueError("timeout_seconds must be positive")
|
|
if max_chars < 1:
|
|
raise ValueError("max_chars must be positive")
|
|
self._timeout_seconds = timeout_seconds
|
|
self._max_chars = max_chars
|
|
|
|
def tools(self) -> list[Tool]:
|
|
return [Tool(self.read_web_page, name="read_web_page", strict=False)]
|
|
|
|
async def read_web_page(self, url: str) -> str:
|
|
"""Read a public web page in a Playwright browser and return cleaned Markdown.
|
|
|
|
Use this before analyzing, summarizing, or quoting a URL. Unlike web_search,
|
|
this loads the page itself and can render JavaScript-driven content.
|
|
"""
|
|
safe_url = await _ensure_public_url(url)
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
_crawl_with_playwright(safe_url, self._timeout_seconds),
|
|
timeout=self._timeout_seconds + 5,
|
|
)
|
|
except TimeoutError as exc:
|
|
raise RuntimeError("Web page crawl timed out") from exc
|
|
|
|
final_url = str(getattr(result, "redirected_url", None) or getattr(result, "url", safe_url))
|
|
await _ensure_public_url(final_url)
|
|
if not getattr(result, "success", False):
|
|
error = str(getattr(result, "error_message", None) or "unknown crawl error")
|
|
raise RuntimeError(f"Could not read web page: {error}")
|
|
|
|
markdown_result = getattr(result, "markdown", None)
|
|
markdown = str(markdown_result or "").strip()
|
|
if not markdown:
|
|
raise RuntimeError("Web page did not contain readable content")
|
|
|
|
truncated = len(markdown) > self._max_chars
|
|
if truncated:
|
|
markdown = markdown[: self._max_chars].rstrip()
|
|
|
|
metadata = getattr(result, "metadata", None)
|
|
title = metadata.get("title") if isinstance(metadata, dict) else None
|
|
return json.dumps(
|
|
{
|
|
"url": final_url,
|
|
"title": title,
|
|
"status_code": getattr(result, "status_code", None),
|
|
"truncated": truncated,
|
|
"markdown": markdown,
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
|
|
async def _crawl_with_playwright(url: str, timeout_seconds: float) -> Any:
|
|
from crawl4ai import AsyncWebCrawler, BrowserConfig, CacheMode, CrawlerRunConfig
|
|
|
|
browser_config = BrowserConfig(
|
|
browser_type="chromium",
|
|
headless=True,
|
|
text_mode=True,
|
|
light_mode=True,
|
|
enable_stealth=True,
|
|
verbose=False,
|
|
)
|
|
run_config = CrawlerRunConfig(
|
|
cache_mode=CacheMode.BYPASS,
|
|
page_timeout=max(1, int(timeout_seconds * 1000)),
|
|
wait_until="domcontentloaded",
|
|
remove_overlay_elements=True,
|
|
remove_consent_popups=True,
|
|
exclude_social_media_links=True,
|
|
process_iframes=False,
|
|
scan_full_page=False,
|
|
verbose=False,
|
|
log_console=False,
|
|
)
|
|
async with AsyncWebCrawler(config=browser_config) as crawler:
|
|
return await crawler.arun(url=url, config=run_config)
|
|
|
|
|
|
async def _ensure_public_url(value: str) -> str:
|
|
url = value.strip()
|
|
if not url or len(url) > 2048:
|
|
raise ValueError("url must contain at most 2048 characters")
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme not in {"http", "https"}:
|
|
raise ValueError("Only http and https URLs are allowed")
|
|
if parsed.username is not None or parsed.password is not None:
|
|
raise ValueError("Credentials in URLs are not allowed")
|
|
hostname = parsed.hostname
|
|
if not hostname:
|
|
raise ValueError("URL must include a hostname")
|
|
|
|
try:
|
|
addresses = [ipaddress.ip_address(hostname)]
|
|
except ValueError:
|
|
addresses = await asyncio.to_thread(_resolve_addresses, hostname, parsed.port)
|
|
if not addresses or any(not address.is_global for address in addresses):
|
|
raise ValueError("Local and private network URLs are not allowed")
|
|
return url
|
|
|
|
|
|
def _resolve_addresses(hostname: str, port: int | None) -> Sequence[ipaddress.IPv4Address | ipaddress.IPv6Address]:
|
|
try:
|
|
records = socket.getaddrinfo(hostname, port or 443, type=socket.SOCK_STREAM)
|
|
except socket.gaierror as exc:
|
|
raise ValueError("URL hostname could not be resolved") from exc
|
|
return tuple({ipaddress.ip_address(record[4][0]) for record in records})
|