shadow-vault

A bi-directional, format-preserving reversible PII proxy for AI agents, LangChain/LangGraph, and Model Context Protocol (MCP) tools.

Stop breaking your AI Agent tools with [REDACTED]

Most PII masking libraries were designed for static document redaction, not autonomous AI agents. Replacing sensitive data with [REDACTED] or <EMAIL_1> completely breaks tool-calling workflows:

1. Schema & Validator Rejection: When your AI agent attempts to invoke a Stripe API, PostgreSQL tool, or CRM integration, strict downstream validators (Pydantic, regex schemas, database constraints) immediately crash when handed [REDACTED_CARD] instead of a 16-digit numeric card string.
2. One-Way Irreversibility: Once data is destroyed by one-way redaction, local tools cannot execute real actions. You cannot process a refund, send an email, or dispatch an SMS with [EMAIL]. Bi-directional pseudonymization securely restores real PII at the tool boundary and re-masks outputs before returning to the model.
3. SSE Streaming Token Tearing: Naive regex replacements leak torn fragments when modern LLMs stream tokens across Server-Sent Events (SSE) chunks (e.g. streaming "u_8f", then "9a1@shadow", then ".vault"). shadow-vault provides a lookahead holdback buffer that reassembles split tokens seamlessly without token tearing.

Architecture Flow

                     +---------------------------------------+
                     |         END USER / CLIENT             |
                     +---------------------------------------+
                                  |                 ^
                1. Raw User PII   |                 |  8. Hydrated Real PII
                   (e.g. Card)    v                 |     (StreamHydrator SSE)
                     +---------------------------------------+
                     |         SHADOW-VAULT PROXY            |
                     +---------------------------------------+
                        |                          ^
      2. Dehydrated     |                          |  7. Dehydrated Tool
         Safe Prompt    v                          |     Return Value
+------------------------------------+    +------------------------------------+
|         CLOUD LLM / AGENT          |    |          LOCAL MCP TOOL            |
| (OpenAI / Anthropic / LangChain)   |    |      (Stripe / SQL / API)          |
+------------------------------------+    +------------------------------------+
           |                                                 ^
           |  3. Model Tool Call with                        |  4. @vault_guard
           |     Synthetic Tokens                            |     Reversible Hydration
           +-------------------------------------------------+

Quickstart (12 Lines)

Install the core library (zero heavy NLP dependencies):

pip install shadow-vault

Protect any Python function or agent tool in seconds:

from shadow_vault import ShadowVault, vault_guard

vault = ShadowVault()

# 1. Protect your internal tool: automatically receives real PII, returns masked PII
@vault_guard(vault=vault, session_id_arg="session_id")
def charge_customer(email: str, card: str, session_id: str = "default") -> dict:
    # Function executes with REAL PII: email='alice@example.com', card='4532 0151 1283 0366'
    return {"status": "PAID", "receipt": email, "backup_phone": "+91 9876543210"}

# 2. Dehydrate safe prompt for the LLM:
prompt = vault.dehydrate("Charge alice@example.com on card 4532 0151 1283 0366", session_id="s1")
# Prompt: "Charge u_c09753@shadow.vault on card 4000009995970590"

# 3. Model executes tool using synthetic tokens -> returns sanitized output back to agent state:
result = charge_customer(email="u_c09753@shadow.vault", card="4000009995970590", session_id="s1")
# Result: {'status': 'PAID', 'receipt': 'u_c09753@shadow.vault', 'backup_phone': '+919000597059'}

Interactive Zero-Key Terminal Demo

Run the self-contained 4-stage interactive walkthrough (requires NO API keys):

python examples/demo_agent_roundtrip.py

Supported Entities (Global + India)

shadow-vault uses zero-dependency regexes coupled with strict mathematical checksum algorithms so random numbers are never falsely flagged:

Why shadow-vault?

Framework Integrations

1. LangChain & LangGraph

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from shadow_vault import ShadowVault, protect_tool, ShadowVaultChatWrapper

vault = ShadowVault()
session_id = "agent_session"

# Protect Tools: Inputs hydrated before execution, outputs dehydrated before returning
@tool
def lookup_db(card_number: str) -> dict:
    """Fetch record from database."""
    return {"card": card_number, "balance": 1500}

safe_tool = protect_tool(lookup_db, vault=vault, session_id=session_id)

# Protect ChatModel: Prompts dehydrated before cloud LLM, tool-call arguments kept synthetic
llm = ChatOpenAI(model="gpt-4o")
safe_llm = ShadowVaultChatWrapper(llm=llm, vault=vault, session_id=session_id)

2. Model Context Protocol (MCP)

Intercept MCP tools/call JSON-RPC requests and responses:

from shadow_vault import ShadowVault, MCPVaultInterceptor

vault = ShadowVault()
interceptor = MCPVaultInterceptor(vault=vault)

# 1. Intercept incoming request from LLM:
safe_request = interceptor.intercept_request(mcp_request, session_id="user_session")

# 2. Intercept outgoing tool response before it reaches LLM:
safe_response = interceptor.intercept_response(mcp_response, session_id="user_session")

3. OpenAI-Compatible FastAPI Proxy Server

Deploy a drop-in proxy for any OpenAI SDK client:

import uvicorn
from shadow_vault import ShadowVault, create_proxy_app

vault = ShadowVault()
app = create_proxy_app(vault=vault, upstream_base_url="https://api.openai.com/v1")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Point your client to http://localhost:8000/v1:

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="your-api-key",
    default_headers={"X-Session-ID": "session_42"}
)

# Streaming SSE response with real-time token re-hydration:
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Send bill to alice@example.com"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Benchmark & Performance Profile (1,000 Iterations)

Measured on a realistic 2,000-token prompt containing 12 distinct PII entities, structured MCP tool-call dictionaries, and 500-chunk SSE streams using benchmarks/run_benchmarks.py:

Latency Profile

Memory Overhead (tracemalloc)

Checksum Accuracy & False-Positive Rejection

To re-run the benchmark suite locally:

python benchmarks/run_benchmarks.py

Running Tests

pytest -v
# 54 passed in 1.37s (100% pass rate)

Contributing

Contributions are welcome! Read CONTRIBUTING.md to see how you can add a new regional PII detector in under 20 lines of code.

License

Distributed under the MIT License. See LICENSE for details.