Build a Source-Led Deep Research Workflow With MrScraper
GuideBuild a deep research workflow that turns a question into a source plan, collects evidence from approved pages, handles conflicting sources and writes a cited report.
Most "deep research" tools start with a web search and read whatever comes back. That is fast, but you do not control which sources shape the answer.
A source-led workflow works the other way around. You decide which sources count before any page is read. Then the workflow finds the right pages on those sources, pulls out evidence and writes a report where every claim points back to a quote.
In this tutorial, you will build that workflow in Python as your own application. It has five stages:
- Turn the question into a source plan
- Find pages on each approved source
- Extract evidence from those pages
- Handle sources that disagree
- Write a cited report MrScraper handles page discovery and extraction. Your application handles planning, checking and writing.
The example question
How long do three project management tools keep audit logs on their business plans?
This question is a good fit for a source-led approach. The best answers live on each vendor's own docs and pricing pages. Third-party blog posts may be out of date.
How do you turn a question into a source plan?
A source plan answers three things before any page is fetched:
- What are the sub-questions? Break the main question into parts that one page could answer.
- Which sources may answer them? List domains and starting URLs.
- How much do you trust each source? Mark each one as official or secondary.
Draft the plan with a model
Ask your model to return a plan in a fixed JSON shape. draft_source_plan is your own model call.
{
"question": "How long do three project management tools keep audit logs on their business plans?",
"sub_questions": [
{"id": "q1", "text": "Audit log retention on Vendor A's business plan"},
{"id": "q2", "text": "Audit log retention on Vendor B's business plan"},
{"id": "q3", "text": "Audit log retention on Vendor C's business plan"}
],
"sources": [
{"id": "s1", "domain": "docs.vendor-a.example", "start_url": "https://docs.vendor-a.example/",
"authority": "official", "answers": ["q1"], "include_patterns": [".*(audit|log|retention|plan).*"]},
{"id": "s2", "domain": "help.vendor-b.example", "start_url": "https://help.vendor-b.example/",
"authority": "official", "answers": ["q2"], "include_patterns": [".*(audit|log|retention|plan).*"]},
{"id": "s3", "domain": "vendor-c.example", "start_url": "https://vendor-c.example/pricing",
"authority": "official", "answers": ["q3"], "include_patterns": [".*(audit|security|pricing).*"]}
]
}
Check the plan in code
The model drafts the plan. Your code decides whether it is allowed to run.
from urllib.parse import urlparse
APPROVED_DOMAINS = {"docs.vendor-a.example", "help.vendor-b.example", "vendor-c.example"}
MAX_SOURCES = 6
def check_plan(plan: dict) -> list[str]:
problems = []
if len(plan["sources"]) > MAX_SOURCES:
problems.append(f"too many sources ({len(plan['sources'])})")
covered = set()
for source in plan["sources"]:
url = urlparse(source["start_url"])
if url.scheme != "https":
problems.append(f"{source['id']}: start_url must use https")
if source["domain"] not in APPROVED_DOMAINS or url.netloc != source["domain"]:
problems.append(f"{source['id']}: {source['domain']} is not approved")
if source["authority"] not in ("official", "secondary"):
problems.append(f"{source['id']}: unknown authority level")
covered.update(source["answers"])
for q in plan["sub_questions"]:
if q["id"] not in covered:
problems.append(f"{q['id']} has no source")
return problems
If the check finds problems, send them back to the model for a new draft, or ask a person to fix the plan. For high-stakes research, have a person approve every plan before it runs.
The approved list is where your team decides which sources it has permission to use. Keep it in config, not in the prompt.
How do you find pages on each source?
Each source has a starting URL. The workflow now needs the specific pages that answer its sub-questions.
The MrScraper Scrape Sitemap endpoint uses the map agent to collect URLs from a starting page. The includePatterns field keeps only URLs that match your patterns, and limit caps how many come back.
import os
import requests
MRSCRAPER_URL = "https://api.mrscraper.com/"
TOKEN = os.environ["MRSCRAPER_API_TOKEN"]
def call_mrscraper(body: dict, **params) -> dict:
response = requests.post(
MRSCRAPER_URL,
params={"token": TOKEN, **params},
headers={"x-api-token": TOKEN, "Content-Type": "application/json"},
json={**body, "maxRetries": 2},
timeout=330,
)
response.raise_for_status()
return response.json()
def discover(source: dict, max_urls: int = 50) -> list[str]:
result = call_mrscraper({
"url": source["start_url"],
"agent": "map",
"maxDepth": 2,
"maxPages": 3,
"limit": max_urls,
"includePatterns": source["include_patterns"],
})
urls = (result.get("data") or {}).get("urls", [])
return [u for u in urls if urlparse(u).netloc == source["domain"]]
Pick the most likely pages
Discovery may return dozens of URLs. Read only the best few. A simple keyword score on the URL path works well as a first step.
import re
def rank_urls(urls: list[str], sub_questions: list[dict], keep: int = 4) -> list[str]:
words = {w for q in sub_questions for w in re.findall(r"[a-z]{4,}", q["text"].lower())}
def score(url: str) -> int:
path = urlparse(url).path.lower()
return sum(1 for w in words if w in path)
return sorted(urls, key=score, reverse=True)[:keep]
How do you extract evidence from discovered pages?
Now the workflow reads each chosen page and pulls out evidence for its sub-questions.
Ask for quotes, not summaries
The MrScraper Scrape JSON endpoint takes a URL and a prompt with a JSON schema. It returns structured data. By default, the response also includes the page html, which you will use to check the quotes.
import json
EVIDENCE_SCHEMA = {
"type": "object",
"properties": {
"evidence": {
"type": "array",
"items": {
"type": "object",
"properties": {
"sub_question_id": {"type": "string"},
"value": {"type": ["string", "null"]},
"quote": {"type": "string"},
"page_updated": {"type": ["string", "null"]},
},
},
}
},
}
def extract_evidence(url: str, sub_questions: list[dict]) -> dict:
questions = "\n".join(f"- {q['id']}: {q['text']}" for q in sub_questions)
prompt = (
"For each sub-question below, find the sentence on this page that answers it.\n"
"Copy that sentence exactly into 'quote'. Put the short answer in 'value'.\n"
"If the page shows a last-updated date, put it in 'page_updated'.\n"
"Skip sub-questions the page does not answer.\n"
f"{questions}\n"
"Json Schema:\n" + json.dumps(EVIDENCE_SCHEMA)
)
return call_mrscraper({"url": url, "prompt": prompt, "agent": "general"})
Verify every quote against the page
An extraction model can paraphrase, even when you ask for exact quotes. So check each quote against the page text before you trust it.
from datetime import datetime, timezone
from html.parser import HTMLParser
BLOCK_TAGS = {"p", "div", "li", "br", "tr", "td", "th", "h1", "h2", "h3", "h4", "section"}
class TextOnly(HTMLParser):
def __init__(self):
super().__init__()
self.parts, self.skip = [], False
def handle_starttag(self, tag, attrs):
if tag in ("script", "style"):
self.skip = True
elif tag in BLOCK_TAGS:
self.parts.append(" ")
def handle_endtag(self, tag):
if tag in ("script", "style"):
self.skip = False
elif tag in BLOCK_TAGS:
self.parts.append(" ")
def handle_data(self, data):
if not self.skip:
self.parts.append(data)
def page_text(html: str) -> str:
parser = TextOnly()
parser.feed(html or "")
return " ".join("".join(parser.parts).split())
def normalize(text: str) -> str:
return " ".join((text or "").split()).lower()
def collect_evidence(url: str, source: dict, sub_questions: list[dict]) -> list[dict]:
result = extract_evidence(url, sub_questions)
text = normalize(page_text(result.get("html", "")))
retrieved_at = datetime.now(timezone.utc).isoformat()
items = []
for item in (result.get("data") or {}).get("evidence", []):
quote, value = item.get("quote", ""), item.get("value")
verified = bool(quote) and normalize(quote) in text and (
value is None or normalize(value) in normalize(quote)
)
items.append({
**item,
"url": url,
"source_id": source["id"],
"authority": source["authority"],
"retrieved_at": retrieved_at,
"verified": verified,
})
return items
The parser adds a space only around block tags like <p> and <li>. That way, a sentence with a bold word inside it still matches its quote. Keep unverified items, but do not use them in the report's answers. They are useful when you debug the prompt.
Run the collection stage
def run_collection(plan: dict) -> list[dict]:
evidence = []
for source in plan["sources"]:
targets = [q for q in plan["sub_questions"] if q["id"] in source["answers"]]
for url in rank_urls(discover(source), targets):
evidence.extend(collect_evidence(url, source, targets))
return evidence
Each source only answers the sub-questions the plan assigned to it. That keeps a vendor's page from being used as evidence about a different vendor.
How do you handle conflicting sources?
Sources disagree more often than you might expect. A help article says 90 days. A pricing page says 180 days. A third-party review says 30.
Do not let the workflow quietly pick one. Make the conflict visible.
Group evidence by sub-question
For each sub-question, collect the distinct values from verified evidence.
- One value: the sources agree.
- More than one value: there is a conflict.
- No verified values: there is a gap.
Rank, but keep both sides
When values conflict, choose a preferred answer with clear rules. Then show the other values too.
- Official sources rank above secondary sources.
- Within the same level, a page with a newer
page_updateddate ranks higher. - If there is still a tie, mark the answer as unresolved.
AUTHORITY_RANK = {"official": 0, "secondary": 1}
def resolve(sub_question: dict, evidence: list[dict]) -> dict:
items = [e for e in evidence if e["verified"] and e["sub_question_id"] == sub_question["id"]]
if not items:
return {"id": sub_question["id"], "status": "gap", "answer": None, "items": []}
values = {normalize(e["value"]) for e in items}
if len(values) == 1:
return {"id": sub_question["id"], "status": "agreed", "answer": items[0]["value"], "items": items}
newest_first = sorted(items, key=lambda e: e.get("page_updated") or "", reverse=True)
ranked = sorted(newest_first, key=lambda e: AUTHORITY_RANK.get(e["authority"], 9))
best, runner_up = ranked[0], ranked[1]
tie = (best["authority"] == runner_up["authority"]
and best.get("page_updated") == runner_up.get("page_updated")
and normalize(best["value"]) != normalize(runner_up["value"]))
return {
"id": sub_question["id"],
"status": "unresolved" if tie else "conflict",
"answer": None if tie else best["value"],
"items": ranked,
}
The first sort puts newer pages first, with undated pages last. The second sort puts official sources first and keeps the date order inside each group. Dates must use one format, such as 2026-09-30, for this to work. Normalize them when you collect evidence.
A conflict is a finding, not a failure. Readers often care most about where sources disagree.
How do you produce a cited report?
The report should be easy to read and easy to check. Every answer points to numbered sources, and every source shows the quote it supports.
Build the report in code
Writing the core report in code, not with a model, means every citation is correct by design.
def build_report(plan: dict, evidence: list[dict]) -> str:
sources, lines = [], [f"# {plan['question']}", ""]
def cite(item: dict) -> int:
key = (item["url"], item["quote"])
for i, s in enumerate(sources, start=1):
if (s["url"], s["quote"]) == key:
return i
sources.append(item)
return len(sources)
gaps = []
for q in plan["sub_questions"]:
result = resolve(q, evidence)
lines.append(f"## {q['text']}")
if result["status"] == "gap":
lines.append("No verified answer found in the approved sources.")
gaps.append(q["text"])
elif result["status"] == "agreed":
refs = "".join(f"[{cite(e)}]" for e in result["items"])
lines.append(f"**{result['answer']}** {refs}")
else:
if result["answer"]:
lines.append(f"**{result['answer']}** [{cite(result['items'][0])}]")
lines.append("Sources disagree:")
for e in result["items"]:
lines.append(f"- {e['value']} ({e['authority']}) [{cite(e)}]")
lines.append("")
if gaps:
lines += ["## Open questions", *[f"- {g}" for g in gaps], ""]
lines.append("## Sources")
for i, s in enumerate(sources, start=1):
lines.append(f"{i}. {s['url']} (retrieved {s['retrieved_at'][:10]})")
lines.append(f" > {s['quote']}")
return "\n".join(lines)
A section of the finished report looks like this:
## Audit log retention on Vendor A's business plan
**180 days** [1]
Sources disagree:
- 180 days (official) [1]
- 90 days (official) [2]
## Sources
1. https://docs.vendor-a.example/security/audit-logs (retrieved 2026-10-16)
> Business plans keep audit logs for 180 days.
2. https://docs.vendor-a.example/plans/compare (retrieved 2026-10-16)
> Audit logs are retained for 90 days.
Add a summary, then check it
You can ask a model to write a short summary at the top. Give it only the resolved findings, not the raw pages. Then check that every sentence cites a source number that exists.
def check_summary(summary: str, source_count: int) -> list[str]:
problems = []
for sentence in re.split(r"(?<=[.!?])\s+", summary.strip()):
refs = [int(n) for n in re.findall(r"\[(\d+)\]", sentence)]
if sentence and not refs:
problems.append(f"no citation: {sentence}")
if any(n < 1 or n > source_count for n in refs):
problems.append(f"unknown source number: {sentence}")
return problems
If the summary fails the check, ask for a new one or leave it out. The cited sections below it already answer the question.
Hand off to your application
At the end of a run, your app has three useful outputs:
- The approved source plan
- The full evidence list, verified and unverified, with URLs and retrieval times
- The report in Markdown Store all three together. When someone asks how you reached an answer, you can show the plan, the pages and the exact quotes.
FAQs
How do you turn a question into a source plan?
Break the question into sub-questions that one page could answer. Have a model draft a list of sources for each one, with a domain, a starting URL and an authority level. Then check the plan in code against your approved domains, and have a person approve it when the stakes are high.
How do you produce a cited report?
Build the report in code from verified evidence. Number each source, show the exact quote it supports and cite those numbers next to each answer. Show conflicts and gaps openly. If you add a model-written summary, check that every sentence cites a real source number.
Which parts does MrScraper provide, and which parts must my application implement?
MrScraper is a developer-focused web data API. In this workflow, it discovers URLs from each starting page with the map agent and extracts structured evidence from each page with a prompt and schema. It also returns the page HTML, which you use to check quotes.
This guide does not use a single "deep research" call. Your application builds each stage itself: the source plan and its approval, URL ranking, quote checks, conflict rules and the report. Your team also decides which sources it has permission to use.
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
Choose an Agent Framework for a Web Data Application
Compare LangGraph, the OpenAI Agents SDK, Google ADK, CrewAI, Mastra and the Claude Agent SDK on too…
Design AI Research Loops That Stop, Recover and Produce Evidence
Learn how to build an AI research loop with clear states, firm stop conditions, safe retries after f…
How to Build a Custom Amazon Price Monitor Without Code
Learn how to build a custom Amazon price tracker in under two minutes without code. Automate price m…