DEV Community

Mitiku Yohannes
Mitiku Yohannes

Posted on

mcp-pvp — Privacy Vault Protocol for MCP

#privacy #agents #security #ai

Inspiration

We were building an AI agent that handled customer support emails. The agent used MCP to call tools like send_email, lookup_account, and create_ticket. It worked beautifully — until we looked at the logs.

Every customer's email address, phone number, and account ID was sitting in plain text across four different places: the LLM prompt, the debug logs, the agent's conversation history, and the tool responses being fed back into the context window. One user request created four copies of their PII that we had no control over.

The obvious fix — redacting PII before the LLM sees it — breaks everything. If you replace alice@example.com with [REDACTED], the agent tries to call send_email(to="[REDACTED]") and the workflow fails. You're trading a privacy problem for a functionality problem.

We needed a way to keep sensitive data hidden from the LLM while still letting tools use it. That's what mcp-pvp does.

What it does

mcp-pvp is an open-source Python library that adds automatic PII protection to any MCP server. Instead of redacting sensitive data, it tokenizes it — replacing real values with typed, opaque references while storing the originals in a local vault.

Here's what happens:

User says:  "Send the report to alice@example.com"
LLM sees:   "Send the report to [[PII:EMAIL:tkn_x7k9m2]]"
Tool gets:  to="alice@example.com"  (resolved at execution time)
LLM gets:   to="[[PII:EMAIL:tkn_n3w]]"  (re-tokenized result)
Enter fullscreen mode Exit fullscreen mode

The LLM can still reason perfectly — it knows there's an email address and it should pass it to the email tool. But it never possesses the actual value. The real email only exists in memory for the brief moment the tool function executes (~50ms).

Key features:

  • Drop-in integration — FastPvpMCP subclasses FastMCP. Wrap your existing server in 3 lines.
  • Default-deny policy — You declare exactly which PII types each tool can access, at which argument paths. Everything else is blocked.
  • HMAC-signed capabilities — Every disclosure is authorized by a cryptographic token bound to the session, PII reference, and target tool.
  • Session-scoped storage — Each MCP connection gets an isolated vault. Tokens can't leak across sessions. Everything is cleaned up on disconnect.
  • Automatic re-tokenization — Tool results are scanned for PII and re-tokenized before they reach the agent.
  • Pluggable detection — Built-in regex detector for zero-dependency use, optional Microsoft Presidio integration for production NLP-based detection.
  • Full audit trail — Every tokenization, resolution, and policy denial is logged (never with raw PII values).

How we built it

Architecture

The system has six core components:

Client ──► FastPvpMCP Server
               │
               ├── Detector Pipeline (Regex / Presidio)
               │     Finds PII spans in text
               │
               ├── Vault (central coordinator)
               │     Tokenizes content, resolves tokens, manages lifecycle
               │
               ├── Session Store (in-memory, TTL-bounded)
               │     Isolated per-connection storage for PII values
               │
               ├── Policy Engine (declarative, default-deny)
               │     Controls which tools see which PII types
               │
               ├── Capability Manager (HMAC-SHA256)
               │     Cryptographic authorization for each disclosure
               │
               └── Audit Logger (structured, PII-free)
                     Records every operation for compliance
Enter fullscreen mode Exit fullscreen mode

Tech stack

  • Python 3.11+ with full type annotations
  • Pydantic v2 for all data models
  • MCP SDK 1.26+ — standard Model Context Protocol
  • Microsoft Presidio (optional) for NLP-based PII detection with spaCy
  • structlog for structured audit logging
  • HMAC-SHA256 for capability signing with constant-time verification

The tokenization flow

When content arrives, the vault:

  1. Serializes — Recursively flattens nested structures (dicts, lists, Pydantic models) into scannable text
  2. Detects — Runs the detector pipeline to find PII spans with type and confidence
  3. Stores — Generates unique references (tkn_<random>), stores raw values in the session
  4. Replaces — Swaps PII spans with typed tokens, working right-to-left to preserve character positions

When a tool is called with tokenized arguments:

  1. Scans — An O(n) state machine extracts token references (no regex backtracking)
  2. Evaluates policy — Checks if this PII type is allowed for this tool at this argument path
  3. Issues capability — Creates an HMAC-signed, time-limited authorization
  4. Resolves — Retrieves the real value from the session store
  5. Executes — Runs the tool with real values (the only moment PII is in memory)
  6. Re-tokenizes — Scans the result for any PII and replaces it with fresh tokens

Server code 

from mcp_pvp.bindings.mcp.server import FastPvpMCP
from mcp_pvp.models import PIIType, Policy, PolicyAllow, SinkPolicy
from mcp_pvp.vault import Vault

# Declare what each tool is allowed to see
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"]),
        ]
    ),
})

mcp = FastPvpMCP(name="my-app", vault=Vault(policy=policy))

@mcp.tool()
def send_email(to: str, subject: str, body: str) -> dict:
    # 'to' is the real email — resolved automatically by the vault
    return {"status": "sent", "recipient": to}

