DEV Community: Zenrows

Build a Live Web Data Agent With Agno and Zenrows

Tiioluwani — Tue, 03 Nov 2026 20:03:23 +0000

This article was originally published on the Zenrows blog.

In this tutorial, you'll build a two-agent Agno workflow that reports only the product data a live page supports. You'll need Python 3.11 to 3.13, a Zenrows API key, and an OpenAI API key. The free Zenrows plan includes 5,000 credits a month.
A normal HTTP request may return a blocked response or miss product values added by JavaScript, which leaves an agent with incomplete information. Three parts of the workflow keep the agent from filling those gaps.

The web data agent fetches the page through a typed Zenrows tool that only accepts a URL, so it gets the full page and the API key never appears in the tool schema.
The market intelligence agent has no tools, extracts only values the page supports, quotes evidence, and uses null for anything missing.
Pydantic validates each handoff, so malformed data stops before the next agent sees it. On a Cloudflare-protected page, a direct request got HTTP 403 while the same URL through the Agno tool returned HTTP 200 with the page content. The code is on GitHub.

How It Works

Agent	Tool	Output	Job
Web data agent	`fetch_with_zenrows`	`FetchedPage`	Fetches the page through ZenRows
Market intelligence agent	None	`MarketSnapshot`	Extracts fields with quoted evidence

1. Create the Project

mkdir agno-zenrows-live-web-agent
cd agno-zenrows-live-web-agent
python -m venv .venv
source .venv/bin/activate

On Windows PowerShell, activate with this command.

.venv\Scripts\Activate.ps1

Create the source package.

mkdir src
mkdir src/live_web_agent
python -c "from pathlib import Path; Path('src/live_web_agent/__init__.py').touch()"

Create pyproject.toml. The project pins Agno 3.0.10, the version the tutorial was tested with.

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "agno-zenrows-live-web-agent"
version = "0.1.0"
description = "A two-agent Agno workflow that retrieves protected web data through Zenrows."
readme = "README.md"
requires-python = ">=3.11,<3.14"
dependencies = [
  "agno==3.0.10",
  "beautifulsoup4==4.14.2",
  "openai==3.16.2",
  "pydantic==2.12.5",
  "python-dotenv==1.2.1",
  "requests==2.32.5",
]
[project.optional-dependencies]
dev = [
  "pytest==9.0.2",
  "pytest-cov==7.0.0",
  "ruff==0.15.6",
]
[project.scripts]
live-web-agent = "live_web_agent.cli:main"
[tool.hatch.build.targets.wheel]
packages = ["src/live_web_agent"]
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.ruff]
line-length = 88
target-version = "py311"

python -m pip install --upgrade pip
python -m pip install -e ".[dev]"

The finished project uses this structure. The tests/ folder and .env.example come with the GitHub repository.

agno-zenrows-live-web-agent/
├── src/
│   └── live_web_agent/
│       ├── __init__.py
│       ├── agents.py
│       ├── cli.py
│       ├── config.py
│       ├── schemas.py
│       ├── web.py
│       └── workflow.py
├── tests/
├── .env
├── .env.example
├── .gitignore
└── pyproject.toml

Add your keys to .env and list .env in .gitignore.

ZENROWS_API_KEY=replace_with_your_zenrows_api_key
OPENAI_API_KEY=replace_with_your_openai_api_key
OPENAI_MODEL=gpt-5-mini
TARGET_URL=https://www.scrapingcourse.com/cloudflare-challenge

2. Define the Zenrows Request

Create src/live_web_agent/web.py.

from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
import requests
from bs4 import BeautifulSoup
ZENROWS_ENDPOINT = "https://api.zenrows.com/v1/"
def page_to_text(html: str, limit: int = 12_000) -> tuple[str | None, str]:
    soup = BeautifulSoup(html, "html.parser")
    for element in soup(["script", "style", "noscript", "svg"]):
        element.decompose()
    title = soup.title.get_text(" ", strip=True) if soup.title else None
    text = " ".join(soup.get_text(" ", strip=True).split())
    metadata: list[str] = []
    for element in soup.select("[itemprop][content]"):
        itemprop = element.get("itemprop")
        content = element.get("content")
        if itemprop in {"priceCurrency", "availability"} and content:
            value = f"{itemprop}: {content}"
            if value not in metadata:
                metadata.append(value)
    if metadata:
        text = f"{text} Structured metadata: {'; '.join(metadata)}"
    return title, text[:limit]
def fetch_with_zenrows(
    url: str,
    api_key: str,
    *,
    session: Any = requests,
) -> dict[str, object]:
    """Fetch a public page through Zenrows and return compact page text."""
    response = session.get(
        ZENROWS_ENDPOINT,
        params={
            "url": url,
            "apikey": api_key,
            "mode": "auto",
        },
        timeout=60,
    )
    response.raise_for_status()
    title, text = page_to_text(response.text)
    return {
        "source_url": url,
        "status_code": response.status_code,
        "title": title,
        "text": text,
        "fetched_at": datetime.now(UTC).isoformat(),
    }
def fetch_direct(url: str, *, session: Any = requests) -> dict[str, object]:
    """Fetch the same page without Zenrows for a transparent comparison."""
    response = session.get(
        url,
        headers={"User-Agent": "Mozilla/5.0 (compatible; tutorial-validator/1.0)"},
        timeout=30,
    )
    title, text = page_to_text(response.text, limit=500)
    return {
        "source_url": url,
        "status_code": response.status_code,
        "title": title,
        "text_preview": text,
    }

fetch_with_zenrows sends the URL with mode=auto, so Fetch selects the access configuration for each target and applies JavaScript rendering or Premium Proxies when needed. page_to_text strips non-content elements, caps the text at 12,000 characters, and appends the Schema.org currency and availability values the processing agent needs. The tool reads HTML because those values sit in itemprop attributes, which a Markdown response doesn't carry. fetch_direct sends the same URL without Zenrows so the CLI can print both responses side by side.

3. Load Settings

Create src/live_web_agent/config.py.

from __future__ import annotations
import os
from dataclasses import dataclass
@dataclass(frozen=True)
class Settings:
    zenrows_api_key: str
    openai_api_key: str
    openai_model: str = "gpt-5-mini"
    @classmethod
    def from_env(cls) -> "Settings":
        # stop before creating either agent when a credential is missing
        missing = [
            name
            for name in ("ZENROWS_API_KEY", "OPENAI_API_KEY")
            if not os.getenv(name)
        ]
        if missing:
            names = ", ".join(missing)
            raise RuntimeError(f"Missing required environment variables: {names}")
        # use gpt-5-mini unless OPENAI_MODEL overrides it
        return cls(
            zenrows_api_key=os.environ["ZENROWS_API_KEY"],
            openai_api_key=os.environ["OPENAI_API_KEY"],
            openai_model=os.getenv("OPENAI_MODEL", "gpt-5-mini"),
        )

from_env stops the application early if either key is missing, and the frozen dataclass keeps the loaded values unchanged.

4. Define the Output Schemas

Create src/live_web_agent/schemas.py.

from __future__ import annotations
from datetime import datetime
from urllib.parse import urlparse
from pydantic import BaseModel, ConfigDict, Field, field_validator
def _validate_http_url(value: str) -> str:
    parsed = urlparse(value)
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise ValueError("source_url must be a valid HTTP or HTTPS URL")
    return value
def _keep_source_url_schema_plain(schema: dict[str, object]) -> None:
    properties = schema.get("properties")
    if isinstance(properties, dict):
        source_url = properties.get("source_url")
        if isinstance(source_url, dict):
            # remove metadata that the structured-output schema does not need
            source_url.pop("title", None)
class FetchedPage(BaseModel):
    # keep source_url as a plain string in the OpenAI output schema
    model_config = ConfigDict(json_schema_extra=_keep_source_url_schema_plain)
    source_url: str
    status_code: int
    title: str | None = None
    text: str
    fetched_at: datetime
    # validate the URL in Python without adding format: "uri" to the schema
    _source_url_is_http = field_validator("source_url")(_validate_http_url)
class MarketSnapshot(BaseModel):
    model_config = ConfigDict(json_schema_extra=_keep_source_url_schema_plain)
    product_name: str | None = None
    current_price: float | None = None
    currency: str | None = None
    availability: str | None = None
    source_url: str
    collected_at: datetime
    evidence: list[str] = Field(default_factory=list, max_length=5)
    _source_url_is_http = field_validator("source_url")(_validate_http_url)

source_url is a plain string. In the tested Agno 3.0.10 setup with gpt-5-mini, Pydantic's HttpUrl added format: "uri" to the generated JSON Schema, and OpenAI rejected the structured-output request. _validate_http_url() keeps HTTP and HTTPS validation without that keyword.

5. Create Both Agents

Create src/live_web_agent/agents.py.

from __future__ import annotations
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from .config import Settings
from .schemas import FetchedPage, MarketSnapshot
from .web import fetch_with_zenrows
def build_agents(settings: Settings) -> tuple[Agent, Agent]:
    # both agents use the model selected in the application settings
    model = OpenAIChat(
        id=settings.openai_model,
        api_key=settings.openai_api_key,
    )
    # capture the API key here so the model supplies only the URL
    def fetch_with_zenrows_tool(url: str) -> dict[str, object]:
        """Fetch a public page through Zenrows and return compact page text."""
        try:
            return fetch_with_zenrows(
                url,
                api_key=settings.zenrows_api_key,
            )
        except Exception:
            # request exceptions can include the complete query string
            # keep Zenrows credentials out of logs and model tool results
            raise RuntimeError("Zenrows request failed") from None
    # expose a stable public tool name to Agno
    fetch_with_zenrows_tool.__name__ = "fetch_with_zenrows"
    # require the first agent to fetch the supplied page through Zenrows
    fetch_agent = Agent(
        name="Web data agent",
        model=model,
        tools=[fetch_with_zenrows_tool],
        output_schema=FetchedPage,
        instructions=[
            "Always call fetch_with_zenrows for the supplied URL.",
            "Return only the fetched page data. Do not invent missing content.",
        ],
    )
    # give the second agent no tools because it only processes fetched data
    processing_agent = Agent(
        name="Market intelligence agent",
        model=model,
        output_schema=MarketSnapshot,
        instructions=[
            "Extract only values supported by the supplied page data.",
            "Use null for missing fields and quote short evidence snippets.",
        ],
    )
    return fetch_agent, processing_agent

The nested fetch_with_zenrows_tool reads the API key from settings when called, so the schema Agno 3.0.10 generates for the tool has a single required url field.

{
  "type": "object",
  "properties": {
    "url": {
      "type": "string"
    }
  },
  "required": [
    "url"
  ]
}

Errors are replaced with a generic message, because request exceptions can include the full query string.

6. Connect the Agents

Create src/live_web_agent/workflow.py.

from __future__ import annotations
import json
from typing import Any
from pydantic import ValidationError
from .schemas import FetchedPage, MarketSnapshot
class WorkflowAgentError(RuntimeError):
    """Raised when an Agno agent or its model provider fails."""
def _run_agent(agent: Any, prompt: str, role: str) -> Any:
    try:
        response = agent.run(prompt)
    except Exception as exc:
        # include the failed agent's role in the application error
        raise WorkflowAgentError(f"{role} failed: {exc}") from exc
    status = getattr(response, "status", None)
    status_value = getattr(status, "value", status)
    # catch provider errors returned as Agno responses
    if isinstance(status_value, str) and status_value.upper() == "ERROR":
        detail = getattr(response, "content", None) or "unknown provider error"
        raise WorkflowAgentError(f"{role} failed: {detail}")
    return response
def _content(
    response: Any,
    schema: type[FetchedPage] | type[MarketSnapshot],
) -> Any:
    value = response.content
    if isinstance(value, schema):
        return value
    try:
        # parse JSON strings before handling dictionary-like values
        if isinstance(value, str):
            return schema.model_validate_json(value)
        return schema.model_validate(value)
    except ValidationError as exc:
        raise WorkflowAgentError(
            f"Agent returned invalid {schema.__name__} data: {value!r}"
        ) from exc
def run_workflow(
    url: str,
    fetch_agent: Any,
    processing_agent: Any,
) -> MarketSnapshot:
    fetch_response = _run_agent(
        fetch_agent,
        f"Fetch this public product page through Zenrows: {url}",
        "Web data agent",
    )
    # validate the fetch result before sending it to the processing agent
    page = _content(fetch_response, FetchedPage)
    # pass the validated page as JSON instead of raw HTML
    processing_response = _run_agent(
        processing_agent,
        "Extract a market snapshot from this fetched page data:\n"
        + json.dumps(page.model_dump(mode="json"), ensure_ascii=False),
        "Market intelligence agent",
    )
    return _content(processing_response, MarketSnapshot)

run_workflow validates the fetch agent's output as FetchedPage before the processing agent sees it, then validates the final MarketSnapshot. Any failure raises WorkflowAgentError with the agent's role.

7. Add the CLI

Create src/live_web_agent/cli.py.

from __future__ import annotations
import argparse
import json
import os
from dotenv import load_dotenv
from .agents import build_agents
from .config import Settings
from .web import fetch_direct, fetch_with_zenrows
from .workflow import run_workflow
def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run the Agno and Zenrows workflow"
    )
    # use TARGET_URL unless the reader passes --url
    parser.add_argument("--url", default=os.getenv("TARGET_URL"))
    parser.add_argument("--skip-direct", action="store_true")
    parser.add_argument("--fetch-only", action="store_true")
    return parser.parse_args()
def main() -> None:
    # load the local environment before reading TARGET_URL or API keys
    load_dotenv()
    args = parse_args()
    if not args.url:
        raise SystemExit("Pass --url or set TARGET_URL")
    # print the direct response unless --skip-direct is present
    if not args.skip_direct:
        print("Direct request comparison:")
        print(
            json.dumps(
                fetch_direct(args.url),
                indent=2,
                ensure_ascii=False,
            )
        )
    # print the Zenrows tool result without running the agents
    if args.fetch_only:
        api_key = os.getenv("ZENROWS_API_KEY")
        if not api_key:
            raise SystemExit("Set ZENROWS_API_KEY to run --fetch-only")
        print("\nZenrows tool result:")
        page = fetch_with_zenrows(args.url, api_key=api_key)
        print(json.dumps(page, indent=2, ensure_ascii=False))
        return
    settings = Settings.from_env()
    fetch_agent, processing_agent = build_agents(settings)
    result = run_workflow(args.url, fetch_agent, processing_agent)
    print("\nTwo-agent result:")
    print(result.model_dump_json(indent=2))
if __name__ == "__main__":
    main()

The CLI runs the direct comparison unless you pass --skip-direct. With --fetch-only, it prints the Zenrows tool result and stops before either agent runs, so you can check the web access layer before spending model tokens.

8. Run It

The TARGET_URL in .env points to ScrapingCourse's Cloudflare challenge page, which answers automated requests with a challenge. Check both access paths without running the agents.

live-web-agent --fetch-only

Direct request comparison:
{
  "source_url": "https://www.scrapingcourse.com/cloudflare-challenge",
  "status_code": 403,
  "title": "Just a moment...",
  "text_preview": "Just a moment..."
}
Zenrows tool result:
{
  "source_url": "https://www.scrapingcourse.com/cloudflare-challenge",
  "status_code": 200,
  "title": "Cloudflare Challenge - ScrapingCourse.com",
  "text": "Cloudflare Challenge - ScrapingCourse.com Scraping Course Cloudflare Challenge You bypassed the Cloudflare challenge! :D",
  "fetched_at": "2026-10-21T08:11:16.464568+00:00"
}

The direct request got 403 and a "Just a moment..." challenge. Through Zenrows, the tool got 200, and the content included You bypassed the Cloudflare challenge! :D.

Now run both agents on a product page.

live-web-agent --url "https://www.scrapingcourse.com/ecommerce/product/chaz-kangeroo-hoodie/"

The web data agent calls the same tool. Printed with --fetch-only --skip-direct, this is the tool result it receives for that page.

Zenrows tool result:
{
  "source_url": "https://www.scrapingcourse.com/ecommerce/product/chaz-kangeroo-hoodie/",
  "status_code": 200,
  "title": "Chaz Kangeroo Hoodie – Ecommerce Test Site to Learn Web Scraping",
  "text": "Chaz Kangeroo Hoodie – Ecommerce Test Site to Learn Web Scraping Skip to navigation Skip to content Ecommerce Test Site to Learn Web Scraping ScrapingCourse.com Search for: Search Menu Shop Home Cart Checkout My account $ 0.00 0 items Home / Clothing / Men / Tops / Hoodies & Sweatshirts|Clothing / Collections / Eco Friendly|Clothing / Chaz Kangeroo Hoodie Chaz Kangeroo Hoodie $ 52.00 This is a variable product called a Chaz Kangeroo Hoodie Size Choose an option XS S M L XL Color Choose an option Black Gray Orange Clear Chaz Kangeroo Hoodie quantity Add to cart SKU: MH01 Category: Eco Friendly|Clothing Description Additional information Description Ideal for cold-weather training or work outdoors, the Chaz Hoodie promises superior warmth with every wear. Thick material blocks out the wind as ribbed cuffs and bottom band seal in body heat. • Two-tone gray heather hoodie. • Drawstring-adjustable hood. • Machine wash/dry. Additional information Size XS, S, M, L, XL Color Black, Gray, Orange Related products Frankie Sweatshirt $ 60.00 Select options Bruno Compete Hoodie $ 63.00 Select options Frankie Sweatshirt Bruno Compete Hoodie Search Search © Ecommerce Test Site to Learn Web Scraping 2026 Built with WooCommerce . My Account Search Search for: Search Cart 0",
  "fetched_at": "2026-10-21T08:11:24.399910+00:00"
}

