How do I call IdenticAPI PII detection from Python?

POST to /api/v1/security/pii-secrets with httpx or requests, passing text and optional redact in JSON and your API key in the Authorization header. Parse verdict, findings, and redacted_text from the JSON response.

Where should scanning run in FastAPI or Django apps?

In the route or service layer that assembles the LLM prompt—before OpenAI, Anthropic, or local model clients. Apply the same scan to RAG chunk preprocessing in ingestion workers.

How do I scan RAG chunks in Python?

Call scan_pii_secrets on each chunk with redact true before embedding. Quarantine or skip chunks that return unsafe verdicts instead of indexing secrets into your vector store.

Should LangChain memory store raw or redacted messages?

Prefer storing redacted text in memory and databases when possible. If you store raw messages, re-scan the full assembled prompt on every LLM call because memory can reintroduce unredacted PII.

PII Detection in Python

PII detection in Python means invoking a text-scanning HTTP API from your backend—FastAPI, Django, Flask, or Celery workers—before user content reaches an LLM client, then applying policy based on the JSON verdict, findings, and optional redacted_text fields. Python dominates ML and data pipelines, so this guard often sits alongside LangChain, LlamaIndex, or custom RAG code that assembles large context strings from documents and chat history.

This guide provides a complete Python integration with IdenticAPI. Start with What Is PII Detection? and How to Detect PII in Text with an API for API semantics.

Prerequisites

Python 3.10+
httpx or requests for HTTP
API key in environment variable IDENTICAPI_API_KEY

Documentation: PII & Secrets Detection docs. Product: PII & Secrets Detection.

Install dependencies

pip install httpx

Core client module

Save as pii_secrets_client.py:

from __future__ import annotations

import os
from dataclasses import dataclass
from enum import Enum
from typing import Any, Literal, Optional

import httpx

BASE_URL = os.getenv("IDENTICAPI_BASE_URL", "https://www.identicapi.com")
API_KEY = os.getenv("IDENTICAPI_API_KEY")

Verdict = Literal["safe", "suspicious", "unsafe"]
Risk = Literal["low", "medium", "high"]


class PrivacyAction(str, Enum):
    ALLOW = "allow"
    REDACT = "redact"
    BLOCK = "block"


@dataclass
class PiiFinding:
    category: str
    reason: str
    confidence: Optional[float] = None
    start: Optional[int] = None
    end: Optional[int] = None


@dataclass
class PiiSecretsResult:
    request_id: str
    verdict: Verdict
    risk: Risk
    confidence: float
    findings: list[PiiFinding]
    reasons: list[str]
    redacted_text: Optional[str]
    usage_units: int
    processing_time_ms: int
    detector_version: str


class PiiScanError(Exception):
    def __init__(self, message: str, status: Optional[int] = None, body: Any = None):
        super().__init__(message)
        self.status = status
        self.body = body


def resolve_privacy_action(verdict: Verdict) -> PrivacyAction:
    if verdict == "unsafe":
        return PrivacyAction.BLOCK
    if verdict == "suspicious":
        return PrivacyAction.REDACT
    return PrivacyAction.ALLOW


def _parse_result(payload: dict[str, Any]) -> PiiSecretsResult:
    findings = [
        PiiFinding(
            category=f["category"],
            reason=f["reason"],
            confidence=f.get("confidence"),
            start=f.get("start"),
            end=f.get("end"),
        )
        for f in payload.get("findings", [])
    ]
    return PiiSecretsResult(
        request_id=payload["request_id"],
        verdict=payload["verdict"],
        risk=payload["risk"],
        confidence=payload["confidence"],
        findings=findings,
        reasons=payload.get("reasons", []),
        redacted_text=payload.get("redacted_text"),
        usage_units=payload.get("usage_units", 1),
        processing_time_ms=payload.get("processing_time_ms", 0),
        detector_version=payload.get("detector_version", "unknown"),
    )


def scan_pii_secrets(text: str, *, redact: bool = False, timeout: float = 5.0) -> PiiSecretsResult:
    if not API_KEY:
        raise RuntimeError("IDENTICAPI_API_KEY is not configured")

    url = f"{BASE_URL}/api/v1/security/pii-secrets"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    body = {"text": text, "redact": redact}

    with httpx.Client(timeout=timeout) as client:
        response = client.post(url, headers=headers, json=body)

    try:
        payload = response.json()
    except Exception as exc:
        raise PiiScanError("Invalid JSON response", response.status_code) from exc

    if response.status_code >= 400:
        raise PiiScanError(
            f"PII scan failed: HTTP {response.status_code}",
            status=response.status_code,
            body=payload,
        )

    return _parse_result(payload)

