Web Scraping for AI Agents: Feeding Real-Time Data to LLMs Using Residential Proxies

[Autonomous AI Agent]
       |
       v
[Attempt 1: Direct Cloud IP Fetch] ---> WAF Block (HTTP 403 / Cloudflare Turnstile)
       |
       v
[Attempt 2: Route via Residential Proxy] ---> [HTML DOM Received (1.8 MB Payload)]
       |
       v
[Attempt 3: Dump Raw HTML into LLM] ---> Context Window Exhausted (45,000+ Tokens Burned!)

[User Query / Agent Task]
       |
       v
[Step 1: Geo-Targeted Residential Proxy Fetch (asyncio + httpx)]
       |
       v
[Step 2: DOM Sanitization & Boilerplate Removal (readability / BeautifulSoup)]
       |
       v
[Step 3: Markdown / JSON Schema Conversion (html2text)]
       |
       v
[Step 4: Inject Clean Token-Optimized Payload into LLM Context]

import asyncio
import httpx
import html2text
from bs4 import BeautifulSoup
from readability import Document
from dataclasses import dataclass
from typing import Optional

# BytesFlows Residential Gateway
PROXY_HOST = "p1.bytesflows.com:8001"
BASE_USER = "your-sub-user"
PASSWORD = "your-password"

@dataclass
class AgentScrapeResult:
    url: str
    title: str
    clean_markdown: str
    word_count: int
    estimated_tokens: int
    status: str
    duration_ms: int

class AIWebReaderTool:
    def __init__(self, default_country: str = "us"):
        self.default_country = default_country
        self.html_converter = html2text.HTML2Text()
        self.html_converter.ignore_links = False
        self.html_converter.ignore_images = True
        self.html_converter.ignore_tables = False
        self.html_converter.body_width = 0

    def _get_proxy_url(self, country: str, session_id: Optional[str] = None) -> str:
        """Generates localized residential proxy credentials."""
        auth = f"{BASE_USER}-loc-{country}"
        if session_id:
            auth += f"-session-{session_id}-time-10"
        return f"http://{auth}:{PASSWORD}@{PROXY_HOST}"

    def _estimate_tokens(self, text: str) -> int:
        """Rough token estimation (approx. 4 characters per token for English text)."""
        return max(1, len(text) // 4)

    async def fetch_and_distill(self, url: str, country: Optional[str] = None) -> AgentScrapeResult:
        target_geo = country or self.default_country
        proxy_url = self._get_proxy_url(target_geo)
        
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.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.9" if target_geo == "us" else "en;q=0.8",
            "Accept-Encoding": "gzip, deflate, br"
        }

        start_ts = asyncio.get_event_loop().t

ime()
        try:
            async with httpx.AsyncClient(proxies={"http://": proxy_url, "https://": proxy_url}, verify=False) as client:
                resp = await client.get(url, headers=headers, timeout=15.0, follow_redirects=True)
                duration_ms = round((asyncio.get_event_loop().time() - start_ts) * 1000)
                
                if resp.status_code != 200:
                    return AgentScrapeResult(url, "", f"HTTP Error {resp.status_code}", 0, 0, "error", duration_ms)

                # Step 1: Extract primary content readability tree
                doc = Document(resp.text)
                article_html = doc.summary()
                title = doc.title()

                # Step 2: Clean residual DOM noise with BeautifulSoup
                soup = BeautifulSoup(article_html, "html.parser")
                for tag in soup(["script", "style", "nav", "footer", "iframe", "noscript"]):
                    tag.decompose()

                # Step 3: Convert clean DOM to Markdown
                clean_md = self.html_converter.handle(str(soup)).strip()
                words = len(clean_md.split())
                tokens = self._estimate_tokens(clean_md)

                return AgentScrapeResult(
                    url=url,
                    title=title,
                    clean_markdown=clean_md,
                    word_count=words,
                    estimated_tokens=tokens,
                    status="success",
                    duration_ms=duration_ms
                )
        except Exception as e:
            duration_ms = round((asyncio.get_event_loop().time() - start_ts) * 1000)
            return AgentScrapeResult(url, "", f"Execution Failed: {str(e)}", 0, 0, "failed", duration_ms)

# Example Usage inside an AI Agent Tool execution
async def main():
    reader = AIWebReaderTool(default_country="us")
    target_url = "https://httpbin.org/html"
    
    print(f"[Agent] Fetching live knowledge from {target_url}...")
    result = a

wait reader.fetch_and_distill(target_url)
    
    print(f"\n--- Scraping Result ({result.status.upper()}) ---")
    print(f"Title: {result.title}")
    print(f"Latency: {result.duration_ms} ms | Word Count: {result.word_count} | Estimated LLM Tokens: {result.estimated_tokens}")
    print(f"\nMarkdown Payload Preview:\n{result.clean_markdown[:400]}...")

if __name__ == "__main__":
    asyncio.run(main())

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class WebScraperInput(BaseModel):
    url: str = Field(description="The live webpage URL to fetch and read.")
    country: str = Field(default="us", description="2-letter ISO country code for geographic targeting.")

class ResidentialWebScraperTool(BaseTool):
    name = "live_web_reader"
    description = "Fetches live webpages using residential proxies and returns clean Markdown for reading."
    args_schema = WebScraperInput

    def _run(self, url: str, country: str = "us") -> str:
        # Synchronous wrapper around AIWebReaderTool
        reader = AIWebReaderTool(default_country=country)
        result = asyncio.run(reader.fetch_and_distill(url, country))
        if result.status == "success":
            return f"# {result.title}\n\n{result.clean_markdown}"
        return f"Error reading page: {result.clean_markdown}"