The web data agent returns that data as a FetchedPage. The processing agent then returns a MarketSnapshot with product_name, current_price, currency, availability, source_url, collected_at, and up to five evidence snippets quoted from the page text. Fields the page doesn't support stay null.
The repository's test suite covers the request parameters, the tool schema, the schemas, and the agent handoff without spending API credits.

pytest -q
ruff check .

21 passed in 1.75s
All checks passed!

Wrapping Up

You now have an Agno workflow that fetches protected pages through a key-safe tool and validates every handoff. To adapt it to another public market-data source, change the target URL and update the MarketSnapshot fields. The current model represents one product, so a page with many products needs a list-based response model.

What's Next

Agno agents can also use the Zenrows MCP server through MCPTools at https://mcp.zenrows.com/mcp. This tutorial uses a function tool because the workflow needs the typed FetchedPage handoff.
An agent without a Zenrows account can pay for Fetch per request over x402, MPP, or a card.
Track prices across many pages by submitting them as one Batch job.
See the same idea with Mastra, Google ADK, or AG2.

Build a Web Research Agent With Google ADK and Zenrows

Tiioluwani — Tue, 03 Nov 2026 19:36:17 +0000

This article was originally published on the Zenrows blog.

Your research agent might be reading fake pages. When a target sits behind Cloudflare, a plain requests call returns a 403 and a "Just a moment..." challenge page in place of the content. Unless the tool checks for that, the agent can treat the challenge markup as the answer and report on a page it never saw.

In this tutorial, you'll see how that happens and how to fix it in a Google Agent Development Kit (ADK) agent. You'll give the agent a web tool that routes each fetch through Zenrows Fetch and returns the page as Markdown, then connect a second agent to the hosted Zenrows MCP server with no function code. You'll need Python 3.10 or later, a Google AI Studio API key, and a Zenrows API key. The free plan includes 5,000 credits a month. The complete code is on GitHub.

What You'll Build

You'll register a function tool, fetch_protected_page, on a Gemini-powered ADK agent.
You'll write a comparison script that fetches one URL with plain HTTP and with Zenrows.
You'll add a second agent that loads the Zenrows MCP tools through ADK's MCPToolset.

1. Set Up the Project

python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt

The pinned requirements.txt installs google-adk 2.9.2 with its MCP extra.

google-adk[mcp]==2.9.2
mcp==1.30.0
python-dotenv==1.2.3
requests==2.34.2
pytest==9.1.1
ruff==0.16.8
-e .

Copy the environment template and add your keys. The .gitignore excludes .env.

cp .env.example .env

GOOGLE_API_KEY=your_google_api_key
ZENROWS_API_KEY=your_zenrows_api_key

Here's the project layout.

google-adk-zenrows-research-agent/
├── scripts/
│   ├── compare_fetch.py
│   └── list_mcp_tools.py
├── tests/
│   ├── test_agent.py
│   ├── test_mcp_agent.py
│   └── test_web.py
├── zenrows_mcp_agent/
│   ├── __init__.py
│   └── agent.py
├── zenrows_research_agent/
│   ├── __init__.py
│   ├── agent.py
│   └── web.py
├── .env.example
├── .gitignore
├── LICENSE
├── pyproject.toml
├── README.md
└── requirements.txt

2. Write the Fetch Function

Create zenrows_research_agent/web.py.

"""Web retrieval tools used by the Google ADK agent."""
import os
import requests
ZENROWS_ENDPOINT = "https://api.zenrows.com/v1/"
DEFAULT_TIMEOUT = 180
MAX_TOOL_CONTENT = 12_000
def fetch_with_http(url: str) -> dict:
    """Fetch a public URL with a standard HTTP request for comparison."""
    try:
        # send the request without a protected-site access service.
        response = requests.get(
            url,
            headers={"User-Agent": "google-adk-research-agent/0.1"},
            timeout=20,
        )
    except requests.RequestException as exc:
        return {"status": "error", "url": url, "error": str(exc)}
    return {
        "status": "success" if response.ok else "blocked",
        "url": url,
        "status_code": response.status_code,
        "content": response.text[:2_000],
    }
def fetch_protected_page(url: str) -> dict:
    """Fetch a protected or JavaScript-rendered page through Zenrows.
    Args:
        url: The complete public URL the research agent needs to read.
    Returns:
        A dictionary containing the request status, source URL, and page content.
    """
    api_key = os.getenv("ZENROWS_API_KEY")
    if not api_key:
        return {
            "status": "error",
            "url": url,
            "error": "ZENROWS_API_KEY is not configured.",
        }
    # let Adaptive Stealth Mode select the request configuration.
    params = {
        "apikey": api_key,
        "url": url,
        "mode": "auto",
        "response_type": "markdown",
        "original_status": "true",
    }
    try:
        # give Adaptive Stealth Mode enough time to retrieve the page.
        response = requests.get(
            ZENROWS_ENDPOINT,
            params=params,
            timeout=DEFAULT_TIMEOUT,
        )
        response.raise_for_status()
    except requests.RequestException as exc:
        # remove the API key from the error text before it reaches the model.
        error = str(exc).replace(api_key, "[REDACTED]")
        return {"status": "error", "url": url, "error": error}
    # do not return a successful result without page content.
    if not response.text.strip():
        return {
            "status": "error",
            "url": url,
            "error": "Zenrows returned an empty response.",
        }
    return {
        "status": "success",
        "url": url,
        "content": response.text[:MAX_TOOL_CONTENT],
    }

fetch_protected_page sends the URL to Zenrows with mode=auto, which turns on Adaptive Stealth Mode. It starts with the lightest configuration likely to work and escalates only when the target needs it. A few safeguards keep the agent honest.

response_type=markdown returns readable text in place of raw HTML.
original_status=true passes through the target's status code, so an error page from the target counts as an error.
The 180-second timeout gives mode=auto room to finish on protected targets.
Content is capped at 12,000 characters, and an empty response is reported as an error.
The API key is redacted from any error text before it reaches the model. fetch_with_http is a plain request used only for the comparison. ## 3. Register the Function as an ADK Tool Create zenrows_research_agent/agent.py.

"""Google ADK agent configuration."""
from pathlib import Path
from dotenv import load_dotenv
from google.adk.agents import Agent
from .web import fetch_protected_page
# load the API keys stored in the project environment file.
load_dotenv(Path(__file__).resolve().parents[1] / ".env")
# register the Zenrows retrieval function as an ADK tool.
root_agent = Agent(
    model="gemini-3.6-flash",
    name="web_research_agent",
    description="Researches public web pages that ordinary HTTP tools cannot read.",
    instruction=(
        "You are a web research agent. Use fetch_protected_page whenever a task "
        "requires content from a supplied URL. Base the answer only on returned "
        "page content, mention the source URL, and state clearly when retrieval "
        "fails or the page does not contain the requested information."
    ),
    tools=[fetch_protected_page],
)

ADK builds the tool schema from the function's name, type hints, and docstring, so the model only has to supply a URL.

4. Compare Plain HTTP With Zenrows

Create scripts/compare_fetch.py.

"""Compare standard HTTP access with Zenrows access."""
import json
from dotenv import load_dotenv
from zenrows_research_agent.web import fetch_protected_page, fetch_with_http
TARGET_URL = "https://www.scrapingcourse.com/cloudflare-challenge"
def summarize(result: dict) -> dict:
    """Reduce a fetch result to fields that are safe to print."""
    return {
        "status": result["status"],
        "url": result["url"],
        "status_code": result.get("status_code"),
        "content_preview": result.get("content", "")[:300],
        "error": result.get("error"),
    }
if __name__ == "__main__":
    # load the Zenrows API key from the project environment.
    load_dotenv()
    # retrieve the same URL through both access methods.
    comparison = {
        "standard_http": summarize(fetch_with_http(TARGET_URL)),
        "zenrows": summarize(fetch_protected_page(TARGET_URL)),
    }
    # print the comparison without exposing the API key.
    print(json.dumps(comparison, indent=2))

python scripts/compare_fetch.py

{
  "standard_http": {
    "status": "blocked",
    "url": "https://www.scrapingcourse.com/cloudflare-challenge",
    "status_code": 403,
    "content_preview": "<!DOCTYPE html><html lang=\"en-US\"><head><title>Just a moment...</title><meta http-equiv=\"Content-Type\" content=\"text/html; charset=UTF-8\"><meta http-equiv=\"X-UA-Compatible\" content=\"IE=Edge\"><meta name=\"robots\" content=\"noindex,nofollow\"><meta name=\"viewport\" content=\"width=device-width,initial-scal",
    "error": null
  },
  "zenrows": {
    "status": "success",
    "url": "https://www.scrapingcourse.com/cloudflare-challenge",
    "status_code": null,
    "content_preview": "[![](https://www.scrapingcourse.com/assets/images/logo.svg) Scraping Course](http://www.scrapingcourse.com/)\n\n# Cloudflare Challenge\n\n![](https://www.scrapingcourse.com/assets/images/challenge.svg)\n\n## You bypassed the Cloudflare challenge! :D",
    "error": null
  }
}

The plain request got a 403 and the challenge page. Zenrows returned the page heading and the verification message. The Zenrows status_code reads null because the function doesn't return it.

5. Run the Agent

adk run zenrows_research_agent

Submit this task.

Fetch https://www.scrapingcourse.com/cloudflare-challenge and report the page title, whether the challenge was passed, and the visible verification text. Cite the source URL.

[web_research_agent]: Based on the content retrieved from [https://www.scrapingcourse.com/cloudflare-challenge](https://www.scrapingcourse.com/cloudflare-challenge):
* **Page Title / Main Heading:** Cloudflare Challenge
* **Challenge Status:** Passed / Bypassed
* **Visible Verification Text:** "You bypassed the Cloudflare challenge! :D"
* **Source URL:** `https://www.scrapingcourse.com/cloudflare-challenge`

Run adk web if you'd prefer to test in ADK's browser UI.

6. Connect the Zenrows MCP Server

Create zenrows_mcp_agent/agent.py.

"""Google ADK agent that uses the hosted Zenrows MCP server."""
import os
from pathlib import Path
from dotenv import load_dotenv
from google.adk.agents import Agent
from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams
# load the API keys stored in the project environment file.
load_dotenv(Path(__file__).resolve().parents[1] / ".env")
# connect to the hosted Zenrows MCP server with the API key as a bearer token.
zenrows_toolset = MCPToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://mcp.zenrows.com/mcp",
        headers={"Authorization": f"Bearer {os.getenv('ZENROWS_API_KEY', '')}"},
    )
)
# the agent chooses between the Zenrows scrape and browser tools on its own.
root_agent = Agent(
    model="gemini-3.6-flash",
    name="web_research_mcp_agent",
    description="Researches public web pages through the Zenrows MCP server.",
    instruction=(
        "You are a web research agent. Use the Zenrows tools whenever a task "
        "requires content from a supplied URL. Base the answer only on returned "
        "page content, mention the source URL, and state clearly when retrieval "
        "fails or the page does not contain the requested information."
    ),
    tools=[zenrows_toolset],
)

MCPToolset connects to https://mcp.zenrows.com/mcp with your API key as a bearer token, and ADK loads the server's tools. Check which ones arrived with scripts/list_mcp_tools.py.

"""Check which Zenrows MCP tools Google ADK loads for the MCP agent."""
import asyncio
from zenrows_mcp_agent.agent import zenrows_toolset
RESEARCH_TOOLS = ["scrape", "extract", "crawl_create", "batch_create", "browser_navigate"]
async def main() -> None:
    # ask the hosted Zenrows MCP server which tools it exposes.
    tools = await zenrows_toolset.get_tools()
    await zenrows_toolset.close()
    names = {tool.name for tool in tools}
    # confirm the tools a research agent relies on are loaded.
    print(f"Loaded {len(names)} Zenrows tools")
    for name in RESEARCH_TOOLS:
        print(f"{name}: {'available' if name in names else 'not loaded'}")
if __name__ == "__main__":
    asyncio.run(main())

python scripts/list_mcp_tools.py
adk run zenrows_mcp_agent

Loaded 62 Zenrows tools
scrape: available
extract: available
crawl_create: available
batch_create: available
browser_navigate: available

ADK prints experimental-feature warnings before that output.

Function Tool or MCP Server

	Function tool	Zenrows MCP
What the agent calls	`fetch_protected_page`, one typed function	`scrape`, `extract`, Crawl, Batch, and browser tools
Request settings	Fixed in code (`mode=auto`, Markdown)	Chosen by the MCP server per task
Code you maintain	The function and its tests	One `MCPToolset` block
Pick it when	You want one fixed, typed request	You want the agent to choose among scraping, crawling, and browser actions

Wrapping Up

Your ADK agent now reads protected pages as Markdown through a typed function, and a second agent reaches the full Zenrows toolset over MCP. Change the URL and the prompt to point it at your own research tasks.

What's Next

Use Extract with extract=auto when you need structured JSON fields.
Discover linked pages with Crawl or run large URL lists as one job with Batch.
Build the same idea with Agno and Zenrows or as a multi-agent system with AG2.

We Cut Our Agent’s MCP Stack. Here’s What Changed.

Goodness E. Eboh — Tue, 03 Nov 2026 15:56:01 +0000

This article was originally published on the Zenrows blog.

Cutting an agent's MCP stack down to the servers it uses changes how the agent behaves. Every connected tool takes up context, and a crowded menu leads the agent to pick the wrong one. Two servers for the same job, such as two browser servers, leave it choosing between near-identical tools. Keep one server per capability, and each tool the agent sees has a clear job.
This post shows what to keep. It maps the six capabilities a production agent can need to the MCP server that covers each one, gives three ready-made stacks, and shows how to connect remote and local servers to Claude Code, Cursor, and other clients. A typical agent needs only two or three servers.

The Stack at a Glance

Capability	MCP server	What your agent gets	Skip it when
Web data access	Zenrows MCP	Pages, structured JSON, crawls, and batch jobs on protected and JavaScript-heavy sites	Your agent never needs information from the web
Browser automation	Zenrows MCP	A cloud browser it clicks, types, and navigates in, with residential proxies and anti-bot bypass	Your agent only reads pages
Memory	Mem0 (Zep for facts that change over time)	Context that persists across sessions	Your agent is stateless
Code and dev tools	GitHub MCP, with Context7 for library docs	Repositories, issues, pull requests, and Actions runs	Your agent doesn't touch code
Databases	Supabase MCP	Tables, SQL queries, migrations, and logs for a Postgres project	Your agent has no database
Productivity and SaaS	Google Drive, Notion, Slack, Atlassian	Documents, wikis, conversations, and tickets	Your agent has no workplace data to act on

Each server on this list is first-party or officially maintained, actively developed, and covers a job no other server in the stack covers.

Connect the Servers

Remote servers run on the provider's infrastructure. You add the URL to your client and sign in with OAuth, so there's nothing to install or update. Local servers run on your machine as subprocesses, usually through npx or Docker, and read credentials from environment variables. Claude Code, Cursor, VS Code, and Claude Desktop accept both.
The remote Zenrows server is https://mcp.zenrows.com/mcp, and the local one runs with one command.

npx -y @zenrows/mcp

The local server starts without an API key and creates a free account on its first call, which you can claim later. The Zenrows CLI prints the exact config for your client, and the CLI docs list every supported client. Supabase's hosted server lives at https://mcp.supabase.com/mcp and signs you in through the browser.

What Each Server Does

Zenrows MCP handles web data. It exposes scrape, extract, crawl, batch, and browser tools. Adaptive Stealth Mode escalates the request setup only when a site needs it, and when the agent names the fields it needs, Zenrows returns just those as structured JSON at a fraction of the tokens a full page costs. In a benchmark of four scraping APIs, Zenrows succeeded on all 800 requests against DataDome, Cloudflare, and Akamai targets. Agents can also pay per request over x402 or MPP without an account, and the free plan includes 5,000 credits a month.
Zenrows MCP also handles browser automation. The cloud browser runs behind residential proxies with anti-bot bypass, and sessions keep cookies, local storage, and tabs between calls. It clicks, types, fills forms, picks dropdowns, scrolls, hovers, drags, and presses keys. It reads pages through the accessibility tree as text, so no vision model is needed, and it can take screenshots, save PDFs, and run JavaScript, with several actions chained in a single call. Playwright MCP, Microsoft's server, runs a local browser and suits testing your own apps.
Mem0 handles memory. The current official Mem0 MCP server is hosted. It stores memories under user, session, and agent scopes and updates existing ones. The original self-hosted mem0-mcp repo is archived, and Mem0 itself is Apache 2.0 licensed. Use Zep, with its Graphiti engine, for facts that change over time.
GitHub MCP handles code. The first-party server had over 33k stars as of October 2026 and covers code search, branches, issues, pull requests, and Actions runs. Running and testing code still happens elsewhere, such as CI. Context7 by Upstash adds version-specific library docs.
Supabase MCP handles databases. It lists tables, runs SQL, applies migrations, queries logs, checks security and performance advisors, deploys Edge Functions, and generates TypeScript types. Scope it to one project and turn on read-only mode to run queries as a read-only Postgres user.
Official servers handle workplace tools. Google Drive searches and reads files and spreadsheets, Notion searches and updates pages and databases, Slack reads channels and threads and sends messages, and Atlassian works with Jira tickets and Confluence docs.

Three Sample Stacks

Agent	Servers
Web research agent	Zenrows and Mem0 (memory is optional for one-off tasks)
Coding agent	GitHub MCP, Playwright MCP, and Zenrows
Internal company assistant	Mem0, Slack, Google Drive, and Notion

