Build a Context Layer for an AI Agent Using Web Data

An AI agent is only as good as the information you give it. When that information comes from the web, three problems show up fast.

The agent needs different pages for different tasks. Pages change while a session is running. And when someone asks "where did that answer come from?", you need to know which version of which page the agent saw.

A context layer solves these problems in one place. In this tutorial, you will build a small one in Python. It fetches web data for each task, saves page versions, tracks which versions each session used and throws out stale data.

How does a context layer differ from a vector store?

A vector store holds a large index of text chunks. You send it a query, and it returns the chunks that look most similar. It is built for search across a big, long-lived collection.

A context layer works per task and per session. It knows which sources a task needs, how fresh they must be and which version each session saw. It can use a vector store inside it, but it does a different job.

If you need a long-lived search index, read our guide to building a RAG pipeline with web data. This tutorial covers the session layer that sits in front of your agent.

What you will build

The example is a support agent that answers questions using a company's public help pages. It handles two kinds of questions:

• Billing questions use the pricing page and the billing FAQ.
• Incident questions use the status page. The context layer has four parts:

1. A task registry that lists the sources and freshness rules for each task
2. A snapshot store that saves each version of each page
3. A fetcher that calls MrScraper when a page is missing or stale
4. A get_context function that the agent calls The example uses SQLite to keep things simple. You can swap in Postgres or another database later.

Step 1: Define tasks and sources

Each task lists its sources, how old each source may be and the terms used to pick passages.

from datetime import timedelta
 
TASKS = {
    "billing_question": {
        "sources": [
            {"url": "https://www.example.com/pricing", "max_age": timedelta(hours=24)},
            {"url": "https://help.example.com/billing-faq", "max_age": timedelta(hours=24)},
        ],
        "terms": ["price", "plan", "invoice", "refund", "billing"],
    },
    "incident_question": {
        "sources": [
            {"url": "https://status.example.com", "max_age": timedelta(minutes=10)},
        ],
        "terms": ["incident", "outage", "degraded", "resolved"],
    },
}

Notice the different freshness rules. A pricing page rarely changes within a day. A status page can change in minutes. Your application sets these rules based on what each task needs.

Step 2: Create the snapshot store

Two tables do the work. snapshots stores every version of every page. session_sources logs which version each session used.

import sqlite3
 
db = sqlite3.connect("context_layer.db")
db.executescript("""
CREATE TABLE IF NOT EXISTS snapshots (
    url TEXT NOT NULL,
    version INTEGER NOT NULL,
    content_hash TEXT NOT NULL,
    fetched_at TEXT NOT NULL,
    markdown TEXT NOT NULL,
    PRIMARY KEY (url, version)
);
CREATE TABLE IF NOT EXISTS session_sources (
    session_id TEXT NOT NULL,
    url TEXT NOT NULL,
    version INTEGER NOT NULL,
    used_at TEXT NOT NULL,
    PRIMARY KEY (session_id, url, version)
);
""")

Step 3: Fetch pages with MrScraper

The fetcher uses the MrScraper Scrape Markdown endpoint. It returns the page as clean Markdown, which keeps headings and drops HTML tags. For pages that load content in the browser, set browserRendering to true.

import os
import requests
 
MRSCRAPER_URL = "https://api.mrscraper.com/"
TOKEN = os.environ["MRSCRAPER_API_TOKEN"]
 
 
def fetch_markdown(url: str, render: bool = False) -> str:
    params = {"token": TOKEN, "markdown": "true"}
    if render:
        params["browserRendering"] = "true"
 
    response = requests.post(
        MRSCRAPER_URL,
        params=params,
        headers={"x-api-token": TOKEN, "Content-Type": "application/json"},
        json={"url": url, "maxRetries": 2},
        timeout=330,
    )
    response.raise_for_status()
    return response.json().get("markdown", "")

Keep this call on your server. Your API token should never reach the browser or the model's context.

Step 4: Save new versions only when content changes

Each time you fetch a page, hash its content. If the hash matches the latest version, update the fetch time. If it is different, save a new version.

