95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Mechanical web-fetch helper for the /research skill.
|
|
|
|
Talks to the local SearXNG (search) and Firecrawl (scrape) services. Stdlib
|
|
only — no third-party runtime dependencies. The /research skill (Claude) does
|
|
all the thinking; this file only fetches and parses.
|
|
|
|
Endpoints resolve from env (WEBTOOLS_HOST / SEARXNG_PORT / FIRECRAWL_PORT) with
|
|
defaults for the `webtools` box. Any failure exits nonzero with a stderr message
|
|
so the calling skill can react.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
class WebToolsError(Exception):
|
|
"""Raised on any unrecoverable search/scrape failure."""
|
|
|
|
|
|
def endpoints() -> tuple[str, str]:
|
|
"""Return (searxng_base, firecrawl_base) from env, with webtools defaults."""
|
|
host = os.environ.get("WEBTOOLS_HOST", "10.10.20.37")
|
|
searx_port = os.environ.get("SEARXNG_PORT", "8080")
|
|
fire_port = os.environ.get("FIRECRAWL_PORT", "3002")
|
|
return (f"http://{host}:{searx_port}", f"http://{host}:{fire_port}")
|
|
|
|
|
|
def _request(url, *, data=None, headers=None, timeout=30, retries=2, backoff=1.5):
|
|
"""GET (data=None) or POST JSON; parse JSON; retry transient errors."""
|
|
headers = dict(headers or {})
|
|
body = None
|
|
if data is not None:
|
|
body = json.dumps(data).encode("utf-8")
|
|
headers.setdefault("Content-Type", "application/json")
|
|
|
|
last = None
|
|
for attempt in range(retries + 1):
|
|
try:
|
|
req = urllib.request.Request(
|
|
url, data=body, headers=headers,
|
|
method="POST" if body is not None else "GET",
|
|
)
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
except (urllib.error.URLError, TimeoutError, ValueError) as exc:
|
|
last = exc
|
|
if attempt < retries:
|
|
time.sleep(backoff ** attempt)
|
|
raise WebToolsError(f"request failed after {retries + 1} attempt(s): {url}: {last}")
|
|
|
|
|
|
def search(query, max_results=12, timeout=30):
|
|
"""Query SearXNG and return up to max_results normalized hits."""
|
|
searx_base, _ = endpoints()
|
|
qs = urllib.parse.urlencode({"q": query, "format": "json"})
|
|
payload = _request(f"{searx_base}/search?{qs}", timeout=timeout)
|
|
hits = []
|
|
for r in payload.get("results", []):
|
|
url = r.get("url")
|
|
if not url:
|
|
continue
|
|
hits.append({
|
|
"title": r.get("title", ""),
|
|
"url": url,
|
|
"snippet": r.get("content", ""),
|
|
"engine": r.get("engine", ""),
|
|
})
|
|
return hits[:max_results]
|
|
|
|
|
|
def scrape(url, timeout=60):
|
|
"""Scrape a URL to markdown via Firecrawl. Raises WebToolsError on failure."""
|
|
_, fire_base = endpoints()
|
|
payload = _request(
|
|
f"{fire_base}/v1/scrape",
|
|
data={"url": url, "formats": ["markdown"]},
|
|
timeout=timeout,
|
|
)
|
|
if not payload.get("success", True):
|
|
raise WebToolsError(f"firecrawl failed for {url}: {payload.get('error')}")
|
|
markdown = payload.get("data", {}).get("markdown", "")
|
|
if not markdown:
|
|
raise WebToolsError(f"firecrawl returned no markdown for {url}")
|
|
return markdown
|