FastAPI endpoint example

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from pii_secrets_client import (
    PrivacyAction,
    PiiScanError,
    resolve_privacy_action,
    scan_pii_secrets,
)

app = FastAPI()


class ChatRequest(BaseModel):
    message: str = Field(..., min_length=1, max_length=32000)


class ChatResponse(BaseModel):
    reply: str
    privacy_action: PrivacyAction
    request_id: str


async def call_llm(prompt: str) -> str:
    # Replace with OpenAI, Anthropic, or local model client
    return f"Assistant response to: {prompt[:100]}"


@app.post("/api/chat", response_model=ChatResponse)
async def chat(req: ChatRequest) -> ChatResponse:
    try:
        scan = scan_pii_secrets(req.message, redact=True)
    except PiiScanError as exc:
        raise HTTPException(
            status_code=503,
            detail={"error": "privacy_scan_unavailable", "message": str(exc)},
        ) from exc

    action = resolve_privacy_action(scan.verdict)

    if action == PrivacyAction.BLOCK:
        raise HTTPException(
            status_code=400,
            detail={
                "error": "secrets_detected",
                "message": "Remove API keys, tokens, and private keys from your message.",
                "request_id": scan.request_id,
            },
        )

    prompt_for_llm = scan.redacted_text if action == PrivacyAction.REDACT else req.message
    reply = await call_llm(prompt_for_llm)

    return ChatResponse(
        reply=reply,
        privacy_action=action,
        request_id=scan.request_id,
    )

Run with:

uvicorn main:app --reload

RAG preprocessing helper

Scan each chunk before embedding:

from pii_secrets_client import scan_pii_secrets, resolve_privacy_action, PrivacyAction


def sanitize_chunk(chunk: str) -> tuple[str, bool]:
    """Returns (text_for_embedding, was_redacted)."""
    result = scan_pii_secrets(chunk, redact=True)
    action = resolve_privacy_action(result.verdict)

    if action == PrivacyAction.BLOCK:
        raise ValueError(f"Chunk contains secrets; request_id={result.request_id}")

    if action == PrivacyAction.REDACT and result.redacted_text:
        return result.redacted_text, True

    return chunk, False


def preprocess_documents(chunks: list[str]) -> list[str]:
    clean: list[str] = []
    for i, chunk in enumerate(chunks):
        try:
            text, _ = sanitize_chunk(chunk)
            clean.append(text)
        except ValueError as exc:
            # Skip or quarantine — policy choice
            print(f"quarantine chunk {i}: {exc}")
    return clean

Pair with ingestion checklists in PII and Secrets Leakage Checklist.

Async variant with httpx

For async FastAPI routes under high concurrency:

import httpx

async def scan_pii_secrets_async(text: str, *, redact: bool = False) -> PiiSecretsResult:
    if not API_KEY:
        raise RuntimeError("IDENTICAPI_API_KEY is not configured")

    url = f"{BASE_URL}/api/v1/security/pii-secrets"
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.post(url, headers=headers, json={"text": text, "redact": redact})

    payload = response.json()
    if response.status_code >= 400:
        raise PiiScanError(f"HTTP {response.status_code}", response.status_code, payload)

    return _parse_result(payload)

Synthetic test script

def main() -> None:
    cases = [
        ("Hello!", "safe"),
        ("Contact user@example.com", "suspicious"),
        ("sk-test_abcdefghijklmnopqrstuvwxyz123456", "unsafe"),
    ]
    for text, expected in cases:
        result = scan_pii_secrets(text)
        status = "OK" if result.verdict == expected else "FAIL"
        print(status, expected, result.verdict, text[:50])


if __name__ == "__main__":
    main()

Use the PII Checker for manual validation before CI integration.

Logging guidance

import logging

logger = logging.getLogger("privacy")

def log_scan(result: PiiSecretsResult, action: PrivacyAction) -> None:
    logger.info(
        "pii_scan",
        extra={
            "request_id": result.request_id,
            "verdict": result.verdict,
            "action": action.value,
            "categories": [f.category for f in result.findings],
        },
    )

Never log raw text when findings are present.

Limitations

Adds network round-trip latency per scan—batch or cache where safe
Does not replace Python-side secret management (python-dotenv in repos is still dangerous)
LangChain memory may reintroduce unredacted history—scan assembled prompt, not only latest input
Obfuscated PII may bypass regex detectors

Node.js equivalent: PII Detection in Node.js. Architecture patterns: LLM Privacy Filter.