How to set up PII redaction for AI workflows.

Analysis

Every time your team pastes a customer email, a support transcript, or a spreadsheet into an AI tool, a copy of that text leaves the building. It travels to someone else's servers. For a lot of Australian businesses that is fine right up until the day it isn't: a name, a phone number, a Medicare reference, or a client's home address ends up in a prompt log somewhere you have no control over.

The fix is not "stop using AI". It is to scrub the personal details out of the text before it goes anywhere, then stitch them back in when the answer comes home. The customer's name becomes a placeholder on the way out, and the placeholder becomes their name again on the way back. The AI does its job. The private bits never leave your control.

Microsoft built an open-source tool for exactly this, called Presidio, and it already knows how to spot dozens of kinds of personal data out of the box. The rest of this piece is the build: how to detect personal data, how to teach the tool about your own internal codes, how to make the redaction reversible, and how to keep an audit trail your compliance people will actually accept. It is written for engineers, but the idea underneath is simple enough to explain to anyone signing off the budget.

Analysis

Prerequisites

• Python 3.10+
• Presidio: pip install presidio-analyzer presidio-anonymizer (presidio-anonymizer on PyPI)
• spaCy model: python -m spacy download en_core_web_lg (Presidio installation docs)
• FastAPI for the redaction service

Step-by-Step Framework

Step 1: Basic PII Detection

Presidio splits the work in two. The `AnalyzerEngine` finds the personal data, and the AnonymizerEngine decides what to do with each piece it finds. You point it at some text, it hands back the entities it spotted, and you tell it how to mask each type.

# redaction/basic.py
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def redact_pii(text: str) -> dict:
    # Analyse
    results = analyzer.analyze(text=text, language='en')

    # Anonymise
    anonymized = anonymizer.anonymize(
        text=text,
        analyzer_results=results,
        operators={
            "DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
            "PERSON": OperatorConfig("mask", {"type": "hash", "hash_type": "sha256"}),
            "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "<EMAIL>"})
        }
    )

    return {
        "original": text,
        "redacted": anonymized.text,
        "entities_found": [{"type": r.entity_type, "start": r.start, "end": r.end} for r in results],
        "entity_count": len(results)
    }

# Test
text = "Contact John Smith at john.smith@example.com or 555-123-4567. His SSN is 123-45-6789."
result = redact_pii(text)
print(result["redacted"])
# Contact <REDACTED> at <EMAIL> or <REDACTED>. His SSN is <REDACTED>.

One caution before you copy that block into production. The PERSON line above mixes up two different Presidio operators. The `mask` operator only understands chars_to_mask, masking_char and from_end; the type and hash_type keys actually belong to the separate hash operator. As written, this will not SHA-256 the name and will probably error or quietly do nothing. If you want hashing, switch the operator to "hash". Presidio ships five operators in total: replace, redact, mask, hash and encrypt, so pick the one that matches what you mean.

Step 2: Custom Entity Recognisers

Presidio's built-in list covers the usual suspects: names, emails, phone numbers, credit cards, addresses. What it does not know is your business. Your employee IDs, your project codes, your internal record numbers all look like ordinary text to it. You teach it those with a `PatternRecognizer` and a regex.

# redaction/custom_entities.py
from presidio_analyzer import PatternRecognizer, Pattern

# Employee ID recogniser
employee_id_pattern = Pattern(
    name="employee_id",
    regex=r"EMP-[0-9]{5,8}",
    score=0.9
)
employee_recognizer = PatternRecognizer(
    supported_entity="EMPLOYEE_ID",
    patterns=[employee_id_pattern]
)

# Project code recogniser
project_pattern = Pattern(
    name="project_code",
    regex=r"PRJ-[A-Z]{2,4}-\d{4}",
    score=0.85
)
project_recognizer = PatternRecognizer(
    supported_entity="PROJECT_CODE",
    patterns=[project_pattern]
)

# Medical record number (HIPAA)
mrn_pattern = Pattern(
    name="mrn",
    regex=r"MRN\d{8,10}",
    score=0.95
)
mrn_recognizer = PatternRecognizer(
    supported_entity="MEDICAL_RECORD_NUMBER",
    patterns=[mrn_pattern]
)

# Register custom recognisers
from presidio_analyzer import RecognizerRegistry

registry = RecognizerRegistry()
registry.load_predefined_recognizers()
registry.add_recognizer(employee_recognizer)
registry.add_recognizer(project_recognizer)
registry.add_recognizer(mrn_recognizer)

analyzer = AnalyzerEngine(registry=registry)

The score on each pattern is your confidence dial. A medical record number with a strict format earns a 0.95; a looser pattern that might catch false positives deserves something lower. Load the predefined recognisers first, then stack your own on top.

Step 3: Reversible Redaction (Deanonymisation)

Masking is enough when you only need an answer about the text. But plenty of workflows need the real values back: a booking confirmation has to name the actual customer, not <PERSON_xxxx>. That is the redact-then-restore pattern Presidio is built around. You swap each entity for a unique token, keep a private map of token-to-original, and reverse the swap after the LLM replies.

# redaction/reversible.py
import hashlib
import json
from typing import Dict