Mistakes to Avoid

Connecting too many tools inflates context and leads the agent to pick the wrong one.
Running two servers for the same job, such as two browser servers, makes the agent choose between near-identical tools.
Granting broad credentials, such as a GitHub token with write access to every repository, gives the agent more reach than its task needs. Scope each token and use read-only access where the agent only reads.
Letting an agent send messages or push changes without an approval step leaves it open to prompt injection.

Wrapping Up

Pick servers from your agent's workflow. Connect the ones it uses and leave the rest off.

What's Next

Start free with Zenrows and its 5,000 monthly credits.
Compare scraping-focused options in the roundup of scraping MCP servers.
Build a local agent with Ollama and the Zenrows MCP server.

We Compared 5 Scraping APIs for Live Data. Here’s the Result

Praise James — Tue, 03 Nov 2026 11:02:16 +0000

This article was originally published on the Zenrows blog.

We compared five web scraping APIs for real-time data, meaning the current page content returned at the moment your application asks for it. Three of them, Zenrows, Zyte, and ScraperAPI, ran through Zenrows' benchmark of 100 requests per target against four protected sites. Here's the result. Zenrows succeeded on 99.00% of requests at an average of 9.1 seconds, Zyte on 74.75% at 14.3 seconds, and ScraperAPI on 49.00% at 36.0 seconds.

Bright Data and Oxylabs weren't benchmarked, so they're compared on how they handle bot detection, rendering, and pricing. You'll also get the best fit for each use case and two Python scripts for collecting fresh data on demand and in bulk. The scripts are on GitHub.

The Short Version

API	Success rate	Approach	Output	Entry price
Zenrows	100% (AI agent benchmark), 99% (API benchmark)	Adaptive Stealth Mode	HTML, Markdown, JSON via `extract=auto`	$16/mo for 45,000 credits, success-only billing
Bright Data	Not benchmarked	Managed browsers, 400M+ proxy network	JSON, NDJSON, CSV	$1.50 per 1,000 records
Zyte	74.75%	AI proxy selection and retries	JSON with AI extraction	$0.13 to $1.27 per 1,000 HTTP requests
ScraperAPI	49%	ML proxy and header selection	JSON or CSV via `autoparse`	$49/mo for 100,000 credits
Oxylabs	Not benchmarked	ML proxy rotation, 175M+ residential IPs	HTML, Markdown, PNG, JSON	$49/mo

All benchmark figures come from Zenrows' own tests, so run your target sites through a free plan before you commit.

How Success Was Measured

A request counted as successful when it returned HTTP 200 and a valid title or meaningful data. The web scraping API benchmark sent 100 requests per target at 2 requests per second to Zim, Glassdoor, Idealista, and Indeed.

API	Success rate	Average response time
Zenrows	99.00%	9.1 s
Zyte	74.75%	14.3 s
ScraperAPI	49.00%	36.0 s

A separate AI agent benchmark recorded 100% for Zenrows on DataDome, Cloudflare, and Akamai targets.

Why Real-Time Scraping Fails

Protected sites check IP reputation, browser identity, headers, and behavior. A request that fails those checks gets a block, a challenge page, or stale cached content. Managed APIs handle IP rotation, TLS and header alignment, sessions, and CAPTCHAs. Pages that load data with JavaScript also need a headless browser.

Fetch One Page on Demand

# pip install requests python-dotenv
import html
import os
import re

import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ.get("ZENROWS_API_KEY")
TARGET_URL = "https://www.amazon.com/dp/B0CHWRXH8B"

params = {
    "url": TARGET_URL,
    "apikey": API_KEY,
    "mode": "auto",
}

response = requests.get(
    "https://api.zenrows.com/v1/",
    params=params,
    timeout=180,
)
response.raise_for_status()

title_match = re.search(
    r"<title[^>]*>(.*?)</title>",
    response.text,
    re.IGNORECASE | re.DOTALL,
)
if not title_match:
    raise ValueError("The response does not contain a page title.")

page_title = html.unescape(title_match.group(1)).strip()

print(f"Status code: {response.status_code}")
print(f"Page title: {page_title}")

mode=auto starts with a lightweight request and adds Premium Proxies and JavaScript rendering only when the page needs them. The script confirms the right page came back by reading its title.

Status code: 200
Page title: Amazon.com: Apple AirPods Pro (2nd Generation) Wireless Ear Buds with USB-C Charging, Up to 2X More Active Noise Cancelling Bluetooth Headphones, Transparency Mode, Adaptive, Personalized Spatial Audio, White : Electronics

Credit costs scale with what the page needs. A basic request costs 1 credit, rendering costs 5, Premium Proxies cost 10, and both together cost 25.

Collect Many Pages as One Batch Job

# pip install requests python-dotenv
import os

import requests
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv("ZENROWS_API_KEY")
batch_url = "https://async.api.zenrows.com/v1/jobs"

headers = {
    "X-API-Key": api_key,
    "Content-Type": "application/json",
}

payload = {
    "type": "regular",
    "status": "closed",
    "zenrows_params": {
        "mode": "auto",
        "extract": "auto",
    },
    "tasks": [
        {"url": "https://www.amazon.com/dp/B0CHWRXH8B"},
        {"url": "https://www.amazon.com/dp/B09HM94VDS"},
    ],
}

response = requests.post(
    batch_url,
    headers=headers,
    json=payload,
    timeout=30,
)
response.raise_for_status()

job = response.json()
print(job["job_id"], job["latest_run"]["status"])

The script submits two product URLs as one Batch job with extract=auto, so each result comes back as structured JSON with pricing, availability, specifications, and reviews.

01M5ERCYJ7TSGP713TNMR1E8XY running

Your job ID will differ. Batch manages the queue, retries, schedules, and webhook notifications while the pages process.

How the Other APIs Work

Bright Data offers 100+ pre-built scrapers for sites such as Amazon, TikTok, LinkedIn, and Google Maps, plus 5,000 free records a month. It prices per record regardless of page complexity.
Zyte runs a managed browser with page actions and network capture, AI parsing for products, articles, and job postings, and up to 3,000 requests per minute on standard plans. Its trial gives $5 of credit for 30 days.
ScraperAPI needs render=true for JavaScript pages, and its documentation says CAPTCHAs embedded in forms need a third-party solver. Rendering costs 10 credits, rendering with premium proxies 25, and rendering with ultra-premium proxies 75.
Oxylabs provides preconfigured endpoints, geo_location targeting, the OxyCopilot AI parser, and browser instructions for clicks and form input. Its free plan covers 2,000 results, and 100,000 rendered results cost $135.

Which API Fits Which Job

Use case	Best fit	Why
Dynamic pricing	Zenrows	Protected pages and scheduled Batch jobs
Financial data and news	Zyte	AI article extractor returns headlines, bodies, and dates as JSON
AI training data	Zenrows	`mode=auto` across mixed sources, with Markdown or JSON
Inventory monitoring	Zenrows	Batch with a proxy country per market
SEO and SERP tracking	ScraperAPI	Google SERP endpoint with country and domain parameters

When Another Option Makes Sense

Pick Bright Data if you need direct control over proxy routing by ASN, carrier, or ZIP code.
Pick Zyte if you already run Scrapy spiders, though Zenrows also has a Scrapy integration.
Build your own stack with Requests and Beautiful Soup if your targets are static and you can maintain the infrastructure.

Start Collecting Real-Time Data With Zenrows Fetch

For protected, JavaScript-heavy targets, Zenrows led every benchmark here and bills only successful requests. Each alternative has a niche, from Bright Data's proxy control to ScraperAPI's SERP endpoint.

What's Next

Sign up for Zenrows and test your own targets with 5,000 free credits a month, no card required.
Clone the companion repo and swap in your own URLs.
Read about building a self-hosted web scraper to see where an in-house stack stops working.

We benchmarked 4 web scraping APIs for AI agents against DataDome, Cloudflare, and Akamai

Goodness E. Eboh — Sun, 11 Oct 2026 17:14:55 +0000

This article was originally published on the Zenrows blog.

This post compares five web scraping APIs for AI agents using 800 benchmark requests against DataDome, Cloudflare, and Akamai. It then shows you how to connect the strongest performer to Claude Desktop, the OpenAI Agents SDK, and LangChain in a few lines of code.

The APIs are Zenrows, Firecrawl, Jina AI Reader, CrawlForge, and Bright Data. Every number below comes from a public benchmark repository with the raw per-request results, so you can rerun any of it yourself.

The short version

Provider	Protected-target success rate	Avg. response time
Zenrows	100.00%	5,215 ms
Jina	99.33%	1,257 ms
Firecrawl	66.67%	1,497 ms
CrawlForge	0.00%	867 ms

All four tested APIs scored 100 percent on an unprotected Wikipedia page. The protected targets are where they separated. Bright Data wasn't part of the test because it requires enterprise KYC.

Why test anti-bot access first

Scraping APIs for agents usually get compared on output format, MCP support, and JavaScript rendering. Those features matter once the agent has the page. The harder question is whether the agent gets the page at all when Cloudflare, DataDome, or Akamai protects the target.

Public docs, company blogs, and news sites rarely cause trouble. E-commerce pricing pages, job boards, and industry directories are different. They return challenge pages, empty responses, or partial content, and the agent has nothing useful to work with. That's why this comparison measures anti-bot success rate before anything else.

How the benchmark worked

Zenrows, Jina AI Reader, and CrawlForge ran on August 20, 2026, with Zenrows using Fetch in mode=auto.
Firecrawl ran on August 25, 2026, in proxy=auto mode, which starts in standard mode and retries in stealth mode.
Each provider got 50 requests per target at a fixed rate of two requests per second, for 800 requests in total.
The targets were JobLookup behind DataDome, Chrono24 behind Cloudflare, Reuters behind Akamai, and Wikipedia with no protection as a baseline.
A request counted as successful only if it returned an HTTP 200 status code and a valid page title or meaningful data.

Every figure was recomputed from the raw dataset on October 1, 2026, so each one traces back to a logged request.

Full results

Provider	Target	Protection	Success rate	Avg. response time (ms)
Zenrows	JobLookup	DataDome	100%	10,053
Zenrows	Chrono24	Cloudflare	100%	4,590
Zenrows	Reuters	Akamai	100%	4,103
Zenrows	Wikipedia	Unprotected	100%	2,115
Jina	JobLookup	DataDome	98%	1,029
Jina	Chrono24	Cloudflare	100%	1,417
Jina	Reuters	Akamai	100%	1,242
Jina	Wikipedia	Unprotected	100%	1,340
Firecrawl	JobLookup	DataDome	0%	1,299
Firecrawl	Chrono24	Cloudflare	100%	1,688
Firecrawl	Reuters	Akamai	100%	1,488
Firecrawl	Wikipedia	Unprotected	100%	1,512
CrawlForge	JobLookup	DataDome	0%	781
CrawlForge	Chrono24	Cloudflare	0%	709
CrawlForge	Reuters	Akamai	0%	728
CrawlForge	Wikipedia	Unprotected	100%	1,251

Zenrows returned a usable page on every request across all four targets. Jina came closest and failed once against JobLookup. Firecrawl hit a reCAPTCHA page on all 50 JobLookup requests and scored 100 percent on the other three targets. CrawlForge got 403s on JobLookup and Chrono24 and a 401 on Reuters, so it only succeeded on Wikipedia.

Provider	Avg. success rate (3 protected targets)	Avg. response time (4 targets)	Observed consumption per request (4 targets)
Zenrows	100.00%	5,215 ms	4.59 credits
Jina	99.33%	1,257 ms	24,922.58 tokens
Firecrawl	66.67%	1,497 ms	1 credit
CrawlForge	0.00%	867 ms	1 credit

Jina was the fastest API that returned content on the protected targets. CrawlForge's low average reflects how quickly it was turned away. Zenrows was slower on average, especially on the protected targets, where it was the only API to clear every request.

Consumption uses each provider's native unit, so the figures don't compare directly on cost. They're dashboard readings taken before and after the run, divided by the 200 requests each provider served, and the balances are in the repository. A low consumption figure says little about value when the requests behind it fail.

How Zenrows credits scale with protection

The 4.59-credit average for Zenrows blends four very different targets, because mode=auto escalates only as far as each target forces it to.

The unprotected Wikipedia page cost one credit per request. Reuters, behind Akamai, cost ten. Cloudflare-protected Chrono24 averaged 2.36 because not every request needed the full escalation. For an agent whose targets mix easy and hard pages, you pay for the protection each page puts up.

The five APIs

Zenrows

Zenrows fits agents that need protected and dynamic web sources. It scored 100 percent across the protected targets in this benchmark, and Zenrows publishes a 99.93 percent success rate across its wider target set. Zenrows Fetch handles Cloudflare, DataDome, Akamai, and PerimeterX, along with dynamic website support, in one API call.

It returns pages as LLM-ready Markdown, plain text, or PDF, and Zenrows Extract returns structured JSON without pre-written CSS selectors. The Agent Toolkit ships on every plan and includes the following:

An MCP server exposes scrape, extract, and more than 30 browser_* tools, with first-party setup guides for Claude Desktop, Claude Code, Cursor, Windsurf, VS Code, Zed, JetBrains IDEs, OpenCode, Google Antigravity, the Claude Agent SDK, the OpenAI Agents SDK, Google ADK, and Mastra. You can run it locally over npx or use the hosted endpoint, and the tool surface is identical either way.
A CLI gives terminal-native agents zenrows fetch and zenrows extract, and zenrows mcp config --client claude-code writes the MCP configuration for a client.
Agentic payments let an agent buy its own access. Zenrows publishes a machine-readable API catalogue with prices an agent can read without a login, and it settles through x402, MPP, or a card. An agent that hits a 402 challenge can complete the purchase and carry on.

Zenrows Batch adds async jobs of up to 100,000 URLs with concurrency, retries, and webhook delivery. The free tier gives you 5,000 credits a month with no card. Paid plans run from the Build tier at $16/month for 45,000 credits to the Scale tier at $458/month for five million credits, plus a custom Enterprise tier. Billing is pay per successful request, so blocked requests cost nothing.

Firecrawl

Firecrawl suits agents that access unprotected or lightly protected sources, and it averaged 66.67 percent across the protected targets here. Its output is built for LLMs, with Markdown, JSON Schema generation, and metadata extraction designed for RAG and chunking. It also offers /agent and /interact endpoints for multi-page exploration and stateful browser control.

All 50 of its JobLookup responses came back as a reCAPTCHA page, even with stealth retries. Its AI Extract feature needs a separate subscription on top of the base plan. Firecrawl has a free plan with 1,000 credits a month and paid plans from $19/month.

Jina AI Reader

Jina is a zero-setup gateway that converts URLs into LLM-ready Markdown when you prepend r.jina.ai/ to them. It includes image captioning for multimodal agents. It placed second in this benchmark at 99.33 percent across the protected targets.

Its token consumption varied with the amount of content returned and averaged 24,922.58 tokens per request, which makes billing harder to predict on large crawls. Jina offers 10 million free tokens, and paid usage starts at $50, at $0.050 per million tokens. It's a good match for quick lookups of documentation, news, and public web content.

CrawlForge

CrawlForge is MCP-native and ships 31 specialized tools an agent can discover without wrapper code, including fetch_url, deep_research, map_site, and scrape_template. It produces token-efficient Markdown, can run extraction schemas locally through Ollama at no cost, and charges a predictable one credit for a basic fetch and up to 10 for deep research.

It scored zero percent on all three protected targets and 100 percent on Wikipedia. CrawlForge has a free tier of 1,000 credits and paid plans from $19/month for 5,000 credits. It works best for agents on MCP-native hosts that mostly access unprotected sources.

Bright Data

Bright Data is the enterprise option for protected-site access. It runs a pool of more than 150 million IPs and holds SOC 2 Type II and ISO 27001 certifications. Its Web Unlocker, Browser API, and Residential Proxies products each have $499/month Scale plans, so the full stack costs more than a typical self-service tool. It fits enterprise teams with five-figure monthly data budgets, compliance requirements, and dedicated data engineering staff.

Connecting Zenrows to your agent

You can give an agent access to live web data through the Zenrows MCP server or through a framework integration. Each pattern below needs a Zenrows API key, which you can get from the signup page.

Claude Desktop and other MCP hosts

{
  "mcpServers": {
    "zenrows": {
      "command": "npx",
      "args": ["-y", "@zenrows/mcp"],
      "env": {
        "ZENROWS_API_KEY": "YOUR_ZENROWS_API_KEY"
      }
    }
  }
}

Add that block to claude_desktop_config.json and restart Claude Desktop. Claude can then call the scrape and browser_* tools whenever a task needs them. To see how this server compares with the alternatives, read the test of seven MCP servers for web scraping.

OpenAI Agents SDK

import asyncio
import os

from agents import Agent, HostedMCPTool, Runner

ZENROWS_API_KEY = os.environ["ZENROWS_API_KEY"]

agent = Agent(
    name="Web Research Assistant",
    instructions="Use live web data when you need current information.",
    tools=[
        HostedMCPTool(
            tool_config={
                "type": "mcp",
                "server_label": "zenrows",
                "server_description": "Web data access through Zenrows.",
                "server_url": "https://mcp.zenrows.com/mcp",
                "authorization": ZENROWS_API_KEY,
                "require_approval": "never",
            }
        )
    ],
)

async def main():
    result = await Runner.run(
        agent,
        "Visit https://news.ycombinator.com/ and summarize the three most recent posts.",
    )
    print(result.final_output)

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

