Securing AI Agent Pipelines: A Privacy Vault Architecture for the Model Context Protocol

Abstract

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

---

1. Introduction

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:

1. LLM prompt — the cloud-hosted model sees raw PII
2. Observability logs — debugging and telemetry capture tool arguments
3. Agent memory — conversation history persists PII to disk
4. Tool responses — results echo PII back into the context window

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.

---

2. Architecture Overview

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.

---

3. Data Flow: End-to-End Walkthrough

3.1 Tokenization Phase

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.

3.2 Tool Execution Phase

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.

---

4. Security Model

4.1 Default-Deny Policy

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.

4.2 HMAC-Signed Capabilities

Each disclosure is authorized by a capability — a cryptographic token that binds together:

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.

4.3 Session Isolation

Each MCP connection receives an isolated VaultSession:

• Unique session ID: vs_<random> (cryptographically random, URL-safe)
• TTL-bounded: Sessions expire after a configurable period (default: 1 hour)
• Token ownership: Every stored PII value is tagged with its owning session ID
• Cross-session checks: 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 session

When the MCP connection closes, the session and all its stored PII are garbage-collected.

---

5. Implementation Details

5.1 Token Formats

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.

5.2 O(n) Token Scanner

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.

5.3 Recursive Serialization

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:

• Nested dictionaries and lists
• Pydantic models (via .model_dump())
• Custom objects (via str())
• Exception objects
• None and primitive types

After detection, the vault maps character positions back to the original structure for accurate token replacement.

5.4 Result Re-Tokenization

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.

---

6. Integration: The FastPvpMCP Binding

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.

6.1 Server Setup

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:

1. Creates a vault session per MCP connection
2. Registers pvp_tokenize as a built-in tool
3. Exposes pvp://session as a built-in resource
4. Wraps every @mcp.tool() to resolve tokens → execute → re-tokenize

6.2 Client Integration

import 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.

---

7. Detector Pipeline

7.1 RegexDetector (Built-in)

The default detector uses compiled regex patterns for:

Regex detection runs with no external dependencies, making it suitable for development and low-latency environments.

7.2 PresidioDetector (Optional)

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.

7.3 Pluggable Interface

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.

---

8. Audit Trail

Every operation produces an AuditEvent with a unique audit_id, timestamp, and structured metadata:

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.

---

9. Comparison with Alternative Approaches

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.

---

10. Threat Model Summary

The full threat model is maintained in the project repository with versioned attack scenarios and mitigations.

---

11. Getting Started

Installation

# Core library (regex-based detection)
pip install mcp-pvp

# With Presidio NLP detection
pip install mcp-pvp[presidio]

Minimal Example

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")

Running

# 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"]
    }
  }
}

---

12. Conclusion

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.