Build a Live Web Data Agent With Agno and Zenrows

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

.venv\Scripts\Activate.ps1 

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

[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]" 

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 

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 

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, } 

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 __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) 

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 

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

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" +Source：https://dev.to/zenrows/build-a-live-web-data-agent-with-agno-and-zenrows-2h7k
My favorites