DEV Community: Benny

Going outside was never my problem. Saying hello was.

Benny — Sun, 11 Oct 2026 12:14:05 +0000

This is a submission for the Hacktoberfest Open-Source AI Challenge Week 1: Touch Grass

What I Built

I'm an introvert by nature but an extrovert by society. In a meeting or anything official, I'm fine, mainly because there's a reason to talk. What I struggle with is talking to people when I don't have to, or what we call small talk, especially with someone I don't necessarily need to talk to.

So when the theme was "Touch Grass", I didn't want another app that tells you to go for a walk. Going outside isn't the hard part for introverts. Talking to the people out there is. Many social apps help you talk to people, but almost all of them keep you talking through a screen. This app is the opposite. It gets you talking to the world around you: the people you already pass every day and never really speak to. It’s about realizing that every random passerby is living a life just as vivid, complex, and emotionally rich as your own. That’s the idea behind the app.

Sonder is a weekly game for that. It makes it easier to connect with the people around you, without the pressure of walking up to a stranger and figuring out what to say. Because sometimes, all it takes to turn a stranger into a friend is a little nudge. Every week you get a few quests on sticky notes set in places you actually go. It comes with two difficulty levels (easy, fine, and scary)

You do the quest out in the world, come back, and write what you found out. Gemma replies like a friend, and your notes build into a wall of people who aren't strangers anymore. It's for anyone who walks past the same faces every day and never says more than "good morning".

How it works

Tell it your normal week. Pick the places you actually go (market, office, gym, bus stop, church or mosque, beach and more), or add your own.
Pick your challenge. Easy, Fine or Scary (see below).
Get your week. Gemma writes a mix of quests for your places. Each has a one-line opener you can actually say.
Go and do them. Write down what you found out. Gemma replies to what you learned, not to the quest.
Watch your wall grow. The People tab keeps every note, week by week.

Settings let you change places, level, and quests per week at any time. New quests swaps the ones you haven't started and keeps the ones you've done.

Three levels

Level	What a quest looks like	Example
Easy	One quick question, under a minute	Ask someone at the bus stop which bus goes somewhere you know
Fine	A light chat with one simple follow-up	Ask a regular at the gym to show you one exercise they swear by
Scary	A real conversation	Ask an older trader at the market what has changed most over the years

Every week mixes quest types, so no two weeks feel the same. There are 14 types, from Learn a name and Ask the way up to Origin story and Return visit (going back to someone you met before).

A real quest, start to finish

Gemma generates the quests:

Quests are displayed.

Give feedback on the quest. What did you learn?

See all the strangers you have interacted with:

"Evening, because he tried morning for a week and was sleepy at work." That's one person at my gym who isn't a stranger anymore.

Not feeling it? Skip with a reason. "Didn't feel safe" switches you to Play it safe automatically.

Or change the settings.

Demo

Sonder runs on my laptop, so there's no public link. It also stores notes about real people you've met, so it runs on your own laptop instead of a server. To try it, clone the repo and follow the three steps in the README. It takes about five minutes if you have Ollama installed.

The video below shows how it works on my MacBook.

I took it outside

My first quest was an Easy one: ask your barber which style everyone is asking for right now. So I went for a haircut and asked Tobi, my barber. I've been going to him for a while, and I'd never really asked him anything beyond "the usual". His answer: everyone wants to dye their hair. 😂

That turned into a whole conversation. He reminded me that I'd talked about dyeing my hair when I came in a month ago, and dragged me for never coming back to do it.

I wrote it down in Sonder, and Gemma replied:

"Seriously? Everyone's going for a dye job? That's wild, what color were people asking for specifically?"

Which is fair. I didn't ask. That's the follow-up for my next haircut. Sonder has a gallery of all your conversations.

Code

Bennykillua / sonder

A weekly game that gets you talking to the people you walk past every day. Small quests on sticky notes, written by Gemma running locally. Your notes never leave your laptop.

Sonder

Going outside was never the hard part. Saying hello was.

sonder (n.): the realisation that every stranger you pass has a life as vivid and complex as your own.

Sonder is a weekly game that gets you talking to the people you walk past every day. You get a few small quests on sticky notes, set in the places you actually go. You do them out in the world, write down what you found out, and Gemma replies like a friend would. Over time, strangers turn into names.

It runs entirely on your own machine, with Gemma running locally through Ollama.

The problem

I'm an introvert. In a meeting or anything official, I'm fine, because there's a reason to talk. Talking to people when I don't have to is the hard part.

Most "touch grass" apps tell you to go outside. Going outside was never my problem. Sonder is…

View on GitHub

MIT-licensed and built during the challenge. It's one Python file for the backend. The README explains how to run it locally with Ollama in three commands.

How I Built It

The stack is small on purpose:

Gemma, running locally through Ollama
FastAPI and SQLite on my laptop
One HTML page for the whole interface, opened on my phone. I made it whimsical and delightful

The interesting part wasn't the stack. It was getting a small model to write quests worth doing.

What Gemma does

Writes your week. It turns your places and your challenge level into a set of quests, each with an opener you can actually say and, for Fine and Scary, a follow-up question.
Replies to what you found out. When you finish a quest and write a note, Gemma responds to the person you met, like a friend would over text.

How the app is built around a small local model

Small models are fast and private, but they copy examples and drift off-brief. So the app gives Gemma structure instead of freedom:

Each week is planned before Gemma writes anything. The app splits the week into slots, and each slot gets a quest type (14 types, from "Learn a name" to "Origin story") and one of your places. Gemma writes one quest per slot, so the week stays varied and tied to where you actually go.
Gemma returns structured JSON. Ollama's structured output means every quest arrives with the same fields: task, opener, follow-up, and place.
Every quest is checked in code before you see it. It's dropped if it's a label instead of a task, small talk, a copy of its own brief, too long to say out loud, set somewhere you didn't pick, or a repeat.
Safety rules are enforced in code, not just in the prompt:
- never asks about marriage, tribe, politics, money, or age, and never asks for phone numbers or photos
- no faith questions at all
- hospital quests are staff-only
- no alcohol at bars or drink shops
- a "Play it safe" mode keeps everything in shops and queues
83 backup quests fill any gaps, so a bad run from the model never leaves you with an empty week.
Weeks are written in the background. The phone checks back every few seconds, so a slow model never freezes the app.

Why Does Open Innovation Matter?

Because of what Sonder stores. It's a diary of every person you talked to, what they told you, and which conversations you skipped because they felt awkward or unsafe. I wouldn't send that to a server I don't control, and I don't think anyone else should either.

That's also why there's no hosted version: the moment I host it, your notes live on my server.

With an open model running locally:

Nothing leaves my laptop. Gemma, the quests, and the notes all live on my machine.
It costs nothing to run. That matters in Nigeria, where paying per request adds up fast.
I could choose my own speed and quality trade-off. I tried gemma3:1b, which wrote a week in about 2 seconds but gave plainer quests. I went back to gemma3:4b for better ones, and kept it loaded in memory so it's ready when you open the app. With a closed API, that choice wouldn't be mine.

The safety filters are in my code, and anyone can read them in the repo.

My Agent Session

I don't have a DevRelay session to share. I didn't build Sonder with Claude Code. I used Claude in chat as a pair programmer: it drafted code, helped me plan the app, and helped me fix bugs. I ran everything, tested it on my phone and with various friends, debugged with Claude Code, and decided what stayed.

Prize Categories

Best Use of Gemma

Gemma does two jobs in Sonder. Both run locally through Ollama, using gemma3:4b on my laptop (a MacBook):

The quest writer. Each week, the app gives Gemma one brief per quest: a quest type, one of your places, and your challenge level. Gemma writes the task, a one-line opener you can actually say and, on Fine and Scary, a follow-up question. It returns them as structured JSON, and code checks every quest before you see it.
The friend who replies. When you finish a quest and write what you found out, Gemma responds to the person you met, not the quest, like a friend over text.

