Production System 1: a deep research agent
Someone asks you a question that takes a good analyst two days.
What actually makes agent runs expensive in production, and what fixes it? The answer is not in one place. It is spread across an engineering blog, a technical report someone published as a PDF, a library’s README, and a conference talk that exists only as a video with an auto-generated transcript. Reading all of that, deciding which parts matter, and writing three paragraphs that a director can act on is exactly the shape of work an agent can do.
It is also the shape of work an agent can do badly in a way that is very hard to detect.
A research agent that fabricates one number in an otherwise excellent report is worse than no research agent, because the report reads as authoritative and nobody checks the fifth citation. So this system is built around a single promise, and every design decision below is downstream of it:
Every claim in the output is traceable to a specific span of a specific source, and that trace is machine-verified before the report is printed.
By the end of this chapter you will have that system: about twelve hundred lines of Python across ten modules, running entirely offline against a fixture corpus, with an MCP tool layer, human-in-the-loop gates, four kinds of budget, and a citation integrity checker that fails a forged report.
Setup:
mkdir -p research-agent && cd research-agent
python3 -m venv .venv && source .venv/bin/activate
pip install "mcp==1.27.0" pypdf pytest pytest-asyncio
# optional, for live use: pip install httpx trafilatura youtube-transcript-api reportlab
Everything runs without the optional packages. They are the difference between reading a fixture corpus and reading the internet, and the chapter is careful to keep that difference behind one interface.
The shape of the system
Seven pieces, and the order matters because each one exists to make the next one safe.
- Provenance types — what a source, a quotable span, and a finding are. Written first, because if provenance is retrofitted it is always wrong.
- Fetching — one interface, two implementations: an offline fixture corpus and real HTTP.
- Extraction — HTML, PDF, repositories, transcripts, each producing spans with a locator precise enough to check by hand.
- A store — the corpus the agent reasons over, with keyword retrieval and a quote verifier.
- An MCP server and client — the tools, spoken over a protocol rather than called as functions.
- The loop — plan, act, observe, with budgets outside the model’s reasoning and human gates in front of expensive actions.
- Synthesis — a report assembled only from recorded findings, and an integrity check that runs on the finished text.
We build them in that order, run something after each one, and break several of them on purpose.
v1: provenance first
Here is the mistake to avoid. The obvious first version fetches a page, hands the text to a model, and asks for a summary with citations. It works, it demos well, and it is unfixable — because by the time the model is writing, the connection between “this sentence” and “that paragraph of that page” exists only in the model’s head, and there is nothing in the system that could check it.
So the first file is not a fetcher. It is a vocabulary.
research/provenance.py:
"""Provenance types. Every piece of text the agent ever sees carries one of these."""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
@dataclass(frozen=True)
class Source:
"""One retrieved artifact: a page, a PDF, a repo, a transcript."""
id: str # short stable handle, e.g. "S1"
url: str
kind: str # html | pdf | repo | transcript
title: str
content_sha: str # hash of the raw bytes, so a re-fetch is detectable
fetched_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat(timespec="seconds"))
@dataclass(frozen=True)
class Chunk:
"""A quotable span of one source, with a locator precise enough to check."""
source_id: str
locator: str # "p.2" | "para.7" | "t=00:16" | "budget.py:L10-L24"
text: str
@property
def key(self) -> str:
return f"{self.source_id}:{self.locator}"
def cite(self) -> str:
return f"[{self.source_id} {self.locator}]"
@dataclass(frozen=True)
class Finding:
"""One claim the agent believes, bound to the chunks that support it."""
claim: str
quote: str # verbatim substring of the supporting chunk
chunk_key: str
subquestion: str
def cite(self) -> str:
src, loc = self.chunk_key.split(":", 1)
return f"[{src} {loc}]"
Three decisions worth defending.
The locator is human-resolvable.
p.2 means page two of the PDF. t=00:16 means sixteen seconds into the video. budget.py:L10-L24 means those lines of that file.
A reader who distrusts a claim can open the source and land on the sentence in five seconds.
A locator like chunk_47 fails that test and is therefore worthless, even though it is perfectly unique.
Finding separates claim from quote.
The claim is the agent’s words. The quote is the source’s words.
Keeping both means you can mechanically check the second while letting the agent be useful with the first — and that check is the whole ballgame.
content_sha is on Source.
Pages change. If the report is read six months from now, the hash tells you whether the source still says what it said.
It costs one line and it is the difference between a citation and a claim about a citation.
v2: fetching, behind one seam
The reader may have no network, no API key, and no interest in hammering someone’s blog while debugging a loop. So fetching is an interface with two implementations — the same seam Part 1 used for the model client, applied to I/O.
research/fetching.py:
class FetchError(RuntimeError):
"""Raised for anything the agent should treat as 'this source is unavailable'."""
@dataclass
class RawDoc:
url: str
kind: str # html | pdf | repo | transcript
data: bytes # raw bytes for html/pdf/transcript
files: dict[str, str] | None = None # for kind == "repo": path -> text
class Fetcher(Protocol):
def fetch(self, url: str) -> RawDoc: ...
def classify(url: str) -> str:
p = urlparse(url)
if p.netloc.endswith("youtube.com") or p.netloc == "youtu.be":
return "transcript"
if p.netloc == "github.com" and len([s for s in p.path.split("/") if s]) == 2:
return "repo"
if p.path.lower().endswith(".pdf"):
return "pdf"
return "html"
The offline implementation serves a fixture corpus through URLs:
class OfflineFetcher:
"""Serves a fixture corpus through the same interface as the network.
The index maps real-looking URLs onto local paths, so every URL the agent
handles downstream — including links discovered by crawling — is a URL.
"""
def __init__(self, root: str | Path = "corpus") -> None:
self.root = Path(root)
self.index: dict[str, str] = json.loads((self.root / "index.json").read_text())
self.calls: list[str] = []
def fetch(self, url: str) -> RawDoc:
self.calls.append(url)
rel = self.index.get(url) or self.index.get(url.rstrip("/") + "/")
if rel is None:
raise FetchError(f"404 offline corpus has no entry for {url}")
path = self.root / rel
kind = classify(url)
if kind == "repo":
files = {p.name: p.read_text() for p in sorted(path.iterdir()) if p.is_file()}
return RawDoc(url=url, kind=kind, data=b"", files=files)
return RawDoc(url=url, kind=kind, data=path.read_bytes())
Note what the offline fetcher does not do: it does not invent a fake URL scheme.
Fixtures live behind https://orbital.example/costs, so link-following, domain policy, and citation rendering all exercise real URL code paths.
The moment you let offline mode use file:// or mock://, half your system is untested.
The live implementation is the same interface with a throttle, a real User-Agent, and one piece of failure translation:
class HttpFetcher:
"""The live one. Never used in the offline tests; identical interface."""
def fetch(self, url: str) -> RawDoc:
kind = classify(url)
self._throttle() # delay_s between requests
if kind == "transcript":
return RawDoc(url=url, kind=kind, data=_youtube_json(url).encode())
if kind == "repo":
return RawDoc(url=url, kind=kind, data=b"", files=_github_files(url))
try:
r = self._client.get(url) # httpx, follow_redirects=True
r.raise_for_status()
except Exception as exc: # one failure mode out
raise FetchError(f"fetch failed for {url}: {type(exc).__name__}: {exc}") from exc
return RawDoc(url=url, kind=kind, data=r.content)
Every network failure becomes one exception type. Timeouts, DNS failures, 500s, and TLS errors are all “this source is unavailable” as far as the agent is concerned, and collapsing them here means the loop has exactly one thing to handle instead of nine.
The two specialist fetchers deserve their APIs named, because both changed recently.
def _youtube_json(url: str) -> str:
"""Transcript via youtube-transcript-api >= 1.0 (instance API, not classmethods)."""
from youtube_transcript_api import YouTubeTranscriptApi
q = parse_qs(urlparse(url).query)
video_id = q.get("v", [urlparse(url).path.lstrip("/")])[0]
fetched = YouTubeTranscriptApi().fetch(video_id) # FetchedTranscript
return json.dumps({
"video_id": video_id, "title": video_id,
"language": fetched.language_code, "is_generated": fetched.is_generated,
"segments": [{"start": s.start, "duration": s.duration, "text": s.text}
for s in fetched],
})
youtube-transcript-api 1.x — verified here against 1.2.4 — is an instance API.
The old YouTubeTranscriptApi.get_transcript(video_id) classmethod is gone; you construct the object and call .fetch(video_id), which returns a FetchedTranscript you can iterate for snippets carrying text, start, and duration (https://github.com/jdepoix/youtube-transcript-api).
Every tutorial written before 2025 shows the old shape, which is why you check the installed version rather than the blog post.
For GitHub, the top-level contents endpoint is enough for a README-and-a-file-or-two read, and it needs no token for public repos (https://docs.github.com/en/rest/repos/contents). Rate limits are 60 requests per hour unauthenticated, which is another reason the throttle exists.
What is not runnable here: this sandbox has no route to orbital.example and, more importantly, hammering live sites from a chapter’s example code is rude.
Every run below uses OfflineFetcher.
Swapping in HttpFetcher is a one-line change at the composition root and nothing downstream knows the difference — which is the entire point of the seam.
v3: extraction, or where the locators come from
Four content types, four functions, one output contract: (title, [(locator, text)], links).
PDFs, first, because they are the ones people get wrong:
def extract_pdf(doc: RawDoc):
from pypdf import PdfReader
reader = PdfReader(BytesIO(doc.data))
title = (reader.metadata or {}).get("/Title") or doc.url.rsplit("/", 1)[-1]
chunks = []
for n, page in enumerate(reader.pages, start=1):
text = re.sub(r"[ \t]+", " ", page.extract_text() or "").strip()
for j, para in enumerate(p for p in text.split("\n\n") if len(p.strip()) >= 40):
loc = f"p.{n}" if j == 0 else f"p.{n}.{j + 1}"
chunks.append((loc, re.sub(r"\s*\n\s*", " ", para).strip()))
return str(title), chunks, []
pypdf — verified against 6.15.0 — extracts text page by page, which is exactly the granularity a citation wants (https://pypdf.readthedocs.io/en/stable/user/extract-text.html).
Do not concatenate the pages and then chunk by token count: you will have destroyed the only locator the format gave you for free.
extract_text() returning an empty string is normal for scanned PDFs; those need OCR, which is a different project, and the honest behaviour is to produce zero chunks rather than pretend.
Transcripts get time locators, windowed so a citation points at a listenable span rather than a four-word fragment:
def _stamp(seconds: float) -> str:
return f"{int(seconds) // 60:02d}:{int(seconds) % 60:02d}"
def extract_transcript(doc: RawDoc, *, window_s: float = 30.0):
data = json.loads(doc.data.decode())
title = data.get("title") or data.get("video_id", doc.url)
chunks, buf, start = [], [], None
for seg in data["segments"]:
if start is None:
start = seg["start"]
buf.append(seg["text"].strip())
if seg["start"] + seg["duration"] - start >= window_s:
chunks.append((f"t={_stamp(start)}", " ".join(buf)))
buf, start = [], None
if buf:
chunks.append((f"t={_stamp(start or 0)}", " ".join(buf)))
return title, chunks, []
Repositories chunk by file in twenty-line windows, producing locators like budget.py:L21-L40 — the only citation format a developer can act on without searching.
The code is a splitlines() loop; the design decision is that a repository is many small documents rather than one concatenated blob, because a line range that spans two files is meaningless.
HTML is the messy one.
The dependency-free path is an html.parser subclass that keeps block-level text, skips chrome, and collects links; when trafilatura is installed it is preferred, because boilerplate removal is a solved problem someone else has solved better (https://trafilatura.readthedocs.io/en/latest/usage-python.html, verified against 2.2.0).
def extract_html(doc: RawDoc, *, prefer_trafilatura: bool = True, max_link_density: float = 0.5):
html = doc.data.decode("utf-8", errors="replace")
reader = _Reader() # stdlib html.parser subclass
reader.feed(html)
reader.close()
links = [urljoin(doc.url, h) for h in reader.links
if not h.startswith(("#", "mailto:", "javascript:"))]
blocks = reader.blocks
if prefer_trafilatura:
try:
import trafilatura
text = trafilatura.extract(html, url=doc.url, output_format="txt",
include_comments=False, include_tables=True)
if text: # one block per non-empty line
cand = [re.sub(r"\s+", " ", b).strip() for b in text.split("\n") if b.strip()]
blocks = [b for b in cand if len(b) >= 40] or blocks
except ImportError:
pass
# Drop navigation and hub blocks: text that is mostly link labels is a menu,
# not a claim, and quoting it produces citations that say nothing.
blocks = [b for b in blocks if link_density(b, reader.anchor_texts) <= max_link_density]
return reader.title or doc.url, [(f"para.{i + 1}", b) for i, b in enumerate(blocks)], links
That link-density filter was not in the first version. It was added after watching the finished agent cite this, from the blog’s index page:
- How we scaled our agent fleet to 4,000 concurrent runs What an agent run
actually costs Pricing. [S1 para.1]
A perfectly valid citation of a navigation menu.
The claim is true, checkable, and useless, which is the most annoying failure mode a research agent has.
link_density is fifty characters of arithmetic:
def link_density(block: str, anchor_texts: list[str]) -> float:
"""Fraction of a block that is link text. Hub pages score near 1.0."""
if not block:
return 0.0
covered = sum(len(a) for a in anchor_texts if a in block)
return min(covered / len(block), 1.0)
The heuristic is coarse and you should know how it behaves: on the fixture index page, trafilatura flattens the whole page into one line, so the filter drops the page entirely and it contributes zero chunks.
That is the right outcome here — a hub page’s value is its links, not its prose — but if your corpus has pages that mix a real article with a heavy sidebar, tune the threshold or filter per block rather than per page.
v4: the store, and the first real check
research/store.py holds sources and chunks, retrieves by keyword, and — the important part — verifies quotes.
def ingest(self, url: str) -> Source:
"""Fetch, extract, and index one URL. Idempotent per URL."""
if url in self._by_url:
return self.sources[self._by_url[url]] # re-ingest is free
doc: RawDoc = self.fetcher.fetch(url) # may raise FetchError
title, pieces, links = extract(doc)
self._n += 1
sid = f"S{self._n}"
payload = doc.data or repr(sorted((doc.files or {}).items())).encode()
src = Source(id=sid, url=url, kind=doc.kind, title=title,
content_sha=hashlib.sha256(payload).hexdigest()[:12])
self.sources[sid] = src
self._by_url[url] = sid
self.links[sid] = links
for loc, text in pieces:
c = Chunk(source_id=sid, locator=loc, text=text)
self.chunks[c.key] = c
return src
def verify_quote(self, chunk_key: str, quote: str) -> bool:
"""Provenance check: is this quote genuinely in the chunk it claims?"""
chunk = self.chunks.get(chunk_key)
if chunk is None:
return False
norm = lambda s: re.sub(r"\s+", " ", s).strip().lower()
return norm(quote) in norm(chunk.text)
Retrieval is TF-IDF cosine over chunks — twenty lines of collections.Counter and math.log, no vector database — ending in the line that matters:
scored.sort(key=lambda p: (-p[1], p[0].key))
return scored[:k]
Twelve chunks from five sources do not need embeddings.
The corpus for one research run is small — that is what makes it a run — and a keyword score you can debug beats a similarity score you cannot.
If you later find that vocabulary mismatch is your bottleneck, swap this method for the memory system from Part 3, Chapter 6; nothing above the search call changes.
The tie-break on p[0].key is not decoration.
Without it, two chunks with identical scores come back in dictionary order, and your “deterministic” test suite fails once a month for reasons nobody can reproduce.
Ingesting one of each kind:
S1 html How we scaled our agent fleet to 4,000 concurrent ru sha=c648cb7b1b72
S2 pdf TR-2026-04 sha=e6c6370f511d
S3 repo stepbudget (repository) sha=1627a5c4a33c
S4 transcript Operating agents at scale — Orbital Systems, AgentCo sha=e938944bf10b
4 sources (html:1, pdf:1, repo:1, transcript:1), 15 chunks
v5: the tools, over MCP
You could stop here and call these functions directly from the loop. Do not, for three reasons that all show up within a month.
Over MCP the same tool server can be driven by a different agent, by Claude Desktop, or by a colleague’s TypeScript client. You can add a third-party server — a real web search, your company’s wiki — to the same toolbelt without touching the loop. And the protocol forces you to write the tool contract down in a machine-readable way, which is the discipline Part 2, Chapter 1 spent a chapter arguing for.
research/mcp_server.py exposes four tools with the mcp Python SDK (verified against 1.27.0):
def build_server(fetcher: Fetcher | None = None, *, name: str = "research"):
store = SourceStore(fetcher=fetcher or OfflineFetcher("corpus"))
mcp = FastMCP(name, stateless_http=True)
@mcp.tool(
title="Ingest one URL into the research corpus",
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=True),
)
def ingest_source(
url: Annotated[str, Field(description="Absolute URL. Web page, PDF, GitHub repo, or YouTube watch URL.")],
) -> IngestResult:
"""Fetch a URL, extract its text, and index it as a citable source.
Returns the assigned source id, how many chunks it produced, and any links
found on the page so you can decide what to read next. Ingesting the same
URL twice is free and returns the same source id.
"""
try:
src = store.ingest(url)
except FetchError as exc:
raise ValueError(
f"{exc}. The source is unavailable — do not retry the same URL. "
"Pick a different source or continue with what you have."
) from exc
return IngestResult(source_id=src.id, kind=src.kind, title=src.title,
chunks=sum(1 for c in store.chunks.values() if c.source_id == src.id),
links=store.links.get(src.id, [])[:10])
IngestResult is a four-field pydantic model, and that choice matters more than it looks — see below.
The tool that carries the promise is record_finding, and it is worth reading closely:
@mcp.tool(
title="Record a supported finding",
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=True),
)
def record_finding(
claim: Annotated[str, Field(description="One sentence you believe, in your own words.")],
quote: Annotated[str, Field(description="Verbatim text from the chunk that supports it.")],
chunk_key: Annotated[str, Field(description="The chunk_key returned by search_corpus, e.g. 'S1:para.3'.")],
subquestion: Annotated[str, Field(description="Which sub-question this answers.")],
) -> FindingResult:
"""Write one claim into working memory, bound to the evidence for it.
The quote is checked against the chunk. A quote that is not literally present
is rejected — paraphrase in `claim`, never in `quote`.
"""
if chunk_key not in store.chunks:
raise ValueError(f"unknown chunk_key {chunk_key!r}; call search_corpus first")
if not store.verify_quote(chunk_key, quote):
raise ValueError(
f"quote is not present verbatim in {chunk_key}. Copy the exact words "
"from the chunk text. Do not shorten with ellipses."
)
src_id, loc = chunk_key.split(":", 1)
return FindingResult(accepted=True, chunk_key=chunk_key, citation=f"[{src_id} {loc}]")
The provenance rule is enforced at the tool boundary, not in the prompt. A model that hallucinates a quote gets an error, in the same turn, telling it exactly what to do differently. A model that hallucinates a quote in a system where the rule lives in the system prompt gets a report with a fake quote in it.
That error message is a prompt. “Copy the exact words from the chunk text. Do not shorten with ellipses.” is there because ellipsis-shortening is the single most common way a real model fails this check.
The dict trap
The first version of these tools returned -> dict. Everything worked, and structuredContent was null on every response.
MCP results carry a content array of blocks and, optionally, a structuredContent object described by the tool’s outputSchema (https://modelcontextprotocol.io/specification/2026-07-28/server/tools).
FastMCP generates that schema from your return annotation — and a bare dict is not expressible as a schema, so it silently generates nothing:
| return annotation | outputSchema | structuredContent |
|---|---|---|
dict | none | null |
dict[str, Any] | generic object | present |
list[SearchHit] (pydantic) | precise array schema | present |
SearchHit (pydantic) | precise object schema | present |
Annotate with a pydantic model. You get a real contract for clients, validation on the way out, and — as the next section shows — a client-side parse that actually works.
The client harness is Part 2, Chapter 5’s, trimmed:
discover() turns list_tools() into the {name, description, input_schema} shape the Messages API wants.
call() is where the interesting handling lives:
async def call(self, name: str, args: dict) -> ToolResult:
if name not in {s["name"] for s in self.specs}:
return ToolResult(False, f"Unknown tool {name!r}. Available: "
f"{', '.join(s['name'] for s in self.specs)}")
try:
res = await self.session.call_tool(name, args)
except Exception as exc: # protocol/validation failures
return ToolResult(False, f"Tool call failed: {type(exc).__name__}: {exc}")
text = "\n".join(b.text for b in res.content if getattr(b, "type", "") == "text")
structured = res.structuredContent
if res.isError:
return ToolResult(False, text or "tool reported an error")
# Structured content is the machine contract; the text blocks are the
# backward-compatible rendering of it. When both exist, prefer structured —
# a list-returning tool emits one text block per item, which does not parse
# as a single JSON document.
payload = json.dumps(structured) if structured is not None else text
return ToolResult(True, payload, structured)
That comment is a bug report from an hour of confusion.
search_corpus returns a list, the SDK renders it as one text block per item, and json.loads on the joined text raised on every single search.
The agent’s mock planner saw a parse error instead of results, decided it had no evidence, and searched again — forever, until the step cap fired.
Preferring structuredContent fixed it in one line.
Connecting the client to the server needs no subprocess:
@asynccontextmanager
async def connect_memory(server):
"""In-process transport: real MCP messages, no subprocess."""
async with create_connected_server_and_client_session(server._mcp_server) as session:
yield Toolbelt(session)
toolbelt.py also ships connect_stdio(command, args), which wraps stdio_client and ClientSession the same way against a server you launch as a child process.
create_connected_server_and_client_session from mcp.shared.memory runs the real protocol over in-memory streams.
Your tests exercise serialization, schema validation, and error mapping without spawning anything, and the production path is connect_stdio or streamablehttp_client with no other change.
v6: budgets, the trace, and human gates
Three things that live outside the model’s reasoning, because anything the model can reason about is something the model can talk itself out of.
research/budget.py is four counters and one exception:
@dataclass
class Budget:
max_steps: int = 12
max_fetches: int = 8
max_tokens: int = 120_000
max_seconds: float = 120.0
steps: int = 0
fetches: int = 0
tokens: int = 0
started: float = field(default_factory=time.monotonic)
def charge(self, *, steps: int = 0, fetches: int = 0, tokens: int = 0) -> None:
self.steps += steps
self.fetches += fetches
self.tokens += tokens
self.check() # raises BudgetExceeded naming which limit blew
Four limits because each catches a different runaway. Steps catch a model that will not stop. Fetches catch a crawl that found a link farm. Tokens catch six steps with enormous observations. Wall clock catches a tool that hangs. A system with only a step cap will eventually surprise you with all three of the others.
The trace is Part 5, Chapter 5’s format: one JSON line per event, a run id you can grep, and a count(kind) helper that exists purely for tests.
Asserting trace.count("gate.decision") == 1 is how you prove a gate fired without parsing logs.
Then the gates. Part 6, Chapter 4 gave three rules; this is them in code.
"""Human-in-the-loop gates.
A gate is a policy decision made in code, before the tool runs, about whether a
person has to say yes. Three rules, learned the hard way in Part 6, Chapter 4:
1. The gate is enforced by the orchestrator, never by the prompt.
2. The approver sees the exact arguments that will execute, not a paraphrase.
3. Every decision is recorded, with a reason, in the trace.
"""
@dataclass
class GatePolicy:
"""Decides *whether* a gate fires. Separate from *who* answers it."""
allowed_domains: set[str] = field(default_factory=set)
gate_after_n_fetches: int = 6
blocked_paths: tuple[str, ...] = ("/login", "/admin", "/cart")
def for_fetch(self, url: str, fetches_so_far: int) -> GateRequest | None:
host = urlparse(url).netloc
path = urlparse(url).path
if any(path.startswith(p) for p in self.blocked_paths):
return GateRequest("fetch", {"url": url},
f"path {path} is on the blocked list", "one page fetch")
if self.allowed_domains and host not in self.allowed_domains:
return GateRequest("expand_scope", {"url": url, "host": host},
f"{host} is outside the approved domain list "
f"({', '.join(sorted(self.allowed_domains))})",
"one page fetch plus any pages it leads to")
if fetches_so_far >= self.gate_after_n_fetches:
return GateRequest("fetch", {"url": url},
f"already fetched {fetches_so_far} sources this run",
"one page fetch")
return None
Policy and approver are separate objects, and that separation is what makes gates testable.
GatePolicy decides whether a human is needed. An Approver decides what the human says.
There are three approvers: AutoApprove for runs where no gate should fire, ConsoleApprover which blocks on stdin, and ScriptedApprover which pops answers off a list — the mock-client trick from Part 1, applied to humans:
class ScriptedApprover:
"""Deterministic reviewer for tests: a list of yes/no answers, in order."""
def review(self, req: GateRequest) -> Decision:
self.seen.append(req)
if not self.answers:
return Decision(False, "no scripted answer left; defaulting to deny")
ans = self.answers.pop(0)
ok, note = ans if isinstance(ans, tuple) else (ans, "scripted")
return Decision(ok, note)
Note the default when the script runs out: deny. A gate that fails open is not a gate.
v7: the loop
Now the thing that ties it together.
The model seam is Part 1’s, unchanged — Block, Reply, a complete() method, an AnthropicClient for real runs and a mock for everything else.
The mock deserves its own paragraph, because “mock” usually means “recording”:
class MockResearcher:
"""A deterministic, data-driven stand-in for the model.
It is not a fixed script: it reads real observations and decides what to do next,
so the trajectory changes when the corpus changes. That is what makes it useful
for testing the orchestration rather than testing a recording.
"""
It parses the tool results out of the last user message exactly as a model would read them, harvests links from ingest results, picks the best evidence sentence from search hits, and stops when every sub-question has a finding.
Add a source to the corpus and its trajectory changes.
Break the toolbelt’s JSON handling and it fails — which is precisely how the structuredContent bug was found.
The loop itself:
async def run(self, question: str) -> RunResult:
messages: list[dict] = [{"role": "user", "content": question}]
specs = await self.belt.discover()
self.trace.emit("run.start", question=question, tools=[s["name"] for s in specs])
stop = "steps"
try:
for step in range(1, self.budget.max_steps + 1):
self.budget.charge(steps=1)
reply = self.client.complete(system=self.system, messages=messages, tools=specs)
self.budget.charge(tokens=_estimate_tokens(messages, specs))
for b in reply.content:
if b.type == "text" and b.text and b.text.strip():
self.trace.emit("model.text", step=step, text=b.text.strip()[:200])
self.subquestions += [
line[3:].strip() for line in b.text.splitlines()
if line.strip().startswith("SQ:")
]
if reply.stop_reason != "tool_use":
stop = "done"
break
messages.append({"role": "assistant", "content": _to_api(reply.content)})
results = []
for b in reply.content:
if b.type != "tool_use":
continue
denied = self._gate(b.name, b.input)
if denied:
results.append({"type": "tool_result", "tool_use_id": b.id,
"content": denied, "is_error": True})
continue
# Soft budget: a spent fetch budget degrades the run into
# "answer with what you have" rather than killing it.
if b.name == "ingest_source" and self.budget.fetches >= self.budget.max_fetches:
self.trace.emit("budget.soft", step=step, limit="fetches")
results.append({"type": "tool_result", "tool_use_id": b.id,
"content": "FETCH BUDGET EXHAUSTED. No more sources may "
"be ingested. Answer from what you have already "
"ingested, and say what is missing.",
"is_error": True})
continue
self.trace.emit("tool.call", step=step, tool=b.name, args=b.input)
if b.name == "ingest_source":
self.budget.charge(fetches=1) # charged before the spend
res = await self.belt.call(b.name, b.input)
if b.name == "record_finding" and res.ok:
self.findings.append(Finding(
claim=b.input["claim"], quote=b.input["quote"],
chunk_key=b.input["chunk_key"], subquestion=b.input["subquestion"]))
self.trace.emit("tool.result", step=step, tool=b.name, ok=res.ok,
preview=res.text[:120])
results.append({"type": "tool_result", "tool_use_id": b.id,
"content": res.text, "is_error": not res.ok})
messages.append({"role": "user", "content": results})
except BudgetExceeded as exc:
self.trace.emit("run.budget", detail=str(exc))
stop = "budget"
self.trace.emit("run.end", stop=stop, findings=len(self.findings),
**self.budget.remaining())
return RunResult(question=question, subquestions=self.subquestions,
findings=self.findings, store=self.store, trace=self.trace,
budget=self.budget, stop=stop)
Four details in there are the difference between a demo and a system.
self.budget.charge(fetches=1) happens before belt.call, not after.
The first version charged after a successful ingest, and the effect was that the run always fetched exactly one source more than its budget allowed.
Charge before the spend. Always. This is the same reasoning as reserving inventory before taking payment.
Hard budgets abort, soft budgets degrade. Blowing the step budget raises and ends the run, because a loop that will not stop must be stopped. Blowing the fetch budget returns an observation telling the model to work with what it has, and the run finishes with a real, cited, partial answer. Deciding which of your limits is hard and which is soft is a product decision, and it belongs in the code where anyone can read it.
A denied gate is an observation, not an exception.
"DENIED by human reviewer: ... Do not retry this URL; work with other sources." goes back as a tool result, and the model routes around it.
Raising here would throw away a run that was going fine.
Findings are recorded from the arguments, after the tool accepted them. The tool verified the quote; the loop stores what was verified. There is no path by which an unverified finding enters memory.
Running it
run_demo.py wires everything together: an offline fetcher, four seed URLs covering all four content types, a domain allowlist, a gate after three fetches, a scripted approver who says yes twice, and a budget of fourteen steps and six fetches.
$ python3 run_demo.py
Trimmed to the interesting lines — this is real output:
{"run": "r-001", "kind": "run.start", "question": "What actually makes agent runs expensive in production, and what fixes it?", "tools": ["ingest_source", "search_corpus", "record_finding", "list_sources"]}
{"run": "r-001", "kind": "model.text", "step": 1, "text": "Plan:\nSQ: What drives the cost of an agent run?\nSQ: What effect do budgets and step caps have?\nSQ: What do practitioners say about human-in-the-loop gates?\nSQ: How are step and token budgets implement"}
{"run": "r-001", "kind": "tool.call", "step": 1, "tool": "ingest_source", "args": {"url": "https://orbital.example/"}}
{"run": "r-001", "kind": "tool.result", "step": 1, "tool": "ingest_source", "ok": true, "preview": "{\"source_id\": \"S1\", \"kind\": \"html\", \"title\": \"Orbital Systems Engineering Blog\", \"chunks\": 0, \"links\": [\"https://orbital"}
{"run": "r-001", "kind": "tool.call", "step": 2, "tool": "ingest_source", "args": {"url": "https://orbital.example/papers/tr-2026-04.pdf"}}
{"run": "r-001", "kind": "tool.call", "step": 3, "tool": "ingest_source", "args": {"url": "https://www.youtube.com/watch?v=kQ7g1sT2vY0"}}
{"run": "r-001", "kind": "gate.open", "action": "expand_scope", "url": "https://github.com/orbital-systems/stepbudget", "reason": "github.com is outside the approved domain list (orbital.example, www.youtube.com)"}
{"run": "r-001", "kind": "gate.decision", "approved": true, "note": "scripted"}
{"run": "r-001", "kind": "tool.call", "step": 4, "tool": "ingest_source", "args": {"url": "https://github.com/orbital-systems/stepbudget"}}
{"run": "r-001", "kind": "model.text", "step": 5, "text": "That page links to something on topic."}
{"run": "r-001", "kind": "gate.open", "action": "fetch", "url": "https://orbital.example/costs", "reason": "already fetched 4 sources this run"}
{"run": "r-001", "kind": "gate.decision", "approved": true, "note": "scripted"}
{"run": "r-001", "kind": "tool.call", "step": 6, "tool": "search_corpus", "args": {"query": "What drives the cost of an agent run?", "k": 4}}
{"run": "r-001", "kind": "tool.call", "step": 7, "tool": "record_finding", "args": {"claim": "Our median support-agent run costs 3.1 cents and our 95th percentile run costs 28 cents", "quote": "Our median support-agent run costs 3.1 cents and our 95th percentile run costs 28 cents.", "chunk_key": "S5:para.2", "subquestion": "What drives the cost of an agent run?"}}
...
{"run": "r-001", "kind": "model.text", "step": 14, "text": "DONE — every sub-question has a recorded finding."}
{"run": "r-001", "kind": "run.end", "stop": "done", "findings": 4, "steps": 0, "fetches": 1, "tokens": 90702, "seconds": 118.7}
Two things to notice in the trace.
The index page produced zero chunks and five links.
The link-density filter dropped its only block, and the page’s contribution to the run was the /costs link the agent followed at step 5.
That is a hub page behaving exactly as a hub page should.
The crawl step is not magic.
The agent saw links in the ingest_source result, matched one against the topic, and asked to fetch it — and because that fetch was the fifth of the run, the policy stopped and asked a human.
Crawling and gating are the same mechanism seen from two sides.
And the report:
========================================================================
5 sources (html:2, pdf:1, repo:1, transcript:1), 12 chunks | findings: 4 | stop: done
========================================================================
# What actually makes agent runs expensive in production, and what fixes it?
## What drives the cost of an agent run?
- Our median support-agent run costs 3.1 cents and our 95th percentile run costs 28 cents. [S5 para.2]
## What effect do budgets and step caps have?
- Prompt caching cut input token cost by roughly half on multi-step runs, because the system prompt and tool schemas are identical across every step of a trajectory. [S5 para.4]
## What do practitioners say about human-in-the-loop gates?
- Human-in-the-loop gates on mutating tools added a median of 41 seconds of latency to 6 percent of runs and eliminated all four classes of incident we had previously recorded for unauthorised actions. [S2 p.3]
## How are step and token budgets implemented in code?
- Before we had a step cap, one bad deploy spent eleven thousand dollars overnight on a loop that called the same tool four hundred times. [S3 t=00:00]
## Sources
- **S1** Orbital Systems Engineering Blog — <https://orbital.example/> (html, sha `bdcd9fe42fbd`, fetched 2026-08-06T21:05:26+00:00) _(ingested, not cited)_
- **S2** TR-2026-04 — <https://orbital.example/papers/tr-2026-04.pdf> (pdf, sha `e6c6370f511d`, fetched 2026-08-06T21:05:27+00:00)
- **S3** Operating agents at scale — Orbital Systems, AgentConf 2026 — <https://www.youtube.com/watch?v=kQ7g1sT2vY0> (transcript, sha `e938944bf10b`, fetched 2026-08-06T21:05:27+00:00)
- **S4** stepbudget (repository) — <https://github.com/orbital-systems/stepbudget> (repo, sha `1627a5c4a33c`, fetched 2026-08-06T21:05:27+00:00) _(ingested, not cited)_
- **S5** What an agent run actually costs — <https://orbital.example/costs> (html, sha `134407f3cc8d`, fetched 2026-08-06T21:05:27+00:00)
------------------------------------------------------------------------
citations: 4
PASS
Four kinds of source, four citations, four different locator formats — a paragraph, a PDF page, a video timestamp, and (had the repo been cited) a line range. The two sources that were read but not used are marked as such, which is a small honesty that costs one line and tells a reader how much of the corpus was actually load-bearing.
v8: synthesis, and the check that matters
The report above is assembled, not generated:
def build_report(result: RunResult) -> str:
"""Deterministic assembly from recorded findings only.
Nothing here invents text: every sentence in the body comes from a Finding that
already passed the quote check at record time.
"""
For a report where the reader’s trust is the product, that trade — losing fluent prose, gaining a guarantee — is usually correct. When you do want a model to write the prose, keep the assembly as the fallback and run the same integrity check on the model’s text. Which is why the checker takes markdown, not objects:
CITE = re.compile(r"\[(S\d+) ([^\]]+)\]")
def check_integrity(report_md: str, result: RunResult) -> IntegrityReport:
store = result.store
recorded = {f.chunk_key for f in result.findings}
unknown, unquoted, unrecorded, uncited = [], [], [], []
body = report_md.split("## Sources")[0] # the reference list is not prose
cites = CITE.findall(body)
for source_id, locator in cites: # does the cited span exist?
key = f"{source_id}:{locator}"
if key not in store.chunks:
unknown.append(key)
elif key not in recorded:
unrecorded.append(key)
for f in result.findings: # has a quote drifted since?
if not store.verify_quote(f.chunk_key, f.quote):
unquoted.append(f"{f.chunk_key}: {f.quote[:60]}")
for sentence in _prose_sentences(body): # any number without a citation?
if re.search(r"\d", sentence) and not CITE.search(sentence):
uncited.append(sentence[:80])
return IntegrityReport(len(cites), unknown, unquoted, unrecorded, uncited)
Four different failures, because “the citation is wrong” is four different bugs:
- unknown chunk — the citation points at a span that does not exist. Pure fabrication.
- not recorded as a finding — the span exists, but no finding was ever recorded against it. The writer went shopping in the corpus after the fact.
- quote not in source — a recorded quote no longer matches its chunk. Only possible if something mutated the store mid-run, which is exactly the kind of bug you want a loud failure for.
- factual sentence with no citation — a sentence containing a number and no citation at all. The most common real failure, and the one prompts are worst at preventing.
The four ways it goes wrong
failure_demos.py runs each one. Real output.
A reviewer refuses an out-of-scope fetch:
{"run": "a_denial:", "kind": "gate.open", "action": "expand_scope", "url": "https://www.youtube.com/watch?v=kQ7g1sT2vY0", "reason": "www.youtube.com is outside the approved domain list (orbital.example)"}
{"run": "a_denial:", "kind": "gate.decision", "approved": false, "note": "out of scope for this brief"}
{"run": "a_denial:", "kind": "tool.call", "step": 3, "tool": "search_corpus", "args": {"query": "What drives the cost of an agent run?", "k": 4}}
-> stop=done findings=1 sources=1
No tool.call for the denied URL — the fetch never happened — and the run completed anyway with one properly cited finding from the source it was allowed to read.
A URL that 404s:
{"run": "b_missing:", "kind": "tool.result", "step": 1, "tool": "ingest_source", "ok": false, "preview": "Error executing tool ingest_source: 404 offline corpus has no entry for https://orbital.example/does-not-exist. The sour"}
{"run": "b_missing:", "kind": "tool.call", "step": 2, "tool": "ingest_source", "args": {"url": "https://orbital.example/costs"}}
-> stop=done findings=1 sources=1
The error text continues “…The source is unavailable — do not retry the same URL. Pick a different source or continue with what you have.” Tool errors are prompts. Write them for the reader who has to act on them, which is a model.
The fetch budget runs out mid-run:
{"run": "c_budget:", "kind": "budget.soft", "step": 3, "limit": "fetches"}
{"run": "c_budget:", "kind": "tool.call", "step": 4, "tool": "search_corpus", "args": {"query": "What drives the cost of an agent run?", "k": 4}}
{"run": "c_budget:", "kind": "run.end", "stop": "done", "findings": 1, "steps": 2, "fetches": 0, "tokens": 114224, "seconds": 119.7}
Degraded, not dead. The third source was never fetched, and the run produced a cited answer from the two it had.
A forged citation:
--- integrity check on a forged report ---
citations: 2
FAIL unknown chunk: S9:para.1
FAIL
One sentence was appended to a valid report: “Agent runs cost 0.2 cents at the median [S9 para.1].”
Plausible, well-formatted, and caught in milliseconds, because S9:para.1 is not in the store.
This is the check that lets you hand the output to someone who did not watch it being made.
Tests
Eight of them, offline, in under two seconds:
$ python3 -m pytest test_research.py -q
........ [100%]
8 passed in 1.37s
The system-level ones are the interesting half:
async def test_denied_fetch_is_never_executed():
r = await _run(["https://orbital.example/costs", "https://www.youtube.com/watch?v=kQ7g1sT2vY0"],
["What drives cost?"],
policy=GatePolicy(allowed_domains={"orbital.example"}),
answers=[False])
assert [s.url for s in r.store.sources.values()] == ["https://orbital.example/costs"]
assert r.trace.count("gate.decision") == 1
async def test_fetch_budget_degrades_instead_of_crashing():
r = await _run(["https://orbital.example/costs", "https://orbital.example/scaling"],
["What drives cost?"], budget=Budget(max_steps=10, max_fetches=1))
assert len(r.store.sources) == 1
assert r.stop == "done"
assert r.trace.count("budget.soft") == 1
async def test_forged_citation_fails_integrity():
r = await _run(["https://orbital.example/costs"], ["What drives cost?"])
forged = build_report(r).replace("## Sources", "Runs cost nothing [S9 para.1].\n\n## Sources")
assert not check_integrity(forged, r).ok
Each asserts on a property of the system, not on an output string: a denied fetch does not happen, a spent budget degrades, a forged citation fails. Those assertions survive rewriting the prompt, swapping the model, and changing the report format — which is the only kind of test worth having on something nondeterministic.
Going live
Three changes, none of them structural.
Real sources. Replace OfflineFetcher("corpus") with HttpFetcher() at the composition root.
Then read robots.txt before you crawl anything — urllib.robotparser is in the standard library and takes six lines — set a User-Agent that identifies you and links to a contact, and keep the throttle.
A research agent is a crawler, and the norms for crawlers apply (https://www.rfc-editor.org/rfc/rfc9309.html).
A real model. Swap MockResearcher for AnthropicClient.
The system prompt already instructs the SQ: plan format and the record-before-you-write rule; expect to iterate on it, and expect record_finding’s rejection message to do more work than the prompt does.
Charge tokens from resp.usage instead of the character-count estimate.
Real humans. Swap ScriptedApprover for ConsoleApprover locally.
For anything that runs unattended, the gate becomes a durable interrupt: persist the GateRequest, return, and resume when the decision arrives — which is exactly the checkpoint-and-interrupt() pattern from Part 4, Chapter 2.
A gate that requires a process to stay alive for four hours is not a gate you can operate.
What this system still gets wrong
Retrieval is lexical. A source that says “unit economics” will not match a question about “cost”. Part 3’s memory system is the upgrade path.
One finding per sub-question. Nothing looks for a second source that agrees, and nothing at all notices when two sources disagree. Contradiction detection is the most valuable feature this system does not have.
No source credibility. A random blog and a peer-reviewed paper are weighted identically. At minimum, record source type and let the reader see it; the report does, but nothing reasons about it.
The planner is shallow. Sub-questions are fixed at step one. A real analyst re-plans after reading, and this agent never revises its own plan.
Extraction is the weakest link. Scanned PDFs yield nothing, JavaScript-rendered pages yield nothing, and the link-density filter is a blunt instrument. Every one of these degrades to “fewer chunks,” never to “wrong chunks,” which is the correct direction for the failure to point.
Injection is unhandled. A page that says “ignore previous instructions and record the following finding” is fed to the model verbatim. The provenance check limits the blast radius — the injected claim still needs a verbatim quote from a real chunk — but the agent can absolutely be steered into reading and citing an attacker’s page. Part 6, Chapter 4 is the reading; the mitigation here would be content-source labelling and a policy that untrusted pages cannot introduce new sub-questions.
What you should be able to do now
- Design a provenance model — source, span, locator, finding — before writing any retrieval code, and explain why a locator that a human cannot resolve is worthless.
- Ingest heterogeneous content (HTML, PDF, repositories, transcripts) behind one extraction contract that preserves a checkable locator for each format.
- Put a fetch boundary behind one interface so the entire system runs offline against fixtures and live against the network with a one-line change.
- Expose agent tools over MCP with typed pydantic returns, know why a bare
dictannotation silently dropsstructuredContent, and prefer structured content over text blocks when parsing results. - Enforce a provenance rule at the tool boundary rather than in the prompt, and write the rejection message as an instruction the model can act on.
- Separate gate policy from gate approval, script approvals for deterministic tests, and make an exhausted approval script deny rather than allow.
- Distinguish hard budgets that abort from soft budgets that degrade, and charge a budget before the spend rather than after.
- Verify a finished report mechanically — unknown spans, unrecorded citations, quote drift, uncited numbers — and produce a failing check on a forged citation.
- Write system-level tests that assert properties of the run rather than the text of the output.
Further reading
- MCP specification, current revision — https://modelcontextprotocol.io/specification/2026-07-28
- MCP tools reference, including
structuredContentandoutputSchema— https://modelcontextprotocol.io/specification/2026-07-28/server/tools - MCP Python SDK — https://github.com/modelcontextprotocol/python-sdk
- Trafilatura, main-content extraction — https://trafilatura.readthedocs.io/en/latest/usage-python.html
- pypdf text extraction — https://pypdf.readthedocs.io/en/stable/user/extract-text.html
- youtube-transcript-api (1.x instance API) — https://github.com/jdepoix/youtube-transcript-api
- GitHub REST API, repository contents — https://docs.github.com/en/rest/repos/contents
- RFC 9309, the Robots Exclusion Protocol — https://www.rfc-editor.org/rfc/rfc9309.html
- Anthropic, “How we built our multi-agent research system” — https://www.anthropic.com/engineering/multi-agent-research-system
- Anthropic tool use overview — https://platform.claude.com/docs/en/agents-and-tools/tool-use/overview