How to Build a Price Monitoring Agent with Pydantic AI and ZenRows
Introduction
Most price monitoring systems look simple on paper: point a scraper at a product page, ask an LLM to extract the price, and store the result in a database. The problem arises when the same workflow runs continuously against heavily protected targets such as Amazon and Walmart. At scale, reliability becomes the real challenge.
The why comes down to two failures that happen quietly: retrieval and extraction. Retrieval fails when a protected site returns a blank JavaScript shell, incomplete data, and a block page instead of the product data. Extraction fails when the LLM returns a field in a data type or shape that your downstream database write action does not expect. Both failures pass for normal output, so the pipeline keeps running and drops records downstream without raising an error.
This tutorial shows you how you can build a price monitoring agent using ZenRows and Pydantic AI. ZenRows retrieves the page and returns product details with a 99.93% success rate against protected websites, while Pydantic AI extracts and validates LLM output against a defined schema. The tutorial walks through the two-step pipeline from a single product page to a multi-site price monitoring run across Amazon and Walmart, two heavily protected e-commerce sites.
The two failures in a price monitoring pipeline
Two failure modes break the price monitoring pipeline: retrieval and extraction.
1. Retrieval returns a block page
You send a request to the protected site, and the response looks successful at the HTTP level. But the page still contains no usable product data or encounters an error 1020. The code below shows this failed retrieval directly.
import requests
# Demo page that behaves like a protected target
PROTECTED_URL = "https://www.scrapingcourse.com/antibot-challenge"
resp = requests.get(PROTECTED_URL, timeout=30)
print(f"status code: {resp.status_code}")
print(f"bytes returned: {len(resp.text)}")
print("contains 'price'?", "price" in resp.text.lower())
print(resp.text[:200])
The request returns a 403, so the pipeline does not receive the product page required for price monitoring.
benny@Mac price_monitoring % python3 test.py
status code: 403
bytes returned: 5507
contains 'price'? False
<!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 nam
benny@Mac price_monitoring %
When this happens, your pipeline would assume that your script has received data and move on. But without any prices, you can't monitor the pages for any product.
2. Extraction returns inconsistent information
Your first run can return data successfully. Across runs, the same field can drift into different formats. For example, one run may return a price with currency symbols attached. Another may return the value as text, and the third as a number. The example below demonstrates how an LLM behaves when this type of data drift occurs.
# Simulated raw LLM extraction responses from two runs
raw_llm_responses = [
{ "product": "Echo Dot (5th Gen)", "price": 29.99 },
{ "product": "Echo Dot (5th Gen)", "price": "$29.99" }
]
def write_to_db(price: float) -> None:
if not isinstance(price, float):
raise TypeError(f"expected float, got {type(price).__name__}")
for i, response in enumerate(raw_llm_responses, start=1):
price = response["price"]
print(f"run {i}: price={price!r}, "
f"type={type(price).__name__}")
for i, response in enumerate(raw_llm_responses, start=1):
try:
write_to_db(response["price"])
print(f"run {i}: write OK")
except TypeError as e:
print(f"run {i}: write FAILED -> {e}")
Below is the output of the extraction script. It returns valid-looking data, but the output schema can drift between runs. Without validation, downstream writes fail when the data type does not match the expected format.
benny@Mac price_monitoring % python3 test_llm.py
run 1: price=29.99, type=float
run 2: price='$29.99', type=str
run 1: write OK
run 2: write FAILED -> expected float, got str
benny@Mac price_monitoring %
Such inconsistencies can break validation logic and downstream workflows, causing pipelines to fail unpredictably during automated or scheduled executions.
Next, let's take a look at how ZenRows handles the retrieval problem.
Prerequisites for this tutorial
Use Python 3.10 or newer for this tutorial. If you have an older Python version, you need to create a dedicated virtual environment with a current Python installation to avoid dependency conflicts.
Install the required packages using:
python -m pip install "pydantic-ai-slim[anthropic]" requests python-dotenv
This tutorial was tested locally with Pydantic AI 0.8.1 and Anthropic SDK 0.111.0.
- ZenRows API key. Create an account at zenrows.com and copy your key from the dashboard. This key authenticates every scrape the researcher runs.
- Anthropic API key. The extraction agent uses Claude to read the scraped markdown and return a validated record. Generate a key in the Anthropic Console.
Create a .env file and save your API keys there.
Setting up ZenRows for e-commerce retrieval
One way to address retrieval failures is to use a scraping API to retrieve data from protected pages and pass the page content to the extraction step. ZenRows web scraping API handles the retrieval layer for protected pages. It renders the pages, uses proxies, and returns markdown in a single request. You can then pass this output to your extraction agent for the next stage of your pipeline, which you will see in the next section of this piece. The code below shows this flow with js_render and premium_proxy.
import os
import requests
from dotenv import load_dotenv
load_dotenv()
ZENROWS_API_KEY = os.environ["ZENROWS_API_KEY"]
# Target page (NO markdown formatting)
url = "https://www.scrapingcourse.com/ecommerce/"
params = {
"url": url,
"apikey": ZENROWS_API_KEY,
"js_render": "true",
"premium_proxy": "true",
"proxy_country": "us",
"response_type": "markdown",
}
response = requests.get("https://api.zenrows.com/v1/", params=params, timeout=60)
print(f"status code: {response.status_code}")
print(f"bytes returned: {len(response.text)}")
print("\n".join(response.text.splitlines()[:30]))
ZenRows manages anti-bot bypass via "js_render": "true" which ensures ZenRows loads the page like a real user browser, and "premium_proxy": "true" which enables residential proxies. For this use case, set proxy_country=us to route requests through a US IP address. That keeps our price comparisons consistent across runs. The response type is markdown, which is easier for LLMs to parse and read.
Your response after running the code should be similar to the output below.
benny@Mac price_monitoring % python3 zenrows_store.py
status code: 200
bytes returned: 6777
[Skip to navigation](http://www.scrapingcourse.com#site-navigation)
[ Skip to content](http://www.scrapingcourse.com#content)
[Ecommerce Test Site to Learn Web Scraping](https://www.scrapingcourse.com/ecommerce/)
ScrapingCourse.com Search for: Search Menu - [Shop](https://www.scrapingcourse.com/ecommerce/)<!--THE END-->
- [Home](https://www.scrapingcourse.com/ecommerce/)
- [Cart](https://www.scrapingcourse.com/ecommerce/cart/)
- [Checkout](https://www.scrapingcourse.com/ecommerce/checkout/)
- [My account](https://www.scrapingcourse.com/ecommerce/my-account/)<!--THE END-->
- [$ 0.00 0 items](https://www.scrapingcourse.com/ecommerce/cart/ "View your shopping cart")
- No products in the cart.
# Shop Default sorting Sort by popularity Sort by latest Sort by price: low to high Sort by price: high to low Showing 1-16 of 188 results
benny@Mac price_monitoring %
Now, let's define your price schema to ensure your LLM reads the extracted content correctly and maps each product field into a predictable format for downstream processing.
Defining the price schema and building the extraction agent
Now that ZenRows has retrieved the data from the website, Pydantic AI turns the markdown into a validated price record. You define the schema first, then run the agent against it. For a price monitoring agent, Pydantic protects your pipeline from messy extraction output.
You will start by defining your schema, which acts as a contract between the extraction layer and the rest of your pipeline. In the code below, that is PriceRecord. After that, you build the agent against the schema. So if the field has the wrong type, Pydantic AI reprompts the model with the validation error and returns the right output.
from datetime import datetime, timezone
from pydantic import BaseModel, Field
from dotenv import load_dotenv
load_dotenv()
class PriceRecord(BaseModel):
name: str
price: float
currency: str
marketplace: str
availability: str
scraped_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc)
)
from pydantic_ai import Agent
price_agent = Agent(
"anthropic:claude-sonnet-4-6",
output_type=PriceRecord,
instructions=(
"Extract the product price record from the Markdown. "
"price is a number with no currency symbol. "
"currency is the ISO code, for example USD. "
"marketplace is the site name, for example Amazon."
),
)
Now, let's run it against our ZenRows output.
import os
import requests
from datetime import datetime, timezone
from typing import Optional
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from pydantic_ai import Agent
load_dotenv()
ZENROWS_API_KEY = os.environ["ZENROWS_API_KEY"]
PRODUCT_URL = "https://www.scrapingcourse.com/ecommerce/"
# --- Step 1 retrieve with ZenRows ---
def fetch_markdown(url: str) -> Optional[str]:
params = {
"url": url,
"apikey": ZENROWS_API_KEY,
"js_render": "true",
"premium_proxy": "true",
"proxy_country": "us",
"response_type": "markdown",
}
resp = requests.get(
"https://api.zenrows.com/v1/",
params=params,
timeout=60,
)
if resp.status_code != 200:
print(f"retrieval blocked: {resp.status_code}")
return None
return resp.text
# --- Step 2 schema + agent ---
class PriceRecord(BaseModel):
name: str
price: float
currency: str
marketplace: str
availability: str
scraped_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc)
)
price_agent = Agent(
"anthropic:claude-sonnet-4-6",
output_type=PriceRecord,
instructions=(
"Extract the product price record from the Markdown. "
"price is a number with no currency symbol. "
"currency is the ISO code, for example USD. "
"marketplace is the site name, for example ScrapingCourse."
),
)
# --- Step 3 run pipeline ---
markdown = fetch_markdown(PRODUCT_URL)
if markdown is None:
raise Exception("No markdown returned from ZenRows")
print(f"markdown length: {len(markdown)}")
result = price_agent.run_sync(markdown)
record = result.output
print("\n--- extracted record ---")
print(record.model_dump())
print("\n--- type checks ---")
print("price is float?", isinstance(record.price, float), "->", repr(record.price))
print("name is str?", isinstance(record.name, str))
print("scraped_at set?", record.scraped_at.isoformat())
From the output below, you can see that the ZenRows returned page content instead of a block or challenge response, hence the markdown length of 6750. The LLM also converted the messy markdown into your expected fields and shows the schema validation.
(venv) benny@Mac price_monitoring % python zenrows_pydantic.py
markdown length: 6750
--- extracted record ---
{'name': 'Abominable Hoodie', 'price': 69.0, 'currency': 'USD', 'marketplace': 'ScrapingCourse', 'availability': 'In Stock', 'scraped_at': datetime.datetime(2026, 6, 22, 23, 5, 8, 154880, tzinfo=datetime.timezone.utc)}
--- type checks ---
price is float?True -> 69.0
name is str? True
scraped_at set?2026-06-22T23:05:08.154880+00:00
(venv) benny@Mac price_monitoring %
Running the pipeline across multiple marketplaces
The script below uses the same retrieve-then-extract framework as in the previous two sections. However, now it is a loop over multiple URLs, collecting a validated PriceRecord from each site. To demonstrate multi-marketplace price monitoring and ZenRows' capability, this tutorial runs the workflow against Amazon and Walmart. Amazon and Walmart are among the most aggressively protected e-commerce sites on the web, making them a meaningful test of retrieval reliability. Successfully extracting product data from these targets demonstrates ZenRows' ability to handle real-world protected websites, not just scraper-friendly demo pages.
import os
import requests
from datetime import datetime, timezone
from typing import Optional
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from pydantic_ai import Agent, UnexpectedModelBehavior
load_dotenv()
ZENROWS_API_KEY = os.getenv("ZENROWS_API_KEY", "")
MIN_MARKDOWN_LENGTH = 200 # a real product page dwarfs any block or error blob
# The same product across three marketplaces, one product page per URL,
# all on US domains so the prices share a currency.
TARGETS = [
("Amazon", "https://www.amazon.com/Apple-iPhone-Version-256GB-Titanium/dp/B0DHJDPYYR"),
("Walmart", "https://www.walmart.com/ip/Restored-Apple-iPhone-16-Pro-Carrier-Unlocked-256GB-Black-Titanium-Refurbished/13323059232"),
]
# --- Step 1: retrieve with ZenRows ---
def fetch_markdown(url: str) -> Optional[str]:
params = {
"url": url,
"apikey": ZENROWS_API_KEY,
"js_render": "true",
"premium_proxy": "true",
"proxy_country": "us",
"response_type": "markdown"
Comments
No comments yet. Start the discussion.