Dev.to AI 🤖 Ai Tue, 03 Nov 2026 20:03:23 +0000 👁 0 📖 12 min read

Build a Live Web Data Agent With Agno and Zenrows

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 AP

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.

📰 Read the original article on Dev.to AI

Originally published by Dev.to AI. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.

← News 🤖 Ai 🏠 Home