Python FastAPI PII Masking: Lightweight Local Sidecar (<1.2ms vs 40ms Presidio)

import os
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI

app = FastAPI(title="Zero-Trust AI Gateway")
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "your-api-key"))

SIDECAR_URL = os.environ.get("ZTDS_SIDECAR_URL", "http://127.0.0.1:3100")

class PromptRequest(BaseModel):
    prompt: str
    profile: str = "General"

@app.post("/api/safe-chat")
async def safe_chat(req: PromptRequest):
    async with httpx.AsyncClient(timeout=2.0) as http:
        # 1. Sanitize prompt in local sidecar (<1.2ms latency)
        scrub_resp = await http.post(
            f"{SIDECAR_URL}/scrub",
            json={"text": req.prompt, "profile": req.profile}
        )
        if scrub_resp.status_code != 200:
            raise HTTPException(status_code=500, detail="Sanitization failed")
        
        scrub_data = scrub_resp.json()
        sanitized_prompt = scrub_data["sanitizedText"]
        token_map = scrub_data["tokenMap"]

        # 2. Forward sanitized prompt to OpenAI (zero PII leaves process)
        completion = openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": sanitized_prompt}],
            temperature=0.2
        )
        ai_response = completion.choices[0].message.content or ""

        # 3. Deterministically restore original customer identifiers
        unscrub_resp = await http.post(
            f"{SIDECAR_URL}/unscrub",
            json={"text": ai_response, "tokenMap": token_map}
        )
        final_text = unscrub_resp.json().get("restoredText", ai_response)

        return {
            "response": final_text,
            "maskedEntitiesCount": len(token_map),
            "telemetry": scrub_data.get("telemetry", {})
        }

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