MCP Presidio Server

import asyncio
import logging, logging.config, uuid
from typing import List, Dict, Any

from mcp.server.fastmcp import FastMCP, Context
from app.services import storage
from app.services.anonymizer import content_anonymizer, reverse_mapping, content_deanonymizer
from collections import defaultdict

logging.config.fileConfig("logging.conf")
logger = logging.getLogger(__name__)

mcp = FastMCP("mcp-presidio-server")

@mcp.tool()
async def generate_session_id(ctx: Context) -> Dict[str, str]:
    """Generate and persist a new session id."""
    sid = str(uuid.uuid4())
    storage.create_session(sid)
    return {"session_id": sid}

@mcp.tool()
async def anonymize(session_id: str, texts: List[str]) -> Dict[str, Any]:
    """Anonymize a list of texts and store mappings under session_id."""
    storage.create_session(session_id)
    entity_mapping = defaultdict(dict)
    entity_counter = defaultdict(int)
    anon_texts: List[str] = []
    for t in texts:
        anon, entity_mapping, entity_counter = content_anonymizer(t, entity_mapping, entity_counter)
        anon_texts.append(anon)
    rev = reverse_mapping(entity_mapping)  # placeholder -> original
    storage.save_mappings(session_id, rev)
    return {"session_id": session_id, "texts": anon_texts}

@mcp.tool()
async def deanonymize(session_id: str, text: str) -> Dict[str, str]:
    """Given session_id and anonymized text, return deanonymized text."""
    rev = storage.get_mapping(session_id)
    if not rev:
        return {"error": "No mapping for session_id", "text": text}
    return {"text": content_deanonymizer(text, rev), "session_id": session_id}

if __name__ == "__main__":
    # Run an stdio MCP server
    mcp.run()