feat: add QR login and browser-backed web tools

app/web_tools.py

@@ -0,0 +1,124 @@
+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})