Running Gemma locally means your notes about real people never leave your machine, and Sonder costs nothing per request.

Built solo.

If you run Sonder yourself, tell me what features you'd like to see or how I can improve it.

Also, if you have some doubts about giving it a try, think of this viral internet quote: "It’s okay to embarrass yourself a little in the pursuit of human connection," which stemmed from a quote by Bianca Sparacino

In a world that seeks connection, we oddly avoid eye contact, we time our text responses to protect ourselves from seeming too eager or too interested, and we hold our feelings back because we don’t want to seem overly emotional or unreasonable. We silence our instincts, and at the end of the day instead of feeling good about ourselves, we feel alone, we feel misunderstood. Remember - it is okay to be emotional, to seek help, to confidently tell someone you enjoy being around that you are infatuated with them. There is nothing wrong with vulnerability, with being human, for that is what creates depth within our relationships, and that is what ultimately unities us.

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.

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

Build a web-aware agent with OpenAI Agents SDK and Zenrows

Benny — Fri, 04 Sep 2026 07:27:21 +0000

This article was originally published on the Zenrows blog. Read the original here: https://www.zenrows.com/blog/web-aware-agent-openai-agents-sdk-zenrows

This guide shows you how to register Zenrows Fetch as a @function_tool next to WebSearchTool in a single OpenAI Agents SDK agent, so the model routes between discovery and live-page retrieval on its own each turn. You need Python 3.9+, an OpenAI API key, and a Zenrows API key.

All the code is on GitHub.

Before you start

Python 3.9 or later
An OpenAI API key from your OpenAI developer dashboard
A Zenrows API key from your Zenrows dashboard

Install dependencies:

python3 -m pip install openai-agents python-dotenv

Add both keys to a .env file and add it to .gitignore:

OPENAI_API_KEY=your_openai_api_key_here
ZENROWS_API_KEY=your_zenrows_api_key_here

Load them in your script:

from dotenv import load_dotenv

load_dotenv()

Set up a baseline agent with the built-in search

WebSearchTool is a hosted tool that runs on OpenAI's infrastructure. Enabling it takes one line.

import asyncio
from dotenv import load_dotenv
from agents import Agent, Runner, WebSearchTool

load_dotenv()

agent = Agent(
    name="Product Researcher",
    instructions=(
        "You research products and web pages. "
        "Report exactly what you find, and state plainly when "
        "specific information is missing from your results."
    ),
    tools=[WebSearchTool()],
)

async def main():
    result = await Runner.run(
        agent,
        "Which laptops did Apple release most recently, and where can I buy them?",
    )
    print(result.final_output)

asyncio.run(main())

This works for broad discovery queries. The agent returns relevant pages with source links. The gap appears on live or protected targets.

Where the built-in search stops

Issue 1: indexed snippets, not live pages

Built-in search returns OpenAI's cached copy of a page, not the current version. For documentation, that's usually fine. For prices, stock levels, or anything that changes frequently, the cached copy lags.

Issue 2: JavaScript-rendered content

Many pages ship an HTML shell and fill in prices, inventory, and reviews via JavaScript at runtime. The search index stores the shell. The data you want was never captured.

import asyncio
from dotenv import load_dotenv
from agents import Agent, Runner, WebSearchTool

load_dotenv()

agent = Agent(
    name="Product Researcher",
    instructions=(
        "You retrieve product details from web pages. "
        "Report exactly what you find, and say so plainly if content is missing."
    ),
    tools=[WebSearchTool()],
)

async def main():
    result = await Runner.run(
        agent,
        "What is the current price and stock status of "
        "https://www.amazon.com/dp/B0GR1JKMBV/ref=fs_a_mbt2_us1?th=1",
    )
    print(result.final_output)

asyncio.run(main())

The agent finds the page. It returns the product title and a clear statement that the price wasn't accessible:

I visited the Amazon product page for ASIN B0GR1JKMBV...
The page displays: "To see product details, add this item to your cart."
Therefore I could not retrieve the current price or availability.

Add Zenrows Fetch as a @function_tool

import os
import requests
from agents import function_tool

@function_tool
def fetch_page_content(url: str) -> str:
    """Fetch the full, current content of a specific web page as Markdown.

    Use this when you need the complete content of a known URL rather than
    a search result summary, such as live prices, stock levels, or data
    that loads via JavaScript after the page opens.

    Args:
        url: The full URL of the page to retrieve.
    """
    response = requests.get(
        "https://api.zenrows.com/v1/",
        params={
            "url": url,
            "apikey": os.getenv("ZENROWS_API_KEY"),
            # mode=auto lets Zenrows pick the right settings per site
            "mode": "auto",
            "response_type": "markdown",
        },
        timeout=90,
    )

    # return the error as a string so the agent can react instead of crashing
    if response.status_code != 200:
        return f"Zenrows returned {response.status_code} for {url}"

    return response.text

Four things to note:

@function_tool builds the schema from the function signature automatically. No JSON spec to write.
mode=auto activates Adaptive Stealth Mode: JavaScript rendering and premium proxy escalation only when the target actually needs them.
response_type=markdown strips markup so the model reads clean text.
The docstring is what the model reads every turn to decide whether to call this tool. The phrase "rather than a search result summary" is what routes it away from built-in search when you have a specific URL.

Run both tools in a single agent loop

import asyncio
import os
import requests
from dotenv import load_dotenv
from agents import Agent, Runner, WebSearchTool, function_tool

load_dotenv()

@function_tool
def fetch_page_content(url: str) -> str:
    """Fetch the full, current content of a specific web page as Markdown.

    Use this when you need the complete content of a known URL rather than
    a search result summary, such as live prices, stock levels, or data
    that loads via JavaScript after the page opens.

    Args:
        url: The full URL of the page to retrieve.
    """
    response = requests.get(
        "https://api.zenrows.com/v1/",
        params={
            "url": url,
            "apikey": os.getenv("ZENROWS_API_KEY"),
            "mode": "auto",
            "response_type": "markdown",
        },
        timeout=90,
    )

    if response.status_code != 200:
        return f"Zenrows returned {response.status_code} for {url}"

    return response.text

# the model routes between both tools from their descriptions alone
agent = Agent(
    name="Web-Aware Researcher",
    instructions=(
        "You research products on the web. "
        "Use web search to find the product page URL. "
        "Then call fetch_page_content on that URL and report the price "
        "and stock status from the fetched page content. "
        "State which tool each figure came from."
    ),
    tools=[WebSearchTool(), fetch_page_content],
)

def print_trace(items):
    for item in items:
        raw = getattr(item, "raw_item", None)
        name = getattr(raw, "name", None) or getattr(raw, "type", "")
        print(f"[{item.type}] {name}")

        if item.type == "tool_call_item" and name == "fetch_page_content":
            print("  args:", getattr(raw, "arguments", ""))

        if item.type == "tool_call_output_item":
            out = str(getattr(item, "output", ""))
            print(f"  output: {len(out)} chars")
            print(f"  body: {out[:500]}")

async def main():
    result = await Runner.run(
        agent,
        "Find the PriceOye page for the iPhone 17 Pro "
        "and report its current price and stock status.",
    )

    print_trace(result.new_items)
    print("\n---\n")
    print(result.final_output)

asyncio.run(main())

The trace shows the two-step routing. Turn one: web_search_call for discovery. Turn two: fetch_page_content with the URL it found.

[tool_call_item] web_search_call
[message_output_item] message
[tool_call_item] fetch_page_content
  args: {"url":"https://priceoye.pk/mobiles/apple/apple-iphone-17-pro"}
[tool_call_output_item]

---

Here are the details for the iPhone 17 Pro from its PriceOye product page:

