feat(tools): run_python, web search/fetch, market data, options, portfolio

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016DvdRZkyCy9NJpHYDdPqmS

tools/web.py

@@ -0,0 +1,114 @@
+"""Our own web tools — no search API key.
+
+web_search hits DuckDuckGo's keyless HTML endpoint and parses results ourselves (lxml,
+which ships with trafilatura). Falls back to the `ddgs` library if the HTML layout shifts
+or we get rate-limited. web_fetch pulls a page and extracts readable text with trafilatura.
+"""
+
+from __future__ import annotations
+
+from urllib.parse import parse_qs, unquote, urlparse
+
+import httpx
+import trafilatura
+from langchain_core.tools import tool
+from lxml import html as lxml_html
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+_UA = (
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/125.0 Safari/537.36"
+)
+_MAX_RESULTS = 6
+_FETCH_CHAR_LIMIT = 12000  # ~3k tokens
+
+
+def _clean_url(href: str) -> str:
+    """DDG HTML wraps results in a /l/?uddg=<encoded> redirect — unwrap to the real URL."""
+    if href.startswith("//"):
+        href = "https:" + href
+    if "duckduckgo.com/l/" in href or href.startswith("/l/"):
+        qs = parse_qs(urlparse(href).query)
+        if "uddg" in qs:
+            return unquote(qs["uddg"][0])
+    return href
+
+
+@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=8), reraise=True)
+def _ddg_html_search(query: str, max_results: int = _MAX_RESULTS) -> list[dict]:
+    resp = httpx.post(
+        "https://html.duckduckgo.com/html/",
+        data={"q": query},
+        headers={"User-Agent": _UA},
+        timeout=15,
+        follow_redirects=True,
+    )
+    resp.raise_for_status()
+    doc = lxml_html.fromstring(resp.text)
+    anchors = doc.xpath('//a[contains(@class, "result__a")]')
+    snippets = doc.xpath('//a[contains(@class, "result__snippet")]')
+    results = []
+    for i, a in enumerate(anchors[:max_results]):
+        title = a.text_content().strip()
+        url = _clean_url(a.get("href", ""))
+        snippet = snippets[i].text_content().strip() if i < len(snippets) else ""
+        if title and url:
+            results.append({"title": title, "url": url, "snippet": snippet})
+    return results
+
+
+def _ddgs_search(query: str, max_results: int = _MAX_RESULTS) -> list[dict]:
+    from ddgs import DDGS
+
+    with DDGS() as d:
+        return [
+            {"title": r.get("title", ""), "url": r.get("href", ""), "snippet": r.get("body", "")}
+            for r in d.text(query, max_results=max_results)
+        ]
+
+
+@tool
+def web_search(query: str) -> str:
+    """Search the web; returns the top results as title + URL + snippet.
+
+    Use for news, analyst opinions, macro context, catalysts — anything the structured
+    market-data tools don't cover. Follow up with web_fetch to read a promising result.
+    """
+    results: list[dict] = []
+    try:
+        results = _ddg_html_search(query)
+    except Exception:
+        pass
+    if not results:
+        try:
+            results = _ddgs_search(query)
+        except Exception as e:
+            return f"ERROR: web_search failed (both backends): {e}"
+    if not results:
+        return f"No results for {query!r}."
+    return "\n".join(
+        f"{i}. {r['title']}\n   {r['url']}\n   {r['snippet']}" for i, r in enumerate(results, 1)
+    )
+
+
+@retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=1, max=6), reraise=True)
+def _http_get(url: str) -> str:
+    resp = httpx.get(url, headers={"User-Agent": _UA}, timeout=20, follow_redirects=True)
+    resp.raise_for_status()
+    return resp.text
+
+
+@tool
+def web_fetch(url: str) -> str:
+    """Fetch a URL and return its main readable text (truncated).
+
+    Use after web_search to read an article or page in depth. Strips nav/ads/boilerplate.
+    """
+    try:
+        html_text = _http_get(url)
+    except Exception as e:
+        return f"ERROR: fetch failed: {e}"
+    text = trafilatura.extract(html_text, include_comments=False, include_tables=False)
+    if not text:
+        return "ERROR: no readable content extracted from that URL."
+    return text[:_FETCH_CHAR_LIMIT]