__init__.py

from __future__ import annotations

import json

from .agent_web_search.reader import DirectReader, HeadlessReader
from .agent_web_search.search import DirectBackend, HeadlessBrowserBackend

SCHEMA = {
    "name": "agent_web_search",
    "description": "Search the public web without an API key. Current default direct backend uses Bing only because Baidu and Yandex currently require human verification on this machine; blocked engines are never treated as results. Request 1-30 results.",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query."},
            "limit": {"type": "integer", "minimum": 1, "maximum": 30, "default": 5, "description": "Results in this batch."},
            "offset": {"type": "integer", "minimum": 0, "default": 0, "description": "Zero-based result offset. Use 10 for the second batch after a 10-result first batch."},
            "exclude_urls": {"type": "array", "items": {"type": "string"}, "maxItems": 100, "description": "URLs returned by earlier batches; these are removed from this batch."},
            "backend": {"type": "string", "enum": ["auto", "direct", "headless_browser"], "default": "auto"},
        },
        "required": ["query"],
    },
}

READ_SCHEMA = {
    "name": "agent_web_read",
    "description": "Read article text from up to 10 public URLs without opening the user's visible browser. Direct HTTP first, headless Chromium fallback. Captcha, login, and paywall pages are reported, never fabricated.",
    "parameters": {
        "type": "object",
        "properties": {
            "urls": {"type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 10},
            "backend": {"type": "string", "enum": ["auto", "direct", "headless_browser"], "default": "auto"},
        },
        "required": ["urls"],
    },
}


def _rows(items):
    rows = []
    for item in items:
        if isinstance(item, dict):
            rows.append(item)
        else:
            rows.append({"title": item.title, "url": item.url, "snippet": item.snippet, "engine": item.engine})
    return rows


def web_search(args: dict, **_: object) -> str:
    query = str(args.get("query", "")).strip()
    limit = max(1, min(int(args.get("limit", 5)), 30))
    offset = max(0, int(args.get("offset", 0)))
    exclude_urls = tuple(str(url).strip() for url in (args.get("exclude_urls") or []) if str(url).strip())
    backend = str(args.get("backend", "auto"))
    if not query:
        return json.dumps({"success": False, "error": "query is required"})
    attempts = [("direct", DirectBackend)] if backend == "direct" else [("headless_browser", HeadlessBrowserBackend)] if backend == "headless_browser" else [("direct", DirectBackend), ("headless_browser", HeadlessBrowserBackend)]
    errors = []
    for name, factory in attempts:
        try:
            instance = factory()
            if offset or exclude_urls:
                results = instance.search(query, limit, offset=offset, exclude_urls=exclude_urls)
            else:
                results = instance.search(query, limit)
            return json.dumps({
                "success": True,
                "backend": name,
                "offset": offset,
                "next_offset": offset + limit,
                "results": _rows(results),
                "engines": [item.__dict__ for item in getattr(instance, "attempts", [])],
            }, ensure_ascii=False)
        except Exception as exc:
            errors.append(f"{name}: {exc}")
    return json.dumps({"success": False, "error": "; ".join(errors)}, ensure_ascii=False)


def web_read(args: dict, **_: object) -> str:
    urls = [str(url).strip() for url in (args.get("urls") or []) if str(url).strip()][:10]
    backend = str(args.get("backend", "auto"))
    if not urls:
        return json.dumps({"success": False, "error": "urls is required"})
    factories = [("direct", DirectReader)] if backend == "direct" else [("headless_browser", HeadlessReader)] if backend == "headless_browser" else [("direct", DirectReader), ("headless_browser", HeadlessReader)]
    articles, failures = [], []
    for url in urls:
        for index, (name, factory) in enumerate(factories):
            try:
                article = factory().read(url)
                articles.append({"url": article.url, "title": article.title, "text": article.text, "backend": name})
                break
            except Exception as exc:
                if index == len(factories) - 1:
                    failures.append({"url": url, "error": str(exc)})
    return json.dumps({"success": bool(articles), "articles": articles, "failures": failures}, ensure_ascii=False)


def prefer_agent_web_search(**_: object) -> dict[str, str]:
    return {
        "context": (
            "Web tool routing policy: for public web search, current facts, documentation discovery, and reading public URLs, "
            "use agent_web_search first and agent_web_read for page text. These tools use direct HTTP with a temporary private "
            "headless browser fallback and do not open the user's visible browser. Do not use legacy browser automation merely "
            "to search or read public pages. Use a visible/interactive browser only when the task genuinely requires login, forms, "
            "clicking, user account state, downloads, or another interactive page action. If agent_web_search or agent_web_read "
            "reports a CAPTCHA, login wall, or unavailable page, report that honestly instead of bypassing verification."
        )
    }


def register(ctx) -> None:
    ctx.register_tool(name="agent_web_search", toolset="agent-web-search", schema=SCHEMA, handler=web_search)
    ctx.register_tool(name="agent_web_read", toolset="agent-web-search", schema=READ_SCHEMA, handler=web_read)
    ctx.register_hook("pre_llm_call", prefer_agent_web_search)