Build an Agent Harness for Repeatable Web Research
AI & Machine LearningBuild a small agent harness in Python that saves job state, enforces tool and page budgets, and checks every claim against the page it came from.
Give a model a web tool and a question, and it can do research. Run it twice, though, and you may get two different answers. It may read 40 pages when 4 would do. It may cite a page that never said what it claims.
The fix is not a better prompt. It is a harness: the code around the model that controls what it can do, saves its progress and checks its work.
In this tutorial, you will build a small harness in Python. It runs one research task, saves state after every step, enforces budgets and only reports facts it can prove.
The example task
For each library on a list, find its latest stable release version and release date, using only that project's official website.
The input is a short list:
JOB_INPUT = [
{"name": "example-lib", "domain": "docs.example-lib.org",
"start_url": "https://docs.example-lib.org/changelog"},
{"name": "sample-sdk", "domain": "sample-sdk.dev",
"start_url": "https://sample-sdk.dev/releases"},
]
The output should be a version, a date, the page it came from and the exact sentence that proves it. If the harness cannot prove an answer, it should say "not found" instead of guessing.
What does the harness control?
The model decides what to try next. The harness decides everything else.
| The harness controls | The model controls |
|---|---|
| The job input and the output format | Which allowed page to read next |
| Which tools exist | What to extract from a page |
| Which domains each item may use | When it thinks an item is done |
| Budgets for steps, pages and tokens | |
| Saving and loading state | |
| Checking every claim before it is reported |
This split is the main idea. The model handles the parts that need judgment. Code handles the parts that need to be the same every time.
The model gets two tools:
read_page(url)returns a page as Markdown.submit_claim(field, value, quote, url)records a fact plus the exact sentence that supports it. It can also returnfinish_itemwhen it is done with the current library.
How do you persist job state?
A research job can take minutes. Networks fail, processes restart and deploys happen. If the job lives only in memory, you lose everything.
So the harness saves state after every step. If the job stops, it picks up where it left off.
Define the state
from dataclasses import dataclass, field, asdict
@dataclass
class ItemState:
name: str
domain: str
start_url: str
status: str = "pending" # pending, running, done, not_found
pages_read: list = field(default_factory=list)
claims: list = field(default_factory=list)
@dataclass
class JobState:
job_id: str
items: list
status: str = "running" # running, verifying, done, budget_exhausted
steps: int = 0
pages: int = 0
tokens: int = 0
history: list = field(default_factory=list)
Save state and pages to SQLite
Two tables are enough. jobs holds the state as JSON. pages keeps a copy of every page the job read. You need those copies later to check quotes.
import json
import sqlite3
from datetime import datetime, timezone
db = sqlite3.connect("harness.db")
db.executescript("""
CREATE TABLE IF NOT EXISTS jobs (
job_id TEXT PRIMARY KEY,
state TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS pages (
job_id TEXT NOT NULL,
url TEXT NOT NULL,
fetched_at TEXT NOT NULL,
markdown TEXT NOT NULL,
PRIMARY KEY (job_id, url)
);
""")
def now() -> str:
return datetime.now(timezone.utc).isoformat()
def save_state(state: JobState) -> None:
db.execute(
"INSERT OR REPLACE INTO jobs VALUES (?, ?, ?)",
(state.job_id, json.dumps(asdict(state)), now()),
)
db.commit()
def load_state(job_id: str):
row = db.execute("SELECT state FROM jobs WHERE job_id = ?", (job_id,)).fetchone()
if row is None:
return None
data = json.loads(row[0])
data["items"] = [ItemState(**item) for item in data["items"]]
return JobState(**data)
def stored_page(job_id: str, url: str):
row = db.execute(
"SELECT markdown, fetched_at FROM pages WHERE job_id = ? AND url = ?",
(job_id, url),
).fetchone()
return {"markdown": row[0], "fetched_at": row[1]} if row else None
Make steps safe to repeat
If the job restarts in the middle of a step, it may run that step again. That is fine as long as repeating it does no harm. Here, read_page checks the pages table first. A page the job already read costs nothing the second time.
How do you enforce tool and page budgets?
Without limits, an agent can loop forever or spend far more than the task is worth. Set budgets in code and check them before every tool call.
Set the budgets
BUDGET = {
"max_steps": 30, # Total model turns for the job
"max_pages_total": 12, # Total pages fetched for the job
"max_pages_per_item": 4, # Pages fetched per library
"max_tokens": 200, # Total MrScraper tokens for the job
"token_cap_per_page": 20, # Passed to MrScraper as tokenCap
}
class BudgetExceeded(Exception):
pass
Pick numbers that fit your task. Start small. You can raise a limit once you see that the job needs it.
Fetch pages through the harness
read_page is the only way the model can reach the web. It checks the domain, checks the budgets, calls MrScraper and records the cost.
The call uses the MrScraper Scrape Markdown endpoint. The tokenCap field limits how many tokens one page and its retries can use. The response reports the actual token_usage, which the harness adds to the job total.
import os
import requests
from urllib.parse import urlparse
MRSCRAPER_URL = "https://api.mrscraper.com/"
TOKEN = os.environ["MRSCRAPER_API_TOKEN"]
def fetch_markdown(url: str) -> dict:
response = requests.post(
MRSCRAPER_URL,
params={"token": TOKEN, "markdown": "true"},
headers={"x-api-token": TOKEN, "Content-Type": "application/json"},
json={"url": url, "maxRetries": 2, "tokenCap": BUDGET["token_cap_per_page"]},
timeout=330,
)
response.raise_for_status()
return response.json()
def read_page(state: JobState, item: ItemState, url: str) -> str:
if urlparse(url).netloc != item.domain:
return f"Refused: {url} is outside {item.domain}."
cached = stored_page(state.job_id, url)
if cached:
return cached["markdown"]
if len(item.pages_read) >= BUDGET["max_pages_per_item"]:
return "Refused: page budget for this item is used up. Submit claims or finish."
if state.pages >= BUDGET["max_pages_total"]:
raise BudgetExceeded("job page budget")
if state.tokens >= BUDGET["max_tokens"]:
raise BudgetExceeded("job token budget")
result = fetch_markdown(url)
markdown = result.get("markdown", "")
db.execute("INSERT OR REPLACE INTO pages VALUES (?, ?, ?, ?)",
(state.job_id, url, now(), markdown))
db.commit()
state.pages += 1
state.tokens += result.get("token_usage") or 0
item.pages_read.append(url)
return markdown
Notice the two kinds of limits. Hitting an item limit sends a message back to the model, so it can wrap up that item. Hitting a job limit stops the whole job, because there is nothing left to spend.
Run the loop
The loop asks the model for one action at a time. next_action is your own model call. It sends the task, the current item, recent history and the tool descriptions, then returns a dict like {"tool": "read_page", "args": {"url": "..."}}.
def run_job(job_id: str, next_action) -> JobState:
state = load_state(job_id) or JobState(
job_id=job_id, items=[ItemState(**i) for i in JOB_INPUT]
)
try:
for item in state.items:
if item.status in ("done", "not_found"):
continue
item.status = "running"
while True:
if state.steps >= BUDGET["max_steps"]:
raise BudgetExceeded("step budget")
state.steps += 1
action = next_action(task="Find the latest stable release version and date.",
item=item, history=state.history[-6:])
tool, args = action.get("tool"), action.get("args", {})
if tool == "read_page":
observation = read_page(state, item, args.get("url", ""))
elif tool == "submit_claim":
item.claims.append({**args, "item": item.name})
observation = "Claim recorded. It will be checked before the report."
elif tool == "finish_item":
item.status = "done"
observation = "Item finished."
else:
observation = f"Unknown tool {tool}."
state.history.append({"item": item.name, "tool": tool,
"args": args, "observation": observation[:500]})
save_state(state)
if item.status == "done":
break
except BudgetExceeded as e:
state.status = "budget_exhausted"
state.history.append({"stopped": str(e)})
save_state(state)
return state
state.status = "verifying"
save_state(state)
return state
Each step saves state before the next one starts. If the process dies after step 7, the next run_job call with the same job_id resumes at step 8.
How do you verify the final answer?
The model can say anything. The harness only reports what it can prove. Every claim must pass four checks before it reaches the report:
- The page was read in this job. The URL must be in the
pagestable for thisjob_id. - The quote is real. The quote must appear, word for word, in the stored page text.
- The value is in the quote. The version or date must appear inside the quoted sentence.
- The domain is allowed. The URL must belong to the item's official domain.
def normalize(text: str) -> str:
return " ".join(text.split()).lower()
def verify_claim(state: JobState, item: ItemState, claim: dict):
url, quote, value = claim.get("url", ""), claim.get("quote", ""), str(claim.get("value", ""))
if urlparse(url).netloc != item.domain:
return False, "outside allowed domain"
page = stored_page(state.job_id, url)
if page is None:
return False, "page was not read in this job"
if not quote or normalize(quote) not in normalize(page["markdown"]):
return False, "quote not found on page"
if normalize(value) not in normalize(quote):
return False, "value not found in quote"
return True, page["fetched_at"]
Build the report from verified claims only
FIELDS = ("version", "release_date")
def build_report(state: JobState) -> list[dict]:
report = []
for item in state.items:
row = {"library": item.name, "status": "not_found", "rejected": []}
for claim in item.claims:
ok, detail = verify_claim(state, item, claim)
if ok and claim.get("field") in FIELDS:
row[claim["field"]] = {
"value": claim["value"],
"quote": claim["quote"],
"url": claim["url"],
"retrieved_at": detail,
}
else:
row["rejected"].append({"claim": claim, "reason": detail})
if all(f in row for f in FIELDS):
row["status"] = "verified"
elif any(f in row for f in FIELDS):
row["status"] = "partial"
report.append(row)
if state.status == "verifying":
state.status = "done"
save_state(state)
return report
A verified row looks like this:
{
"library": "example-lib",
"status": "verified",
"version": {
"value": "4.2.1",
"quote": "Version 4.2.1 was released on 2026-09-30.",
"url": "https://docs.example-lib.org/changelog",
"retrieved_at": "2026-10-15T06:42:10+00:00"
},
"release_date": {
"value": "2026-09-30",
"quote": "Version 4.2.1 was released on 2026-09-30.",
"url": "https://docs.example-lib.org/changelog",
"retrieved_at": "2026-10-15T06:42:10+00:00"
},
"rejected": []
}
Keep the rejected list. It shows where the model made claims it could not back up. That is useful data when you tune your prompts or tools.
These checks catch made-up quotes and wrong pages. They do not prove that a quote means what the model says it means. For example, a changelog might list a pre-release above the stable release. For important jobs, send partial rows and a sample of verified rows to a person for review.
Run it, stop it, resume it
state = run_job("release-check-2026-10-15", next_action=my_model_step)
if state.status == "verifying":
report = build_report(state)
elif state.status == "budget_exhausted":
print("Stopped early:", state.history[-1])
report = build_report(state) # Still report what was verified
To resume after a crash, call run_job again with the same job ID. Finished items are skipped, pages already read come from the cache and the step count picks up where it stopped.
To repeat the job next week, use a new job ID. The new run fetches fresh pages and keeps its own evidence, so you can compare the two reports.
FAQs
What does the harness control?
The harness controls the job input, the tools, the allowed domains, the budgets, saving and loading state, and the checks on every claim. The model only chooses the next action inside those limits.
How do you verify the final answer?
Check every claim against the stored copy of the page it cites. The page must have been read in this job, the quote must appear word for word, the value must appear in the quote and the URL must be on an allowed domain. Report only claims that pass, and send partial results to a person for review.
Which parts does MrScraper provide, and which parts must my application implement?
MrScraper is a developer-focused web data API. It fetches pages, can capture content that loads in the browser and returns clean Markdown or HTML. You can cap tokens and retries per request, and each response reports token usage and runtime.
Your application builds the harness. That includes the job state, the database, the loop, the model calls, the budgets, the domain rules, claim checks and the final report. Your team also decides which sources it has permission to collect from.
Summarize this post
Open it in your assistant of choice with the prompt ready to send.
Take a Taste of Easy Scraping!
Find more insights here
Authenticate Web Data Tools for AI Agents Without Exposing Secrets
Learn when to use API keys or OAuth for AI agent tools, where credentials should live, how to separa…
Design Reliable Web Data Tools for AI Agents
Learn how to define clear inputs, structured results and honest errors for the web data tools your A…
Build a Context Layer for an AI Agent Using Web Data
Build a small context service that fetches web data for each agent task, tracks source versions per…