Current Price: Rs 471,999
Stock Status: Only 1 left in stock

Price and stock status are directly extracted from the fetched page content
(functions.fetch_page_content tool).

No conditionals. No orchestration code. The routing lives entirely in the tool descriptions.

When to use each tool

Use WebSearchTool for discovery: finding pages, open-ended research, answering questions from indexed public content.

Use Zenrows as the retrieval layer when you have a specific URL and need the live contents — protected targets, JavaScript-rendered pages, anything where freshness matters.

For structured field extraction instead of full-page Markdown, use Zenrows Extract. For high-volume or scheduled retrieval across many URLs, use Zenrows Batch.

The same pattern carries over to multi-agent setups: building a web research multi-agent system with AG2 and Zenrows. And to smolagents, under a @tool decorator: Zenrows smolagents guide.

Conclusion

You now have an agent with two retrieval paths. WebSearchTool handles discovery. Zenrows reads the full live page once the URL is known, including JavaScript-rendered and anti-bot-defended targets. The model routes between them from the tool descriptions alone.

What's next

Zenrows Batch for high-volume or scheduled retrieval
Zenrows MCP server if you prefer ready-made tools over a custom wrapper
OpenAI Agents SDK integration guide for the full reference

To read

Benny — Thu, 03 Sep 2026 16:30:00 +0000

Prosper Otemuyiwa for Valyu AI

Aug 26

Investment Research APIs in 2026: 7 DeepResearch Workflows for Deal Teams

#ai #api #finance #python

12 min read

[Boost]

Benny — Thu, 03 Sep 2026 16:27:11 +0000

Prosper Otemuyiwa for Valyu AI

Aug 26

Investment Research APIs in 2026: 7 DeepResearch Workflows for Deal Teams

#ai #api #finance #python

12 min read

How to Build a Price Monitoring Agent with Pydantic AI and ZenRows

Benny — Wed, 22 Jul 2026 20:46:40 +0000

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. However, before we start, let's see everything you need to follow along.

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

    }
    resp = requests.get(
        "https://api.zenrows.com/v1/",
        params=params,
        timeout=60,
    )

    # ZenRows returns a non-200 when it cannot deliver the page.
    if resp.status_code != 200:
        return None

    markdown = resp.text.strip()

    # Too short to be a product page. A block or error blob lands here.
    if len(markdown) < MIN_MARKDOWN_LENGTH:
        return None

    # An error came back as JSON instead of Markdown.
    if markdown.startswith("{") and '"code"' in markdown:
        return None

    return markdown


# --- 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."
    ),
)


# --- Step 3: run the pipeline across every marketplace ---

def run_monitor() -> tuple[list[PriceRecord], int]:
    results: list[PriceRecord] = []
    failed = 0

    for marketplace, url in TARGETS:
        print(f"{marketplace}: {url}")

        # Retrieval failure: ZenRows returns a blocked response. Log and skip.
        markdown = fetch_markdown(url)
        if markdown is None:
            print("  retrieval blocked, skipping")
            failed += 1
            continue

        # Extraction failure: the model never returns a valid record and
        # Pydantic AI exhausts its retries. Log and skip.
        try:
            result = price_agent.run_sync(
                f"Marketplace: {marketplace}\n\n{markdown}"
            )
        except UnexpectedModelBehavior as e:
            print(f"  extraction failed after retries, skipping ({e})")
            failed += 1
            continue

        record = result.output
        record.marketplace = marketplace   # use the known label, not the LLM's casing
        results.append(record)
        print(f"  ok -> {record.name} {record.price} {record.currency}")

    return results, failed


if __name__ == "__main__":
    records, failed_count = run_monitor()

    print(f"\nvalidated records: {len(records)}")
    print(f"failed URLs: {failed_count}")

    for record in records:
        print(record.model_dump())

    # The decision: cheapest marketplace first.
    print()
    for record in sorted(records, key=lambda r: r.price):
        print(f"{record.marketplace:<10} {record.price} {record.currency}")

From our output below, you will notice that both sites ran and the product's information was retrieved from Amazon and Walmart. This tells us how reliable ZenRows is for retrieval and scraping.

