Large Language Models integrated into agentic workflows via the Model Context Protocol (MCP) create a systemic privacy problem: user-provided PII flows through LLM prompts, debug logs, telemetry, and persistent conversation history. Conventional approaches — full redaction or post-hoc filtering — either break tool execution or fail to cover the entire data surface. We present mcp-pvp (Privacy Vault Protocol), an open-source Python library that introduces a tokenize-then-resolve architecture. PII is replaced with typed opaque tokens before the LLM sees it, stored in a session-scoped local vault, and resolved only at the moment of tool execution under a default-deny, capability-gated policy. This article walks through the architecture, security model, implementation details, and integration patterns.
Keywords: MCP, privacy, PII tokenization, capability security, LLM agents, tool calling
The Model Context Protocol (MCP) standardizes how LLMs interact with external tools. An agent receives a user message like "Send the quarterly report to alice@example.com", reasons about which tool to call, and dispatches a send_email invocation. MCP made this painless — but it also created a data pipeline where sensitive values flow through four uncontrolled surfaces:
The core tension is between privacy (the LLM shouldn't see sensitive data) and functionality (tools need real values to execute). Redacting PII solves the first problem but breaks the second — send_email(to="[REDACTED]") doesn't work.
mcp-pvp resolves this tension with tokenization: replacing PII with opaque, typed references that preserve semantic meaning while hiding raw values. The LLM can still reason about the data ("there's an email address here, I should pass it to the email tool"), but it never possesses the actual value.
The system consists of six core components arranged in a pipeline:
┌─────────────────────────────────────────────────────────────┐
│ FastPvpMCP │
│ (MCP Server Binding) │
│ │
│ ┌──────────┐ ┌──────────┐ ┌────────────┐ │
│ │ Detector │──▶│ Vault │──▶│ Executor │ │
│ │ Pipeline │ │ │ │ │ │
│ └──────────┘ │ ┌──────┐ │ └────────────┘ │
│ │ │Store │ │ │
│ │ └──────┘ │ │
│ │ ┌──────┐ │ ┌────────────┐ │
│ │ │Policy│ │ │ Audit │ │
│ │ └──────┘ │ │ Logger │ │
│ │ ┌──────┐ │ └────────────┘ │
│ │ │ Caps │ │ │
│ │ └──────┘ │ │
│ └──────────┘ │
│ │
│ Built-in Tools: pvp_tokenize | pvp://session resource │
└─────────────────────────────────────────────────────────────┘
Detector Pipeline — Identifies PII spans in arbitrary text using pluggable backends (regex for zero-dependency use, Microsoft Presidio for production NLP-based detection).
Vault — Central coordinator. Accepts text, runs detection, stores raw values in the session store, returns tokenized text. On tool execution, resolves tokens back to real values under policy control.
Session Store — In-memory, TTL-bounded key-value store. Each MCP connection gets an isolated session. Tokens are keyed by session, preventing cross-session data leakage.
Policy Engine — Declarative rules specifying which PII types may be disclosed to which sinks (tools), at which argument paths. Default-deny for all sinks. LLM and engine sinks are permanently blocked.
Capability Manager — Issues HMAC-SHA256 signed, time-limited capability tokens that authorize a specific PII disclosure to a specific sink. Capabilities cannot be forged, replayed, or transferred across sessions.
Audit Logger — Records every tokenization, resolution, and policy decision with structured metadata. Never logs raw PII values — only types, counts, references, and sink names.
When a client sends content containing PII, the vault performs a multi-step tokenization:
User Input: "Contact alice@example.com or call 555-0123"
│
▼
┌─────────────────────┐
│ serialize_for_pii │ ◄── Recursively flattens
│ _detection() │ dicts, lists, nested
└─────────┬───────────┘ structures to text
│
▼
┌─────────────────────┐
│ Detector.detect() │ ◄── Returns PIIDetection
│ (Regex/Presidio) │ spans with type,
└─────────┬───────────┘ position, confidence
│
▼
┌─────────────────────┐
│ store.store_pii() │ ◄── Generates unique ref
│ │ (tkn_<random>), stores
└─────────┬───────────┘ value in session
│
▼
┌─────────────────────┐
│ Build token text │ ◄── Replaces spans
│ │ right-to-left to
└─────────┬───────────┘ preserve positions
│
▼
Output: "Contact [[PII:EMAIL:tkn_x7k9m2]] or call [[PII:PHONE:tkn_a1b2c3]]"
Key implementation detail: replacements are applied right-to-left (from the end of the string backward) so that earlier character positions remain valid as the string changes length.
When the LLM calls a tool with tokenized arguments:
Tool Call: send_email(to="[[PII:EMAIL:tkn_x7k9m2]]", subject="Report")
│
▼
┌─────────────────────┐
│ TokenScanner.scan │ ◄── O(n) state machine
│ _tokens() │ extracts all token refs
└─────────┬───────────┘
│
▼
┌─────────────────────┐
│ Policy.evaluate() │ ◄── Checks: is EMAIL
│ │ allowed for tool
└─────────┬───────────┘ "send_email" at
│ arg_path "to"?
▼
┌─────────────────────┐
│ CapabilityManager │ ◄── Issues HMAC-signed
│ .create() │ capability with TTL
└─────────┬───────────┘
│
▼
┌─────────────────────┐
│ CapabilityManager │ ◄── Verifies signature,
│ .verify() │ checks expiry,
└─────────┬───────────┘ validates session
│
▼
┌─────────────────────┐
│ store.get_pii() │ ◄── Retrieves raw value
│ │ with session check
└─────────┬───────────┘
│
▼
┌─────────────────────┐
│ executor.execute() │ ◄── Raw PII in memory
│ │ for ~50ms
└─────────┬───────────┘
│
▼
┌─────────────────────┐
│ vault.tokenize_ │ ◄── Re-scans result,
│ tool_result() │ replaces any PII
└─────────────────────┘ with fresh tokens
The raw PII value exists in process memory only during executor.execute() — typically for the duration of a single function call. Before and after that window, only opaque tokens exist.
The policy engine uses a declarative, allowlist-only model:
from mcp_pvp.models import PIIType, Policy, PolicyAllow, SinkPolicy policy = Policy(sinks={ "tool:send_email": SinkPolicy( allow=[ PolicyAllow(type=PIIType.EMAIL, arg_paths=["to", "cc"]), ] ), "tool:lookup_user": SinkPolicy( allow=[ PolicyAllow(type=PIIType.EMAIL, arg_paths=["email"]), PolicyAllow(type=PIIType.PHONE, arg_paths=["phone"]), ] ), })
If a PII type or argument path isn't explicitly listed, disclosure is denied. There is no wildcard or "allow all" shortcut by design.
The SinkKind enum distinguishes between tool sinks, LLM sinks, engine sinks, and local sinks. LLM and engine sinks are permanently blocked — there is no policy configuration that permits raw PII disclosure to a model or agent engine.
Each disclosure is authorized by a capability — a cryptographic token that binds together:
| Field | Purpose |
|---|---|
vault_session | Scopes capability to one session |
pii_ref | The specific token reference (e.g., tkn_x7k9m2) |
pii_type | The PII type (EMAIL, PHONE, etc.) |
sink | The target tool and argument path |
exp | Expiration timestamp (default: 300s TTL) |
The capability is serialized as base64url(json_payload).base64url(hmac_sha256_signature). The HMAC key is a 32+ byte secret generated per vault instance.
Verification uses constant-time comparison (hmac.compare_digest) to prevent timing attacks. Expired capabilities are rejected. Session mismatch rejects prevent cross-session replay.
Each MCP connection receives an isolated VaultSession:
vs_<random> (cryptographically random, URL-safe)store.get_pii() validates that the requesting session matches the token's owner — even if an attacker guesses a valid tkn_ reference, it won't resolve in a different sessionWhen the MCP connection closes, the session and all its stored PII are garbage-collected.
mcp-pvp supports two token formats for different integration patterns:
TEXT format (default): [[PII:EMAIL:tkn_x7k9m2]]
Designed for string interpolation. The double-bracket syntax is chosen to be unlikely to collide with natural text while remaining human-readable for debugging.
JSON format: {"$pii_ref": "tkn_x7k9m2", "type": "EMAIL", "cap": "..."}
Designed for structured API payloads where tokens need to be machine-parsable.
Rather than using regex for token extraction (which is vulnerable to catastrophic backtracking on malformed input), the scanner uses a single-pass character-by-character state machine:
States: TEXT → BRACKET1 → BRACKET2 → PII → COLON1 → TYPE → COLON2 → REF → CLOSE1
Each character is processed exactly once. Invalid sequences immediately reset to the TEXT state, and the current character is checked for starting a new potential token. This guarantees O(n) time complexity regardless of input characteristics.
The performance benchmark suite demonstrates that on pathological input (thousands of nested brackets), the scanner maintains linear performance where regex would backtrack exponentially.
Tool arguments and results can be arbitrarily nested (dicts within lists within dicts). The serialize_for_pii_detection() function recursively flattens any Python structure into a string for PII scanning, handling:
.model_dump())str())None and primitive typesAfter detection, the vault maps character positions back to the original structure for accurate token replacement.
After tool execution, the vault re-scans the result for PII. This catches a common pattern where tools echo input data:
# Tool returns: {"status": "sent", "to": "alice@example.com", "subject": "Report"} # After re-tokenization: {"status": "sent", "to": "[[PII:EMAIL:tkn_n3w_r3f]]", "subject": "Report"}
The re-tokenization uses fresh token references, maintaining a clean separation between input and output tokens in the audit trail.
FastPvpMCP is a drop-in subclass of the MCP SDK's FastMCP class. It intercepts the tool execution lifecycle to inject privacy protection without requiring any changes to tool implementations.
from mcp_pvp.bindings.mcp.server import FastPvpMCP from mcp_pvp.models import PIIType, Policy, PolicyAllow, SinkPolicy from mcp_pvp.vault import Vault # Define disclosure policy policy = Policy(sinks={ "tool:send_email": SinkPolicy( allow=[PolicyAllow(type=PIIType.EMAIL, arg_paths=["to"])] ), "tool:lookup_user": SinkPolicy( allow=[ PolicyAllow(type=PIIType.EMAIL, arg_paths=["email"]), PolicyAllow(type=PIIType.PHONE, arg_paths=["phone"]), ] ), }) # Create vault with policy vault = Vault(policy=policy) # Create MCP server — privacy protection is automatic mcp = FastPvpMCP(name="secure-app", vault=vault) @mcp.tool() def send_email(to: str, subject: str, body: str) -> dict: """Send an email. 'to' receives the real email address.""" return {"status": "sent", "recipient": to} @mcp.tool() def lookup_user(email: str) -> dict: """Look up a user by email.""" return {"name": "Alice", "email": email}
FastPvpMCP automatically:
pvp_tokenize as a built-in toolpvp://session as a built-in resource@mcp.tool() to resolve tokens → execute → re-tokenizeimport json from mcp import ClientSession async def demo(session: ClientSession): # Step 1: Tokenize user input result = await session.call_tool("pvp_tokenize", { "content": "Email alice@example.com about the report" }) response = json.loads(result.content[0].text) # response["redacted"] = "Email [[PII:EMAIL:tkn_x7k9m2]] about the report" # response["tokens"] = ["[[PII:EMAIL:tkn_x7k9m2]]"] # Step 2: Pass token to tool — vault resolves it server-side result = await session.call_tool("send_email", { "to": response["tokens"][0], "subject": "Quarterly Report", "body": "Please find attached." }) # Result has re-tokenized PII — safe for LLM context
The LLM sees only the token string throughout the entire interaction. No code changes are needed to existing tool implementations.
The default detector uses compiled regex patterns for:
| PII Type | Pattern | Confidence |
|---|---|---|
| RFC 5322-like address pattern | 0.8 | |
| PHONE | US/international formats with optional country code | 0.8 |
| IPV4 | Dotted quad with octet validation | 0.8 |
| CC | Visa, MasterCard, Amex, Discover, JCB formats | 0.8 |
Regex detection runs with no external dependencies, making it suitable for development and low-latency environments.
For production use, mcp-pvp integrates with Microsoft Presidio, which uses NLP models (spaCy) for context-aware detection:
pip install mcp-pvp[presidio]
Presidio provides higher accuracy, context-sensitive detection, and support for additional entity types (names, addresses, SSN, etc.). The confidence scores from Presidio's ML models replace the fixed 0.8 confidence of the regex detector.
Both detectors implement the PIIDetector abstract base class:
class PIIDetector(ABC): @abstractmethod def detect(self, content: str, types: list[PIIType] | None = None) -> list[PIIDetection]: ... @abstractmethod def supports_type(self, pii_type: PIIType) -> bool: ...
Custom detectors (domain-specific patterns, third-party APIs) can be passed to the Vault constructor.
Every operation produces an AuditEvent with a unique audit_id, timestamp, and structured metadata:
| Event Type | Recorded Data |
|---|---|
TOKENIZE | Session ID, PII types detected, token count, reference IDs |
RESOLVE | Session ID, token ref, target sink, policy decision |
DELIVER | Session ID, tool name, token count, linked parent audit |
POLICY_DENIED | Session ID, requested type, target sink, denial reason |
SESSION_CREATED | Session ID, TTL |
SESSION_CLOSED | Session ID, token count at close |
Raw PII values are never included in audit records. The audit trail can be directed to any backend via the AuditLogger interface — structured logging (default), files, SIEM systems, or compliance databases.
Parent-child linking (via parent_audit_id) enables tracing the full lifecycle of a piece of PII from tokenization through resolution to tool execution.
| Approach | Privacy | Functionality | Implementation Complexity |
|---|---|---|---|
| No protection | ✗ PII everywhere | ✓ Full | None |
| Full redaction | ✓ PII removed | ✗ Tools break | Low |
| Post-hoc log scrubbing | ~ Partial | ✓ Full | Medium |
| Encryption at rest | ~ Partial (in-memory exposure) | ✓ Full | High |
| mcp-pvp tokenization | ✓ PII never in prompts | ✓ Tools work via resolution | Low (drop-in) |
The tokenization approach uniquely preserves both privacy and functionality. The LLM retains enough information to reason about the data (it knows [[PII:EMAIL:tkn_x7k9]] is an email) without possessing the actual value.
| Threat | Mitigation |
|---|---|
| LLM prompt exfiltration | LLM never sees raw PII — only opaque tokens |
| Log/telemetry leakage | Tokens in arguments; audit trail excludes raw values |
| Cross-session data theft | Session-scoped tokens + ownership validation |
| Capability forgery | HMAC-SHA256 signatures with constant-time comparison |
| Token replay across sessions | Session ID bound into capability; mismatch = rejection |
| ReDoS via malformed tokens | O(n) state machine scanner, no regex backtracking |
| Tool result leakage | Automatic re-tokenization of tool outputs |
| Policy bypass via wildcards | No wildcard support; every disclosure must be explicit |
The full threat model is maintained in the project repository with versioned attack scenarios and mitigations.
# Core library (regex-based detection) pip install mcp-pvp # With Presidio NLP detection pip install mcp-pvp[presidio]
from mcp_pvp.bindings.mcp.server import FastPvpMCP from mcp_pvp.models import PIIType, Policy, PolicyAllow, SinkPolicy from mcp_pvp.vault import Vault policy = Policy(sinks={ "tool:greet": SinkPolicy( allow=[PolicyAllow(type=PIIType.EMAIL, arg_paths=["email"])] ), }) mcp = FastPvpMCP(name="demo", vault=Vault(policy=policy)) @mcp.tool() def greet(email: str) -> str: return f"Hello, {email}!" if __name__ == "__main__": mcp.run(transport="stdio")
# With MCP Inspector npx @modelcontextprotocol/inspector python my_server.py # With Claude Desktop (add to claude_desktop_config.json) { "mcpServers": { "my-app": { "command": "python", "args": ["path/to/my_server.py"] } } }
mcp-pvp addresses a structural gap in the MCP ecosystem: the protocol's tool-calling primitives have no built-in concept of data sensitivity. By introducing a tokenize-then-resolve architecture with session-scoped storage, default-deny policy, and cryptographic capabilities, the library provides privacy guarantees without requiring changes to existing tool implementations or the MCP protocol itself.
The design prioritizes local-first operation (no network calls to resolve tokens), minimal trust surface (raw PII only exists during tool execution), and auditability (every disclosure is logged). These properties make it suitable for environments with regulatory requirements (GDPR, HIPAA, CCPA) where demonstrable data minimization is necessary.
The library is open source under the Apache-2.0 license.
Repository: github.com/Hidet-io/mcp-pvp
Documentation: hidet-io.github.io/mcp-pvp
PyPI: pypi.org/project/mcp-pvp
mcp-pvp is maintained by Hidet. Contributions, threat model reviews, and detector integrations are welcome.