Chapter 8: Data Access Patterns
Data access is where many MCP servers become genuinely useful for the first time—and genuinely dangerous for the first time.
The danger comes from confusing “the model can reach the data” with “the model should have a direct interface to the data store.” Those are not the same thing. A raw SQL tool feels powerful in a demo. In production, it is an invitation for the model to write a SELECT * with no LIMIT, join five tables on unindexed columns, or—worse—interpret a user’s casual request as a DELETE. Good data-access servers mediate. They expose information in shapes that are safe, comprehensible, and aligned with actual user intent rather than database internals.
The five patterns below share one rule: the model says what it wants, and the server decides what query runs. Each pattern moves that decision a step further from the model, from intent-shaped query tools, through search and fetch, filesystem projections and precomputed snapshots, to narrow write paths.
Pattern 1: The Database Wrapper
The naive version of this pattern is a single run_sql tool that accepts arbitrary SQL. It feels maximally flexible and is usually a mistake. The model does not know your schema intimately, will produce expensive queries without realizing it, and has no intuition about which tables contain sensitive data. The flexibility you think you are giving the model is actually a liability you are handing to whoever writes the prompt.
The better approach is intent-shaped tools: named operations that express what the user wants to accomplish, backed by parameterized queries that the server controls.
Intent-Shaped Query Tools
Consider an MCP server for a customer support database. Instead of exposing raw SQL, you expose tools that match the questions a support agent actually asks:
import aiosqlite
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
mcp = MCPServer("Support Database")
DB_PATH = "/data/support.db"
MAX_CUSTOMERS = 10
def connect_ro() -> aiosqlite.Connection:
"""Read path: SQLite refuses every write on this connection."""
return aiosqlite.connect(f"file:{DB_PATH}?mode=ro", uri=True)
@mcp.tool()
async def find_customer(
email: str = "",
name: str = "",
account_id: str = "",
) -> dict:
"""Find a customer by email, name, or account ID.
Provide at least one search parameter. Returns matching
customers with their account status and recent ticket count.
"""
if not any([email, name, account_id]):
raise ToolError("Provide at least one of: email, name, account_id.")
conditions, params = [], []
if email:
conditions.append("c.email = ?")
params.append(email.lower().strip())
if name:
conditions.append("c.name LIKE ?")
params.append(f"%{name}%")
if account_id:
conditions.append("c.account_id = ?")
params.append(account_id)
where = " AND ".join(conditions)
query = f"""
SELECT c.account_id, c.name, c.email, c.status,
COUNT(t.id) as recent_tickets
FROM customers c
LEFT JOIN tickets t ON t.customer_id = c.id
AND t.created_at > datetime('now', '-90 days')
WHERE {where}
GROUP BY c.id
LIMIT {MAX_CUSTOMERS + 1}
"""
async with connect_ro() as db:
db.row_factory = aiosqlite.Row
async with db.execute(query, params) as cursor:
rows = await cursor.fetchall()
return {
"results": [
{
"account_id": row["account_id"],
"name": row["name"],
"email": row["email"],
"status": row["status"],
"recent_tickets": row["recent_tickets"],
}
for row in rows[:MAX_CUSTOMERS]
],
"more_matches": len(rows) > MAX_CUSTOMERS,
}
@mcp.tool()
async def recent_tickets(
account_id: str,
limit: int = 10,
) -> dict:
"""List recent support tickets for a customer account.
Returns ticket ID, subject, status, and creation date,
ordered by most recent first.
"""
limit = min(limit, 50) # Hard cap
query = """
SELECT t.id, t.subject, t.status, t.priority,
t.created_at, t.resolved_at
FROM tickets t
JOIN customers c ON c.id = t.customer_id
WHERE c.account_id = ?
ORDER BY t.created_at DESC
LIMIT ?
"""
async with connect_ro() as db:
db.row_factory = aiosqlite.Row
async with db.execute(query, [account_id, limit]) as cursor:
rows = await cursor.fetchall()
return {
"account_id": account_id,
"tickets": [
{
"ticket_id": row["id"],
"subject": row["subject"],
"status": row["status"],
"priority": row["priority"],
"created": row["created_at"],
"resolved": row["resolved_at"],
}
for row in rows
],
"showing": len(rows),
"note": f"Showing up to {limit} most recent tickets.",
}
The numbers in this code are decisions, not defaults. find_customer stops at ten because a lookup that matches more than ten customers is a query too vague to act on. Nobody picks the right Jane Smith out of 200 rows, so the tool reports more_matches and lets the model ask for an email instead. recent_tickets caps at 50 because 50 ticket summaries come to roughly 3,000 tokens, enough history to spot a pattern and small enough to leave room for the conversation. The only f-string in the SQL interpolates text the server wrote (where, the cap). Every value the model supplies travels as a parameter. An empty result is a successful call with zero rows, not an error. The model should say “no match”, not retry.
Schema as a Resource, and as a Tool
Even with intent-shaped tools, the model sometimes needs the data model to answer well. Publish a curated schema that covers only the tables the tools can reach:
EXPOSED_TABLES = ("customers", "tickets", "ticket_comments")
@mcp.resource("db://schema")
async def database_schema() -> str:
"""Schema for the tables this server's tools can read."""
marks = ", ".join("?" for _ in EXPOSED_TABLES)
async with connect_ro() as db:
async with db.execute(
"SELECT sql FROM sqlite_master "
f"WHERE type='table' AND name IN ({marks}) ORDER BY name",
EXPOSED_TABLES,
) as cursor:
rows = await cursor.fetchall()
return (
"\n\n".join(row[0] for row in rows)
+ "\n\nRelationships:\n"
"- tickets.customer_id → customers.id\n"
"- ticket_comments.ticket_id → tickets.id\n"
)
@mcp.tool()
async def describe_schema() -> str:
"""Tables, columns and relationships the query tools can read."""
return await database_schema()
Resources are host-controlled. The host decides whether the schema is ever attached, and many hosts attach a resource only when the user picks it. A model that needs the schema on its own needs a tool, so the server exposes both, backed by one function. The resource suits hosts that attach context up front. The tool guarantees the model can orient itself anywhere.
The Constrained Query Escape Hatch
Some use cases genuinely need ad hoc queries. A data analyst using an LLM to explore a dataset should not be limited to five predefined tools. For these cases, offer a constrained query tool—but constrain it aggressively:
import re
BLOCKED = ("DROP", "DELETE", "INSERT", "UPDATE", "ALTER",
"CREATE", "ATTACH", "DETACH", "PRAGMA")
@mcp.tool()
async def query_readonly(
sql: str,
max_rows: int = 25,
) -> dict:
"""Execute a read-only SQL query against the support database.
Only SELECT statements are allowed. Results are limited
to prevent expensive queries from consuming resources.
Call describe_schema to see the available tables.
"""
max_rows = min(max_rows, 100)
sql = sql.strip().rstrip(";")
normalized = sql.upper()
problems = []
if not normalized.startswith("SELECT"):
problems.append("Only SELECT queries are allowed.")
# Whole tokens only: CREATE must not match CREATED_AT
problems += [f"Disallowed keyword: {kw}" for kw in BLOCKED
if re.search(rf"\b{kw}\b", normalized)]
if problems:
raise ToolError(" ".join(problems))
# Fetch one extra row so "truncated" is never a guess
limited_sql = f"SELECT * FROM ({sql}) LIMIT {max_rows + 1}"
try:
async with connect_ro() as db:
db.row_factory = aiosqlite.Row
await db.execute("PRAGMA busy_timeout = 5000")
async with db.execute(limited_sql) as cursor:
rows = await cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
except aiosqlite.Error as e:
raise ToolError(f"Query failed: {e}")
return {
"columns": columns,
"rows": [dict(row) for row in rows[:max_rows]],
"row_count": min(len(rows), max_rows),
"truncated": len(rows) > max_rows,
}
This is a compromise. The keyword blocklist is not a security boundary—a sufficiently creative query can work around string matching. The real protection comes from the read-only connection (mode=ro, or a separate read-only role on a server database) and the row limit enforced by the wrapping subquery. Think of the keyword filter as a fast-fail for obvious mistakes, not a fortress wall. Even a word-boundary blocklist fails open. That is why the connection above is opened read-only.
The max_rows + 1 fetch is the idiom to copy. With len(rows) == max_rows, a query that returns exactly 25 rows gets reported as truncated, and the model goes looking for rows that don’t exist.
For PostgreSQL deployments, use a dedicated read-only role with SET statement_timeout = '5s' at the session level. SQLite has no statement timeout: busy_timeout bounds only how long a query waits for a lock. To bound run time, install a progress handler that aborts the query after a deadline.
Read-Write Separation
Separate reads and writes at the connection: a SELECT-only user or mode=ro connection for every query tool and resource, and a write-capable one used only by explicit mutation tools. The database enforces this, not your Python code. If the read connection cannot execute a DELETE, no prompt injection can make a query tool delete data.
Pattern 2: The RAG Gateway
The weak version of a retrieval gateway is a single search tool that returns chunks. The strong version separates search from retrieval, manages context budget, and gives the model enough metadata to decide what to read further.
Search and Fetch as Separate Capabilities
Search is exploratory (“things related to X”); retrieval is precise (“the full content of document Y”). Collapsing them into one tool forces the model to over-fetch full documents on every search or under-fetch snippets with no way to get more.
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
mcp = MCPServer("Knowledge Base")
# vector_store is your vector database client (pgvector, Qdrant, ...);
# extract_section is a small helper that slices a document by heading.
@mcp.tool()
async def search_documents(
query: str,
collection: str = "default",
limit: int = 5,
min_score: float = 0.7,
) -> dict:
"""Search the knowledge base for documents related to a query.
Collections: default, product-docs, eng-wiki, legal.
Returns document titles, IDs, relevance scores, and short
excerpts. Use get_document to retrieve full content for
any result.
"""
limit = min(limit, 20)
results = await vector_store.search(
query=query,
collection=collection,
top_k=limit,
min_score=min_score,
)
return {
"query": query,
"collection": collection,
"results": [
{
"doc_id": r.id,
"title": r.metadata.get("title", "Untitled"),
"score": round(r.score, 3),
"excerpt": r.text[:200] + "..." if len(r.text) > 200 else r.text,
"source": r.metadata.get("source_url", ""),
"updated": r.metadata.get("updated_at", ""),
}
for r in results
],
"total_found": len(results),
}
@mcp.tool()
async def get_document(
doc_id: str,
section: str = "",
) -> dict:
"""Retrieve the full content of a document by its ID.
Optionally specify a section heading to retrieve only that
part of the document. Use search_documents first to find
relevant document IDs.
"""
doc = await vector_store.get_by_id(doc_id)
if not doc:
raise ToolError(f"Document {doc_id} not found. "
"Use an ID returned by search_documents.")
content = doc.text
if section:
content = extract_section(content, section) or content
return {
"doc_id": doc_id,
"title": doc.metadata.get("title", "Untitled"),
"content": content,
"source": doc.metadata.get("source_url", ""),
"last_updated": doc.metadata.get("updated_at", ""),
"word_count": len(content.split()),
}
The sizes here are budget arithmetic. A 200-character excerpt is about 50 tokens, so five results with titles and scores cost around 400 tokens. That’s enough for the model to judge relevance, and less than a tenth of one full document. The default of five results comes from the same trade: past five, the model rarely reads the tail, and each extra result costs as much as the first.
Collection Metadata as Resources
For knowledge bases with multiple collections (product docs, engineering wiki, legal policies), expose the collection catalog as a resource for hosts that attach it, and name the collections in search_documents’s description, as above, so the model sees them either way.
Chunking Sets the Ceiling
Chunking is an ingestion concern, but it sets the ceiling on every answer this server gives: bad chunks become bad search results, which the model presents as authoritative. Debug the ingestion pipeline and the server as one system.
Pattern 3: The Filesystem Projection
Filesystem access is the most deceptively simple pattern. The model asks to read a file. You read the file and return the contents. What could go wrong?
Path traversal, for one. Symlink following. Reading /etc/shadow because the model was told to “check the system configuration.” Returning a 500 MB log file because the model did not know it was large. Exposing .env files with production credentials because the sandbox was not properly scoped.
The reference filesystem server in the modelcontextprotocol/servers repository shipped the classic containment bug in 2025. CVE-2025-53110 was a string-prefix check: any path that began with the allowed directory’s name, such as /private/tmp/allow_dir_evil, counted as inside /private/tmp/allow_dir.
Scoped Projections
A filesystem MCP server should expose a projection—a curated, bounded view of a directory tree—not the raw filesystem. Define the root at startup and reject everything outside it:
from pathlib import Path
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
mcp = MCPServer("Project Docs")
# The only directory this server can see
PROJECT_ROOT = Path("/srv/project/docs").resolve()
MAX_FILE_SIZE = 512 * 1024 # 512 KB
def safe_path(relative: str) -> Path | None:
"""Resolve a relative path and verify it is inside the project root."""
try:
target = (PROJECT_ROOT / relative).resolve()
except (ValueError, OSError):
return None
# resolve() follows symlinks—check the resolved path, not the input
if not target.is_relative_to(PROJECT_ROOT):
return None
return target
@mcp.resource("docs://index")
def document_index() -> str:
"""Index of all documentation files in the project."""
files = sorted(PROJECT_ROOT.rglob("*.md"))
lines = ["Documentation Index", "=" * 19, ""]
for f in files:
rel = f.relative_to(PROJECT_ROOT)
size_kb = f.stat().st_size / 1024
lines.append(f"- {rel} ({size_kb:.1f} KB)")
return "\n".join(lines)
@mcp.tool()
def read_document(path: str) -> dict:
"""Read a documentation file by its path relative to the project root.
Call list_directory to see available files.
"""
target = safe_path(path)
if target is None:
raise ToolError("Invalid path. Path must be inside the "
"project documentation directory.")
if not target.is_file():
raise ToolError(f"No such file: {path}")
size = target.stat().st_size
if size > MAX_FILE_SIZE:
raise ToolError(
f"File too large ({size / 1024:.0f} KB; limit "
f"{MAX_FILE_SIZE / 1024:.0f} KB). Try list_directory for "
"smaller files on the same topic."
)
content = target.read_text(encoding="utf-8", errors="replace")
return {
"path": path,
"content": content,
"size_bytes": size,
}
The critical function here is safe_path. It calls .resolve() to follow symlinks before checking whether the resolved path is still inside the project root. This prevents traversal attacks that use ../ sequences or symlinks that point outside the sandbox. Without .resolve(), a symlink at docs/escape pointing to /etc/ would pass a naive containment check. A string prefix check is not enough either: /srv/project/docs-private starts with /srv/project/docs. That is the CVE-2025-53110 bug, and is_relative_to compares path components, not characters.
The Index Resource Needs a Tool Twin
The docs://index resource gives a host that attaches it the full map in one read, and saves the model from walking the tree one directory at a time. But the host controls resources. In a client that attaches them only on user action, the model never sees the index, so the model-callable path is a list_directory tool. For large trees, keep the index to top-level directories and file counts, and let the tool drill down:
@mcp.tool()
def list_directory(path: str = "") -> dict:
"""List files and subdirectories at a path relative to the project root."""
target = safe_path(path) if path else PROJECT_ROOT
if target is None or not target.is_dir():
raise ToolError("Invalid or non-existent directory.")
entries = []
for item in sorted(target.iterdir()):
if item.name.startswith("."):
continue # Skip hidden files
rel = item.relative_to(PROJECT_ROOT)
if item.is_dir():
count = sum(1 for _ in item.rglob("*") if _.is_file())
entries.append({"name": str(rel), "type": "directory",
"file_count": count})
else:
entries.append({"name": str(rel), "type": "file",
"size_kb": round(item.stat().st_size / 1024, 1)})
return {"path": path or "/", "entries": entries}
Pattern 4: The Analytical Snapshot
To answer “how are we doing this quarter?” with query tools, the model must pick tables, write aggregations and interpret the results, spending tokens and calls on work a nightly batch job does better.
Analytical snapshots return pre-computed digests: a sales summary, a deployment status rollup, a risk register, a customer health dashboard. A scheduled job runs the aggregation SQL and writes one row to a report_snapshots table. The server only reads that row:
import json
async def latest_snapshot(name: str) -> dict:
async with connect_ro() as db:
async with db.execute(
"SELECT body, computed_at FROM report_snapshots "
"WHERE name = ? ORDER BY computed_at DESC LIMIT 1",
[name],
) as cursor:
row = await cursor.fetchone()
if row is None:
raise ToolError(f"No '{name}' report has been computed yet.")
return {"report": name, "computed_at": row[1], **json.loads(row[0])}
@mcp.resource("reports://weekly-summary")
async def weekly_summary() -> str:
"""Tickets opened and resolved and SLA compliance, last 7 days."""
return json.dumps(await latest_snapshot("weekly-summary"))
@mcp.tool()
async def get_report(name: str = "weekly-summary") -> dict:
"""Latest precomputed report. computed_at says how fresh it is."""
return await latest_snapshot(name)
Snapshots are effective for three reasons. First, the aggregation runs once, on a schedule, rather than on every model request. Second, the model sees a clean summary rather than raw rows that it must aggregate in its reasoning—saving tokens and reducing hallucination risk. Third, expensive queries that touch large tables are bounded to batch execution, not interactive tool calls. None of that holds if the resource recomputes on read, which is why the read path here contains no aggregation at all.
computed_at travels with every response, so the model can say “as of 06:00 UTC” instead of presenting yesterday’s numbers as live. If the user needs a fresh computation, add a refresh_report tool that triggers the job—but rate-limit it, because the underlying queries may be expensive.
Pattern 5: The Write Path With Guardrails
If the model misreads intent and reads the wrong record, the cost is a confusing answer. If it writes the wrong record, the cost is corrupted data. Write tools should be narrow in scope and explicit in parameters. Contrast these two designs:
from datetime import datetime, timezone
# Bad: wide-open update tool
@mcp.tool()
async def update_record(table: str, id: str, fields: dict) -> dict:
"""Update any field on any table."""
...
# Good: specific mutation with clear semantics
@mcp.tool()
async def resolve_ticket(
ticket_id: str,
resolution_note: str,
) -> dict:
"""Mark a support ticket as resolved with a resolution note.
The ticket must be in 'open' or 'in_progress' status.
Returns the updated ticket state.
"""
now = datetime.now(timezone.utc).isoformat()
async with aiosqlite.connect(DB_PATH) as db: # write path
cursor = await db.execute(
"UPDATE tickets SET status = 'resolved', "
"resolution_note = ?, resolved_at = ? "
"WHERE id = ? AND status IN ('open', 'in_progress')",
[resolution_note, now, ticket_id],
)
await db.commit()
if cursor.rowcount == 0:
async with db.execute(
"SELECT status FROM tickets WHERE id = ?", [ticket_id]
) as check:
row = await check.fetchone()
if row is None:
raise ToolError(f"Ticket {ticket_id} not found.")
raise ToolError(f"Ticket {ticket_id} is already {row[0]}. "
"Only open or in-progress tickets can be resolved.")
return {
"ticket_id": ticket_id,
"new_status": "resolved",
"resolved_at": now,
"resolution_note": resolution_note,
}
The wide-open version gives the model the power to update any field on any table. This is an invitation for mistakes. The narrow version does exactly one thing, enforces its precondition, and returns the resulting state so the model (and user) can verify what happened. Check and write in one statement. A SELECT followed by an UPDATE lets two concurrent resolves both pass the check, and both succeed. The follow-up SELECT runs only after the write has already failed, to explain why.
For destructive tools, declare destructiveHint: true so clients that honour annotations ask the user first. Chapter 6 explains the hints, and Chapter 20 covers what they do and don’t guarantee.
Return State After Writes
Every write tool should return the resulting state of the affected record: the ticket’s new status and resolution time, the invoice’s ID and total, the old and new email address. The model needs it to confirm what happened. It also supports correction. If the model (or user) notices the returned state is wrong, the next action can fix it immediately rather than requiring a separate lookup.
Failure Modes
Data access servers have a specific failure profile. These are the problems you will encounter in production, roughly ordered by how often they appear.
Unbounded result sets. The model calls a query tool with parameters that match thousands of rows. Without a server-side LIMIT, the response fills the context window and the model loses track of the conversation. Every query tool—intent-shaped or ad hoc—needs a hard cap on returned rows.
Expensive query amplification. The model is asked to “summarize all customer activity for the last year.” It calls a query tool in a loop—once per customer, once per month, once per metric. Each call is individually reasonable, but the aggregate hits the database with hundreds of queries in seconds. Rate limiting at the tool level (see Chapter 9, Pattern 3) helps. So does offering a snapshot that pre-computes the summary.
Schema exposure leaking sensitive tables. Publishing the full database schema is useful for the model but may reveal table names and column structures that carry business-sensitive information. If the schema includes tables like employee_salaries or security_audit_log, the model now knows those exist and may attempt to query them. Expose a curated schema view—only the tables and columns the server’s tools can actually access.
Path traversal in filesystem servers. A file-read tool that does not properly resolve and validate paths can be tricked into reading files outside the project directory. Without resolution, symlinks and ../ sequences bypass the check. With resolution but a string-prefix comparison, a sibling directory that shares the root’s name gets through.
Stale snapshot drift. A snapshot shows data from the last batch run. The user asks a question, gets the snapshot data, then makes a change through a write tool. The snapshot now disagrees with reality. If the response doesn’t carry computed_at, the model has no way to tell the user when the data was last computed.
Check-then-act writes. A write tool that reads the record’s status, decides, and then updates in a second statement races itself. Two concurrent calls both see open, and both write. The same shape produces double refunds and duplicate invoices.
Encoding errors on file reads. A tool that reads a binary file as UTF-8 returns garbage. Check file types, set errors="replace", and refuse files above a size threshold.
Vector search returning confident garbage. Results with high similarity scores turn out to be irrelevant. Return the score so the model can weigh it, and set a minimum threshold.
Implementation Checklist
Before shipping a data-access MCP server, verify each of these:
- The read path is read-only at the database. A test opens
connect_ro()directly, executesDELETE FROM tickets, and gets “attempt to write a readonly database”. - Every list path fetches one past its cap. Responses report
truncatedormore_matchesfrom the extra row, never fromlen(rows) == limit. - Real queries pass the blocklist. A
SELECTthat sorts oncreated_atreturns rows; the escape hatch’s filter matches tokens, not substrings. - Containment tests cover three escapes.
../, a symlink pointing out, and a sibling directory sharing the root’s prefix each return an error. - Every resource the model depends on has a tool twin. Schema, file index and reports are reachable in a host that never attaches resources.
- Search and retrieval are separate tools. Search returns IDs, scores and excerpts. Retrieval by ID returns the full content.
- The snapshot read path contains no aggregation. It reads one row, and
computed_atis in every response. - Every write is one conditional statement that returns post-write state. Destructive tools also carry
destructiveHint: true.
What Comes Next
Every pattern in this chapter keeps the query in the server’s hands: the model states intent, and the server decides what runs against the store. External APIs make the same demand, with sharper edges: credentials that expire, rate limits you share with other clients, payload formats that were never designed to be read by a model. Chapter 9 applies the same rule to the services your server calls.