venv) benny@Mac price_monitoring % python zenrows_multi_copy.py
/Users/benny/Desktop/price_monitoring/venv/lib/python3.9/site-packages/urllib3/__init__.py:35: NotOpenSSLWarning: urllib3 v2 only supports OpenSSL 1.1.1+, currently the 'ssl' module is compiled with 'LibreSSL 2.8.3'. See: https://github.com/urllib3/urllib3/issues/3020
  warnings.warn(
Amazon: https://www.amazon.com/Apple-iPhone-Version-256GB-Titanium/dp/B0DHJDPYYR
  ok -> Apple iPhone 16 Pro, 256GB, Black Titanium - Unlocked (Renewed) 745.93 USD
Walmart: https://www.walmart.com/ip/Restored-Apple-iPhone-16-Pro-Carrier-Unlocked-256GB-Black-Titanium-Refurbished/13323059232
  ok -> Restored Apple iPhone 16 Pro - Carrier Unlocked - 256GB Black Titanium (Refurbished) 697.87 USD

validated records: 2
failed URLs: 0
{'name': 'Apple iPhone 16 Pro, 256GB, Black Titanium - Unlocked (Renewed)', 'price': 745.93, 'currency': 'USD', 'marketplace': 'Amazon', 'availability': 'In Stock', 'scraped_at': datetime.datetime(2026, 6, 22, 23, 27, 51, 184428, tzinfo=datetime.timezone.utc)}
{'name': 'Restored Apple iPhone 16 Pro - Carrier Unlocked - 256GB Black Titanium (Refurbished)', 'price': 697.87, 'currency': 'USD', 'marketplace': 'Walmart', 'availability': 'In Stock', 'scraped_at': datetime.datetime(2026, 6, 22, 23, 28, 6, 863848, tzinfo=datetime.timezone.utc)}

Walmart    697.87 USD
Amazon     745.93 USD
(venv) benny@Mac price_monitoring %

You can find the code snippets in this article in this GitHub repository.

Wrapping up

With ZenRows handling retrieval and Pydantic AI validating each extraction, this pipeline remains reliable even when working with protected targets like Amazon and Walmart, which traditional scrapers often fail to scrape or return incomplete data for. ZenRows' strength lies in its ability to reliably access protected sites, with a 99.93% success rate against sites that block everything else. The result is a list of typed product records across three marketplaces in a consistent shape, ready to insert into a database or feed an alerting layer without any official integration.

Try ZenRows for free to build more reliable price-monitoring workflows!

How to Build a Competitor Intelligence Agent with CrewAI and ZenRows

Benny — Thu, 16 Jul 2026 21:45:31 +0000

A competitor intelligence agent enables real-time pricing visibility, automated positioning tracking, and structured competitor briefs for brands without manual research. At the center of these workflows are a researcher agent responsible for gathering data and an analysis agent responsible for generating a summary and a downstream report.

However, one of the things that makes the researcher agent's output trustworthy is its retrieval layer, since a poor retrieval-layer output can make the analysis agent's recommendation questionable. If the researcher agent searches the webpage and retrieves a challenge page, an empty response, or blocked content, every downstream conclusion becomes less trustworthy. With a 99.93% success rate on protected websites, ZenRows provides the reliable retrieval layer that makes these workflows practical in production.

This tutorial shows why a CrewAI researcher agent can fail on protected competitor pages. It starts with a custom ZenRows-based tool setup, then later shows the MCP server as an alternative approach.

Prerequisites

This tutorial works best with Python 3.10 or newer. If you are on an older Python version, create a dedicated virtual environment with a current Python installation to avoid dependency conflicts.

Python 3.10 to 3.13. CrewAI requires this range.
ZenRows API key. Create an account at zenrows.com and copy your key from the dashboard. This key authenticates every scrape the researcher agent runs for data extraction.
Anthropic API key. The crew uses Claude to drive both agents. Generate a key in the Anthropic Console.
Install dependencies and the required packages using pip install "crewai[anthropic]" crewai-tools zenrows python-dotenv.
Create a .env file and save your API keys there.

Why the built-in ScrapeWebsiteTool fails on competitor pages

The problem, as established earlier, starts before the analysis. A CrewAI workflow depends primarily on the information collected, because a competitor intelligence agent relies on a chain of steps. The researcher AI agent collects information, the analysis agent interprets it, and the final output is built from what the researcher agent retrieved. This means the scraping layer becomes the foundation of the entire workflow. If the researcher agent receives incomplete content, every other agent working with it will work with the wrong data.

CrewAI is an open-source framework that enables faster prototyping of AI-driven data extraction pipelines by orchestrating autonomous multi-agent systems. One of its features is CrewAI's built-in ScrapeWebsiteTool, which, though designed to extract and read website URL content, makes standard HTTP requests without any JavaScript rendering. This is a problem because many competitor pages use Cloudflare, DataDome, or similar anti-bot systems that can block standard HTML requests.

Browsers load the page structure first, then JavaScript fetches and inserts pricing tables and other dynamic content. A standard request using just CrewAI's built-in ScrapeWebsiteTool would receive the page shell but not the data your researcher agent needs. Likewise, it would return a challenge page or a Cloudflare 1020 error instead of the competitor page if that specified website has an anti-bot protection layer. The agent, however, doesn't validate the output. It just takes the incomplete output as the actual content.

Let's set up the ScrapeWebsiteTool for making HTTP requests and target a protected site. Run it yourself and see the challenge it faces.

from crewai_tools import ScrapeWebsiteTool

# Cloudflare-protected site
tool = ScrapeWebsiteTool(
    website_url="https://www.yelp.com"
)

text = tool.run()
print(text)

The output below shows that ScrapeWebsiteTool returned an exception and could not retrieve the relevant results from Yelp because it was behind Cloudflare's protection. The analyst agent will work with bad data. This is one example of the limitation of CrewAI's ScrapeWebsiteTool, not proof that it always returns an empty page.

((venv) ) benny@Mac competitor_intelligence % python scrapertool.py
The following text is scraped website content:

yelp.com Please enable JS and disable any ad blocker
((venv) ) benny@Mac competitor_intelligence %

ZenRows solves the retrieval-layer issue by successfully fetching pages from protected sites. ZenRows is an AI web data infrastructure that combines JavaScript rendering and premium proxy rotation to retrieve data from protected sites. Other recommendations, according to CrewAI's docs, are Scrapfly and Firecrawl. ZenRows is the right fit for this workflow because it is purpose-built for retrieving data from heavily protected targets and maintains a 99.93% success rate across supported domains. Below is a script using ZenRows on the same target (Yelp) to retrieve data.

import os
import requests
from dotenv import load_dotenv

load_dotenv()

ZENROWS_API_KEY = os.environ["ZENROWS_API_KEY"]

# Same protected competitor URL used with ScrapeWebsiteTool
url = "https://www.yelp.com"

params = {
    "url": url,
    "apikey": ZENROWS_API_KEY,
    "js_render": "true",
    "premium_proxy": "true"
}

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

print("status code:", response.status_code)
print("bytes returned:", len(response.text))
print(response.text[:1000])

ZenRows was able to reach the site and successfully retrieved the actual page HTML.

((venv) ) benny@Mac competitor_intelligence % python zenrowss.py

status code: 200
bytes returned: 501630

<html lang="en-US" prefix="og: http://ogp.me/ns#" style="margin: 0; padding: 0; border: 0; font-size: 100%; font: inherit; vertical-align: baseline;" data-cookbook-theme="light" class="">
<head>
<script>document.documentElement.className=document.documentElement.className.replace(/no-js/,"js");</script>
<meta http-equiv="Content-Language" content="en-US">
<meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
<link rel="mask-icon" sizes="any" href="https://s3-media0.fl.yelpcdn.com/assets/srv0/yelp_large_assets/b2bb2fb0ec9c/assets/img/logos/yelp_burst.svg" content="#FF1A1A">
<link rel="icon" href="https://s3-media0.fl.yelpcdn.com/assets/srv0/yelp_large_assets/594b78d06612/assets/img/logos/favicon.ico">

<script>
window.ga = window.ga || function() {
    (ga.q = ga.q || []).push(arguments)
};
ga.l = +new Date;
window.ygaPageStartTime = new Date().getTime();
</script>

<script>
window.yelp = window.yelp || {};
window.yelp.cookieTypePreferencesHeader = '["ANALYTICS"...';
</script>

...

((venv) ) benny@Mac competitor_intelligence %

The difference is visible in the response itself. CrewAI's built-in ScrapeWebsiteTool completed the request but returned a Cloudflare challenge page instead of the competitor's website. From CrewAI's perspective, the task did not fail because content was returned. The analyst CrewAI agent will then analyze the data and draw a false conclusion if you use only ScrapeWebsiteTool. In the next section, you will learn how to build a ZenRows tool for reliable web scraping and to avoid that failure.

Building the ZenRows custom tool for CrewAI web scraping

CrewAI allows you to create a custom tool by turning a Python function into something an agent can call. The function receives the input, performs a calculation, and returns the result to the agent. To add a custom web scraping tool to CrewAI, decorate a Python function with @tool. This will return a scraped page as a string and pass the function into our agent's tools list. You can learn more about this from the official CrewAI documentation.

from crewai.tools import tool

@tool("Scrape competitor page")
def scrape_page(url: str) -> str:
    """Describe what the tool does and when the agent should call it."""
    # call your scraper here
    return "page content as a string"

Now that the tool wrapper exists, you will use it with ZenRows and let the researcher call it instead of making normal HTTP requests. This ensures the agent gets the right extracted data. In the code below, js_render runs the page in a headless browser so JavaScript can load, premium_proxy routes the request through residential proxies, and response_type="markdown" returns clean markdown format for the agent. Using js_render and premium_proxy can increase your credit cost per request. Read more about how to navigate resources from the official ZenRows documentation. The tool created is then added to the researcher agent's tool list.

import os

from crewai import Agent
from crewai.tools import tool
from zenrows import ZenRowsClient

# retries=2 retries failed requests (429 and 5xx) with exponential back-off
client = ZenRowsClient(os.environ["ZENROWS_API_KEY"], retries=2)


@tool("Scrape competitor page as Markdown")
def scrape_page_markdown(url: str) -> str:
    """Fetch the full content of a single competitor web page and return it as
    Markdown. Pass one URL as a string. Use this to read pricing pages, product
    pages, and positioning copy from sites that block standard scrapers."""
    response = client.get(
        url,
        params={
            "js_render": "true",
            "premium_proxy": "true",
            "response_type": "markdown",
        },
    )
    response.raise_for_status()
    return response.text

researcher = Agent(
    role="Competitor Pricing Researcher",
    goal=(
        "Scrape each competitor pricing page and extract product names, "
        "prices, and positioning claims"
    ),
    backstory=(
        "You read competitor pricing pages and report exactly what is on the "
        "page. You never invent numbers."
    ),
    tools=[scrape_page_markdown],
    verbose=True,
)

The researcher can now open the competitor page on the web and return the content as Markdown. In the next section, you will extend the tool by adding an analyst agent before running the whole crew.

Building the competitor intelligence crew

To build a competitor intelligence crew, use two agents: a researcher and an analyst. The researcher uses ZenRows to scrape the competitor pricing page, and the analyst turns the researcher's output into a structured brief. Crew.kickoff() will run them in sequence and return the comparison to you. Let's break everything down.

1 . The researcher agent scrapes target pages and pulls product information. The ZenRows tool appears in the tools list to ensure a page's protection layer doesn't limit web scraping tasks.

from crewai import Agent

from zenrows_tools import scrape_page_markdown

researcher = Agent(
    role="Competitor Pricing Researcher",
    goal=(
        "Scrape each competitor pricing page and extract every plan name, "
        "its price, and any positioning claim on the page"
    ),
    backstory=(
        "You read competitor pricing pages and report exactly what is on the "
        "page. You never invent numbers or fill gaps with assumptions."
    ),
    tools=[scrape_page_markdown],
    verbose=True,
)

2 . The Analyst agent compares and writes the detailed comparison report. It doesn't use the ZenRows tool because it doesn't need to access any webpage. It only works with whatever the researcher gives it. This also ensures you spend fewer tokens and that the analyst uses only the output from your data collection.

analyst = Agent(
    role="Competitive Intelligence Analyst",
    goal="Turn the researcher's scraped data into a structured competitive brief",
    backstory=(
        "You compare pricing and positioning across competitors and write a "
        "brief a product team can act on. You analyze; you do not scrape."
    ),
    verbose=True,
)

Each agent has a task, and the analyst's task uses the research task as context so CrewAI can pass the scraped output forward. The crew holds both the agent and the task and runs everything sequentially. CrewAI uses an LLM to drive agents and print a report from the scraped pages. The code below shows how that sequence works in practice.

import os
from dotenv import load_dotenv

load_dotenv()

from crewai import Agent, Task, Crew, Process, LLM
from crewai.tools import tool
from zenrows import ZenRowsClient

# -----------------------------
# ZenRows Client
# -----------------------------
zenrows = ZenRowsClient(os.environ["ZENROWS_API_KEY"], retries=2)

# -----------------------------
# Scraping Tool (wait added so client-side content renders)
# -----------------------------
@tool("Scrape page as Markdown")
def scrape_page_markdown(url: str) -> str:
    """Scrape a page and return it as Markdown. Return only visible scraped content."""
    response = zenrows.get(
        url,
        params={
            "js_render": True,
            "premium_proxy": True,
            "response_type": "markdown",
            "wait": 3000,  # wait 3s for client-side content to load. requires js_render
        },
    )
    response.raise_for_status()
    return response.text

# -----------------------------
# LLM
# -----------------------------
llm = LLM(model="anthropic/claude-sonnet-4-6", temperature=0.1)

# -----------------------------
# TARGETS (product pages, not category pages)
# Swap these for two comparable products. Confirm they resolve first.
# -----------------------------
TARGETS = [
    "https://www.ikea.com/us/en/p/markus-office-chair-vissle-dark-gray-90289172/",
    "https://www.amazon.com/dp/B00FS3VJAO",  # comparable office chair
]

# -----------------------------
# RESEARCHER (STRICT SCRAPER)
# -----------------------------
researcher = Agent(
    role="Retail Product Data Extractor",
    goal=(
        "Extract ONLY visible product name and price from retail product pages. "
        "No inference, no assumptions."
    ),
    backstory=(
        "You are a deterministic extraction system. "
        "If data is not visible in scraped content, it does not exist."
    ),
    tools=[scrape_page_markdown],
    llm=llm,
    verbose=True,
)

# -----------------------------
# ANALYST (STRICT GROUNDED)
# -----------------------------
analyst = Agent(
    role="Retail Pricing Analyst",
    goal=(
        "Compare product pricing across two retailers using ONLY scraped data. "
        "No external knowledge or assumptions."
    ),
    backstory=(
        "You ONLY use provided scraped outputs. "
        "Missing fields must be marked as 'NOT FOUND IN SCRAPE'."
    ),
    llm=llm,
    verbose=True,
)

# -----------------------------
# TASK 1: SCRAPING
# -----------------------------
research_task = Task(
    description=(
        "Scrape each product page below and extract the product name and its "
        "current price. Use only what is visible on the page. If a value is not "
        "present, write NOT FOUND IN SCRAPE.\n\n"
        + "\n".join(TARGETS)
    ),
    expected_output="Product name and current price for each URL, from scraped content only.",
    agent=researcher,
)

# -----------------------------
# TASK 2: ANALYSIS
# -----------------------------
analysis_task = Task(
    description=(
        "Compare the two products using only the researcher's scraped data. "
        "Do not add outside knowledge. Mark anything missing as NOT FOUND IN SCRAPE.\n\n"
        "Produce:\n"
        "- a table of product name and price per retailer\n"
        "- which product is cheaper\n"
        "- one or two insights drawn from the data"
    ),
    expected_output="A short pricing comparison: a table and a brief takeaway.",
    agent=analyst,
    context=[research_task],
)

# -----------------------------
# CREW
# -----------------------------
crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.sequential,
    verbose=True,
)

