How to Add AI Guardrails in Python

Add AI guardrails in Python by calling IdenticAPI Unified Guard with httpx or requests from FastAPI routes, Django views, Celery workers, or standalone scripts. Run prompt_injection, pii_secrets, and output_safety at the correct pipeline stages, authenticate with Bearer idapi_test_ keys, and route on aggregated decision values.

For PII-only Python patterns, see PII Detection in Python. For output moderation, see AI Output Moderation in Python.

Prerequisites

• Python 3.10+
• httpx (recommended) or requests
• IdenticAPI API key in environment variables
• Server-side LLM integration (OpenAI, Anthropic, local models)

Unified Guard request schema

Validated fields:

Endpoint: POST https://www.identicapi.com/api/v1/guard

Maximum text length: 32,000 characters.

Environment setup

export IDENTICAPI_API_KEY="idapi_test_your_synthetic_key_here"
export IDENTICAPI_BASE_URL="https://www.identicapi.com"

Load with os.environ or pydantic-settings. Never embed keys in source or Jupyter notebooks committed to git.

Guard client module

guard/unified_guard.py:

import os
from typing import Any, Literal, TypedDict

import httpx

GuardCheck = Literal["prompt_injection", "pii_secrets", "output_safety", "agent_action"]
GuardDecision = Literal["allow", "review", "block"]

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

class UnifiedGuardResponse(TypedDict):
    request_id: str
    api: str
    decision: GuardDecision
    checks: list[dict[str, Any]]
    usage_units: int
    processing_time_ms: int
    detector_version: str

FALLBACK_MESSAGE = "I can't process that request. Please try again."

_client: httpx.Client | None = None

def get_client() -> httpx.Client:
    global _client
    if _client is None:
        _client = httpx.Client(base_url=BASE_URL, timeout=10.0)
    return _client

def run_unified_guard(
    *,
    text: str | None = None,
    checks: list[GuardCheck],
    redact: bool | None = None,
    agent_action: dict[str, Any] | None = None,
) -> UnifiedGuardResponse:
    if not API_KEY:
        raise RuntimeError("IDENTICAPI_API_KEY is not configured")

    payload: dict[str, Any] = {"checks": checks}
    if text is not None:
        payload["text"] = text
    if redact is not None:
        payload["redact"] = redact
    if agent_action is not None:
        payload["agent_action"] = agent_action

    response = get_client().post(
        "/api/v1/guard",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json=payload,
    )
    response.raise_for_status()
    return response.json()

Reuse one httpx.Client per worker for connection pooling.

FastAPI chat endpoint

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

from guard.unified_guard import run_unified_guard, FALLBACK_MESSAGE

app = FastAPI()

class ChatRequest(BaseModel):
    message: str = Field(..., min_length=1, max_length=8000)
    history: list[dict[str, str]] = []

def assemble_prompt(message: str, history: list[dict[str, str]]) -> str:
    lines = [f"{m['role']}: {m['content']}" for m in history]
    lines.append(f"user: {message}")
    return "\n".join(lines)

async def call_llm(prompt: str) -> str:
    # Replace with your async provider client
    return f"Assistant reply for: {prompt[:80]}"

@app.post("/v1/chat")
async def chat(req: ChatRequest):
    prompt = assemble_prompt(req.message, req.history)

    try:
        input_guard = run_unified_guard(
            text=prompt,
            checks=["prompt_injection", "pii_secrets"],
            redact=True,
        )
    except Exception:
        return {"content": FALLBACK_MESSAGE, "decision": "unknown"}

    if input_guard["decision"] != "allow":
        return {
            "content": FALLBACK_MESSAGE,
            "decision": input_guard["decision"],
            "request_id": input_guard["request_id"],
        }

    try:
        completion = await call_llm(prompt)
    except Exception as exc:
        raise HTTPException(status_code=502, detail="LLM unavailable") from exc

    try:
        output_guard = run_unified_guard(
            text=completion,
            checks=["output_safety", "pii_secrets"],
        )
    except Exception:
        return {"content": FALLBACK_MESSAGE, "decision": "unknown"}

    if output_guard["decision"] != "allow":
        return {
            "content": FALLBACK_MESSAGE,
            "decision": output_guard["decision"],
            "request_id": output_guard["request_id"],
        }

    return {
        "content": completion,
        "decision": output_guard["decision"],
        "request_id": output_guard["request_id"],
    }

Django view pattern

Wrap the same run_unified_guard calls in a class-based or function view. Keep guard logic in a service module imported by views and Celery tasks — not duplicated across endpoints.

Celery workers

When workers generate user-visible text (email drafts, ticket summaries), run output guard before marking jobs complete:

@celery_app.task
def summarize_ticket(ticket_id: str) -> None:
    raw = generate_summary(ticket_id)
    guard = run_unified_guard(
        text=raw,
        checks=["output_safety", "pii_secrets"],
    )
    if guard["decision"] != "allow":
        queue_for_review(ticket_id, guard["request_id"])
        return
    publish_summary(ticket_id, raw)

RAG ingestion

Scan chunks before embedding:

def ingest_chunk(chunk: str) -> str | None:
    result = run_unified_guard(
        text=chunk,
        checks=["prompt_injection", "pii_secrets"],
        redact=True,
    )
    if result["decision"] == "block":
        return None  # quarantine
    return chunk  # or redacted variant from pii check findings

See Secure RAG Retrieved Documents.

curl reference

curl -X POST https://www.identicapi.com/api/v1/guard \
  -H "Authorization: Bearer idapi_test_your_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "text": "User pasted: api_key=sk_test_DO_NOT_FORWARD",
    "checks": ["prompt_injection", "pii_secrets"],
    "redact": true
  }'

Response includes decision, per-check verdict values, findings, and request_id. Aggregation: block > review > allow.

Decision routing

Map per-check results when you need finer policy — e.g., block on pii_secrets block but review on prompt_injection review.

Error handling

Distinguish HTTP errors from policy blocks:

try:
    guard = run_unified_guard(text=prompt, checks=["prompt_injection"])
except httpx.HTTPStatusError as exc:
    logger.error("guard_http_error", extra={"status": exc.response.status_code})
    # fail-closed: return FALLBACK_MESSAGE

Document fail-open vs fail-closed — Fail Open vs Fail Closed.

Testing with pytest

import pytest
from unittest.mock import patch

def test_blocks_on_input_guard(monkeypatch):
    def mock_guard(**kwargs):
        return {"decision": "block", "request_id": "req_test", "checks": []}

    monkeypatch.setattr("guard.unified_guard.run_unified_guard", mock_guard)
    # call endpoint; assert LLM mock not invoked

Never use real credentials in tests. Use synthetic values from PII and Secrets Leakage Checklist.

LangChain and other frameworks

If you use LangChain, place guard calls in your application service layer — not inside prompt templates. See LangChain Security Guardrails for durable architecture guidance.

Observability

Log request_id, decision, processing_time_ms, and usage_units. Avoid logging full text payloads in production. Correlate guard calls with LLM provider request_id fields when available.

Related resources

• Add Guardrails to an LLM Application
• Combine AI Security Guardrails
• Evaluate AI Guardrails

Summary

Python guardrails are REST calls to POST /api/v1/guard with stage-appropriate checks, environment-based API keys, and explicit decision routing. Split input and output stages, fail closed on transport errors, and scan the full assembled prompt — not just the latest user message. Call Unified Guard from Python · Documentation