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:...