# -----------------------------
# RUN
# -----------------------------
if __name__ == "__main__":
    result = crew.kickoff()
    print("\n\n===== FINAL OUTPUT =====\n")
    print(result)

Here is part of the output of the code below (it was a long report). The report is generated entirely from the scraped content returned by ZenRows. A standard scraper would not work reliably here because the retrieval step would likely return a challenge page, incomplete HTML, or a JavaScript shell instead of the underlying content, leaving nothing for the analyst agent to work with.

From the output below, you will notice that the researcher agent remains grounded in the scraped data. For example, it does not attempt to convert currencies or fill in missing information that is not present in the source. Once you have gotten to this point, you have two options. The first is to pass the full page content to downstream agents and let them perform their own analysis, as we did in the script above. The second option is to extract structured fields immediately after retrieval and store them in a database or comparison table. This is often the better choice for use cases where you care about specific attributes rather than the full page content. ZenRows AutoParse is designed for this workflow, returning structured data that can be inserted directly into downstream systems without requiring a second extraction step.

The code is available in this GitHub repository.

╭────────────────────────────────────────────────────────────────── Crew Completion ──────────────────────────────────────────────────────────────────╮
│                                                                                                                                              │
│  Crew Execution Completed                                                                                                                   │
│  Name: crew                                                                                                                                 │
│  ID: 6972714b-a13f-4f66-8ffe-47c908b5ed01                                                                                                    │
│                                                                                                                                              │
│  Final Output: ## Pricing Comparison: Office Chairs                                                                                         │
│                                                                                                                                              │
│  ### Product & Price Table                                                                                                                  │
│                                                                                                                                              │
│  | Retailer | Product Name | Scraped Price |                                                                                                │
│  |----------|--------------|---------------|                                                                                                │
│  | **IKEA** | MARKUS Office Chair | $299.99 USD |                                                                                          │
│  | **Amazon** | BestOffice Ergonomic Office Chair (Mid-Back, Black) | BOB 290.15 |                                                       │
│                                                                                                                                              │
│  ---                                                                                                                                         │
│                                                                                                                                              │
│  ### Which Product Is Cheaper?                                                                                                              │
│                                                                                                                                              │
│  **Cannot be determined.** The Amazon listing returned a price denominated in **BOB (Bolivian Boliviano)**, while the IKEA listing         │
│  returned a price in **USD**. These two currencies cannot be compared without a conversion rate, and applying one would require             │
│  external knowledge outside the scraped data. No valid apples-to-apples price comparison is possible from the data as returned.             │
│                                                                                                                                              │
│  ---                                                                                                                                         │
│                                                                                                                                              │
│  ### Insights from the Scraped Data                                                                                                         │
│                                                                                                                                              │
│  1. **Currency anomaly on Amazon:** The scraper returned `BOB290.15` for the Amazon listing — BOB is the currency code for the             │
│     Bolivian Boliviano. This is flagged in the source data as a likely geo-detection or display error. The true USD price for this          │
│     listing is **NOT FOUND IN SCRAPE** and cannot be assumed.                                                                               │
│                                                                                                                                              │
│  2. **Product detail disparity:** IKEA's scraped data returned only a short product name (`MARKUS`), while Amazon's scrape returned         │
│     a lengthy, keyword-stuffed title. This reflects a structural difference in how each retailer formats product data—not a reflection      │
│     of product quality or completeness.                                                                                                     │
│                                                                                                                                              │
│  ---                                                                                                                                         │
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯

Returning structured data with autoparse

Instead of returning a full markdown document, you can pass autoparse=True so that ZenRows returns structured output. Autoparsing reads the page's structured markup and hands you a JSON string for the analyst to use. Keep in mind that Autoparse is experimental, domain-limited, and only works on pages with a parser.

To autoparse, modify the tool using the code below.

@tool("Scrape competitor page as structured JSON")
def scrape_page_structured(url: str) -> str:
    """Fetch a single competitor product or pricing page and return auto-parsed
    structured data as a JSON string. Pass one URL as a string. Use this when
    you need discrete fields such as product name and price rather than prose."""
    response = zenrows.get(
        url,
        params={
            "js_render": True,
            "premium_proxy": True,
            "autoparse": True,  # replaces response_type; autoparse returns JSON
        },
    )
    response.raise_for_status()
    return response.text

To switch the tool to structured output, replace "response_type": "markdown" with "autoparse": True. Autoparse cannot be combined with response_type, so it goes in its place, not alongside it.

You can keep js_render and premium_proxy on here, because these pricing pages build their content client-side and sit behind anti-bot protection, so without them, autoparse parses an empty shell. Autoparse itself adds no cost; you only pay for js_render and premium_proxy, so this request's cost stays the same as the Markdown version.

Use the ZenRows MCP server instead of a custom tool

MCP is the right choice when the crew is part of a larger multi-agent system and tool maintenance overhead matters; the custom @tool is right when you need tight control over parameters or when Node.js is not available. The MCP exposes the same scraping capability through an agent tool, but shifts the same integration from code to server configuration. CrewAI can mount any MCP server as an agent tool. CrewAI's MCPServerAdapter starts the server, exposes its tools, and stops the server when the block exits. Everything that uses those tools, the agents, the tasks, and kickoff(), has to live inside the with block, because the tools are only connected while the server is running. The server is open source; here is the GitHub repository. You can read more about it via the Official ZenRows MCP documentation.

The MCP server runs on Node.js, so to use it, you'll need to install Node 20.6 or later. ZenRows also offers a hosted server that you can connect to via URL if you want to avoid the Node dependency, which can simplify deployment.

pip install "crewai-tools[mcp]"

Since ZenRows hosted is the recommended way for any app calling an LLM API directly, let's explore that route. The MCP server is free.

import os
from typing import Any

from dotenv import load_dotenv

load_dotenv()

from crewai import Agent, Task, Crew, Process, LLM
from crewai.tools import BaseTool
from crewai_tools import MCPServerAdapter

# Claude drives every agent. ANTHROPIC_API_KEY is read from the environment.
llm = LLM(model="anthropic/claude-sonnet-4-6", temperature=0.1)

# Product pages to compare. Confirm they resolve first.
TARGETS = [
    "https://www.ikea.com/us/en/p/markus-office-chair-vissle-dark-gray-90289172/",
    "https://www.amazon.com/dp/B00FS3VJAO",  # comparable office chair
]

# Hosted ZenRows MCP server. No Node, no npx, just your API key.
SERVER_PARAMS = {
    "url": "https://mcp.zenrows.com/mcp",
    "transport": "streamable-http",
    "headers": {"Authorization": f"Bearer {os.environ['ZENROWS_API_KEY']}"},
}


class NullStrippedTool(BaseTool):
    """Wraps an MCP tool and removes None arguments before the call.

    The MCP adapter fills unset optional fields with null, which the ZenRows
    server rejects. Stripping None here keeps only the values that were set.
    """

    inner: Any = None

    def _run(self, *args: Any, **kwargs: Any) -> Any:
        clean = {k: v for k, v in kwargs.items() if v is not None}
        return self.inner._run(*args, **clean)


def wrap(tools):
    return [
        NullStrippedTool(
            name=t.name,
            description=t.description,
            args_schema=t.args_schema,
            inner=t,
        )
        for t in tools
    ]


def build_and_run():
    # The MCP tools are only live inside this block.
    with MCPServerAdapter(SERVER_PARAMS, "scrape") as mcp_tools:
        scrape_tools = wrap(mcp_tools)

        researcher = Agent(
            role="Retail Product Data Extractor",
            goal=(
                "Extract ONLY the visible product name and current price from "
                "each product page. No inference, no assumptions."
            ),
            backstory=(
                "You are a deterministic extraction system. If a value is not "
                "visible in the scraped content, it does not exist."
            ),
            tools=scrape_tools,
            llm=llm,
            verbose=True,
        )

        analyst = Agent(
            role="Retail Pricing Analyst",
            goal=(
                "Compare product pricing across two retailers using ONLY "
                "scraped data. No external knowledge or assumptions."
            ),
            backstory=(
                "You ONLY use provided scraped outputs. Missing fields must be "
                "marked as 'NOT FOUND IN SCRAPE'."
            ),
            llm=llm,
            verbose=True,
        )

        research_task = Task(
            description=(
                "Scrape each product page below and extract the product name "
                "and its current price. Use only what is visible on the page. "
                "If a value is not present, write NOT FOUND IN SCRAPE.\n\n"
                "When you call the scrape tool, enable JavaScript rendering and "
                "premium proxy, and wait for the page content to load, since "
                "these pages render product data client-side.\n\n"
                + "\n".join(TARGETS)
            ),
            expected_output=(
                "Product name and current price for each URL, from scraped "
                "content only."
            ),
            agent=researcher,
        )

        analysis_task = Task(
            description=(
                "Compare the two products using only the researcher's scraped "
                "data. Do not add outside knowledge. Mark anything missing as "
                "NOT FOUND IN SCRAPE.\n\n"
                "Produce:\n"
                "- a table of product name and price per retailer\n"
                "- which product is cheaper\n"
                "- one or two insights drawn from the data"
            ),
            expected_output="A short pricing comparison: a table and a brief takeaway.",
            agent=analyst,
            context=[research_task],
        )

        crew = Crew(
            agents=[researcher, analyst],
            tasks=[research_task, analysis_task],
            process=Process.sequential,
            verbose=True,
        )

        result = crew.kickoff()
        print("\n\n===== FINAL OUTPUT =====\n")
        print(result)


if __name__ == "__main__":
    build_and_run()

You can find every code snippet in this tutorial in this GitHub repository.

Wrapping up

You now have a production-ready competitor intelligence workflow built with CrewAI and ZenRows, rather than a scrapers that fail when competitors place their pricing pages behind Cloudflare, DataDome, or other anti-bot systems.

The current system works by having a researcher agent retrieve data directly from competitors' websites using ZenRows, which has a documented 99.93% success rate, while the analyst agent uses that data to generate a report. Whether you choose the custom @tool for total control via Integration or the ZenRows MCP server for easier tool management, the architecture remains the same: reliable retrieval, grounded analysis, and actionable intelligence generated from live web data.

Try ZenRows for free and get your comprehensive competitor analysis today!

A CTO's 5-phase roadmap to AI-native internal tools (and why most pilots stall)

Benny — Wed, 10 Dec 2025 07:20:00 +0000

Executives keep declaring that AI is the future. Reports show that adoption is rising. Yet only a tiny fraction of enterprises ever reach real impact. The numbers look impressive until you ask why so many AI pilots stall before they scale.

In our conversation with Shubham Gupta, CTO at ToolJet, he argued that the real barrier is not model quality or governance. It is the quiet weight of operational debt that sits inside every enterprise. Most critical workflows still run on spreadsheets, disconnected files, and email threads. Gupta put it bluntly that "Our biggest competitor is Excel."