class ReversibleRedactor:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.vault: Dict[str, str] = {}  # hash -> original

    def redact(self, text: str) -> tuple[str, dict]:
        results = self.analyzer.analyze(text=text, language='en')
        vault = {}
        current_pos = 0
        parts = []

        for result in sorted(results, key=lambda x: x.start):
            # Add text before entity
            parts.append(text[current_pos:result.start])

            # Create vault key
            entity_text = text[result.start:result.end]
            vault_key = f"<{result.entity_type}_{self._hash(entity_text)}>"
            vault[vault_key] = entity_text

            parts.append(vault_key)
            current_pos = result.end

        parts.append(text[current_pos:])
        redacted = "".join(parts)

        return redacted, vault

    def restore(self, text: str, vault: dict) -> str:
        restored = text
        for key, value in vault.items():
            restored = restored.replace(key, value)
        return restored

    def _hash(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()[:8]

# Usage
redactor = ReversibleRedactor()
original = "John Smith booked a flight for EMP-12345"
redacted, vault = redactor.redact(original)
# "<PERSON_a1b2c3d4> booked a flight for <EMPLOYEE_ID_e5f6g7h8>"

llm_response = f"Booking confirmed for {redacted.split()[0]}"
final = redactor.restore(llm_response, vault)
# "Booking confirmed for John Smith"

The hashes in those comments (a1b2c3d4 and so on) are stand-ins to show the shape of the output. Real SHA-256 prefixes will look nothing like that. The point is that the token is derived from the value rather than a sequential counter, so an attacker reading the redacted text can't infer how many records you hold or guess the next one.

Step 4: FastAPI Middleware

Doing this by hand on every call gets tedious, so push it into middleware that sits in front of your endpoints. The request body gets redacted on the way in, the vault rides along in request state, and the response gets restored on the way out.

# redaction/middleware.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import json

app = FastAPI()
redactor = ReversibleRedactor()

@app.middleware("http")
async def pii_redaction_middleware(request: Request, call_next):
    # Skip if endpoint opts out
    if request.url.path in ["/health", "/metrics"]:
        return await call_next(request)

    # Redact request body
    body = await request.body()
    if body:
        body_text = body.decode()
        redacted_body, vault = redactor.redact(body_text)

        # Store vault in request state for restoration
        request.state.vault = vault

        # Replace request body
        async def receive():
            return {"type": "http.request", "body": redacted_body.encode()}
        request._receive = receive

    # Process request
    response = await call_next(request)

    # Restore PII in response
    if hasattr(request.state, 'vault') and request.state.vault:
        response_body = b"".join([chunk async for chunk in response.body_iterator])
        restored_body = redactor.restore(response_body.decode(), request.state.vault)
        return JSONResponse(
            content=json.loads(restored_body),
            status_code=response.status_code
        )

    return response

@app.post("/llm/completions")
async def llm_completions(request: Request):
    body = await request.json()
    # Forward to LLM API (body already redacted by middleware)
    # ... LLM call
    return {"completion": "Processed: " + body.get("prompt", "")}

Treat this as a working sketch, not a finished product. Overriding request._receive and draining response.body_iterator are real Starlette techniques, but they're fragile: they assume JSON in and JSON out, and they will break on streaming responses or anything that isn't valid JSON. Fine for learning the pattern, worth hardening before it touches customer traffic.

Step 5: Claude Code Integration

If you want the redaction step to fire automatically from inside an agent workflow, you can wrap the service in a small client that calls your /redact endpoint. The snippet below shows the idea.

// .claude/skills/pii-guard.ts
import { defineSkill } from '@anthropic/claude-sdk';
import { z } from 'zod';

const PII_ENDPOINT = process.env.PII_SERVICE_URL || 'http://localhost:8000';

export default defineSkill({
  name: 'pii-guard',
  description: 'Redacts PII before sending to LLM APIs',

  input: z.object({
    text: z.string(),
    mode: z.enum(['redact', 'reversible', 'check']).default('reversible')
  }),

  output: z.object({
    processed: z.string(),
    entities_found: z.array(z.string()),
    vault: z.record(z.string()).optional()
  }),

  async execute({ text, mode }) {
    const response = await fetch(`${PII_ENDPOINT}/redact`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ text, mode })
    });

    return response.json();
  }
});

Read this one as illustration rather than something you can npm install and run. The @anthropic/claude-sdk package and its defineSkill() function don't appear to be a real, published API. Anthropic's actual TypeScript library is `@anthropic-ai/sdk`, and Claude Code skills are authored as Markdown SKILL.md files rather than registered through a TypeScript function. The useful takeaway is the shape: validate the input, call your redaction service, hand back the cleaned text plus the vault.

Step 6: Audit Logging

The logging is what turns "we redact PII" into something you can prove. Record the entity types, the counts, the source, and a hash of the original. Never write the values themselves to the log, or you've just recreated the leak you were trying to stop.

# redaction/audit.py
import structlog
import json
from datetime import datetime

logger = structlog.get_logger("pii_audit")

def log_redaction(original_text: str, redacted_text: str, entities: list, source_ip: str):
    logger.info(
        "PII redaction performed",
        timestamp=datetime.utcnow().isoformat(),
        source_ip=source_ip,
        entity_types=[e['type'] for e in entities],
        entity_count=len(entities),
        original_length=len(original_text),
        redacted_length=len(redacted_text),
        # NEVER log the actual text or entity values
        text_hash=hashlib.sha256(original_text.encode()).hexdigest()
    )

def log_api_call(redacted_prompt: str, model: str, vault_size: int):
    logger.info(
        "LLM API call with redacted data",
        timestamp=datetime.utcnow().isoformat(),
        model=model,
        vault_entries=vault_size,
        prompt_length=len(redacted_prompt)
    )

Do/Don't

Compliance Matrix

A redaction layer helps you meet these requirements; it does not single-handedly make you compliant. Read the matrix as "this is the part of the puzzle redaction solves", not a tick-box for the whole regulation.

Conclusion

Redacting personal data before it reaches an LLM API is the difference between using AI safely and hoping nobody audits you. Presidio gives you solid coverage from the start, and a few custom recognisers cover the codes specific to your business. The reversible pattern keeps the data protected in transit while you still get usable answers back. Audit everything, keep the real values out of your logs, and re-test the pipeline every quarter so it doesn't quietly rot.