How to Add AI Output Moderation in Python

Add AI output moderation in Python by calling IdenticAPI's output safety endpoint from your chat, RAG, or worker service immediately after the LLM returns — inspect verdict, risk, findings, and reasons, then allow delivery, queue for review, or return a safe fallback before the response reaches users or downstream systems.

This guide covers a reusable client, FastAPI integration, error handling, and placement in common Python LLM stacks.

Endpoint recap

Product details: AI Output Safety, documentation. Interactive testing: AI Output Safety Checker.

Install dependencies

pip install httpx fastapi uvicorn

Use httpx for connection pooling in production services.

Reusable moderation client

output_safety.py:

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Literal, Any

import httpx

IDENTICAPI_BASE = "https://www.identicapi.com"
OUTPUT_SAFETY_PATH = "/api/v1/security/output-safety"

Verdict = Literal["safe", "suspicious", "unsafe"]


@dataclass
class OutputSafetyResult:
    verdict: Verdict
    risk: str
    findings: list[dict[str, str]]
    reasons: list[str]
    request_id: str | None = None

    @classmethod
    def from_api(cls, payload: dict[str, Any]) -> "OutputSafetyResult":
        return cls(
            verdict=payload["verdict"],
            risk=payload.get("risk", "unknown"),
            findings=payload.get("findings", []),
            reasons=payload.get("reasons", []),
            request_id=payload.get("request_id"),
        )


class OutputSafetyClient:
    def __init__(self, api_key: str | None = None, timeout: float = 5.0) -> None:
        key = api_key or os.environ.get("IDENTICAPI_API_KEY")
        if not key:
            raise ValueError("IDENTICAPI_API_KEY is required")
        self._client = httpx.Client(
            base_url=IDENTICAPI_BASE,
            headers={
                "Authorization": f"Bearer {key}",
                "Content-Type": "application/json",
            },
            timeout=timeout,
        )

    def check(self, text: str) -> OutputSafetyResult:
        response = self._client.post(OUTPUT_SAFETY_PATH, json={"text": text})
        response.raise_for_status()
        return OutputSafetyResult.from_api(response.json())

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "OutputSafetyClient":
        return self

    def __exit__(self, *args: object) -> None:
        self.close()


FALLBACK_MESSAGE = (
    "I can't share that response. Please rephrase your question or contact support."
)


def route_verdict(verdict: Verdict) -> Literal["deliver", "review", "block"]:
    if verdict == "safe":
        return "deliver"
    if verdict == "unsafe":
        return "block"
    return "review"

See Block vs Review AI Output for policy design.

FastAPI chat endpoint

main.py:

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

from output_safety import (
    OutputSafetyClient,
    FALLBACK_MESSAGE,
    route_verdict,
)

app = FastAPI()

# Reuse client across requests (connection pooling)
moderation = OutputSafetyClient()


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


class ChatResponse(BaseModel):
    reply: str
    verdict: str | None = None
    request_id: str | None = None
    moderated: bool = True


def call_llm(user_message: str) -> str:
    # Replace with your provider SDK
    return f"Synthetic assistant reply for: {user_message}"


@app.post("/chat", response_model=ChatResponse)
def chat(body: ChatRequest) -> ChatResponse:
    assistant_text = call_llm(body.message)

    try:
        result = moderation.check(assistant_text)
    except Exception as exc:
        # Fail-closed: do not return unmoderated LLM output
        raise HTTPException(
            status_code=503,
            detail="Moderation unavailable",
        ) from exc

    action = route_verdict(result.verdict)

    if action == "block":
        return ChatResponse(
            reply=FALLBACK_MESSAGE,
            verdict=result.verdict,
            request_id=result.request_id,
        )

    if action == "review":
        queue_for_review(body.message, assistant_text, result)
        return ChatResponse(
            reply=FALLBACK_MESSAGE,
            verdict=result.verdict,
            request_id=result.request_id,
        )

    return ChatResponse(
        reply=assistant_text,
        verdict=result.verdict,
        request_id=result.request_id,
    )


def queue_for_review(user_message: str, assistant_text: str, result) -> None:
    # Write to your review store — omit raw text from broad logs
    _ = (user_message, assistant_text, result)

Run locally:

export IDENTICAPI_API_KEY=idapi_test_your_key_here
uvicorn main:app --reload

Example API response

When calling IdenticAPI directly with curl:

curl -X POST https://www.identicapi.com/api/v1/security/output-safety \
  -H "Authorization: Bearer idapi_test_your_key_here" \
  -H "Content-Type: application/json" \
  -d '{"text": "<script>alert(1)</script>Here is your answer."}'

Typical JSON:

{
  "request_id": "req_out_py_77",
  "api": "ai-output-safety",
  "verdict": "unsafe",
  "risk": "high",
  "findings": [
    { "category": "unsafe_markup", "reason": "Script tag detected in text" }
  ],
  "reasons": ["Script tag detected in text"],
  "usage_units": 1
}

Placement in RAG pipelines

retrieve chunks → build prompt → LLM → output_safety.check(answer) → return

Also consider screening retrieved snippets shown to users. Input vs output moderation explains complementary roles. For orchestrated checks, see Unified Guard.

Async services

For async FastAPI, use httpx.AsyncClient:

async def check_async(text: str) -> OutputSafetyResult:
    async with httpx.AsyncClient(base_url=IDENTICAPI_BASE, timeout=5.0) as client:
        r = await client.post(
            OUTPUT_SAFETY_PATH,
            json={"text": text},
            headers={"Authorization": f"Bearer {os.environ['IDENTICAPI_API_KEY']}"},
        )
        r.raise_for_status()
        return OutputSafetyResult.from_api(r.json())

Prefer one long-lived async client per worker process.

Celery / background workers

When workers generate user-visible text (email drafts, ticket summaries):

1. Generate text in worker
2. Moderate before persisting status=ready
3. On unsafe, set status=blocked with static template

Do not publish to message queues consumed by UI before moderation passes.

Logging

Log structured metadata:

logger.info(
    "output_safety",
    extra={
        "verdict": result.verdict,
        "risk": result.risk,
        "request_id": result.request_id,
        "categories": [f["category"] for f in result.findings],
    },
)

Avoid writing full flagged completions to centralized logs unless your retention policy explicitly allows it.

Testing with pytest

def test_blocks_unsafe_verdict(monkeypatch):
    class FakeResult:
        verdict = "unsafe"
        risk = "high"
        findings = []
        reasons = []
        request_id = "test"

    monkeypatch.setattr(
        "main.moderation.check",
        lambda text: FakeResult(),
    )
    # assert endpoint returns FALLBACK_MESSAGE

Add integration tests against a test API key with synthetic payloads from XSS prevention guide.

Pair with web frontends

If a Next.js UI calls your Python backend, moderate in Python before JSON reaches the browser — or moderate in both places if multiple clients exist, using consistent policy. See Output Safety in Next.js.

Related reading

• AI Content Moderation API Guide
• How to Moderate LLM Output
• LLM Output as Untrusted Input
• AI Output Safety Checklist

Limitations

Python integration wraps the same probabilistic screening as other clients:

• Verdicts are not guarantees — maintain human review for suspicious
• Timeouts should map to a documented fail-closed policy
• HTML safety still requires sanitization beyond moderation (safe HTML)

Store keys in environment or secret managers, moderate every user-visible completion, and route on structured verdicts — AI Output Safety handles classification; your Python service enforces policy.