ShadowVault MCP Proxy

                     +---------------------------------------+
                     |         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
           +-------------------------------------------------+

pip install shadow-vault

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'}

# 4. Ephemeral Agent Runs (Auto-purges memory on exit):
with vault.session("run_123", auto_purge=True) as sess:
    safe_prompt = sess.dehydrate("Refund alice@example.com on card 4532 0151 1283 0366")
    # Run agent tools...
# 100% of mapping records are wiped from RAM/SQLite upon exiting the block!

# 5. Prevent Log Leaks with Drop-In Logging Filter:
import logging
from shadow_vault import VaultLogFilter

logging.getLogger().addFilter(VaultLogFilter(vault))
# Accidental logger.info(real_card) statements are scrubbed before reaching log sinks!

# 1. Independent Security & Cryptographic Self-Audit (6 automated checks):
shadow-vault verify
# or: python -m shadow_vault.verify

# 2. End-to-End Real Agent Zero-Leak Wire-Inspection Proof:
python examples/verify_agent_zero_leak.py

# 3. Autonomous Model Context Protocol (MCP) Stdio Gateway & AES-256 Vault:
python examples/demo_mcp_gateway.py

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)

# Sits between your Agent/Claude Desktop and downstream MCP servers over stdio:
shadow-vault mcp-proxy --session-id user_42 -- uvx mcp-server-sqlite --db /path/to/data.sqlite

{
  "mcpServers": {
    "secure-sqlite": {
      "command": "shadow-vault",
      "args": [
        "mcp-proxy",
        "--session-id", "claude_session",
        "--fail-closed",
        "--",
        "uvx", "mcp-server-sqlite", "--db", "/path/to/database.db"
      ]
    }
  }
}

from shadow_vault import ShadowVault, MCPVaultInterceptor

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

# 1. Intercept incoming tool call from model (restores real PII before tool executes):
safe_request = interceptor.intercept_request(mcp_request, session_id="user_session")

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

from shadow_vault import ShadowVault, VaultStore

# Enable AES-256-GCM / authenticated encryption at rest:
vault = ShadowVault(
    db_path="./vault.db",
    encryption_key="my-master-vault-encryption-key", # Or via SHADOW_VAULT_KEY env var
)

from shadow_vault import ShadowVault, ToolPolicy, FailClosedAction, vault_guard

vault = ShadowVault(fail_closed=True)

# 1. Restrict hydration by entity type:
stripe_policy = ToolPolicy(
    tool_name="stripe_charge",
    allowed_entities={"CREDIT_CARD", "EMAIL"},
    fail_closed=True,
    fail_action=FailClosedAction.RAISE,
)

# 2. Block hydration completely for untrusted tools:
slack_policy = ToolPolicy(
    tool_name="post_slack_alert",
    allow_hydration=False, # Receives format-preserving synthetic tokens only
)

@vault_guard(vault=vault, policy=slack_policy)
def post_slack_alert(message: str, card: str, session_id: str = "default"):
    # card is guaranteed to remain synthetic (e.g. 4000009995970590)
    ...

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)

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)

pip install "shadow-vault[presidio]"

from shadow_vault import ShadowVault
from shadow_vault.integrations.presidio import PresidioDetector

vault = ShadowVault()
# Plug in Microsoft Presidio / spaCy analyzer:
vault.register_custom_detector(PresidioDetector())

# Dehydrates names with format-preserving synthetic representations:
prompt = vault.dehydrate("Schedule follow-up with Dr. Gregory House.", session_id="s1")
# Prompt: "Schedule follow-up with Anon_Gregory_House."

# Fully reversible at tool boundaries:
hydrated = vault.hydrate(prompt, session_id="s1")
# Restored: "Schedule follow-up with Dr. Gregory House."

python benchmarks/run_benchmarks.py

pytest -v
# 82 passed in 3.26s (100% pass rate)