Let an agent read your bank statements without handing it your PII

Shipped

local-budget v0.1.0 is a local-first, agent-first spending tool. It imports your bank exports into a local SQLite database, then exposes a set of read tools over a stdio MCP server so a Claude Code session can answer “how am I doing this month” without any of your data leaving the machine. The server runs no inference of its own; it serves deterministic data and lets the client model do the talking.

The interesting part is not the budgeting. It is the shape of the server: how you give an agent useful tools over private data while making it structurally impossible for the model to read the sensitive columns. This guide builds that server from scratch. You can point it at any SQLite database you would not want pasted into a chat window.

What you need first

Python 3.11+ and the official MCP SDK: pip install mcp.
A local SQLite database. This guide assumes a transactions table with a few clean columns the agent may see (posted_date, amount_cents, category, merchant_norm) and a few it must not (payee, memo, raw_ofx, acct_hash) where the raw statement text and account numbers live.
A client that speaks MCP over stdio (Claude Code, or any MCP client).

The whole design rests on one rule: the model is untrusted with your PII, so the server enforces what it can read, not a prompt asking it nicely.

Step 1: Define tools as data, not decorators

Make each tool a plain record: a name, a description the model reads, a JSON Schema for its input, and an async handler. One registry is the single source of truth for both tools/list and tools/call, so the two can never drift.

from dataclasses import dataclass
from typing import Awaitable, Callable

@dataclass(frozen=True)
class ToolSpec:
    name: str
    description: str
    input_schema: dict
    handler: Callable[[dict], Awaitable[dict]]

def obj(props: dict | None = None, required: list[str] | None = None) -> dict:
    """A real JSON-Schema object. The MCP wire format needs this exact shape."""
    return {"type": "object", "properties": props or {}, "required": required or []}

A handler returns a dict with two keys: rendered (human-readable markdown) and data (the machine payload). We will use both in Step 3.

async def get_month_summary(args: dict) -> dict:
    month = args.get("month") or current_month()      # "YYYY-MM"
    rows = query_summary(month)                        # your DB read, PII-free columns
    rendered = format_summary_table(rows)              # a markdown string
    return {"rendered": rendered, "data": {"month": month, "categories": rows}}

TOOL_SPECS = [
    ToolSpec(
        "get_month_summary",
        "Spend summary for a month (YYYY-MM, default current). Call this first for "
        "'how am I doing this month'.",
        obj({"month": {"type": "string"}}),
        get_month_summary,
    ),
]
SPEC_BY_NAME = {s.name: s for s in TOOL_SPECS}

A few helpers are yours to supply: current_month() returns "YYYY-MM", query_summary(month) runs a read over PII-free columns, and format_summary_table(rows) returns markdown. Step 2 adds two more, redact_digits(s) (mask digit runs in a string) and format_rows(rows) (a markdown table), plus a DB_PATH constant.

Step 2: Read private data behind a column-level authorizer

The moment you offer a general run_sql tool, you have handed the model a way to SELECT payee, memo FROM transactions. A system prompt saying “do not read PII” is not a control. SQLite gives you a real one: a connection-scoped authorizer callback that fires for every column access and can veto it.

Python’s set_authorizer calls your callback for each attempted access; return SQLITE_OK to allow, SQLITE_DENY to abort the whole statement with an error, or SQLITE_IGNORE to substitute NULL. For an SQLITE_READ action the callback receives the table and column names as its second and third arguments, which is exactly the hook we need.

import sqlite3
from contextlib import contextmanager

READ_DENY = {("transactions", "payee"), ("transactions", "memo"),
             ("transactions", "raw_ofx"), ("accounts", "acct_hash")}

def _authorizer(action, arg1, arg2, dbname, trigger):
    if action == sqlite3.SQLITE_READ:
        return sqlite3.SQLITE_DENY if (arg1, arg2) in READ_DENY else sqlite3.SQLITE_OK
    if action in (sqlite3.SQLITE_SELECT, sqlite3.SQLITE_FUNCTION):
        return sqlite3.SQLITE_OK
    return sqlite3.SQLITE_DENY   # deny writes, ATTACH, PRAGMA, everything else

@contextmanager
def agent_connect(db_path):
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA query_only = ON")   # set BEFORE the authorizer denies PRAGMA
    conn.set_authorizer(_authorizer)
    try:
        yield conn
    finally:
        conn.close()

Now a run_sql tool can be genuinely useful and still safe. Allow only SELECT/WITH, reject a small denylist of keywords, run through agent_connect, and as defense in depth scrub digit-runs from every string cell in case a future column carries an embedded number.

import re

FORBIDDEN = {"attach", "pragma", "insert", "update", "delete", "drop"}

async def run_sql(args: dict) -> dict:
    q = (args.get("query") or "").strip().rstrip(";")
    low = q.lower()
    if not (low.startswith("select") or low.startswith("with")):
        return {"error": "read-only: only SELECT/WITH permitted"}
    if set(re.split(r"[^a-z]+", low)) & FORBIDDEN:
        return {"error": "forbidden keyword"}
    try:
        with agent_connect(DB_PATH) as conn:
            rows = [dict(r) for r in conn.execute(q).fetchmany(200)]
    except sqlite3.DatabaseError:
        return {"error": "denied: query touched a blocked column (payee/memo/raw_ofx)"}
    rows = [{k: redact_digits(v) if isinstance(v, str) else v for k, v in r.items()}
            for r in rows]
    return {"rendered": format_rows(rows), "data": {"rows": rows}}

