Web Search and Browsing Tools for Research Agents

import json
import os
import re
from urllib.parse import urlparse
 
import httpx
 
SEARCH_API = os.environ.get("SEARCH_API_BASE", "https://api.search.example/v1/search")
SEARCH_KEY = os.environ["SEARCH_API_KEY"]
 
TOOL_SPEC = {
    "type": "function",
    "function": {
        "name": "web_search",
        "description": (
            "Search the public web for recent information. "
            "Returns titles, URLs, and short snippets for citation. "
            "Use for facts that may change; prefer primary sources."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query, 3-200 chars."},
                "max_results": {"type": "integer", "minimum": 1, "maximum": 8, "default": 5},
                "site": {
                    "type": "string",
                    "description": "Optional single domain to restrict results (example.com).",
                },
            },
            "required": ["query"],
            "additionalProperties": False,
        },
    },
}
 
_DOMAIN = re.compile(r"^(?:[a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}$")
 
def web_search(arguments_json: str) -> str:
    args = json.loads(arguments_json or "{}")
    query = (args.get("query") or "").strip()
    if not 3 <= len(query) <= 200:
        return json.dumps({"ok": False, "error_code": "INVALID_ARG", "message": "query length", "retryable": False})
    max_results = min(int(args.get("max_results") or 5), 8)
    site = args.get("site")
    if site and not _DOMAIN.match(site):
        return json.dumps({"ok": False, "error_code": "INVALID_ARG", "message": "bad site", "retryable": False})
 
    q = f"site:{site} {query}" if site else query
    with httpx.Client(timeout=20.0) as client:
        r = client.get(
            SEARCH_API,
            params={"q": q, "n": max_results},
            headers={"Authorization": f"Bearer {SEARCH_KEY}"},
        )
    if r.status_code == 429:
        return json.dumps({"ok": False, "error_code": "RATE_LIMIT", "retryable": True})
    if r.status_code >= 400:
        return json.dumps({"ok": False, "error_code": "UPSTREAM", "retryable": r.status_code >= 500})
 
    hits = []
    for item in (r.json().get("results") or [])[:max_results]:
        url = item.get("url") or ""
        if not url.startswith("https://"):
            continue
        hits.append({
            "title": (item.get("title") or "")[:160],
            "url": url,
            "snippet": (item.get("snippet") or "")[:400],
        })
    return json.dumps({"ok": True, "query": query, "results": hits})

from html.parser import HTMLParser
 
class _Text(HTMLParser):
    def __init__(self):
        super().__init__()
        self.parts: list[str] = []
    def handle_data(self, data):
        t = data.strip()
        if t:
            self.parts.append(t)
 
def _is_public_https(url: str) -> bool:
    p = urlparse(url)
    if p.scheme != "https" or not p.hostname:
        return False
    host = p.hostname.lower()
    if host in {"localhost"} or host.endswith(".local"):
        return False
    # Also block private IPs via DNS resolution policy in production.
    return True
 
def fetch_url(arguments_json: str) -> str:
    args = json.loads(arguments_json or "{}")
    url = (args.get("url") or "").strip()
    if not _is_public_https(url):
        return json.dumps({"ok": False, "error_code": "BLOCKED_URL", "retryable": False})
    with httpx.Client(timeout=15.0, follow_redirects=True, max_redirects=3) as client:
        r = client.get(url, headers={"User-Agent": "ResearchAgent/1.0"})
    if r.status_code >= 400:
        return json.dumps({"ok": False, "error_code": "HTTP", "status": r.status_code, "retryable": False})
    raw = r.text[:200_000]
    parser = _Text()
    parser.feed(raw)
    text = " ".join(parser.parts)[:8000]
    return json.dumps({"ok": True, "url": str(r.url), "text": text})