The OpenAI Agents SDK supports MCP natively through HostedMCPTool, so this connects the hosted Zenrows MCP server with no custom function-calling logic. If you'd prefer to register Fetch as a plain function tool, this guide to building a web-aware agent with the OpenAI Agents SDK covers it.

LangChain

from langchain_zenrows import ZenRowsUniversalScraper
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o-mini")
zenrows_tool = ZenRowsUniversalScraper()

agent = create_react_agent(llm, [zenrows_tool])

result = agent.invoke({
    "messages": [
        {
            "role": "user",
            "content": "Scrape https://www.etsy.com/ in Markdown and return the four cheapest products in JSON."
        }
    ]
})

The langchain-zenrows package exposes Zenrows as a LangChain tool, so the agent calls it whenever it needs live web data. Set ZENROWS_API_KEY as an environment variable before you run it.

Everything at a glance

Criteria	Zenrows	Firecrawl	Jina AI Reader	CrawlForge	Bright Data
Best for	AI agents accessing protected and dynamic web sources	RAG, LLM data ingestion, and unprotected or lightly protected sources	Quick lookups of documentation, news, and public web content	MCP-native agents accessing primarily unprotected sources	Enterprise teams requiring protected web access
Protected-target success rate	100%	66.67%	99.33%	0%	Not tested (requires enterprise-level KYC)
Output formats	Markdown, plaintext, PDF, structured JSON with Extract	Markdown, JSON Schema, metadata	LLM-ready Markdown, visual descriptions	Markdown, extraction schemas	HTML, Markdown, structured data and other formats depending on product
MCP server	Yes	Yes	Yes, official and community	Yes	Yes
SDK support	Python, Node.js, Go	Python, Node.js, and others	HTTP API	No dedicated SDK	Multiple SDK and API integrations
Free tier	5,000 credits/month, no card	1,000 credits/month	10 million tokens	1,000 credits	5,000 credits/month
Entry price	$16/month	$19/month	$50	$19/month	Product-dependent, with Scale plans from $499/month
Success-billed only	Yes	Partial (only API-level failures are free)	No	No	Partial (Web Unlocker, SERP API, and Web Scraper API products only)
KYC required	No	No	No	No	Yes for Residential and Mobile IP networks
Benchmark source	Our benchmark	Our benchmark	Our benchmark	Our benchmark	Not tested

What the benchmark settles

Every API here handles output format and tool coverage well. Protected access decides whether an agent returns an answer or a challenge page, and that's where the five tools differ most.

Across 800 requests, Zenrows was the only API to return a usable page every time on DataDome, Cloudflare, Akamai, and the unprotected baseline. It charged one credit for the page that needed no escalation. Jina came closest and runs faster on open targets. Firecrawl and CrawlForge are reasonable choices for agents whose targets stay public.

Choose based on the targets your agent will hit. If any of them sit behind an anti-bot system, start with 5,000 free credits and run your own list through Fetch before you commit.

What's next

Clone the benchmark repository and swap in your own targets to see how each API handles them.
Build a full agent on top of Fetch with this guide to an AI lead generation agent with the OpenAI Agents SDK.
Compare a wider set of options in the roundup of AI web scraping tools.

FAQ

Can AI agents scrape websites protected by Cloudflare or DataDome?

They can when the API they call handles the protection for them. In this benchmark, two of the four tested APIs returned nothing usable from the DataDome target across 50 requests each. Zenrows returned a valid page on every request to the Cloudflare, DataDome, and Akamai targets.

What's the difference between a web scraping API and a scraping MCP server?

A web scraping API is the service that fetches the page. An MCP server hands that service to an agent as a callable tool. Zenrows offers both, so the same Fetch capability works over REST, through the CLI, or as an MCP tool your agent discovers automatically.

Does an AI agent need JavaScript rendering to scrape a page?

It depends on the target. Pages that build their content in the browser return little usable text without rendering, which is why a plain HTTP fetch often comes back empty. Zenrows' mode=auto decides per request, so the agent doesn't have to.

Can I use Zenrows with CrewAI?

You can. CrewAI and other agent frameworks connect to Zenrows through its MCP server, and LangChain agents can use the langchain-zenrows package.

Build an AI lead generation agent with the OpenAI Agents SDK and Zenrows

Benny — Thu, 08 Oct 2026 09:56:17 +0000

This article was originally published on the Zenrows blog.

In this tutorial, you'll build a lead generation agent that reads a bot-protected business directory, pulls every company listed on the page, enriches each one from its own website, and scores it against your ideal customer profile (ICP). You'll need Python 3.10 or above, an OpenAI API key, and a Zenrows API key.

The OpenAI Agents SDK runs the agent loop, and its @function_tool decorator turns plain Python functions into tools. Zenrows Fetch handles the step that usually breaks, which is getting the page at all. In the test run, one directory page produced 78 companies, and the agent scored each one in under 20 seconds.

The complete code is on GitHub if you'd like to clone it and follow along.

Prerequisites

You need Python 3.10 or above.
You need an OpenAI API key, which the agent, the extraction tool, and the scoring tool all use. Get one from your OpenAI developer dashboard.
You need a Zenrows API key to retrieve the full contents of web pages. Create one on the signup page.

Install the dependencies.

python3 -m pip install openai-agents openai requests pydantic python-dotenv

The SDK runs the agent, requests makes the Zenrows calls, pydantic defines the tool schemas, and python-dotenv loads your keys. Create a .env file in your project root with your OPENAI_API_KEY and ZENROWS_API_KEY, and add .env to .gitignore so the credentials never reach version control.

How the agent is structured

The work splits into four tools, and each one does a single job.

Tool	Input	Output	Why it's separate
`fetch_page`	a URL	the page as Markdown	it's the only job that touches the network, so retries and errors live in one place
`extract_leads`	Markdown plus the source URL	`Lead` records	it runs in chunks because a full directory page exceeds one reliable extraction call
`enrich_lead`	a company and its domain	enrichment signals from the company's site	it makes one fetch per nav link, so it drives the cost and is worth isolating
`score_lead`	enriched lead plus the ICP	a score from 0 to 100 with reasoning	it's pure reasoning with no network calls, so you can rerun it without re-fetching

Splitting the work keeps retries local. When enrichment fails on one lead, the agent retries that tool and keeps the rest of the run intact. Each job's output also stays small enough to hand to the next one.

The agent only sees two composed tools. discover_leads runs fetch_page and extract_leads, and qualify_lead runs enrich_lead and score_lead. If the agent passed an enriched lead between two separate tools, it would have to reproduce every page excerpt as a tool argument, and that overflows the context window on a large directory.

1. Build the Zenrows Fetch tool

Every code block from here goes into one file called tools.py. Append each block to the same file.

import os
import re
import requests
from urllib.parse import urlparse, parse_qs, unquote

from agents import function_tool
from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel

load_dotenv()

ZENROWS_API_KEY = os.getenv("ZENROWS_API_KEY")
ZENROWS_ENDPOINT = "https://api.zenrows.com/v1/"

if not ZENROWS_API_KEY:
    raise RuntimeError("set ZENROWS_API_KEY in your .env file")

client = OpenAI()

# ===========================================================================
# job 1: fetch a page, protected or not
# ===========================================================================

def _fetch_page(url: str) -> str:
    """fetch logic, importable and testable without the agent."""
    params = {
        "url": url,
        "apikey": ZENROWS_API_KEY,
        "mode": "auto",              # start cheap, escalate only when the target needs it
        # markdown strips nav and boilerplate before the model reads it
        "response_type": "markdown"
    }

    try:
        response = requests.get(ZENROWS_ENDPOINT, params=params, timeout=90)
        response.raise_for_status()
    # return failures as text instead of raising, so one bad url does not end the run.
    # each message tells the model whether the failure is worth retrying
    except requests.exceptions.Timeout:
        return "FETCH_ERROR: timed out after 90s. retry once, then move on."
    except requests.exceptions.HTTPError as exc:
        status = exc.response.status_code
        if status in (401, 403):
            return f"FETCH_ERROR: {status}. api key rejected, do not retry."
        if status == 429:
            return "FETCH_ERROR: 429 concurrency limit. wait, then retry."
        if status in (400, 404):
            return f"FETCH_ERROR: {status}. this url is not retrievable, do not retry."
        return f"FETCH_ERROR: {status}. retry once, then report the failure."
    except requests.exceptions.RequestException as exc:
        return f"FETCH_ERROR: {exc}. do not retry."

    content = response.text.strip()
    if not content:
        return "FETCH_ERROR: empty response body. do not retry."

    return content

# the decorator replaces the function with a FunctionTool object, which is not
# callable. keeping the logic in _fetch_page above leaves it testable
@function_tool
def fetch_page(url: str) -> str:
    """Fetch any web page as markdown, including pages behind anti-bot protection.

    Args:
        url: full url of the page to fetch
    """
    return _fetch_page(url)

# ===========================================================================
# job 2: turn a directory page into leads
# ===========================================================================

# matches the tracking links directories wrap around outbound urls
REDIRECT_LINK = re.compile(r"https?://[a-z0-9.-]*/redirect\?[^\s\)\"]+")

# a full directory page is too long for one reliable extraction call
CHUNK_SIZE = 40000

# slices cut mid-listing, and the extraction prompt is told to skip partial
# entries, so a listing straddling a boundary is dropped by both slices.
# overlapping the window carries each boundary listing whole into one of them;
# _key dedupes the entries the overlap sees twice
CHUNK_OVERLAP = 2000

_fetch_page holds the logic, and fetch_page wraps it with @function_tool so the agent can call it. The decorator replaces the function with a FunctionTool object that you can't call directly, so keeping the logic in a plain function leaves it testable.

Failures come back as text strings, so one bad URL doesn't end the run. Each message also tells the model whether a retry is worth it.

mode=auto lets Zenrows pick the access configuration each target needs. Simple pages resolve on the lightest path, and only protected targets escalate. That matters because protected sites rarely block you with an error. A Cloudflare-protected page can return a 200 status code with a challenge page in the body, and your agent will treat that challenge as a valid page with zero leads on it.

response_type=markdown strips navigation and boilerplate, which gives the model cleaner input for extraction. If your directory is an API-like endpoint, use response_type=json and skip the conversion. The three constants at the bottom of the block belong to the extraction step in the next section.

2. Extract leads from the directory page

Append the extraction code to tools.py.

# strict tool schemas reject bare dicts, so every tool input and output is a model
class Lead(BaseModel):
    company: str
    name: str
    website: str
    source_url: str

class LeadList(BaseModel):
    leads: list[Lead]

def _unwrap_redirects(markdown: str) -> str:
    """rewrite directory tracking links to the destination domain."""
    def replace(match):
        url = match.group(0)
        # the real destination sits url-encoded in the u parameter
        target = parse_qs(urlparse(url).query).get("u", [""])[0]
        if not target:
            return url
        parsed = urlparse(unquote(target))
        return f"{parsed.scheme}://{parsed.netloc}"

    return REDIRECT_LINK.sub(replace, markdown)

def _reject_source_domain(leads: list[dict], source_url: str) -> list[dict]:
    """blank any website that points back at the directory itself."""
    directory = urlparse(source_url).netloc.replace("www.", "")
    for lead in leads:
        host = urlparse(lead["website"]).netloc.replace("www.", "")
        if host == directory:
            lead["website"] = ""
    return leads

def _key(lead: dict) -> str:
    """stable identity for dedupe across chunks."""
    # the model returns the same domain with and without www, so normalise before comparing
    site = lead["website"].lower()
    for prefix in ("https://", "http://", "www."):
        site = site.replace(prefix, "")
    site = site.rstrip("/")
    # fall back to company name so leads without a domain are not collapsed into one
    return site or lead["company"].strip().lower()

def _extract_chunk(chunk: str, source_url: str) -> list[dict]:
    """run one extraction call over a slice of the page."""
    response = client.responses.parse(
        model="gpt-4o-mini",
        max_output_tokens=8000,
        input=[
            {
                "role": "system",
                "content": (
                    "extract every company listed in this page fragment. "
                    "website is the company's own domain, taken from that listing's "
                    "visit website link. "
                    "never use a link from the directory's own domain as the website. "
                    "name is a person's name and is usually absent from a directory "
                    "listing, so leave it empty unless a person is actually named. "
                    "use an empty string for any field the fragment does not state, "
                    "and never invent a value. "
                    "the fragment may start or end mid listing, so skip any partial entry."
                )
            },
            {
                "role": "user",
                "content": f"source_url: {source_url}\n\n{chunk}"
            }
        ],
        text_format=LeadList
    )
    return [lead.model_dump() for lead in response.output_parsed.leads]

def _extract_leads(content: str, source_url: str) -> list[dict]:
    """parse markdown into lead dicts, chunking long pages."""
    # a fetch failure is a string, not a page. parsing it would return an empty
    # list that looks identical to a genuinely empty directory
    if content.startswith("FETCH_ERROR"):
        return []

    # resolve tracking links before the model reads them
    content = _unwrap_redirects(content)

    leads, seen = [], set()
    for start in range(0, len(content), CHUNK_SIZE - CHUNK_OVERLAP):
        chunk = content[start:start + CHUNK_SIZE]
        for lead in _extract_chunk(chunk, source_url):
            key = _key(lead)
            if key in seen:
                continue
            seen.add(key)
            leads.append(lead)

    return _reject_source_domain(leads, source_url)

@function_tool
def extract_leads(content: str, source_url: str) -> list[Lead]:
    """Parse a fetched page into a list of leads.

    Args:
        content: markdown returned by fetch_page
        source_url: url the content came from, recorded on every lead
    """
    return [Lead(**lead) for lead in _extract_leads(content, source_url)]

# fetch and extract are one job from the agent's point of view. exposing them
# separately means the whole page passes through the model's context to get
# from one tool to the next, which overflows the window on a large directory
@function_tool
def discover_leads(source_url: str) -> list[Lead]:
    """Fetch a directory page and return the companies listed on it.

    Args:
        source_url: url of the directory or listing page
    """
    markdown = _fetch_page(source_url)
    if markdown.startswith("FETCH_ERROR"):
        return []
    return [Lead(**lead) for lead in _extract_leads(markdown, source_url)]

# ===========================================================================
# job 3: enrich each lead from its own website
# ===========================================================================

Strict tool schemas reject bare dicts, so every tool input and output is a Pydantic model. _unwrap_redirects rewrites the directory's tracking links to each company's own domain before the model reads them, and _reject_source_domain blanks any website that points back at the directory.

A full directory page is too long for one reliable extraction call, so _extract_leads splits it into 40,000-character chunks with a 2,000-character overlap. The overlap carries any listing that straddles a boundary into one chunk whole, and _key removes the duplicates the overlap creates.

discover_leads combines fetching and extraction in one tool because the raw output of fetch_page can exceed the context length by itself.

Test both jobs without the agent. Save this script as test_fetch_extract.py.

# test_fetch_extract.py
import json
import os
from tools import _fetch_page, _unwrap_redirects, _extract_leads

SOURCE_URL = "https://clutch.co/it-services"
FIXTURE = "fixtures/directory_page.md"
LEADS_CACHE = "fixtures/leads.json"

# ---- job 1: fetch ---------------------------------------------------------
# cache the page so extraction can be tuned without paying for a fetch each time
if os.path.exists(FIXTURE):
    with open(FIXTURE) as f:
        markdown = f.read()
    print(f"using cached page, {len(markdown)} chars")
else:
    markdown = _fetch_page(SOURCE_URL)
    if markdown.startswith("FETCH_ERROR"):
        raise SystemExit(markdown)
    os.makedirs("fixtures", exist_ok=True)
    with open(FIXTURE, "w") as f:
        f.write(markdown)
    print(f"fetched and cached, {len(markdown)} chars")

unwrapped = _unwrap_redirects(markdown)
print(f"{len(unwrapped)} chars after unwrapping redirects, "
      f"{len(markdown) - len(unwrapped)} saved\n")

# ---- job 2: extract -------------------------------------------------------
leads = _extract_leads(markdown, SOURCE_URL)

with open(LEADS_CACHE, "w") as f:
    json.dump(leads, f, indent=2)

print(f"{len(leads)} leads extracted\n")
for lead in leads[:5]:
    print(f"{lead['company']:<32} {lead['website'] or '(none)'}")

missing_site = sum(1 for lead in leads if not lead["website"])
print(f"\n{missing_site} of {len(leads)} leads have no website")

Run it from the project root.

python3 test_fetch_extract.py

Here's the output from a local run against an IT services directory.

using cached page, 543637 chars
457647 chars after unwrapping redirects, 85990 saved

78 leads extracted

Infracore                        https://infracore.net
Miles IT                         https://www.milesit.com
Geniusee                         https://geniusee.com
Peeklogic, LLC                   https://www.peeklogic.com
TechQuarter LLC                  https://www.techquarter.io

0 of 78 leads have no website

The page came back at 543,637 characters, and unwrapping the redirects saved 85,990 of them. Counts vary between runs because directory pages change and the model can read the same page differently.

3. Enrich and score each lead

Append the enrichment code to tools.py.

# matches any markdown link, used to read a company's own navigation
MARKDOWN_LINK = re.compile(r"\[([^\]]+)\]\((https?://[^\s\)]+)\)")

# guessing paths costs a fetch per miss, so read the site's nav instead and
# follow only the links it actually has
SIGNAL_KEYWORDS = {
    "hiring": ["career", "job", "join", "hiring", "work with us", "we are hiring"],
    "services": ["service", "what we do", "solutions", "expertise", "capabilities"],
    "portfolio": ["portfolio", "case stud", "our work", "projects", "clients"],
    "about": ["about", "who we are", "our story", "team"],
    "contact": ["contact", "get in touch", "book a call", "let's talk", "talk to us"],
}