If the model asks for payee, the statement aborts and it gets a plain “denied” string it can recover from. It never sees the value. The guarantee lives in the database engine, so a curious model cannot talk its way past it the way it could slip past a system-prompt rule.

Step 3: The server, and the two-channel result

Build the low-level server straight off the registry, so listing and calling both read the same TOOL_SPECS.

import json
from mcp import types
from mcp.server.lowlevel import Server

def jsonable(obj: dict) -> dict:
    """Coerce dates/Decimals so structuredContent can serialize."""
    return json.loads(json.dumps(obj, default=str))

def build_server() -> Server:
    server = Server("budget")

    @server.list_tools()
    async def _list() -> list[types.Tool]:
        return [types.Tool(name=s.name, description=s.description, inputSchema=s.input_schema)
                for s in TOOL_SPECS]

    @server.call_tool()
    async def _call(name: str, arguments: dict):
        spec = SPEC_BY_NAME.get(name)
        if spec is None:
            return {"error": f"unknown tool: {name}"}
        result = await spec.handler(arguments or {})
        rendered = result.get("rendered")
        if rendered:
            structured = {"data": result["data"]} if "data" in result else result
            return [types.TextContent(type="text", text=rendered)], jsonable(structured)
        return jsonable(result)

    return server

Returning a tuple of (content, structuredContent) is deliberate. The MCP spec says structured content is returned as a JSON object in the structuredContent field, and that for backwards compatibility a tool returning structured content should also return the serialized JSON in a text block. So the human-readable table goes in the text block, and the same data goes in structuredContent where the model can resolve a row by id, or a merchant by exact string, without re-parsing your markdown. Run it over stdio:

import asyncio
from mcp.server.stdio import stdio_server

async def run_stdio():
    server = build_server()
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(run_stdio())

Use it, then verify it

claude mcp add budget -- python -m budget.mcp_server

Ask a question in the client (“what did I spend on groceries last month?”) and it will call get_month_summary or run_sql and answer from the returned data. Then verify the guarantee yourself, from the client, with the one query that must fail:

# In the MCP client, call run_sql with:
#   SELECT payee FROM transactions LIMIT 1
# Expected: {"error": "denied: query touched a blocked column ..."}

If that returns rows instead of a denial, your authorizer is not attached to the connection the tool actually uses. A green run is a real signal: the sensitive column is unreadable even to a direct SQL request.

Gotchas

The JSON-Schema shorthand does not survive the wire. Trap: writing an input schema as {"month": str}, the way some higher-level SDK helpers accept. Symptom: tools/list fails to serialize and the client shows your server with no tools. Escape: emit a real JSON Schema object, {"type": "object", "properties": {...}, "required": [...]}. The raw protocol serializes exactly what you hand it.
A “do not read PII” prompt is not access control. Trap: shipping a run_sql tool and trusting the description to keep the model out of payee/memo. Symptom: account numbers and raw statement text land in the model’s context the first time it gets curious. Escape: the column-level authorizer above, plus PRAGMA query_only, plus a digit-run redactor on returned strings. Set the pragma before the authorizer, since the authorizer then denies further pragmas.
structuredContent chokes on dates and Decimals. Trap: putting a datetime or a Decimal straight into the data payload. Symptom: the SDK raises at call time trying to serialize the result. Escape: round-trip through json.dumps(obj, default=str) before returning, as jsonable does.
An unscoped tool floods the context window. Trap: an anomaly or history tool that returns everything by default; over two years of statements that is a wall of rows. Symptom: one call blows the model’s context and the answer degrades. Escape: scope the output with month/limit arguments while keeping the detection baseline on full history, so the statistics stay honest but the payload stays small.
Bank exports are hostile XML. Trap: parsing an OFX/QFX file with the standard library xml.etree. Symptom: a crafted file triggers XXE or a billion-laughs entity expansion at import. Escape: parse untrusted XML with defusedxml, which disables entity expansion and external entities by default.

Sources

MCP: Tools specification — structuredContent, the “also return serialized JSON in a text block” rule, output schema, and tool-execution errors as self-correcting feedback.
Python sqlite3: set_authorizer — the per-access callback, SQLITE_OK/DENY/IGNORE semantics, and the table/column arguments for SQLITE_READ.
SQLite: Compile-Time Authorization Callbacks — the authoritative behavior of the authorizer and DENY vs IGNORE.
defusedxml — secure substitutes for Python’s XML parsers against XXE, billion laughs, and quadratic blowup.

Changelog

local-budget: local-first, agent-first spending agent (703a9ae)
feat(mcp): transmit the data payload via structuredContent (6eb2602)
feat(tools): month/limit scoping for find_anomalies (2ecb55c)
feat(tools): list_categories vocabulary tool + recoverable category errors (0ce5fb4)
fix(security): read-deny raw payee/memo to the agent authorizer (578c608)
fix(security): redact account-number runs in stored payee/memo at import (c4afbc5)
fix(security): reject cross-origin requests on mutating dashboard routes (0fb8e02)
test(security): pin hostile-OFX XML safety (XXE, entity bomb, DOCTYPE) (0c9e8bf)