@mcp.tool()
def lookup_user(email: str) -> dict:
    return {"name": "Alice", "email": email, "plan": "pro"}

if __name__ == "__main__":
    mcp.run(transport="stdio")
Enter fullscreen mode Exit fullscreen mode

That's it. Every @mcp.tool() gets automatic PII interception. The pvp_tokenize tool and pvp://session resource are registered automatically.

Client code 

import json
from mcp import ClientSession

async def run(session: ClientSession):
    await session.initialize()

    # Tokenize — raw PII stays server-side
    result = await session.call_tool("pvp_tokenize", {
        "content": "Contact alice@example.com or call 555-0123"
    })
    data = json.loads(result.content[0].text)
    # data["redacted"] = "Contact [[PII:EMAIL:tkn_abc]] or call [[PII:PHONE:tkn_def]]"

    # Use tokens in tool calls — vault resolves them
    result = await session.call_tool("send_email", {
        "to": data["tokens"][0],
        "subject": "Hello",
        "body": "Monthly update"
    })
    # Result contains re-tokenized PII, safe for LLM context
Enter fullscreen mode Exit fullscreen mode

Challenges we ran into

Token scanning without regex backtracking. Our first implementation used a regex pattern for [[PII:TYPE:REF]] extraction. It worked — until we ran it against adversarial input with thousands of nested brackets. The regex engine would backtrack exponentially. We replaced it with a hand-written state machine (TokenScanner) that processes each character exactly once in O(n) time, regardless of input.

Recursive PII in nested structures. Tool arguments and results aren't always flat strings. They're nested dicts, lists, Pydantic models, and sometimes exception objects. We had to build a recursive serializer (serialize_for_pii_detection) that could flatten any Python structure for PII scanning and then map detected spans back to the original structure for replacement.

Result re-tokenization. Tools frequently echo their input: send_email might return {"to": "alice@example.com"}. If we didn't scan results, that raw email would flow right back into the LLM context. We added automatic result re-tokenization as the final step of every tool execution — it generates fresh token references, maintaining a clean audit trail that separates input tokens from output tokens.

Policy granularity. Early versions had tool-level policies ("this tool can see emails"). But that's too coarse — send_email should see an email in its to argument, but not if someone passes one in the body argument. We added argument path specificity so policies control exactly where within a tool's arguments PII can appear.

Session isolation under concurrent connections. MCP servers handle multiple simultaneous clients. Each client's PII must be completely isolated. We built session-scoped storage with ownership tagging — every stored PII value records which session created it, and cross-session access is rejected even if an attacker guesses a valid token reference.

Accomplishments that we're proud of

  • Zero protocol changes — Works with standard MCP. No custom transport, no protocol extensions. Drop it into any existing MCP server.
  • 3-line integration — Import FastPvpMCP, define a policy, and replace FastMCP with FastPvpMCP. All existing @mcp.tool() decorators keep working unchanged.
  • 257 passing tests covering adversarial inputs, session integrity, recursive scrubbing, policy edge cases, and audit coherence.
  • O(n) guaranteed scanning — The state machine scanner handles pathological input where regex would hang.
  • Complete audit trail with zero raw PII in any log record — every disclosure is traceable from tokenization through resolution to tool execution.
  • Default-deny by design — It's impossible to accidentally over-permit. There's no wildcard, no "allow all," and LLM/engine sinks are permanently blocked.

What we learned

The redaction vs. functionality tradeoff is a false dilemma. Tokenization resolves it elegantly — the LLM gets enough semantic information to reason correctly ("this is an email address, I should pass it to the email tool") without ever possessing the raw value.

Capability-based security > access control lists. Instead of asking "is this caller allowed?", capabilities encode "this specific operation is authorized." They're unforgeable (HMAC-signed), time-limited, and bound to a specific session + PII reference + target tool. This eliminates confused deputy attacks.

Privacy needs to be a middleware concern, not an application concern. If every tool author has to remember to scrub PII from results, someone will forget. Making it automatic and transparent — at the framework level — is the only reliable approach.

Session lifecycle is the natural privacy boundary. MCP connections already have a defined lifecycle (connect → use → disconnect). Binding vault sessions to MCP connections means cleanup is automatic and there's no long-lived token database to worry about.

What's next

  • More detector backends — Support for AWS Comprehend, Google DLP, and Azure AI Language PII detection in addition to Presidio
  • Streaming token resolution — Resolve tokens in streamed responses as chunks arrive, not just batch results
  • Policy-as-code — Load policies from YAML/TOML config files for easier management across environments
  • Metrics and dashboards — Export tokenization/resolution counts, policy denial rates, and session statistics via OpenTelemetry
  • Multi-language SDKs — TypeScript and Go implementations for non-Python MCP servers

Built with

Python, Pydantic, MCP SDK, Microsoft Presidio, spaCy, HMAC-SHA256, structlog

Try it 

pip install mcp-pvp
Enter fullscreen mode Exit fullscreen mode

Top comments (0)