# dict[str, X] is rejected by strict schemas too, which is why signals is a list
class Signal(BaseModel):
    name: str
    found: bool
    url: str
    excerpt: str

class EnrichedLead(BaseModel):
    company: str
    website: str
    signals: list[Signal]

def _discover_links(markdown: str, website: str) -> dict:
    """find real urls for each signal by reading the homepage nav."""
    host = urlparse(website).netloc.replace("www.", "")
    homepage_path = urlparse(website).path.rstrip("/") or "/"
    found = {}

    for text, url in MARKDOWN_LINK.findall(markdown):
        parsed = urlparse(url)

        # only follow links on the company's own domain
        if parsed.netloc.replace("www.", "") != host:
            continue

        # an anchor on the homepage is content already in the homepage excerpt
        if parsed.fragment and (parsed.path.rstrip("/") or "/") == homepage_path:
            continue

        # drop the fragment, it never changes what the server returns
        clean_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"

        haystack = f"{text} {url}".lower()
        for signal, keywords in SIGNAL_KEYWORDS.items():
            if signal not in found and any(k in haystack for k in keywords):
                found[signal] = clean_url

    return found

def _enrich_lead(company: str, website: str) -> EnrichedLead:
    """collect signals from the company's own site."""
    # a lead with no domain still flows through to scoring, judged on what little is known
    if not website:
        return EnrichedLead(company=company, website="", signals=[])

    homepage = _fetch_page(website)
    if homepage.startswith("FETCH_ERROR"):
        return EnrichedLead(company=company, website=website, signals=[])

    # the homepage is the most informative single page, and on a one-page site
    # it holds everything the nav links point at
    signals = [Signal(name="homepage", found=True, url=website, excerpt=homepage[:6000])]

    for name, url in _discover_links(homepage, website).items():
        content = _fetch_page(url)
        ok = not content.startswith("FETCH_ERROR")
        signals.append(Signal(
            name=name,
            found=ok,
            url=url,
            # keep excerpts small, scoring reads all of them in one call
            excerpt=content[:2000] if ok else ""
        ))

    return EnrichedLead(company=company, website=website, signals=signals)

@function_tool
def enrich_lead(company: str, website: str) -> EnrichedLead:
    """Collect signals from a company's own website.

    Args:
        company: company name from the directory listing
        website: company's own domain
    """
    return _enrich_lead(company, website)

# ===========================================================================
# job 4: score each lead against the icp
# ===========================================================================

An extracted lead carries only a company name and a domain, which is too little to judge fit. enrich_lead reads the company's homepage, finds the navigation links the site exposes, and fetches only those pages. Guessing paths like /careers would cost a fetch for every miss. The signals that matter live on the company's own site in its hiring pages, services pages, case studies, and contact pages.

Markdown works well when the model does the reading. If you'd prefer structured fields, Zenrows Extract returns JSON without a selector map.

Next, append the scoring code.

class Score(BaseModel):
    score: int
    reasoning: str

class ScoredLead(BaseModel):
    company: str
    website: str
    score: int
    reasoning: str

def _score_lead(lead: EnrichedLead, icp: str) -> ScoredLead:
    """score an enriched lead against the icp description."""
    # no fetching happens here. every page was already retrieved in job 3
    summary = "\n\n".join(
        f"{s.name} page: {'found at ' + s.url if s.found else 'not found'}\n{s.excerpt}"
        for s in lead.signals
    )

    response = client.responses.parse(
        model="gpt-4o-mini",
        max_output_tokens=500,
        input=[
            {
                "role": "system",
                "content": (
                    "score how well this company matches the ideal customer profile. "
                    "return an integer from 0 to 100 and one sentence of reasoning. "
                    "base the score only on evidence in the signals provided. "
                    "a missing page is weak evidence, not disqualifying. "
                    "if the signals are too thin to judge, score below 30 and say so."
                )
            },
            {
                "role": "user",
                "content": (
                    f"ideal customer profile:\n{icp}\n\n"
                    f"company: {lead.company}\n"
                    f"website: {lead.website}\n\n"
                    f"signals:\n{summary}"
                )
            }
        ],
        text_format=Score
    )

    return ScoredLead(
        company=lead.company,
        website=lead.website,
        score=response.output_parsed.score,
        reasoning=response.output_parsed.reasoning
    )

@function_tool
def score_lead(lead: EnrichedLead, icp: str) -> ScoredLead:
    """Score an enriched lead against an ICP description.

    Args:
        lead: enriched lead returned by enrich_lead
        icp: plain-language description of the ideal customer
    """
    return _score_lead(lead, icp)

# an EnrichedLead carries page excerpts the model has no reason to read. passing
# it between two tools means the agent has to reproduce all of it as an argument,
# and it summarises instead, so scoring reads a summary rather than the pages
@function_tool
def qualify_lead(company: str, website: str, icp: str) -> ScoredLead:
    """Enrich a lead from its own website and score it against an ICP.

    Args:
        company: company name from the directory listing
        website: company's own domain
        icp: plain-language description of the ideal customer
    """
    enriched = _enrich_lead(company, website)
    return _score_lead(enriched, icp)

score_lead reads every signal alongside the ICP and returns a score from 0 to 100 with one sentence of reasoning. The prompt ties the score to evidence, so a lead with only a domain scores lower than one with a careers page and named case studies. No fetching happens in this step, which means you can rerun scoring without paying for the pages again.

qualify_lead pairs enrichment and scoring for the same reason discover_leads pairs fetching and extraction. The bulky intermediate data stays inside one Python call.

Test both halves before you wire the agent, so any later failure belongs to the agent loop. Save this script as test_enrich_score.py.

# test_enrich_score.py

import json
import os
import time
from tools import _fetch_page, _unwrap_redirects, _extract_leads, _enrich_lead, _score_lead

SOURCE_URL = "https://clutch.co/it-services"
FIXTURE = "fixtures/directory_page.md"
LEADS_CACHE = "fixtures/leads.json"

ICP = (
    "IT services agencies that build custom software for B2B clients, "
    "publish detailed case studies with named clients, and offer ai and cloud services"
)

# ---- job 1: fetch ---------------------------------------------------------
if os.path.exists(FIXTURE):
    with open(FIXTURE) as f:
        markdown = f.read()
    print(f"using cached page, {len(markdown)} chars")
else:
    markdown = _fetch_page(SOURCE_URL)
    if markdown.startswith("FETCH_ERROR"):
        raise SystemExit(markdown)
    os.makedirs("fixtures", exist_ok=True)
    with open(FIXTURE, "w") as f:
        f.write(markdown)
    print(f"fetched and cached, {len(markdown)} chars")

unwrapped = _unwrap_redirects(markdown)
print(f"{len(unwrapped)} chars after unwrapping redirects, "
      f"{len(markdown) - len(unwrapped)} saved\n")

# ---- job 2: extract -------------------------------------------------------
if os.path.exists(LEADS_CACHE):
    with open(LEADS_CACHE) as f:
        leads = json.load(f)
    print(f"using {len(leads)} cached leads\n")
else:
    leads = _extract_leads(markdown, SOURCE_URL)
    with open(LEADS_CACHE, "w") as f:
        json.dump(leads, f, indent=2)
    print(f"{len(leads)} leads extracted\n")

for lead in leads[:5]:
    print(f"{lead['company']:<32} {lead['website'] or '(none)'}")

missing_site = sum(1 for lead in leads if not lead["website"])
print(f"\n{missing_site} of {len(leads)} leads have no website\n")
print("-" * 60 + "\n")

# ---- jobs 3 and 4: enrich and score ---------------------------------------
# enrichment is one fetch per nav link found, so start with a few leads
sample = [lead for lead in leads if lead["website"]][:3]

for lead in sample:
    start = time.time()
    enriched = _enrich_lead(lead["company"], lead["website"])
    elapsed = time.time() - start

    found = [s.name for s in enriched.signals if s.found]
    print(f"{enriched.company}  ({elapsed:.1f}s)")
    print(f"  found: {', '.join(found) if found else 'nothing'}")

    scored = _score_lead(enriched, ICP)
    print(f"  score: {scored.score}")
    print(f"  {scored.reasoning}\n")

Run it.

python3 test_enrich_score.py

Here are the first two leads from the output.

Infracore  (12.8s)
  found: homepage, contact, services, portfolio, about, hiring
  score: 80
  Infracore fits well within the ideal customer profile as they offer IT services, including cloud solutions and have case studies showcasing customized work for various industries, but they may lack detailed named client presentations.

Miles IT  (12.1s)
  found: homepage, contact, about, services, portfolio, hiring
  score: 75
  Miles IT provides software development and AI services, aligning well with the ideal customer profile, but lacks detailed case studies with named clients which slightly diminishes the score.

Each lead took around 12 seconds to enrich. The page list reflects what each site exposes in its navigation, so a lead with no careers page has no hiring signal.

4. Wire the agent and run it

Save the agent as agent.py.

import asyncio

from agents import Agent, Runner, set_tracing_disabled

from tools import discover_leads, qualify_lead

set_tracing_disabled(True)

SOURCE_URL = "https://clutch.co/it-services"

ICP = (
    "IT services agencies that build custom software for B2B clients, "
    "publish detailed case studies with named clients, and offer ai and cloud services"
)

INSTRUCTIONS = """You find and qualify sales leads from web directories.

Follow this sequence exactly:

1. call discover_leads on the source url the user gives you
2. for each lead that has a website, call qualify_lead with its company, website,
   and the icp description from the user message
3. return the scored leads as a json array sorted by score descending

rules:
- process only the first 10 leads that have a website, then stop and return them
- skip leads with no website, they cannot be scored fairly
- if a tool returns a string starting with FETCH_ERROR, follow the instruction in
  that message. do not retry a call the message tells you not to retry
- never invent a lead, a website, or a score. every value comes from a tool result
"""

agent = Agent(
    name="lead generation agent",
    instructions=INSTRUCTIONS,
    model="gpt-4o-mini",  # the reasoning here is scoring, not planning
    tools=[discover_leads, qualify_lead],
)

async def main():
    result = await Runner.run(
        agent,
        input=f"source url: {SOURCE_URL}\n\nideal customer profile:\n{ICP}",
        max_turns=50,  # one qualify call per lead, plus discovery and the reply
    )

    # the trace shows which tools ran and in what order
    for item in result.new_items:
        if item.type == "tool_call_item":
            print(item.raw_item.name)

    print()
    print(result.final_output)

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

The agent gets the two composed tools and a fixed sequence in its instructions. A small model is enough here because the sequence is already decided and the heavy reasoning happens inside score_lead. The instructions also cap each run at 10 leads, skip leads with no website, and forbid invented values.

Run the agent.

python3 agent.py

The agent prints the ranked leads as JSON. Here are the first two.

[
    {
        "company": "Geniusee",
        "website": "https://geniusee.com",
        "score": 85,
        "reasoning": "Geniusee offers a range of IT services including custom software development, AI solutions, and cloud services, which align closely with the ideal customer profile; however, the absence of detailed case studies with named clients slightly limits the fit."
    },
    {
        "company": "Sigli",
        "website": "https://www.sigli.com",
        "score": 85,
        "reasoning": "Sigli aligns closely with the ideal customer profile by offering custom software development for B2B clients, specializing in AI and cloud services, and publishing case studies, although the specificity of named clients could not be verified."
    }
]

Every value comes from a tool result, so the tool-call trace is the first thing to check when a run misbehaves.

5. Scale across pages with Batch

Save this script as batch.py.

# batch.py
# Scale discovery across several directory pages with Zenrows Batch.
#
# One Fetch call retrieves one page. Batch takes the whole list as a single
# managed job and handles the queue, concurrency and retries, so a directory
# with many pages costs one submission instead of one call per page.

import time

import requests
from agents import function_tool

from tools import ZENROWS_API_KEY, Lead, _extract_leads

BATCH_ENDPOINT = "https://async.api.zenrows.com/v1/jobs"

# batch authenticates by header, unlike fetch which takes apikey as a query param
BATCH_HEADERS = {"X-API-Key": ZENROWS_API_KEY, "Content-Type": "application/json"}

# a job accepts up to 100,000 urls
MAX_URLS_PER_JOB = 100_000

def _submit_batch(urls: list[str]) -> str:
    """submit a url list as one job, return the job id."""
    if len(urls) > MAX_URLS_PER_JOB:
        raise ValueError(f"a job accepts at most {MAX_URLS_PER_JOB} urls")

    payload = {
        "type": "regular",
        "status": "closed",  # run once, accept no further tasks
        "zenrows_params": {"mode": "auto", "response_type": "markdown"},
        "tasks": [{"url": url} for url in urls],
    }
    response = requests.post(
        BATCH_ENDPOINT, headers=BATCH_HEADERS, json=payload, timeout=30
    )
    response.raise_for_status()
    return response.json()["job_id"]

def _collect_batch(job_id: str, max_attempts: int = 60) -> list[tuple[str, str]]:
    """wait for the job to reach a terminal state, then pull each task's markdown."""
    # terminal run states are completed, stopped and deleted. "failed" is a task
    # status, not a run status, so polling for it never returns
    for _ in range(max_attempts):
        time.sleep(5)
        status = requests.get(
            f"{BATCH_ENDPOINT}/{job_id}", headers=BATCH_HEADERS, timeout=30
        )
        status.raise_for_status()
        run = status.json()["latest_run"]
        if run["status"] in ("completed", "stopped", "deleted"):
            break
    else:
        raise TimeoutError(f"job {job_id} did not reach a terminal state in time")

    response = requests.get(
        f"{BATCH_ENDPOINT}/{job_id}/results", headers=BATCH_HEADERS, timeout=30
    )
    response.raise_for_status()

    pages = []
    for task in response.json()["results"]:
        if task["status"] != "successful":
            continue
        # result_url is presigned and valid for 2 hours; re-list the results for a
        # fresh link rather than storing this one
        markdown = requests.get(task["result_url"], timeout=60).text
        pages.append((task["url"], markdown))

    return pages

@function_tool
def discover_leads_batch(source_urls: list[str]) -> list[Lead]:
    """Fetch several directory pages as one job and return every company listed.

    Args:
        source_urls: directory or listing page urls to fetch
    """
    job_id = _submit_batch(source_urls)

    leads = []
    for url, markdown in _collect_batch(job_id):
        leads.extend(Lead(**lead) for lead in _extract_leads(markdown, url))

    return leads

if __name__ == "__main__":
    # two pages of the same directory, submitted as one job
    urls = [
        "https://clutch.co/it-services",
        "https://clutch.co/it-services?page=2",
    ]
    job_id = _submit_batch(urls)
    print(f"submitted {job_id}")

    for url, markdown in _collect_batch(job_id):
        print(f"{url}: {len(markdown)} chars")

One directory page rarely produces enough leads to matter, and a directory usually has many pages. Running each page as its own Fetch call means you manage concurrency and retries yourself. Zenrows Batch takes the whole URL list as one managed job and handles the queue, concurrency, retries, and result delivery. A single job accepts up to 100,000 URLs.

Batch authenticates with an X-API-Key header, while Fetch takes the key as a query parameter. Each successful task returns a presigned result_url that stays valid for two hours.

You can also reach Batch through the Python SDK, the zenrows batch CLI, and the Zenrows MCP server, and none of them need the polling loop. This example uses raw REST so every step stays visible. For the full implementation guide, read running large-scale scraping jobs with Batch.

Wrapping up

You now have an AI lead generation agent that reads a directory with Zenrows Fetch, pulls out every company listed, visits each company's domain, and scores the results against a plain-language ICP. The run in this tutorial pulled 78 companies off a single page, and agent.py scores the first 10 of them per run at under 20 seconds each.

That's automated lead generation with no vendor list underneath it. The agent reads the source, and the source is whichever directory your market uses.

What's next

Raise or remove the 10-lead cap in INSTRUCTIONS when you want a full pass over the directory.
Put a human review checkpoint between scoring and outreach, then push qualifying leads into a CRM such as HubSpot or Salesforce, an email sequencing tool, or a CSV export.
Wrap agent.py in an HTTP endpoint and trigger it as a webhook step from n8n, Zapier, or Power Automate, passing the source URL and the ICP in the request body.
Skip the wrapper entirely with the Zenrows MCP server, which exposes the same fetch capability to any MCP-compatible agent. This walkthrough of the MCP server covers the setup.

FAQ

How much does it cost to run this agent?

A standard Zenrows request costs one credit. mode=auto selects the configuration each target needs, and a page that requires the full configuration costs up to 25 credits. These weights stay fixed across all plans.

In this tutorial, the directory page resolved at the standard rate. Enrichment drives the cost because it fetches the homepage plus each nav link it finds, which came to five or six fetches per lead. A 10-lead run costs roughly 60 credits. You only pay for successful requests, so a blocked page costs nothing. The Zenrows pricing documentation has the details.

Can I use a different LLM?

You can swap the model behind the agent wrapper because the Zenrows tools are plain Python functions. extract_leads and score_lead depend on OpenAI's structured output API, so you'll need to replace those two calls if you switch providers.

Does this agent work on LinkedIn?

You shouldn't point it at LinkedIn. LinkedIn's user agreement prohibits automated access, and enforcement can lead to account restrictions. Point the agent at sources that permit automated access, such as public directories, business registries, and review platforms.

How does this compare to using Apollo or Clay?

Indexed datasets return the same companies when everyone queries with the same filters, and those filters are limited to the fields the dataset captures. This agent reads the source directly and scores each company on signals from its own site, so two teams with the same ICP can still surface different leads.

Build a web-aware TypeScript agent with Mastra and Zenrows