Once you see the problem through that lens, a sharper question appears. If internal operations remain stuck in manual systems, what does it take for an enterprise to adopt AI native internal tools at scale?

This article brings together industry research and Gupta’s firsthand experience to answer that question. It explores the obstacles slowing internal AI adoption and the role of AI native internal tools in removing those obstacles. It then lays out the five-phase roadmap Gupta believes leaders can use to achieve real progress rather than more pilots.

What the industry currently gets wrong about internal AI adoption

IBM reported that enterprise-wide AI adoption stalls due to poor model customization, complex infrastructure, and regulatory requirements. Gupta argues that the real reasons for internal AI adoption failure are legacy systems, fragmented processes, and deprioritization of internal tool development.

For example, team approvals for new product launches still move through email threads. Employees cannot act on data buried in unstructured processes, and the workflows that AI should automate still exist in spreadsheets and email chains.

The industry sees AI sophistication as the main hurdle, but Gupta insists that the true limitation to internal AI adoption is the lack of software.

"Companies with large operational requirements need software for employees to do their work more efficiently," said Gupta. "Imagine there are about 5,000 employees, in the absence of proper internal software, chaos will ensue," he added.

Relegation of internal tools is another adoption obstacle that industry reports often overlook, and Gupta says most enterprises have yet to overcome. With engineers working on customer-facing assets, internal tool development often takes a back seat.

"Engineering teams are always juggling between delivering consumer-facing products and prioritizing these internal tools, which ultimately causes a lot of operational inefficiency," Gupta shared.

To effectively adopt AI internally, organizations must update their manual practices and increase development time for internal software using low-code platforms like ToolJet.

Why internal tools are the natural first home for AI

Analysts often point out that AI succeeds fastest in environments where work follows a repeatable pattern. Gupta sees the same reality inside enterprises.

“Internal tools are repetitive and do not require a lot of contextual thinking, which makes AI well placed to build and manage them."

Employee sentiments also support this direction. A survey by Talker Research revealed that 62% of employees want to use AI in their roles to sort data in spreadsheets. Routine tasks such as data sorting, CRUD operations, approval chains, and inventory tracking are examples of workflows that follow a fixed pattern. They are the logical starting point to introduce AI-powered internal tools because they generate compounding value quickly and provide a deterministic process that enables AI algorithms to function reliably.

The industry view and Gupta’s experience point toward the same starting point for AI adoption. Enterprises waste momentum when they try to automate strategic or ambiguous work first. Gupta suggests focusing on the internal tools that already run on predictable workflows because these are the environments where AI produces reliable gains and compounds value quickly.

The determinism challenge most AI tool builders ignore

Determinism in AI-native internal tools

Enterprises fear AI's inherently non-deterministic nature. Executives want consistent and auditable outcomes from the AI tools they adopt within their organizations because AI unpredictability triggers mistrust, poses compliance issues, and can harm decision-making.

For example, a single hallucinated permission can expose an entire customer dataset. Analysts focus on establishing evals and guardrails, but the industry still lacks a shared standard for how they should work.

Gupta says the solution is an architecture with three interconnected layers, as illustrated in the image above: a multi-agent structure, an eval system, and a human-in-the-loop process. Determinism increases when enterprises break automation into smaller, single-responsibility agents.

"If you spread every AI action into multiple single responsibility prompts, the agent's output quality improves," he said.

The second layer, an eval system, serves as a feedback loop that ensures behavioral consistency, security, and reliability.

"It puts guardrails in place, so when AI responses fall below a certain standard, organizations can re-loop their agents," Gupta shared.

The final layer, human checkpoints, is what locks the architecture into place.

"No AI action on data should be performed automatically. A human should always authorize and test those actions in a development staging environment before they go to a production environment," Gupta highlighted.

This approach to achieving predictable AI outcomes should serve as a framework for enterprises that want to build trustworthy AI-powered internal tools. But tackling non-determinism is only one part of mitigating the vulnerabilities that can affect an organization's trust in AI tools.

Enterprise architecture is the real battlefield

The hardest part of AI adoption isn't the model, but the architecture that surrounds it and the expectations enterprise teams refuse to compromise.

Enterprises want:

Data control for proprietary data and isolated tools that span different departments
Governance to ensure a tool works transparently and meets regulatory requirements
Predictable Software Development Life Cycle (SDLC) for engineering teams that are resistant to disruptive workflows
Integration into existing systems to accommodate tailored workflows

Analysts believe AI tools will replace traditional workflows within enterprises. Rather than outright replacement, Gupta contends that internal AI tooling must respect an organization's existing architecture, protect that foundation, and extend the system when it requires further customization.

In practice, the right internal AI tools offer self-hosting options, provide flexible SSO, and support extensibility through code, plugins, or third-party integrations.

"You can self-host ToolJet internally inside your VPN or deploy it in your existing cloud services to make sure that none of your sensitive data escapes or reaches our servers. The data stays in your own cloud or database, so you can completely air-gap ToolJet inside your network," Gupta noted.

Opinionated platforms lock organizations into vendor-controlled environments, which risks data exposure, while enterprise-friendly AI tools give organizations full control over where their data resides.

Internal tools often span departments, and each team might build its own application while still requiring cross-departmental coordination. Enterprises want unified access control, and a strong Single Sign-On (SSO) provides secure access to the tools.

"Whatever SSO a company supports is supported by ToolJet. Whether they prefer multiple SSOs for multiple departmental tools or a single SSO to rule them all, that also works with ToolJet," said Gupta.

Opinionated platforms typically require organizations to conform to their specific authentication protocols, while adaptive AI tools integrate with multiple identity providers.

Respecting engineering workflows is another priority for internal development teams that have established processes. Gupta explained,

"Engineering teams are very picky with tools because they have a certain way of working and they don't enjoy people messing with it."

Opinionated platforms dictate a new SDLC model, but flexible AI platforms like ToolJet allow engineers to follow their own SDLC while the platform adopts it. Enterprise AI tooling succeeds when platforms adapt to an organization’s constraints, rather than fighting them.

Understanding the real obstacles to internal AI adoption is an important first step, but enterprises also need an action plan that actually drives adoption.

The roadmap to adopting AI native internal tools

Leadership wants results with AI, but often without a clear strategy or a full picture of how to achieve it. Gupta provided a phased approach for business leaders to know where to begin the adoption process and how to scale it across their organization.

Roadmap to adopting AI native internal tools

Phase 1: Audit where manual work still lives
Industry reports often advise executives to conduct an AI readiness assessment that evaluates whether their organization is ready to integrate and adopt AI. But this assessment is a broad and ongoing process.

Instead, Gupta recommends that executives revisit the manual, high-friction, and data-heavy parts of their existing workflows, then redesign them so AI becomes a natural part of the lifecycle.

"Check how many people are handling a bunch of Excel sheets overflowing with customer records or transactional data in your company," he said.

Engage with teams across product, engineering, operations, or sales departments to identify bottlenecks. For example, customer success teams may be losing track of approval requests buried in emails.

Gupta suggested this is how to spot inefficiencies and opportunities for automation. Transforming this manual process into an internal software will ensure that every request has timestamps and owners attached. Teams still make the higher-level decisions, but with an AI-powered internal tool, they start benefitting from automated routine tasks and smoother feedback.

Phase 2: Ship the first AI-generated internal app fast
Analysts say business leaders should define Key Performance Indicators (KPIs) before introducing an internal AI tool. While KPIs outline business objectives, they can create an illusion of clarity. Instead, Gupta says executives should choose one painful workflow and build a solution fast using platforms like ToolJet, which cut time-to-app through natural-language app creation.

The real success metric to track is "shipping a working internal app in one to two weeks, depending on the use case. Then, iterate based on feedback," he said. When employees experience optimized workflows and regain time to focus on higher-value work, internal tool adoption gains momentum.

