butler/tools/web_search.py
Rama main · réplica de lectura del repositorio original.
"""Web search tool — DuckDuckGo HTML POST + API fallback (no API key)."""
from __future__ import annotations
import json
import logging
import os
import re
import time
from html import unescape
from typing import Any, Callable, cast
from urllib.parse import quote_plus, unquote
from butler.utilities.env_parse import env_truthy, float_env, int_env
logger = logging.getLogger(__name__)
_BUDGET_MAX_SEC = 300.0
_PER_ATTEMPT_TIMEOUT_MAX_SEC = 30.0
_DDG_HTML_URL = "https://html.duckduckgo.com/html/"
_DDG_LITE_URL = "https://lite.duckduckgo.com/lite/"
_DDG_API_URL = "https://api.duckduckgo.com/"
_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
),
"Accept": "text/html,application/xhtml+xml,application/json",
"Accept-Language": "zh-CN,zh;q=0.9,en-US,en;q=0.8",
}
def web_search_enabled() -> bool:
return bool(env_truthy("BUTLER_ENABLE_WEB_SEARCH", default=False))
def _per_attempt_timeout_cap() -> float:
try:
raw = float_env("BUTLER_WEB_SEARCH_TIMEOUT", 15)
return float(min(_PER_ATTEMPT_TIMEOUT_MAX_SEC, max(3.0, raw)))
except ValueError:
return 15.0
def _total_budget() -> float:
try:
raw = float_env("BUTLER_WEB_SEARCH_BUDGET", 60)
return float(min(_BUDGET_MAX_SEC, max(5.0, raw)))
except ValueError:
return 60.0
def _proxy_configured() -> bool:
for key in (
"HTTP_PROXY",
"HTTPS_PROXY",
"ALL_PROXY",
"http_proxy",
"https_proxy",
"all_proxy",
):
if str(os.environ.get(key) or "").strip():
return True
return False
def _try_direct_with_proxy() -> bool:
"""When proxy env is set, skip direct attempts unless explicitly enabled."""
if not _proxy_configured():
return True
return bool(env_truthy("BUTLER_WEB_SEARCH_TRY_DIRECT", default=False))
def _search_strategies() -> list[tuple[str, bool]]:
"""Return (kind, trust_env). Proxy-only hosts skip direct to preserve budget."""
kinds = ("html_post", "html_lite", "api")
if _proxy_configured() and not _try_direct_with_proxy():
return [(kind, True) for kind in kinds]
if _proxy_configured():
return [(kind, False) for kind in kinds] + [(kind, True) for kind in kinds]
return [(kind, True) for kind in kinds] + [(kind, False) for kind in kinds]
def _max_retries() -> int:
try:
return max(1, int(int_env("BUTLER_WEB_SEARCH_RETRIES", 2, min=1)))
except ValueError:
return 2
def _unwrap_ddg_href(href: str) -> str:
raw = str(href or "").strip()
match = re.search(r"uddg=([^&]+)", raw)
if match:
return unquote(match.group(1))
return raw
def _strip_html(text: str) -> str:
return unescape(re.sub(r"<[^>]+>", "", text or "").strip())
def parse_ddg_html_results(html: str, max_results: int) -> list[dict[str, str]]:
"""Parse DuckDuckGo HTML SERP into title/snippet/url rows."""
cap = max(1, max_results)
results: list[dict[str, str]] = []
seen: set[str] = set()
blocks = re.findall(
r'<a[^>]+class="result__a"[^>]+href="([^"]*)"[^>]*>(.*?)</a>.*?'
r'<a[^>]+class="result__snippet"[^>]*>(.*?)</a>',
html,
re.DOTALL | re.IGNORECASE,
)
for href, title_html, snippet_html in blocks:
url = _unwrap_ddg_href(href)
title = _strip_html(title_html)
if not title or not url or url in seen:
continue
seen.add(url)
results.append({
"title": title,
"snippet": _strip_html(snippet_html)[:300],
"url": url,
})
if len(results) >= cap:
return results
# Lite / alternate markup fallback.
for href, title_html in re.findall(
r'<a[^>]+rel="nofollow"[^>]+href="([^"]+)"[^>]*>(.*?)</a>',
html,
re.DOTALL | re.IGNORECASE,
):
url = _unwrap_ddg_href(href)
title = _strip_html(title_html)
if not title or not url.startswith("http") or url in seen:
continue
seen.add(url)
results.append({"title": title, "snippet": "", "url": url})
if len(results) >= cap:
break
return results
def _parse_ddg_api_payload(payload: dict[str, Any], max_results: int) -> list[dict[str, str]]:
cap = max(1, max_results)
results: list[dict[str, str]] = []
seen: set[str] = set()
def _add(text: str, url: str) -> None:
if len(results) >= cap or not url or url in seen:
return
seen.add(url)
results.append({
"title": (text or url)[:200],
"snippet": (text or "")[:300],
"url": url,
})
abstract = str(payload.get("AbstractURL") or "").strip()
abstract_text = str(payload.get("AbstractText") or payload.get("Abstract") or "").strip()
if abstract:
_add(abstract_text or abstract, abstract)
related = payload.get("RelatedTopics") or []
if isinstance(related, list):
for item in related:
if not isinstance(item, dict):
continue
topics = item.get("Topics")
if isinstance(topics, list):
for sub in topics:
if not isinstance(sub, dict):
continue
_add(str(sub.get("Text") or ""), str(sub.get("FirstURL") or ""))
else:
_add(str(item.get("Text") or ""), str(item.get("FirstURL") or ""))
return results
def _fetch_with_httpx(
method: str,
url: str,
*,
trust_env: bool,
timeout: float,
data: dict[str, str] | None = None,
) -> str:
import httpx
per = max(3.0, timeout)
connect_cap = 5.0 if trust_env else 3.0
timeouts = httpx.Timeout(
connect=min(connect_cap, per),
read=per,
write=min(10.0, per),
pool=min(5.0, per),
)
with httpx.Client(
timeout=timeouts,
follow_redirects=True,
trust_env=trust_env,
headers=_HEADERS,
) as client:
if method == "post":
resp = client.post(url, data=data or {})
else:
resp = client.get(url)
resp.raise_for_status()
return str(resp.text)
def _search_ddg_html_post(
query: str,
max_results: int,
*,
trust_env: bool,
timeout: float,
) -> list[dict[str, str]]:
html = _fetch_with_httpx(
"post",
_DDG_HTML_URL,
trust_env=trust_env,
timeout=timeout,
data={"q": query},
)
return parse_ddg_html_results(html, max_results)
def _search_ddg_api(
query: str,
max_results: int,
*,
trust_env: bool,
timeout: float,
) -> list[dict[str, str]]:
url = (
f"{_DDG_API_URL}?q={quote_plus(query)}"
"&format=json&no_html=1&no_redirect=1&skip_disambig=1"
)
text = _fetch_with_httpx("get", url, trust_env=trust_env, timeout=timeout)
payload = json.loads(text)
if not isinstance(payload, dict):
return []
return _parse_ddg_api_payload(payload, max_results)
def _search_ddg_lite_get(
query: str,
max_results: int,
*,
trust_env: bool,
timeout: float,
) -> list[dict[str, str]]:
url = f"{_DDG_LITE_URL}?q={quote_plus(query)}"
html = _fetch_with_httpx("get", url, trust_env=trust_env, timeout=timeout)
return parse_ddg_html_results(html, max_results)
def _search_duckduckgo(query: str, max_results: int = 5) -> list[dict[str, str]]:
"""Try DDG backends within a total time budget; direct-first when proxy env is set."""
q = str(query or "").strip()
if not q:
return []
budget = _total_budget()
deadline = time.monotonic() + budget
strategies = _search_strategies()
max_rounds = _max_retries()
round_idx = 0
from butler.tools.web_search_ops import try_search_attempt_safe
while time.monotonic() < deadline and round_idx < max_rounds:
round_idx += 1
for kind, trust_env in strategies:
remaining = deadline - time.monotonic()
if remaining < 2.5:
break
per_timeout = min(_per_attempt_timeout_cap(), max(3.0, remaining))
def _run(
_kind: str = kind,
_trust_env: bool = trust_env,
_timeout: float = per_timeout,
) -> list[dict[str, str]]:
if _kind == "html_post":
return _search_ddg_html_post(
q,
max_results,
trust_env=_trust_env,
timeout=_timeout,
)
if _kind == "html_lite":
return _search_ddg_lite_get(
q,
max_results,
trust_env=_trust_env,
timeout=_timeout,
)
return _search_ddg_api(
q,
max_results,
trust_env=_trust_env,
timeout=_timeout,
)
rows = try_search_attempt_safe(
_run,
kind=kind,
trust_env=trust_env,
round_idx=round_idx,
query=q,
)
if rows:
return cast(list[dict[str, str]], rows)[:max_results]
return []
def tool_web_search(query: str = "", max_results: int = 5, **_: Any) -> str:
"""Search the web and return structured results."""
if not web_search_enabled():
return json.dumps({
"error": "web_search disabled",
"hint": "设置 BUTLER_ENABLE_WEB_SEARCH=1 启用，或使用 /config set BUTLER_ENABLE_WEB_SEARCH 1",
}, ensure_ascii=False)
q = (query or "").strip()
if not q:
return json.dumps({"error": "query is required"}, ensure_ascii=False)
cap = max(1, min(10, int(max_results or 5)))
started = time.monotonic()
results = _search_duckduckgo(q, max_results=cap)
elapsed = round(time.monotonic() - started, 2)
if not results:
return json.dumps({
"ok": True,
"query": q,
"results": [],
"elapsed_seconds": elapsed,
"budget_seconds": _total_budget(),
"message": "No results found or search service unavailable",
"hint": (
"DuckDuckGo 不可达或零结果（已用尽本轮 web_search 时间预算）。"
"下一步：1) mcp_firecrawl_firecrawl_search 取 URL；"
"2) 对命中 URL 用 mcp_firecrawl_scrape 读正文；"
"3) 直接总结并附完整 https 来源。勿重复 web_search，勿调 feedback/agent。"
),
"fallback": "firecrawl_search_then_scrape",
}, ensure_ascii=False)
return json.dumps({
"ok": True,
"query": q,
"count": len(results),
"results": results,
}, ensure_ascii=False)
def register_web_search_tool(register_fn: Callable[..., None]) -> None:
register_fn(
name="web_search",
description=(
"Search the web for information using a keyword query. "
"Returns titles, snippets and URLs. Use this before web_fetch "
"when you need to find relevant pages."
),
schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query (keywords or question)",
},
"max_results": {
"type": "integer",
"default": 5,
"minimum": 1,
"maximum": 10,
"description": "Maximum number of results to return",
},
},
"required": ["query"],
},
handler=tool_web_search,
toolset="network",
)
__all__ = [
"parse_ddg_html_results",
"register_web_search_tool",
"tool_web_search",
"web_search_enabled",
"_proxy_configured",
"_search_duckduckgo",
"_search_strategies",
"_total_budget",
"_try_direct_with_proxy",
]