import hashlib
from datetime import datetime, timezone
 
 
def now() -> datetime:
    return datetime.now(timezone.utc)
 
 
def row_to_snapshot(url, row):
    return {
        "url": url,
        "version": row[0],
        "content_hash": row[1],
        "fetched_at": datetime.fromisoformat(row[2]),
        "markdown": row[3],
    }
 
 
def latest_snapshot(url: str):
    row = db.execute(
        "SELECT version, content_hash, fetched_at, markdown FROM snapshots "
        "WHERE url = ? ORDER BY version DESC LIMIT 1",
        (url,),
    ).fetchone()
    return row_to_snapshot(url, row) if row else None
 
 
def refresh(url: str) -> dict:
    markdown = fetch_markdown(url)
    content_hash = hashlib.sha256(markdown.encode()).hexdigest()
    latest = latest_snapshot(url)
 
    if latest and latest["content_hash"] == content_hash:
        db.execute(
            "UPDATE snapshots SET fetched_at = ? WHERE url = ? AND version = ?",
            (now().isoformat(), url, latest["version"]),
        )
    else:
        version = latest["version"] + 1 if latest else 1
        db.execute(
            "INSERT INTO snapshots VALUES (?, ?, ?, ?, ?)",
            (url, version, content_hash, now().isoformat(), markdown),
        )
    db.commit()
    return latest_snapshot(url)

Some pages include parts that change on every load, like a "last updated" clock. Those parts create new versions even when nothing important changed. If that happens, hash only the passages your tasks use.

How do you fetch task-specific context?

This is the function your agent calls. It takes a session ID and a task name, and it returns a list of source packets.

For each source in the task, it:

1. Loads the latest snapshot
2. Refreshes the page if it is missing or older than max_age
3. Logs the version this session is using
4. Builds a packet with the URL, version, fetch time and the most relevant passages

import re
 
 
def split_by_headings(markdown: str) -> list[dict]:
    passages, heading, lines = [], "Top of page", []
    for line in markdown.splitlines():
        if re.match(r"^#{1,4} ", line):
            if lines:
                passages.append({"heading": heading, "text": "\n".join(lines).strip()})
            heading, lines = line.lstrip("#").strip(), []
        else:
            lines.append(line)
    if lines:
        passages.append({"heading": heading, "text": "\n".join(lines).strip()})
    return [p for p in passages if p["text"]]
 
 
def make_packet(source_id, snapshot, terms, max_passages=3, max_chars=800):
    def score(p):
        text = (p["heading"] + " " + p["text"]).lower()
        return sum(text.count(t) for t in terms)
 
    ranked = sorted(split_by_headings(snapshot["markdown"]), key=score, reverse=True)
    return {
        "source_id": source_id,
        "url": snapshot["url"],
        "version": snapshot["version"],
        "retrieved_at": snapshot["fetched_at"].isoformat(),
        "evidence": [
            {"heading": p["heading"], "text": p["text"][:max_chars]}
            for p in ranked[:max_passages]
            if score(p) > 0
        ],
    }
 
 
def is_stale(snapshot: dict, max_age: timedelta) -> bool:
    return now() - snapshot["fetched_at"] > max_age
 
 
def get_context(session_id: str, task: str) -> list[dict]:
    spec = TASKS[task]
    packets = []
 
    for i, source in enumerate(spec["sources"], start=1):
        url = source["url"]
        snapshot = latest_snapshot(url)
        if snapshot is None or is_stale(snapshot, source["max_age"]):
            snapshot = refresh(url)
 
        previous = last_version_used(session_id, url)
        log_version(session_id, url, snapshot["version"])
 
        packet = make_packet(f"S{i}", snapshot, spec["terms"])
        packet["changed_during_session"] = (
            previous is not None and previous != snapshot["version"]
        )
        packets.append(packet)
 
    return packets

The changed_during_session flag matters. If the pricing page changed after the agent already answered a billing question, the agent can tell the user. For example: "The pricing page has changed since my earlier answer. Here is the updated information."

