""" S.T.E.W Browser Agent — Real web browsing, form filling, automation. Uses httpx + BeautifulSoup as primary. Falls back to requests. No Playwright/Selenium needed — works on any hosting platform. """ import asyncio import re import logging from typing import Optional from urllib.parse import urljoin, urlparse import httpx from bs4 import BeautifulSoup logger = logging.getLogger(__name__) HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.5", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", } class StewBrowser: """Autonomous web browser for S.T.E.W.""" def __init__(self): self.session_cookies: dict = {} self.history: list[str] = [] self.current_url: Optional[str] = None self.current_html: Optional[str] = None async def fetch(self, url: str, timeout: int = 20) -> dict: """Fetch any URL and return structured content.""" try: async with httpx.AsyncClient( headers=HEADERS, follow_redirects=True, timeout=timeout, cookies=self.session_cookies, ) as client: resp = await client.get(url) self.current_url = str(resp.url) self.current_html = resp.text self.history.append(self.current_url) # Persist cookies self.session_cookies.update(dict(resp.cookies)) return self._parse_page(resp.text, str(resp.url), resp.status_code) except httpx.TimeoutException: return {"error": f"Timeout fetching {url}", "url": url} except Exception as e: logger.error(f"Browser fetch error: {e}") return {"error": str(e), "url": url} def _parse_page(self, html: str, url: str, status: int) -> dict: """Extract structured content from HTML.""" soup = BeautifulSoup(html, "html.parser") # Remove noise for tag in soup(["script", "style", "nav", "footer", "aside", "iframe"]): tag.decompose() title = soup.title.string.strip() if soup.title else "No title" # Main content main = soup.find("main") or soup.find("article") or soup.find("body") text = main.get_text(separator="\n", strip=True) if main else soup.get_text(separator="\n", strip=True) # Clean up excessive whitespace lines = [l.strip() for l in text.splitlines() if l.strip()] clean_text = "\n".join(lines) # Truncate to 8000 chars to stay within LLM context if len(clean_text) > 8000: clean_text = clean_text[:8000] + "\n\n[...content truncated for brevity...]" # Extract links links = [] for a in soup.find_all("a", href=True)[:20]: href = urljoin(url, a["href"]) text_label = a.get_text(strip=True) if href.startswith("http") and text_label: links.append({"text": text_label[:80], "url": href}) # Extract forms forms = self._extract_forms(soup, url) # Extract meta description meta_desc = "" meta = soup.find("meta", attrs={"name": "description"}) if meta: meta_desc = meta.get("content", "") return { "url": url, "status": status, "title": title, "description": meta_desc, "content": clean_text, "links": links, "forms": forms, "word_count": len(clean_text.split()), } def _extract_forms(self, soup: BeautifulSoup, base_url: str) -> list[dict]: """Extract all forms and their fields.""" forms = [] for form in soup.find_all("form"): fields = [] for inp in form.find_all(["input", "textarea", "select"]): field_type = inp.get("type", "text") if field_type in ("hidden", "submit", "button", "reset"): continue fields.append({ "name": inp.get("name", inp.get("id", "unknown")), "type": field_type, "placeholder": inp.get("placeholder", ""), "required": inp.has_attr("required"), }) action = urljoin(base_url, form.get("action", base_url)) method = form.get("method", "GET").upper() forms.append({ "action": action, "method": method, "fields": fields, }) return forms async def submit_form(self, url: str, form_data: dict, method: str = "POST") -> dict: """Submit a form with given data.""" try: async with httpx.AsyncClient( headers=HEADERS, follow_redirects=True, timeout=20, cookies=self.session_cookies, ) as client: if method.upper() == "POST": resp = await client.post(url, data=form_data) else: resp = await client.get(url, params=form_data) self.session_cookies.update(dict(resp.cookies)) return self._parse_page(resp.text, str(resp.url), resp.status_code) except Exception as e: return {"error": str(e), "url": url} async def search_web_fallback(self, query: str) -> dict: """ Fallback browser search when Serper API is unavailable. Uses DuckDuckGo HTML search — no API key required. """ ddg_url = f"https://html.duckduckgo.com/html/?q={query.replace(' ', '+')}" try: async with httpx.AsyncClient(headers=HEADERS, follow_redirects=True, timeout=15) as client: resp = await client.post(ddg_url, data={"q": query}) soup = BeautifulSoup(resp.text, "html.parser") results = [] for r in soup.select(".result")[:8]: title_el = r.select_one(".result__title") snippet_el = r.select_one(".result__snippet") link_el = r.select_one(".result__url") if title_el: results.append({ "title": title_el.get_text(strip=True), "snippet": snippet_el.get_text(strip=True) if snippet_el else "", "url": "https://" + link_el.get_text(strip=True) if link_el else "", }) return { "source": "duckduckgo_fallback", "query": query, "results": results, "count": len(results), } except Exception as e: # Ultimate fallback — Bing HTML return await self._bing_fallback(query) async def _bing_fallback(self, query: str) -> dict: """Third-layer fallback using Bing HTML.""" bing_url = f"https://www.bing.com/search?q={query.replace(' ', '+')}" try: async with httpx.AsyncClient(headers=HEADERS, follow_redirects=True, timeout=15) as client: resp = await client.get(bing_url) soup = BeautifulSoup(resp.text, "html.parser") results = [] for li in soup.select("li.b_algo")[:8]: h2 = li.find("h2") p = li.find("p") a = h2.find("a") if h2 else None if a: results.append({ "title": a.get_text(strip=True), "snippet": p.get_text(strip=True) if p else "", "url": a.get("href", ""), }) return { "source": "bing_fallback", "query": query, "results": results, "count": len(results), } except Exception as e: return {"source": "none", "query": query, "results": [], "error": str(e)} async def navigate_and_answer(self, url: str, question: str) -> str: """Browse a URL and answer a specific question about it.""" page = await self.fetch(url) if "error" in page: return f"Could not access {url}: {page['error']}" content = page.get("content", "") title = page.get("title", "") return f"Page: {title}\nURL: {url}\n\nContent:\n{content[:4000]}" # Singleton browser instance per request lifecycle _browser_instance = None def get_browser() -> StewBrowser: global _browser_instance if _browser_instance is None: _browser_instance = StewBrowser() return _browser_instance