Benny — Wed, 07 Oct 2026 12:54:06 +0000

This article was originally published on the Zenrows blog.

In this tutorial, you'll build a Mastra agent with a custom Zenrows Fetch tool that reads protected and JavaScript-rendered pages, then add a workflow that fetches a list of URLs in parallel. You'll need Node.js 22.13 or later, an OpenAI or Anthropic API key, and a Zenrows API key.

The finished agent compares live Walmart product listings and returns either Markdown to reason over or structured JSON with fields like price and rating. The complete code is on GitHub.

Prerequisites

You need Node.js 22.13 or later.
You need an API key for one model provider. This tutorial uses OpenAI by default and shows the Anthropic alternative. Get an OpenAI key from your OpenAI developer dashboard or an Anthropic key from your Anthropic developer dashboard.
You need a Zenrows API key to retrieve the full contents of web pages. Create one from your Zenrows dashboard.

Why the built-in fetch tool falls short

Mastra's built-in webFetchTool makes plain HTTP requests, doesn't run JavaScript, and returns raw HTML. That's enough for simple pages. On browser-rendered or protected pages, such as a product listing protected by Cloudflare, it breaks quietly.

Cloudflare returns its challenge page with a normal HTTP status code, so the tool raises no error. The agent treats the challenge as the result and reports a missing or wrong price.

The fix belongs in the transport layer, which controls HTTP requests, headers, proxies, and whether JavaScript runs. Zenrows Fetch with mode=auto handles DataDome, Akamai, and Cloudflare in a single call. Wrapping that call in Mastra's createTool keeps typed input and output schemas around the fetch.

1. Set up the Mastra project

npm create mastra@latest

The CLI scaffolds the project structure, config files, and a starter agent, and installs @mastra/core and its dependencies. Follow the prompts to name the project and pick your LLM provider. Then add Zod if the scaffold didn't include it.

npm install zod

Create a .env file in the project root with your keys. You only need the key for the provider you plan to use.

ZENROWS_API_KEY=your_zenrows_key
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key

Add .env to .gitignore so your credentials stay out of version control. Mastra loads .env automatically, but the standalone test scripts in this tutorial run outside Mastra's loader, so install dotenv for them.

npm install dotenv --save

Your project structure should look similar to this.

src/
└── mastra/
    ├── agents/
    │   └── agent.ts
    ├── tools/
    │   └── schedule-tools.ts
    └── index.ts

Start the dev server.

npm run dev

Mastra Studio opens at http://localhost:4111. It reads the registered agents from your Mastra instance and gives you a chat UI to test them.

2. Build the Zenrows Fetch tool

Create src/mastra/tools/zenrows-fetch.ts next to the template's example tools.

import { createTool } from '@mastra/core/tools';
import { z } from 'zod';

export const zenrowsFetchTool = createTool({
  id: 'zenrows-fetch',

  description:
  'Fetches web pages using Zenrows. Use this tool when you need information from a specific URL. Set extractJson to true when you need structured data such as product details, pricing, availability, ratings, or other fields. Leave extractJson false when you need the page content as Markdown.',

  inputSchema: z.object({
    url: z.string().url(),
    extractJson: z.boolean().optional(),
  }),

  outputSchema: z.object({
    content: z.union([z.string(), z.record(z.string(), z.unknown())]),
    statusCode: z.number(),
    url: z.string(),
  }),

  execute: async ({ url, extractJson }) => {
    const zenrowsUrl = new URL('https://api.zenrows.com/v1/');

    zenrowsUrl.searchParams.set('url', url);
    zenrowsUrl.searchParams.set(
      'apikey',
      process.env.ZENROWS_API_KEY!,
    );
    zenrowsUrl.searchParams.set('mode', 'auto');

    if (extractJson) {
      zenrowsUrl.searchParams.set('extract', 'auto');
    } else {
      zenrowsUrl.searchParams.set('response_type', 'markdown');
    }

    const response = await fetch(zenrowsUrl);

    if (!response.ok) {
      const error = await response.text();

      throw new Error(
        `Zenrows request failed with status ${response.status}: ${error}`,
      );
    }

    if (extractJson) {
      const body = await response.json();

      return {
        content: body.parsed,
        statusCode: response.status,
        url,
      };
    }

    const content = await response.text();

    return {
      content,
      statusCode: response.status,
      url,
    };
  },
});

A createTool definition needs five parts. The id names the tool, and the description is the only signal the model uses to decide whether to call it, so keep it brief and exact. The inputSchema validates inputs, the outputSchema types the response, and execute calls Zenrows Fetch with mode=auto.

z.string().url() rejects malformed URLs before they reach the API. extractJson is optional and defaults to undefined, which falls through to the Markdown path.

Test the tool on its own with src/mastra/tools/test-tool.ts. It doesn't need the dev server or the agent.

import 'dotenv/config';
import { zenrowsFetchTool } from './zenrows-fetch';

if (!zenrowsFetchTool.execute) {
  throw new Error('Zenrows tool does not have an execute function');
}

const result = await zenrowsFetchTool.execute(
  {
    url: 'https://www.walmart.com/search?q=bike',
    extractJson: false,
  },
  {} as any
);

if (!result || 'statusCode' in result === false) {
  throw new Error('Zenrows tool returned an unexpected result');
}

console.log(result.statusCode);
console.log(
  typeof result.content === 'string'
    ? result.content.slice(0, 300)
    : result.content
);

Run it.

npx tsx src/mastra/tools/test-tool.ts

The output prints the status code, then the first lines of the page as Markdown.