Phase 3: Create a repeatable build system
Industry experts debate which build system is superior between AI, low-code, and pro-code development. But Gupta says executives must approach their decision as a development spectrum.

"Some enterprises need AI-powered development with a prompt-specific approach for rapid prototyping. Others want low-code solutions, where non-technical teams can drag-and-drop and customize pre-built components in a visual builder. Engineering teams may want to customize features even further and add custom connectors for specific functionalities using pro-code development," he said.

A practical approach is to reduce development backlogs and tool build time using AI and low-code platforms, then leverage the flexibility and precision of pro-code for custom workflows and complex features.

Phase 4: Harden with governance and access control
Reports show that the timeline for implementing and scaling AI governance in an enterprise can take 6 to 18 months. Gupta suggests a different approach: build trust into the tool from day one through determinism and access control. Start with Role-Based Access Control (RBAC) for tools that handle significant business processes, manage sensitive data, and report critical information. IT teams should limit users' permissions to only what is necessary to perform their roles.

"If a specific team doesn't need to have access to a tool, they shouldn't have access to it because the principle of least privilege and regulatory adherence is very critical to large enterprises," said Gupta.

Also, implement SSO early with strong encryption to protect authentication data. Above all, maintain human ownership throughout the entire lifecycle of developing and scaling AI internal tools. Human employees should validate decisions and provide ethical oversight, while AI apps or agents handle the operational load and execution.

Phase 5: Scale through internal evangelism
Industry reports often attribute poor internal tool adoption to governance. But Gupta challenges this view, arguing that scaling is a value-proof problem. When teams can see a tool's value quickly, adoption spreads naturally, and scaling becomes feasible.

Gupta encouraged executives to "regularly share wins and updates on a successful internal tool to build trust, and create internal programs like Hackathons which reward employees who actively build tools that contribute to enterprise-wide integration."

A McKinsey survey reinforces Gupta's suggestions, stating that "employees believe access to AI tools in betas or pilots, along with incentives such as financial rewards and recognition, will accelerate AI adoption." A successful pilot within one team will encourage a culture of experimentation, where employees feel empowered to explore AI solutions across organizational processes.

Gupta shared that the ToolJet team uses its own platform internally across different business functions, achieving productivity gains and reducing development costs.

"We used ToolJet to build an internal CRM that handles licensing keys creation, payments, refunds, and recurring invoices for our customers. This used to take a lot of our manual effort, and if we had to build this product from scratch, it would have taken us another two months. But with ToolJet, we were able to launch the product faster, and our dev team spends more time on actual customer-facing products."

He also shared that the team built their release notes engine on ToolJet in one week.

"We evaluated external software, but it was costing us upwards of $1,000 a month. So we decided to build internally using ToolJet, and it took us a week to spin it off. And now we are completely relying on our internal software to manage and publish release notes."

Enterprises that build internal AI tools early for high-friction tasks will gain faster data analysis, reduced manual error, clearer communication, improved IT support, and productive employees who can focus on more strategic tasks.

What it means to build an AI native internal tools strategy

Organizations that want to be ready for an AI-driven future must start with their internal systems. The roadmap presented in this article provides executives and team leads with concrete steps towards building an AI-native internal tooling network that puts them at an advantage to win at AI.

It doesn't require a radical company-wide overhaul. The key is building targeted internal tools that automate mechanical, multi-step work to compound wins quickly. Then, teams can iteratively expand those tools into well-scoped tasks while investing in guardrails.

The leaders who successfully implement this roadmap will move from pilots to full-scale impact.

Vibe Coding Isn’t Dumb — You’re Just Overthinking It (Especially With Tools Like ToolJet & Lovable Around)

Benny — Sat, 29 Nov 2025 20:42:34 +0000

There’s a growing sentiment among developers today about “vibe coding.” But most of the pushback comes from folks who still believe every project needs enterprise-grade architecture from day one. The reality is simple: if your goal is to ship, not philosophize, vibe coding with AI is one of the fastest ways to get ideas into the world.

A recent post broke down why and how AI works best when you remove friction. By friction, I mean mainstream stacks, short PRDs, version control, and clean context resets.

But here’s the part most people miss. Low-code and AI-native builders are literally designed for this workflow. Tools like ToolJet and Lovable take those same vibe-coding principles and bake them directly into the development experience.

Where ToolJet Shines

If you’re building something with real internal logic, workflows, or data operations, ToolJet thrives. It’s not just a “drag-and-drop UI builder.” Its AI agents understand business rules, connect to databases instantly, generate API integrations, and help scaffold real features without you wrestling with boilerplate.

It make the “use a mainstream stack + provide a clear PRD + ship step by step” workflow feel natural. You give it structure, and it gives you production-ready apps that don’t collapse under real usage.

This is vibe coding for serious apps. The kind you internal team can use enterprise wide.

Where Lovable Shines

Lovable is the opposite energy in the best way. It’s the perfect expression of pure vibe coding:

type an idea → get an app → iterate with natural language.

It’s fast, playful, and incredible for prototypes, validation, hackathon-style creativity, and “I just want to see this working in five minutes” moments.

When the screenshot says:

“Pick a mainstream stack so AI gets it right”

Lovable basically automates that decision for you. It handles the stack, scaffolding, routing, structure — so you stay focused on the idea.

Vibes > setup.

The funny thing is: vibe coding doesn’t mean you avoid complexity — it means you stage it.

So maybe start in Lovable to get the first version fast.

Move to ToolJet when your prototype becomes a real internal product.

Or stay in Lovable if your whole goal was speed, not enterprise depth.

Both tools turn the guidelines from the original post into practical, repeatable workflows.

The Takeaway

Vibe coding isn’t “lazy dev.”
It’s a modern mindset. ship fast, reduce friction, give AI constraints, and build only what matters.

ToolJet and Lovable amplify that mindset. Smartest developers aren’t the ones writing everything from scratch. They’re the ones who know how to combine AI, low-code, and clarity to ship 10× faster without losing their sanity.

I tested ToolJet’s Agent and here is what I found!

Benny — Sat, 22 Nov 2025 15:19:00 +0000

I spent last weekend playing with ToolJet’s new AI agents and… low-code might actually be dead in the “holy crap, this is what we’ve been waiting for” way.

Let me explain.

I ran a nightmare prompt that would've normally triggered a week of meetings and 20 Jira tickets:

“Build an internal tool for our sales team to manage customer contracts: upload PDFs, extract key dates with OCR, route approvals with RBAC, send reminders, and log everything.”

No schema. No PRD. No wireframes. Just the prompt/sentence above.

And here’s what the ToolJet agent team gave me in minutes:

A full PostgreSQL schema with normalized tables, foreign keys, indexes, and a scalable structure
A real UI — file uploads, searchable tables, and a Kanban approval board
RBAC baked in with viewer/editor/admin roles without config hell
and Audit logs for every action, exportable to CSV, and that comes with Slack/Zapier notifications for approvals + reminders

Honestly? It felt like working with a full-stack team without getting burned out!

And that got me thinking.

These agents aren’t “GPT wrapped in a UI.” They behave like specialized teammates. There is an agent for DB design, UI scaffolding, API wiring, logic building, debugging, etc. They talk to each other, hand off tasks, and refine.

That’s the part that feels like the real shift.

Want receipts? Here are the demos that convinced me:

My honest take after testing everything from Retool to Bubble to AI coders. Most tools break when you throw enterprise complexity at them because enterprise tools need permissions, environments, audit trails, and integrations.

ToolJet’s agent system didn’t blink. It felt scalable, deliberate, and surprisingly “developer-aware.”

So if you’re drowning in CRUD requests, or you’ve shipped one too many “temporary” internal tools that outlived your will to maintain them, give ToolJet AI a shot.

Your weekend (and sanity) will thank you.