For more on what to put in each packet and how to pick passages, see our guide to context engineering for agents that use live web data.

How do you track source versions per session?

Every time get_context returns a source, it logs the version in session_sources. These two helpers do that:

def log_version(session_id: str, url: str, version: int) -> None:
    db.execute(
        "INSERT OR IGNORE INTO session_sources VALUES (?, ?, ?, ?)",
        (session_id, url, version, now().isoformat()),
    )
    db.commit()
 
 
def last_version_used(session_id: str, url: str):
    row = db.execute(
        "SELECT MAX(version) FROM session_sources WHERE session_id = ? AND url = ?",
        (session_id, url),
    ).fetchone()
    return row[0]

This log lets you answer three useful questions:

• What did the agent see? Look up the session and load each logged version from snapshots.
• Why did two answers differ? Check whether the version changed between them.
• Can I reproduce a bad answer? Rerun the agent with the exact snapshots from that session.

Choose a version policy

This tutorial uses a "refresh and flag" policy. The session gets the newest fresh version, and the packet says when it changed.

Some tasks need the opposite. Picture an agent that writes a report over several steps. You may want every step to use the same version of each page, even if the page changes. For that, check last_version_used first and load that version instead of refreshing.

Pick the policy that fits each task. You can store it in the task registry next to max_age.

How do you expire stale context?

Stale context leads to wrong answers. Use more than one rule to catch it.

Time-based expiry

The max_age setting already handles this. When a snapshot is older than the limit, the next request fetches a fresh copy.

Change-based expiry

The content hash catches real changes. A new hash means a new version, and the session flag tells the agent.

Scheduled refresh

For pages that many sessions use, you may not want the first user to wait for a fetch. You can refresh those pages ahead of time.

MrScraper offers a Schedule feature for automated runs and a Webhook integration for sending results to your services. Check the docs to see whether they fit your setup. Your webhook handler would then save a new snapshot using the same hashing logic as refresh.

Clean up old data

Old sessions and old versions take up space. Remove them on a regular schedule, but keep any version a recent session still points to.

def cleanup(session_days: int = 7, keep_versions: int = 5) -> None:
    cutoff = (now() - timedelta(days=session_days)).isoformat()
    db.execute("DELETE FROM session_sources WHERE used_at < ?", (cutoff,))
    db.execute(
        """
        DELETE FROM snapshots
        WHERE version <= (
            SELECT MAX(version) FROM snapshots s2 WHERE s2.url = snapshots.url
        ) - ?
        AND NOT EXISTS (
            SELECT 1 FROM session_sources ss
            WHERE ss.url = snapshots.url AND ss.version = snapshots.version
        )
        """,
        (keep_versions,),
    )
    db.commit()

Set these numbers to match your own audit and storage needs.

Putting it together

Here is how the agent uses the layer during a session:

packets = get_context(session_id="sess_123", task="billing_question")
 
prompt = build_prompt(
    question="Can I get a refund if I cancel mid-month?",
    sources=packets,
    rules="Answer only from the sources. Cite each claim with its source_id.",
)
answer = call_your_model(prompt)

build_prompt and call_your_model are your own code. The context layer does not care which model or framework you use. It hands over clean, versioned, task-specific packets, and your agent takes it from there.

FAQs

How does a context layer differ from a vector store?

A vector store finds text similar to a query across a large index. A context layer decides which sources a task needs, checks that they are fresh enough and tracks which version each session used. A context layer can use a vector store inside it.

How do you expire stale context?

Combine a few rules. Set a maximum age for each source. Use a content hash to detect real changes. Refresh busy pages on a schedule. Clean up old sessions and versions that nothing points to.

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, returns clean Markdown or HTML, and can extract structured JSON from a prompt and schema. It also offers scheduling and webhooks.

Your application builds the context layer itself. That includes the task registry, freshness rules, the snapshot store, version tracking, passage selection and cleanup. Your team also decides which sources it has permission to collect from.