200
[](https://www.walmart.com)

[Skip to Main Content](https://www.walmart.com#maincontent)

[](https://www.walmart.com/all-departments)[](https://www.walmart.com/)

[![Walmart](https://i5.walmartimages.com/dfw/63fd9f59-14e2/9d304ce6-96de-4331-b8ec-c5191226d378/v1/spark-icon.svg)](https://www.walmart.com/)

If you'd prefer to skip maintaining a custom tool, the Zenrows MCP server is an alternative integration. This walkthrough of the Zenrows MCP server and its implementation covers the setup.

3. Add structured extraction

// part of the previous script
zenrowsUrl.searchParams.set('mode', 'auto');

if (extractJson) {
  // server side parsing, returns fields instead of page text
  zenrowsUrl.searchParams.set('extract', 'auto');
} else {
  zenrowsUrl.searchParams.set('response_type', 'markdown');
}

// parses as JSON
if (extractJson) {
  const body = await response.json();

  return {
    content: body.parsed,
    statusCode: response.status,
    url,
  };
}

const content = await response.text();

return {
  content,
  statusCode: response.status,
  url,
};

These two excerpts from the tool handle the JSON path. When extractJson is true, the request sends extract=auto, and Zenrows Extract parses the page server side into named fields. The response branch then reads the body as JSON and returns body.parsed. When extractJson is false, the tool requests Markdown and reads the body as text.

Markdown suits pages the agent needs to read and reason over. Product listings work better as fields. Test the JSON path with src/mastra/tools/test-tool-extend-extract.ts.

import 'dotenv/config';
import { zenrowsFetchTool } from './zenrows-fetch';

if (!zenrowsFetchTool.execute) {
  throw new Error('Zenrows tool does not have an execute function');
}

const result = await zenrowsFetchTool.execute(
  {
    url: 'https://www.walmart.com/search?q=bike',
    extractJson: true,
  },
  {} as any,
);

if (!result || 'statusCode' in result === false) {
  throw new Error('Zenrows tool returned an unexpected result');
}

console.log(result.statusCode);
console.log(
  typeof result.content === 'string'
    ? result.content.slice(0, 300)
    : result.content
);

npx tsx src/mastra/tools/test-tool-extend-extract.ts

Here's the parsed response, truncated to two product records.

{
  "products": [
    {
      "badge": null,
      "current_price": 149.99,
      "fulfillment_label": null,
      "is_sponsored": true,
      "item_id": "10479873441",
      "low_stock": null,
      "options_from_price": 149.99,
      "original_price": 207.99,
      "rating": 3.9,
      "review_count": 130,
      "title": "Ktaxon 20\" Mountain Bike, 7 Speed Bike with Disc Brakes, White",
      "walmart_plus_savings": null
    },
    {
      "badge": null,
      "current_price": 227,
      "fulfillment_label": null,
      "is_sponsored": true,
      "item_id": "475992717",
      "low_stock": null,
      "options_from_price": null,
      "original_price": null,
      "rating": 4.4,
      "review_count": 739,
      "title": "Mongoose Rebel X1 BMX Bike, 20-in. Wheels, Kids Ages 7-14 Years, Gray Child Bicycle",
      "walmart_plus_savings": null
    }
  ],
  "search_query": {
    "query": "bike",
    "total_results": null
  }
}

4. Wire the tool into a Mastra agent

Create the agent at src/mastra/agents/market-research-agent.ts.

import { Agent } from '@mastra/core/agent';
import { zenrowsFetchTool } from '../tools/zenrows-fetch';

export const marketResearchAgent = new Agent({
  id: 'market-research-agent',
  name: 'Market Research Agent',

  description:
    'An agent that researches and compares products using live web data.',

  instructions: `
You are a product market research agent.

Your job is to research and compare products using the zenrows_fetch tool.

When the user provides one or more product URLs:

1. Fetch each URL using zenrows_fetch.
2. When product information such as name, price, availability, or rating is needed, use structured extraction.
3. Extract the relevant product information from each result.
4. Never invent or assume information that was not returned by the tool.
5. If information is missing, clearly state that it is unavailable.
6. Preserve the currency returned by the source.
7. Do not directly compare prices when the products use different currencies unless a reliable currency conversion is available.
8. When comparing products, present the results clearly in a table when appropriate.
9. After presenting the data, provide useful observations based only on the retrieved information.

Do not use web search when the user asks you to use zenrows_fetch only.
`,

  model: 'openai/gpt-5.6-terra',

  tools: {
    zenrows_fetch: zenrowsFetchTool,
  },
});

The instructions need to be specific. A vague prompt gives you a confident answer with a price from training data and no way to tell which fields came from the page. These instructions forbid invented values, require the agent to flag missing data, and tell it when to use structured extraction.

You can switch to any supported model by changing the model line. Make sure the matching key is in your .env file.

// changing model. Replace it in the previous script
// openai
model: 'openai/gpt-5.6-terra',

// anthropic
model: 'anthropic/claude-sonnet-4-5',

import { Mastra } from '@mastra/core/mastra';
import { LibSQLStore } from '@mastra/libsql';
import { DuckDBStore } from '@mastra/duckdb';
import { MastraCompositeStore } from '@mastra/core/storage';
import {
  MastraStorageExporter,
  MastraPlatformExporter,
  Observability,
  SensitiveDataFilter,
} from '@mastra/observability';
import { agent } from './agents/agent';
import { marketResearchAgent } from './agents/market-research-agent';
import { startScheduleTool, stopScheduleTool } from './tools/schedule-tools';

export const mastra = new Mastra({
  bundler: {
    externals: ['@duckdb/node-bindings'],
  },
  agents: {
    agent,
    marketResearchAgent,
  },
  tools: { startScheduleTool, stopScheduleTool },
  storage: new MastraCompositeStore({
    id: 'composite-storage',
    default: new LibSQLStore({
      id: 'mastra-storage',
      url: process.env.TURSO_DATABASE_URL || 'file:./mastra.db',
      authToken: process.env.TURSO_AUTH_TOKEN || undefined,
    }),
    domains: {
      observability: await new DuckDBStore().getStore('observability'),
    },
  }),
  observability: new Observability({
    configs: {
      default: {
        serviceName: 'mastra',
        exporters: [new MastraStorageExporter(), new MastraPlatformExporter()],
        spanOutputProcessors: [new SensitiveDataFilter()],
      },
    },
  }),
});

Test the agent with src/mastra/tools/test-agent.ts.

import 'dotenv/config';
import { mastra } from '..';

const agent = mastra.getAgent('marketResearchAgent');

const result = await agent.generate(
  'Compare the bikes on https://www.walmart.com/search?q=bike. Give me the five cheapest with their ratings, and flag any that are sponsored.',
);

console.log(JSON.stringify(result.toolCalls, null, 2));
console.log(result.text);

npx tsx src/mastra/tools/test-agent.ts

The script prints the tool call first. The agent chose structured extraction on its own.

[
  {
    "type": "tool-call",
    "runId": "018d9545-eddd-4fde-8cd1-1884316f4212",
    "from": "AGENT",
    "payload": {
      "toolCallId": "call_3H76SowFUi3SRknDFYXrJu7W",
      "toolName": "zenrows_fetch",
      "args": {
        "url": "https://www.walmart.com/search?q=bike",
        "extractJson": true
      },
      "providerMetadata": {
        "openai": {
          "itemId": "fc_06708f1f9910751e006a84956972bc81a084b7ebc5cc25f70d"
        }
      }
    }
  }
]

Then it prints the agent's answer.

The five lowest-priced bikes in the retrieved Walmart search results are:

| Rank | Bike | Price* | Rating | Reviews | Sponsored |
|---:|---|---:|---:|---:|---|
| 1 | Dynacraft 16 Inch Suspect Boys BMX Bike for Child 5-7 Years | 108 | 4.2/5 | 772 | No sponsorship flag returned |
| 2 | 18" Kent Bicycle Abyss Boy's Freestyle BMX Child Bicycle, Blue | 128 | 4.2/5 | 2,325 | No sponsorship flag returned |
| 3 | 20" Kent Tempest BMX Bicycle, Fits Riders 4'2"-5', Black/Aqua, Child, Unisex | 138 | 4.3/5 | 1,496 | No sponsorship flag returned |
| 4 | Ktaxon 24" Women's 7-Speed Cruiser Bike with Basket & Rack, Green | 189.99 | 4.2/5 | 78 | **Sponsored** |
| 5 | Mongoose Rebel X1 BMX Bike, 20-in. Wheels, Kids Ages 7-14, Gray | 227 | 4.4/5 | 740 | **Sponsored** |

The agent ranked the five cheapest bikes from the live page. Where is_sponsored came back null, it said so and didn't guess. This run happened at a different time from the earlier tool test, so some values differ.

5. Test the agent in Mastra Studio

npm run dev

Open http://localhost:4111, select the market research agent, and send it this prompt.

Compare the bikes on Walmart. Give me the five cheapest with their ratings, and flag any that are sponsored.

Studio shows the zenrows_fetch call the agent chose and the ranked table it built from the returned fields.

To add memory, follow the steps in the Mastra memory documentation.

6. Add a workflow for bulk URLs

Create src/mastra/workflow/bulk-fetch.ts.

import { createStep, createWorkflow } from '@mastra/core/workflows';
import { z } from 'zod';
import { zenrowsFetchTool } from '../tools/zenrows-fetch';

const prepareUrlsStep = createStep({
  id: 'prepare-urls',

  inputSchema: z.object({
    urls: z.array(z.string().url()),
  }),

  outputSchema: z.array(
    z.object({
      url: z.string().url(),
    }),
  ),

  execute: async ({ inputData }) => {
    return inputData.urls.map(url => ({
      url,
    }));
  },
});

const fetchProductStep = createStep({
  id: 'fetch-product',

  inputSchema: z.object({
    url: z.string().url(),
  }),

  outputSchema: z.object({
    url: z.string(),
    content: z.record(z.string(), z.unknown()),
    statusCode: z.number(),
  }),

  execute: async ({ inputData }) => {
    // createTool types execute as optional, so narrow it before calling
    if (!zenrowsFetchTool.execute) {
      throw new Error('Zenrows tool does not have an execute function');
    }

    const result = await zenrowsFetchTool.execute(
      {
        url: inputData.url,
        extractJson: true,
      },
      {} as any,
    );

    if (!result || 'statusCode' in result === false) {
      throw new Error('Zenrows tool returned an unexpected result');
    }

    return {
      url: inputData.url,
      // the tool returns a union, this workflow always extracts
      content: result.content as Record<string, unknown>,
      statusCode: result.statusCode,
    };
  },
});

export const productResearchWorkflow = createWorkflow({
  id: 'product-research-workflow',

  inputSchema: z.object({
    urls: z.array(z.string().url()),
  }),

  outputSchema: z.array(
    z.object({
      url: z.string(),
      content: z.record(z.string(), z.unknown()),
      statusCode: z.number(),
    }),
  ),
})
  .then(prepareUrlsStep)
  // concurrency controls how many urls run at once
  .foreach(fetchProductStep, { concurrency: 5 })
  .commit();

Mastra workflows are deterministic, so they suit a known list of URLs. prepare-urls turns the input list into one task per URL, and foreach runs fetch-product on five URLs at a time with structured extraction on.

createTool types execute as optional and as taking two arguments. The step narrows it before the call and passes both arguments, because calling it with the input alone doesn't compile.

// add to the imports
import { productResearchWorkflow } from './workflow/bulk-fetch';

// add to the Mastra constructor, alongside agents
workflows: {
  productResearchWorkflow,
},

Test it with src/mastra/workflow/test-workflow.ts.

import 'dotenv/config';
import { mastra } from '..';

const workflow = mastra.getWorkflow('productResearchWorkflow');

const run = await workflow.createRun();

const result = await run.start({
  inputData: {
    urls: [
      'https://www.walmart.com/search?q=tanktop',
      'https://www.walmart.com/search?q=shirts',
      'https://www.walmart.com/browse/home/shop-water-bottles/4044_623679_639999_7751805_4269055_2976621',
    ],
  },
});

console.log(JSON.stringify(result, null, 2));

npx tsx src/mastra/workflow/test-workflow.ts

The truncated output shows two of the three results. Every URL returned status 200, and the results come back in the same order as the input.

{
  "result": [
    {
      "url": "https://www.walmart.com/search?q=tanktop",
      "content": {
        "products": [
          {
            "badge": "rollback",
            "current_price": 11.98,
            "original_price": 29.98,
            "fulfillment_label": "Shipping, arrives today",
            "is_sponsored": true,
            "item_id": "342153793",
            "low_stock": true,
            "rating": 4.5,
            "review_count": 15482,
            "title": null
          }
          // 8 more products
        ],
        "search_query": { "query": "tanktop", "total_results": null }
      },
      "statusCode": 200
    },
    {
      "url": "https://www.walmart.com/browse/home/shop-water-bottles/...",
      "content": {
        "products": [
          {
            "badge": null,
            "current_price": 15.95,
            "original_price": 17.98,
            "is_sponsored": null,
            "item_id": "1604723",
            "rating": 4.7,
            "review_count": 19927,
            "title": "Brita Standard Replacement Water Filter (3 Pack)"
          }
          // 11 more products
        ],
        "search_query": null
      },
      "statusCode": 200
    }
  ]
}

You can also run the workflow from Mastra Studio.

Wrapping up

You now have a Mastra agent backed by Zenrows that fetches data from protected sites and returns structured results. The Mastra documentation covers more advanced patterns, and the Zenrows MCP server gives you a different integration path.

What's next

Add memory so the agent remembers earlier comparisons.
Move large URL lists to Zenrows Batch so the agent stops firing every request at once. The Batch documentation walks through your first job.
See the same fetch pattern in Python with this guide to an AI lead generation agent with the OpenAI Agents SDK.

FAQ

Why does Mastra's built-in webFetchTool fail on some sites?

It sends HTTP requests without a browser and without anti-bot handling. Cloudflare-protected targets return a challenge page or block page and raise no exception, so the agent reasons over the challenge copy or empty markup and returns an incomplete answer.

Does the Zenrows tool work with both OpenAI and Anthropic models?

It does. Only the agent's model line changes when you switch providers. The tool definition stays the same because Mastra passes your Zod inputSchema to the AI SDK in whatever tool-calling format the provider expects.

How do I handle rate limits when the agent calls the tool frequently?

Handle 429 responses by limiting concurrency, queueing requests, and retrying with exponential backoff and jitter. For large workloads, send the URLs through Zenrows Batch so the agent doesn't fire every request at once.

Can I connect Zenrows through MCP?

You can. Zenrows has an official MCP server, and Mastra's MCPClient works with it like any other MCP server. It supports Streamable HTTP and local transport and gives agents Fetch and browser tools without a wrapper. You still need an API key.

How to run large-scale web scraping jobs with Zenrows Batch

Tiioluwani — Fri, 02 Oct 2026 18:12:43 +0000

This article was originally published on the Zenrows blog. Read the original here: https://www.zenrows.com/blog/large-scale-web-scraping-zenrows-batch

This guide shows you how to run a large scraping job as a single managed call with Zenrows Batch, from one Cloudflare-protected page to a 100,000-URL job, without building the queue, retry logic, and delivery pipeline yourself. You need Python 3.9+ and a Zenrows API key.

Every example ran against the live Batch API. Full code is in the GitHub repo.

Before you start

Python 3.9 or later
Zenrows account for your API key
The client.py helper from the repo, which wraps auth, submission, polling, results, reruns, webhooks, HMAC, CSV uploads, and scheduling

Why concurrency alone doesn't scale

A scraper that works on 500 URLs breaks at 50,000. The target rate-limits you, 429 fills your logs, open connections eat memory, and failed requests stay failed because nothing retries them.

asyncio, aiohttp, and Scrapy solve concurrency. They don't solve the infrastructure underneath: the queue, retry logic, failure tracking, and result delivery. Every large-scale system ends up needing the same six pieces, and Batch replaces all six with one API call.

Submitting a Cloudflare-protected page

The simplest real job submits one protected page with mode=auto.

# refer to 01_cloudflare_target.py in repo
import client

payload = {
    "type": "regular",
    "status": "closed",
    "zenrows_params": {"mode": "auto"},  # handles anti-bot per target
    "tasks": [
        {"url": "https://www.scrapingcourse.com/cloudflare-challenge", "external_id": "cloudflare-challenge"}
    ],
}

submitted = client.submit_job(payload)
job_id = submitted["job_id"]
print(job_id, submitted["latest_run"]["status"])

With mode=auto, the job used dynamic website support and advanced network access, consuming 25 credits. The content came back with the page's confirmation text, "You bypassed the Cloudflare challenge! :D"

Attaching a webhook

A webhook pushes results to your server the moment a run completes, instead of you polling. Signed webhooks use an HMAC key. Generate one first:

# refer to 05_webhook.py in repo
key = client.rotate_hmac_key()
# store key["secret"] now, it's returned exactly once

Attach it to the submission.

payload = {
    "type": "regular",
    "status": "closed",
    "zenrows_params": {"mode": "auto"},
    "tasks": [
        {"url": "https://www.scrapingcourse.com/ecommerce/product/abominable-hoodie/", "external_id": "good-1"},
    ],
    "webhook": {"url": "https://your-app.com/webhooks/zenrows", "signature": True},
}
submitted = client.submit_job(payload)

Here's the completed run delivery.

{
  "schema": "zenrows.webhook/v1",
  "event_type": "run.completed",
  "job_id": "JOB_ID",
  "stats": {"total": 3, "completed": 3, "successful": 3, "failed": 0, "spend": {"credits": 3, "cost": 0.00038997}}
}

Verifying the webhook signature

Verify the delivery came from Zenrows with the X-Signature header and your HMAC secret.

import hmac
import hashlib
import base64

def verify(raw_body: bytes, t: str, v1: str, secret_b64: str) -> bool:
    secret = base64.b64decode(secret_b64)
    expected = hmac.new(secret, f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)  # constant-time compare

Handling failures and reruns

We submitted five real product pages plus three URLs built to fail, a nonexistent domain, a dead page, and an unroutable IP. Query the results and filter for failures.

# refer to 03_failure_and_rerun.py in repo
results = client.list_results(job_id, status="all")
failed_rows = [r for r in results["results"] if r["status"] == "failed"]
for row in failed_rows:
    print(row["external_id"], row["url"], row["error"])

Each failure carries a real error code.

error.code=RESP002  detail="The requested URL page returned a 404 HTTP Status Code..."
error.code=RESP007  detail="The requested target domain could not be resolved..."
error.code=REQS004  title="No ip-based target URLs are allowed"

Rerun only the failures, and it doesn't charge for what already succeeded.

rerun = client.rerun_job(job_id, status="failed")
print(rerun["retried_tasks"], rerun["inherited_tasks"])
# 3 retried, 5 inherited

A URL that's malformed at the syntax level rejects the entire submission with a 400. Fix it and resubmit the whole batch. Only well-formed URLs that fail at scrape time become individual failed tasks you can rerun.

Submitting 100,000 URLs with a credit ceiling

We submitted 100,000 URLs and watched credit use as it ran. The monitor stopped the job at 300,787 credits after 14,105 completed. Here's the watch loop:

# refer to 10_scale_run.py in repo
import time
import client

job_id = "YOUR_JOB_ID"
credit_ceiling = 300_000

while True:
    job = client.get_job(job_id)
    stats = job["latest_run"]["stats"]
    spend = stats.get("spend", {}).get("credits", 0)

    if spend >= credit_ceiling:      # stop before overspending
        client.stop_job(job_id)
        print(f"stopped at {spend} credits, {stats['completed']}/{stats['total']} completed")
        break

    if job["latest_run"]["status"] in {"completed", "stopped"}:
        break

    time.sleep(5)

CSV upload for large lists

For a list too large to paste inline, upload a CSV and reference it.

# refer to 07_csv_upload.py in repo
slot = client.create_csv_input_slot(fields={"url": "URL", "external_id": "Customer Ref"}, header=True)
client.upload_csv(slot, csv_bytes)

payload = {
    "type": "regular",
    "status": "closed",
    "file_input_id": slot["file_input_id"],  # reference the uploaded file
    "zenrows_params": {"mode": "auto"},
}
submitted = client.submit_job(payload)

CSV uploads cap at 50MB or 100,000 rows.

Scheduling recurring jobs

A scheduled job stores your URLs as a template and creates a new run each firing.

# refer to 08_scheduling.py in repo
payload = {
    "type": "scheduled",
    "status": "closed",
    "schedule": {"rate": {"every": 24, "unit": "hour"}},
    "tasks": [{"url": u, "external_id": f"sched-{i}"} for i, u in enumerate(urls, start=1)],
}
submitted = client.submit_job(payload)

Structured extraction

Batch returns structured JSON with css_extractor passed as a JSON-encoded string, or automatic output with Zenrows Extract and no selector map.

# refer to 04_extraction.py in repo
import json

selectors = {"title": "h1", "price": ".price", "description": ".woocommerce-product-details__short-description"}

payload = {
    "type": "regular",
    "status": "closed",
    "zenrows_params": {
        "mode": "auto",
        "css_extractor": json.dumps(selectors),  # must be JSON-encoded
    },
    "tasks": [
        {"url": "https://www.scrapingcourse.com/ecommerce/product/abominable-hoodie/", "external_id": "product-1"},
    ],
}
submitted = client.submit_job(payload)

What Batch leaves to you

Batch holds no browser session open between requests, so a target that needs login, clicking through steps, or a form submission before the data appears calls for Browser Sessions.

What's next

Web data for LLM fine-tuning for running collected data into a training pipeline
Zenrows Fetch for the single-URL path with the same zenrows_params
Full project: GitHub repo

The best Browserbase alternative in 2026 for protected web access

Praise James — Fri, 02 Oct 2026 10:41:34 +0000

This article was originally published on the Zenrows blog. Read the original here: https://www.zenrows.com/blog/best-browserbase-alternative-for-protected-web-access

We benchmarked Zenrows against Browserbase Fetch on four protected targets plus one unprotected page, 100 requests per target at 2 req/s. Zenrows returned 100 percent. Browserbase Fetch returned 62.25 percent across the protected targets and a 403 on all 100 TripAdvisor requests. This covers the results, the cost math, and how to migrate a basic Fetch request.

Benchmark script and CSV results is on GitHub.

The core difference

Browserbase gives agents a real browser to drive, which is right when an agent has to navigate, click, and hold session state. For plain data extraction, running that browser adds a configuration layer you manage, and billing tracks browser time rather than successful data. A failed session still consumes billable browser hours and proxy bandwidth.

Zenrows Fetch uses mode=auto to pick the configuration each target needs and bills only for successful requests.

The cost math on a 40 percent failure rate

10,000 daily page attempts on a Cloudflare-protected site with 40 percent failing consumes about 5,000 browser-hours and 300 GB of proxy bandwidth per month.

Browserbase Developer ($20/mo) comes to about $4,196/month for browser execution and proxy bandwidth alone
Browserbase Startup ($99/mo) comes to about $3,499/month, same exclusions
Zenrows Scale, billed on success, comes to about $622/month at 7.5 million credits, assuming every successful request needs both JS rendering and premium proxies (25x multiplier)

Browserbase's calculation starts from session time and bandwidth you estimate up front. Zenrows starts from successful requests and the config each one needed.

Benchmark results

We run this benchmark on September 24, 2026, each platform on its default config. Zenrows ran mode=auto; Browserbase ran default Fetch, where proxies are disabled by default.

Target	Protection	Zenrows	Browserbase Fetch
IKEA product page	Cloudflare	100%	100%
TripAdvisor listings	DataDome	100%	0% (403 on all 100)
Walmart product page	PerimeterX	100%	100%
Amazon product page	Akamai	100%	49%
Python 3.14.7 docs	none	100%	100%

Zenrows returned 100 percent across the four protected sites because mode=auto escalated only when a target needed more than the default path. Browserbase Fetch averaged 62.25 percent across the protected targets. The zero on TripAdvisor follows from the default config, which executes no JavaScript and disables proxies. Reaching DataDome-protected TripAdvisor requires browser fingerprinting through the Sessions API, which adds a browser-session layer to a data extraction task.

On response time, Zenrows averaged 5.36s per target against Browserbase's 1.18s. Read that alongside retrieval. Browserbase averaged 0.61s on TripAdvisor while returning 403 on every request, and a faster response only counts if it returns the content.

Migrating a Fetch request

Browserbase Fetch uses a POST to https://api.browserbase.com/v1/fetch with the URL in the JSON body:

# pip install browserbase
import os
from browserbase import Browserbase

bb = Browserbase(api_key=os.environ.get("BROWSERBASE_API_KEY"))

response = bb.fetch_api.create(
    url="https://www.ikea.com/us/en/p/kallax-shelf-unit-white-80275887/"
)
print(f"status code: {response.status_code}")

Zenrows uses a GET to https://api.zenrows.com/v1/ with config as query parameters:

# pip install requests
import os
import requests

API_KEY = os.environ.get("ZENROWS_API_KEY")
TARGET_URL = "https://www.ikea.com/us/en/p/kallax-shelf-unit-white-80275887/"

params = {
    "url": TARGET_URL,
    "apikey": API_KEY,
    "mode": "auto",  # adapts config to the target's protection level
}

response = requests.get("https://api.zenrows.com/v1/", params=params)
print(f"status code: {response.status_code}")

The migration changes the endpoint, auth, HTTP method, and parameters. The target URL and how you handle the returned content stay the same. Zenrows keeps protected retrieval and structured output inside the Fetch workflow, where Browserbase splits them across Fetch and Extract.

Driving a browser when you need one

When a page needs browser interaction, Zenrows Browser Sessions connects over CDP through Playwright.

# pip install playwright
# playwright install chromium
import os
from playwright.sync_api import sync_playwright

API_KEY = os.environ.get("ZENROWS_API_KEY")
connection_url = f"wss://browser.zenrows.com?apikey={API_KEY}"  # remote browser endpoint

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(connection_url)
    page = browser.new_page()
    try:
        page.goto(
            "https://www.ikea.com/us/en/p/kallax-shelf-unit-white-80275887/",
            wait_until="domcontentloaded"
        )
        page.locator("body").wait_for()
        print(f"page title: {page.title()}")
    finally:
        browser.close()  # always close the remote session

When Browserbase is the right call

Choose Browserbase when an agent needs to operate a browser rather than retrieve content.

Building an agent that navigates, authenticates, and fills forms inside sites
You need session replay and visual debugging through Live View and Session Inspector
Existing Playwright automation with a custom proxy stack you want to keep
Workflows that depend on Chrome extensions or persistent browser state

What's next

Best Bright Data alternative for the proxy-network comparison
Best Firecrawl alternative for the AI-native extraction comparison
Benchmark repo: GitHub

The 10 best web scraping tools for data extraction in 2026

Tiioluwani — Fri, 02 Oct 2026 08:16:28 +0000

This article was originally published on the Zenrows blog. Read the original here: https://www.zenrows.com/blog/best-web-scraping-tools

This is a comparison of the 10 best web scraping tools for data extraction in 2026, ranked on how well each one gets past defenses and turns the page into usable data. Pick by three factors, your target sites, your output format, and your maintenance budget.

The three factors that decide the tool

Target sites: If your targets sit behind Cloudflare, DataDome, or Akamai, most open-source frameworks are off the table. They render JavaScript but provide no anti-bot bypass on their own.
Output: Raw HTML you parse yourself, clean Markdown for an AI pipeline, or structured JSON with no selectors. This decides whether you need a framework, an API, or an AI-native extraction layer.
Maintenance: Open-source frameworks license for free and cost engineering hours to maintain. Managed APIs charge for access and absorb the upkeep.

How the 10 compare

Tool	Best for	Anti-bot	Structured JSON	Entry price
Zenrows	Protected sites at scale, RAG/LLM	Yes (99.93%)	Yes (Extract)	$16/mo
Bright Data	Enterprise scale and compliance	Yes	Yes	$499/mo committed
Scrapy	Full control, Python pipelines	No (native)	Manual	Free
Playwright	Browser automation, SPAs	No (native)	No	Free
ScrapingBee	Managed API, moderate protection	Yes	Yes (AI)	$49/mo
ScraperAPI	Pre-built e-commerce/search endpoints	Yes (68.95%)	Select domains	$49/mo
Firecrawl	RAG/LLM, clean Markdown	Limited (33.69%)	Yes (native)	Free / $16/mo
Apify	Pre-built Actor marketplace	Varies by Actor	Yes (via Actors)	Free / $29/mo
Browserless	Managed headless browser	Yes	Selector-based	Free / ~$25/mo
Octoparse	No-code, non-technical	Limited	Template-based	Free / $69/mo

1. Zenrows

Handles access and extraction in one call. Fetch with mode=auto covers Cloudflare, DataDome, Akamai, and PerimeterX, starting cheap and escalating only when a site needs it. Reports 99.93 percent success across supported targets. Extract returns structured JSON with one parameter:

# pip install requests
import os
import requests

params = {
    "url": "https://www.chrono24.com/rolex/index.htm",
    "apikey": os.environ["ZENROWS_API_KEY"],
    "mode": "auto",     # Cloudflare bypass
    "extract": "auto",  # structured JSON, no CSS selectors
}

response = requests.get("https://api.zenrows.com/v1/", params=params)
data = response.json()

A plain request to that page returns 403; with both params it returns clean per-item fields. Fetch also does response_type=markdown for LLM pipelines, and Batch handles up to 100,000 URLs per job. The free tier holds 5,000 credits/month, and paid plans start at $16/month.

Limitation: Credit-based pricing scales with the access features a target needs, so estimate your open/protected mix before choosing a plan.

2. Bright Data

Enterprise scale, 93.14 percent on Proxyway's 2025 benchmark of 15 protected sites, the top score tested. 72M+ residential IPs, SOC 2 Type II and ISO 27001. Free tier and PAYG now cover Web Unlocker, SERP API, and Web Scraper API.

Limitation: KYC verification is mandatory for residential and mobile proxies, one to three business days per reviewers. Committed plans from ~$499/month.

3. Scrapy

Open-source Python standard for custom crawlers. Async architecture, middleware for proxy rotation and retries, item pipelines for cleaning and storage. Fits Python developers with DevOps capacity to manage proxies separately.

Limitation: No built-in rendering, anti-bot, or proxies. On defended sites, total cost of ownership often beats a managed API once proxy fees and hours are counted. Free; Scrapy Cloud from $9/month per unit.

4. Playwright

Microsoft's browser automation for Chromium, Firefox, WebKit. The standard for SPAs and sites needing clicks, scrolls, or form interaction. Renders JavaScript like a real browser.

Limitation: No anti-bot protection alone. Cloudflare, DataDome, and Akamai detect a standard session. Stealth libraries help but plateau short of a managed API on hard targets. Zenrows Browser Sessions gives the same control as a managed service. Free.

5. ScrapingBee

Managed API with JavaScript rendering, CAPTCHA handling, and proxy rotation, plus AI extraction from a plain-English description. Handles lightly to moderately protected sites.

Limitation: Steep credit multipliers, one credit basic, five for JS, 25 for premium proxies, 75 for stealth, so 250,000 advertised credits shrink to ~3,333 requests with stealth on. Proxyway scored it 84.47 percent. From $49/month.

6. ScraperAPI

Structured endpoints for Amazon, Google, Walmart, and eBay returning parsed JSON or CSV. Handles proxy rotation and CAPTCHA automatically, skips setup for those targets.

Limitation: No general-purpose AI extraction, so protected sites outside its list return raw HTML. Proxyway scored it 68.95 percent. Hobby plan from $49/month.

7. Firecrawl

AI-native API converting any URL into clean Markdown or structured data, built around scrape, crawl, map, and extract, with an MCP server alongside. Fits RAG pipelines wanting Markdown by default. The free tier holds 1,000 credits/month.

Limitation: Limited anti-bot coverage, 33.69 percent on Proxyway, last among tested, though built for the long tail rather than hardened targets. Extraction costs ~5x a scrape and credits don't roll over. Paid from $16/month annual.

8. Apify

Full-stack platform with 30,000+ pre-built Actors for Google Maps, LinkedIn, Amazon, plus serverless compute and scheduling. Fits teams wanting a pre-built scraper for a popular target.

Limitation: No general-purpose anti-bot, and quality varies by Actor. Residential proxies billed separately at ~$8/GB. The free tier gives $5/month credit, and paid plans start at $29/month.

9. Browserless

Cloud-managed headless browser via REST and BQL, a GraphQL stealth API. CAPTCHA solving, WebGL randomization, and residential proxies built in, with self-hosted Docker. Fits developers wanting Playwright-compatible infrastructure without their own browser pool.

Limitation: Structured extraction means selectors or DOM queries you write. No prompt-to-schema extraction. The free tier holds 1,000 units/month, and paid plans start at $25/month annual.

10. Octoparse

No-code point-and-click scraper, Windows desktop app, 500+ templates. Click the fields and it infers the logic. Fits marketers and small teams on open, well-structured sites.

Limitation: A visual scraper rather than an anti-bot specialist. Cloudflare, DataDome, and Akamai block it. Free desktop version; cloud plans from $69/month.

Which fits your workload

Protected sites, structured JSON, no selectors: Zenrows, one call for access and extraction, with CLI and MCP support.
RAG or LLM ingestion: Fetch returns Markdown so protected and unprotected sources share one pipeline. Firecrawl works when nothing sits behind Cloudflare, DataDome, or Akamai.
Full control over unprotected targets: Scrapy for the pipeline, Playwright when a page needs rendering.
Pre-built scraper for a popular target: Apify's marketplace.

What's next

How we scraped the 2026 FIFA World Cup data and what shocked us

Goodness E. Eboh — Tue, 08 Sep 2026 14:46:14 +0000

This article was originally published on the Zenrows blog. Read the original here: https://www.zenrows.com/blog/scrape-2026-fifa-world-cup-data-python

We built a Python pipeline over three sources (Sofascore, FIFA, and Wikipedia) to pull the full 2026 World Cup dataset. Each source delivered its data differently. Each needed a different extraction strategy. This is how we handled all three, and what the data revealed when we had it.

Full source on GitHub.

Before you start

Python 3.10 or later
Zenrows account for your API key
playwright, requests, beautifulsoup4, pandas, python-dotenv

pip install requests playwright beautifulsoup4 pandas python-dotenv

Three sources, three access patterns

No single source had everything. FIFA had team-level metrics. Sofascore had player-level statistics. Wikipedia had tournament metadata.

The access patterns were all different across the three sources.

Sofascore delivered data through a JavaScript-rendered JSON endpoint
FIFA delivered it through a backend API behind a per-session JWT
Wikipedia delivered server-rendered HTML

Standard requests handles the last one. It fails silently on the first two, returning an empty shell or a challenge page instead of data.

Retrieving Sofascore data from a JavaScript endpoint

Sofascore is a single-page application. The front end calls an internal API and JavaScript renders the result into the browser. requests alone returns the application shell with no data.

We used Zenrows Fetch with mode=auto. Because the endpoint returns JSON, there was no HTML parsing step. The response was already structured data.

import os
import time
from urllib.parse import urlencode

import requests

ZENROWS_API_KEY = os.getenv("ZENROWS_API_KEY")

class SofascoreClient:
    BASE_URL = "https://www.sofascore.com/api/v1"

    def _get(self, endpoint: str, params: dict | None = None):
        base_url = f"{self.BASE_URL}/{endpoint.lstrip('/')}"
        target_url = f"{base_url}?{urlencode(params)}" if params else base_url

        retries = 3
        for attempt in range(retries):
            try:
                response = requests.get(
                    "https://api.zenrows.com/v1/",
                    params={
                        "apikey": ZENROWS_API_KEY,
                        "url": target_url,
                        "mode": "auto",  # handles JS rendering and proxy escalation per request
                    },
                    timeout=120,
                )
                response.raise_for_status()
                return response.json()
            except (
                requests.exceptions.ReadTimeout,
                requests.exceptions.ConnectionError,
                requests.exceptions.HTTPError,
            ) as e:
                print(f"Attempt {attempt + 1}/{retries} failed: {e}")
                if attempt == retries - 1:
                    raise
                wait = 2 ** attempt
                print(f"Retrying in {wait}s...")
                time.sleep(wait)

    def fetch_statistics(self, group, order, page=1, limit=20, accumulation="total"):
        endpoint = f"unique-tournament/{TOURNAMENT_ID}/season/{SEASON_ID}/statistics"
        params = {
            "group": group,
            "order": order,
            "page": page,
            "limit": limit,
            "accumulation": accumulation,
        }
        return self._get(endpoint, params=params)

Retrieving FIFA data from an authenticated API

FIFA's data came from a backend API called with a JWT bearer token issued for the current browsing session. Every visitor gets a temporary token; the front end attaches it to subsequent API calls. The endpoint couldn't be queried as a public URL.

We used Zenrows Browser Sessions with Playwright to follow the same authenticated flow a normal browser follows and intercepted the API responses.

import os
from playwright.async_api import async_playwright

class FIFAClient:
    def __init__(self):
        self.api_key = os.getenv("ZENROWS_API_KEY")
        self.connection_url = f"wss://browser.zenrows.com?apikey={self.api_key}"
        self.responses = {}

    async def _handle_response(self, response):
        if "gameday-prod.fifa.mangodev.co.uk" not in response.url:
            return
        self.responses[response.url] = await response.json()

    async def fetch_all(self):
        async with async_playwright() as p:
            browser = await p.chromium.connect_over_cdp(self.connection_url)
            context = browser.contexts[0]
            page = await context.new_page()

            page.on("response", self._handle_response)  # intercept all API responses
            await page.goto(self.URL)
            await self._remove_cookie_banner(page)

            for tab in self.TABS:
                await self._click_tab(page, tab)

            self._save_files()

Retrieving Wikipedia tournament metadata

Wikipedia is server-rendered. Fetch retrieved it with no extra capabilities. BeautifulSoup located the stadium summary table by its headers.

from bs4 import BeautifulSoup
import re

def normalize(self):
    with open(self.INPUT_FILE, encoding="utf-8") as f:
        soup = BeautifulSoup(f, "html.parser")

    target_table = None

    for table in soup.find_all("table", class_="wikitable"):
        headers = [th.get_text(" ", strip=True) for th in table.find_all("th")]
        if "Year" in headers and "Host" in headers and "Stadiums" in headers:
            target_table = table
            break

    records = []
    for row in target_table.find("tbody").find_all("tr")[1:]:
        cells = row.find_all(["th", "td"])
        if len(cells) < 4:
            continue
        year_match = re.search(r"\d{4}", cells[0].get_text())
        if not year_match:
            continue
        records.append({
            "year": int(year_match.group()),
            "hosts": [t.strip() for t in cells[1].stripped_strings if t.strip()],
            "cities": int(re.search(r"\d+", cells[2].get_text()).group()),
            "stadiums": int(re.search(r"\d+", cells[3].get_text()).group()),
        })

Normalizing three sources into one format

Three sources fetched three ways produce three shapes of data. Each had its own normalization layer. Sofascore returned duplicate players across statistic groups, so we keyed on player IDs to deduplicate.

def normalize_player(self, record):
    player = record.get("player", {})
    team = record.get("team", {})
    stats = {k: v for k, v in record.items() if k not in {"player", "team", "rating"}}
    return {
        "player_id": player.get("id"),
        "player": player.get("name"),
        "team": team.get("name"),
        "rating": record.get("rating"),
        "stats": stats,
    }

def normalize_dataset(self, dataset):
    with open(f"data/raw/sofascore/{dataset}.json") as f:
        data = json.load(f)

    players = {}
    for record in data:
        normalized = self.normalize_player(record)
        if normalized["player_id"]:
            players[normalized["player_id"]] = normalized  # keep one record per player

    return list(players.values())

What the data revealed

Ten findings from the dataset that challenged what we expected.

Dembélé scored six goals from just over two expected goals, close to three times his xG
The Golden Glove winner (Unai Simón) didn't appear in our top ten goalkeepers by rating
Bellingham, Rodri, and Tchouaméni dominated the defensive action rankings ahead of centre-backs
Japan recorded the highest shot conversion rate while taking the lowest share of shots from inside the box
Brian Brobbey recorded a 75% conversion rate without missing a single big chance
Michael Olise set a new World Cup record with seven assists and led the tournament in big chances created
More pressing volume didn't produce quicker ball recovery. Türkiye applied fewer pressures and recovered faster than France, England, and Argentina
Argentina reached the final with the most bookings and no VAR interventions against them
Spain led the tournament in defensive line breaks while holding around 90% passing accuracy
The 2026 three-country World Cup used 16 stadiums, fewer than Spain used alone in 1982

The full findings with charts are in the original article.

What's next

Zenrows Batch for large-scale collection across thousands of URLs
Zenrows Browser Sessions for any authenticated scraping workflow
Web data for LLM fine-tuning for feeding scraped data into training pipelines

Web data for LLM fine-tuning: a practical 2026 guide

Benny — Tue, 08 Sep 2026 14:01:42 +0000

This article was originally published on the Zenrows blog. Read the original here: https://www.zenrows.com/blog/web-data-llm-fine-tuning

This guide shows you how to collect a real-data seed for LLM fine-tuning from protected web sources using Zenrows Fetch, format it as instruction-output pairs, validate it, and measure whether the fine-tune held. You need Python 3.9+, a Zenrows API key, and requests and python-dotenv installed.

All code is on GitHub.

Before you start

Python 3.9 or later
Zenrows account for your API key
pypdf if you are collecting from PDFs

python3 -m pip install requests python-dotenv pypdf

# .env
ZENROWS_API_KEY=your_zenrows_api_key_here

Why a real-data seed matters

In a fine-tuning set of a few hundred rows, one mislabeled extraction is a measurable percentage of your entire training signal. Synthetic-only datasets amplify teacher model biases through each retrain cycle. The real-data seed anchors the generator and breaks the drift loop.

Authoritative sources — documentation, regulatory filings, expert-reviewed publications — are almost always behind access controls. Standard crawlers get flagged on the TLS handshake. Zenrows Fetch handles rendering and access in a single call.

Step 1: Collect clean Markdown from protected sources

mode=auto is Adaptive Stealth Mode. Zenrows picks the retrieval strategy per target — JavaScript rendering, proxy escalation, or direct fetch — without configuration on your end.

import os
import requests
from dotenv import load_dotenv

load_dotenv()

apikey = os.getenv("ZENROWS_API_KEY")
url = "https://www.scrapingcourse.com/ecommerce/"

params = {
    "url": url,
    "apikey": apikey,
    "mode": "auto",
    "response_type": "markdown",  # clean Markdown, not raw HTML
}

response = requests.get("https://api.zenrows.com/v1/", params=params)
print(response.text)

Step 2: Handle PDFs and non-HTML sources

Use response_type=pdf for PDFs. js_render=true is required for PDF responses.

params = {
    "url": url,
    "apikey": apikey,
    "js_render": "true",    # required for PDF responses
    "response_type": "pdf", # returns raw bytes, not text
}

response = requests.get("https://api.zenrows.com/v1/", params=params)

with open("source-document.pdf", "wb") as f:
    f.write(response.content)

Extract text and chunk by structure, not by character count.

from pypdf import PdfReader
import io

reader = PdfReader(io.BytesIO(response.content))

def chunk_by_page(reader, max_chars=6000):
    chunks = []
    for page in reader.pages:
        text = page.extract_text()
        if not text or not text.strip():
            continue  # skip blank or image-only pages
        if len(text) <= max_chars:
            chunks.append(text)
        else:
            for i in range(0, len(text), max_chars):
                chunks.append(text[i:i + max_chars])
    return chunks

chunks = chunk_by_page(reader)
print(f"Produced {len(chunks)} chunks from {len(reader.pages)} pages")

Step 3: Track provenance

Every row in the seed needs a source, retrieval date, and license.

from datetime import date

source = {
    "url": pdf_url,
    "retrieved": date.today().isoformat(),
    "license": license_url,  # check the source's terms before collecting at scale
}

Step 4: Run batch collection

Match concurrency to your plan limit. time.sleep(0.5) spaces out requests per worker.

import time
from concurrent.futures import ThreadPoolExecutor

MAX_CONCURRENT = 50  # check your Zenrows dashboard for your plan's limit

def fetch(url):
    params = {
        "url": url,
        "apikey": apikey,
        "response_type": "markdown",
        "mode": "auto",
        "wait": 3000,
    }
    response = requests.get("https://api.zenrows.com/v1/", params=params)
    time.sleep(0.5)  # pace requests within each worker

    if response.status_code != 200:
        return {"url": url, "status": response.status_code, "content": None}
    return {"url": url, "status": response.status_code, "content": response.text}

urls = [
    "https://playboard.co",
    "https://www.samsung.com",
    "https://www.kickstarter.com",
]

with ThreadPoolExecutor(max_workers=MAX_CONCURRENT) as executor:
    results = list(executor.map(fetch, urls))

Step 5: Format as instruction-output pairs

The Markdown is the input. The extracted JSON is the output. Write one JSON object per line in JSONL format.

import json

def to_training_example(instruction, markdown_content, extracted_json):
    return {
        "messages": [
            {"role": "system", "content": instruction},
            {"role": "user", "content": markdown_content},
            {"role": "assistant", "content": json.dumps(extracted_json)},
        ]
    }

instruction = "Extract the product name, price, and availability as JSON."

with open("training_data.jsonl", "w") as f:
    for markdown_content, extracted_json in seed_examples:
        example = to_training_example(instruction, markdown_content, extracted_json)
        f.write(json.dumps(example) + "\n")

Step 6: Validate before training

Every row must clear three checks before it enters the training set — length, schema, and duplicate hash.

import hashlib
from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string", "minLength": 1},
        "price": {"type": "string", "minLength": 1},
    },
    "required": ["name", "price"],
}

seen = set()

def keep(markdown, extracted):
    if len(markdown.strip()) < 200:
        return False  # page rendered to almost nothing
    try:
        validate(instance=extracted, schema=schema)
    except ValidationError:
        return False  # wrong types or missing fields
    h = hashlib.sha256(" ".join(markdown.lower().split()).encode()).hexdigest()
    if h in seen:
        return False  # duplicate or near-duplicate
    seen.add(h)
    return True

clean_examples = [(url, md, js) for url, md, js in seed_examples if keep(md, js)]
print(f"Collected {len(seed_examples)} pages, {len(clean_examples)} passed validation")

Step 7: Hold out real data and measure

Set aside 10 to 20 percent before any synthetic expansion and keep it out of training permanently.

import random

random.seed(42)
random.shuffle(seed_examples)

split = int(len(seed_examples) * 0.85)
train_examples = seed_examples[:split]
holdout_examples = seed_examples[split:]  # never touches the training set

After fine-tuning, run the model against the holdout and measure task accuracy. Record the score after each retrain. A declining score is an early sign of drift.

What's next

How to extract web data for AI training — broader context on training data collection
Zenrows Batch — managed concurrency and retries for large URL lists
Zenrows MCP server — web access for agents without managing the collection layer