Content Moderation API Python: Practical Backend Tutorial

If you run a Python service that accepts user text, you already have the problem. Comments, chat messages, profile bios, support tickets and increasingly LLM output all arrive as strings that need a policy decision attached to them. This tutorial is a hands-on walkthrough of adding a content moderation API Python integration to a service you already ship: one POST request, a JSON body, and a boolean response you can branch on. No tuning loop, no score calibration, no model hosting.

Everything below uses the documented cloud API. The base URL is https://api.cautionlabs.com, the endpoint is POST /moderation, authentication is a bearer Cloud API key, and the request body has a single required field, text. If you want the conceptual overview before the Python specifics, adding moderation in an afternoon covers the shape of the product first.

Why Python teams reach for a moderation API

Moderation rarely shows up as its own project. It shows up as a line item inside a feature you are already building: a comment endpoint that needs to reject a small percentage of submissions, a chat backend that needs to route certain messages to a review queue, an ingestion pipeline that needs to tag documents before indexing, or an LLM wrapper that needs to check model output before it reaches a user.

The CautionLabs API is deliberately narrow. You send text, you get back a categories object with twelve boolean keys and an overall flagged boolean. That is the entire contract. There are no scores, no probabilities, no confidence values to interpret, and no thresholds to tune. That last point matters more than it looks. When a moderation provider returns a float, every team ends up inventing a cutoff, arguing about it in review, and re-tuning it when the distribution shifts. With booleans, the API answers "is this category present in this text?" and the decision about what to do about it stays in your codebase, where your product context lives.

That split is the design pattern this tutorial builds around: the API supplies boolean moderation flags from Python, and your application owns the policy that turns those flags into allow, label, review queue or block.

Before you write code: base URL, key type and request shape

Three things to get right before the first request.

Base URL and endpoint. All cloud requests go to https://api.cautionlabs.com, and the moderation endpoint is POST /moderation. There is one endpoint. The full contract is in the Cloud API Reference and the machine-readable OpenAPI description.

Authentication. Send a Cloud API key from the dashboard as a bearer token in the Authorization header. A private on-premises key does not work with the cloud API, so keep the two credential types separate in your configuration. Read the key from an environment variable, never from a checked-in file.

Request body. A single required field:

{ "text": "..." }

The text value is a string between 1 and 10,000 characters. An empty string is not a valid request, which is worth handling client-side so you do not spend a round trip on a bug you already know about.

Cost shape. Cloud usage is billed at $0.40 USD per 1,000 requests, where a request covers up to 3,000 characters and each started 3,000 characters counts as one request. The arithmetic is covered below. There is no batch endpoint, so one text is sent per request.

Making a content moderation API Python request

The fastest way to see the response shape is a short script. This uses the requests library, which is already in most Python services, and reads the key from the environment.

Minimal example with requests

import os
import requests

API_URL = "https://api.cautionlabs.com/moderation"


def moderate(text: str) -> dict:
    api_key = os.environ["CAUTIONLABS_API_KEY"]

    response = requests.post(
        API_URL,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        json={"text": text},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()


if __name__ == "__main__":
    result = moderate("This is a test comment that should pass moderation.")

    print("flagged:", result["flagged"])
    for category, is_flagged in result["categories"].items():
        print(f"  {category}: {is_flagged}")

Run it with the key exported in your shell:

export CAUTIONLABS_API_KEY="your-cloud-api-key"
python moderate.py

The printout is intentionally boring, and that is the point. You get twelve booleans and an overall flagged value. Nothing to normalize, nothing to compare against a baseline.

One practical note: the json= argument sets the Content-Type header for you in most cases, but sending it explicitly keeps the request readable and avoids surprises if a proxy in your stack rewrites headers.

The same call with httpx for async services

If you are running FastAPI, Starlette or any asyncio-based stack, blocking a worker on a synchronous HTTP call is the wrong shape. httpx gives you the same request with an async client, and a reusable client avoids a new connection pool per request.

import os
import httpx


class ModerationClient:
    def __init__(self, api_key: str | None = None, timeout: float = 10.0):
        api_key = api_key or os.environ["CAUTIONLABS_API_KEY"]
        self._client = httpx.AsyncClient(
            base_url="https://api.cautionlabs.com",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=timeout,
        )

    async def moderate(self, text: str) -> dict:
        response = await self._client.post("/moderation", json={"text": text})
        response.raise_for_status()
        return response.json()

    async def aclose(self) -> None:
        await self._client.aclose()

Wire it into the app lifespan so the pool is created once and closed on shutdown:

from contextlib import asynccontextmanager
from fastapi import FastAPI

moderation = ModerationClient()


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await moderation.aclose()


app = FastAPI(lifespan=lifespan)

For FastAPI content moderation inside a route handler, the call is a single await:

from pydantic import BaseModel, Field


class Comment(BaseModel):
    text: str = Field(min_length=1, max_length=10000)


@app.post("/comments")
async def create_comment(comment: Comment):
    result = await moderation.moderate(comment.text)
    # Your policy decides what happens next; see the next section.
    return {"flagged": result["flagged"], "categories": result["categories"]}

The Pydantic field constraints mirror the documented envelope: at least one character, at most 10,000. Rejecting oversized input before it leaves your process is cheaper than discovering the problem in a response body.

Reading the response: twelve flags and one overall value

The response has exactly two top-level keys, categories and flagged. The categories object always contains the same twelve boolean keys:

• toxic
• profanity
• hate
• harassment
• self_harm
• adult
• violence
• drugs
• weapons
• pii
• spam
• minor

The flagged boolean is true when any category is flagged. Treat it as a fast path: if flagged is false you can allow the text without inspecting the categories at all, and only walk the categories object when it is true.

A typical flagged response looks like this:

{
  "categories": {
    "toxic": true,
    "profanity": true,
    "hate": false,
    "harassment": false,
    "self_harm": false,
    "adult": false,
    "violence": false,
    "drugs": false,
    "weapons": false,
    "pii": false,
    "spam": false,
    "minor": false
  },
  "flagged": true
}

Because there is no numeric output, your moderation policy is application code. That is a feature, not a gap. Keep it in exactly one place so that changing what your product blocks does not require hunting through route handlers.

BLOCK = {"hate", "harassment", "self_harm", "minor", "violence", "weapons"}
REVIEW = {"toxic", "drugs", "adult"}
LABEL = {"profanity", "spam", "pii"}


def decide_action(result: dict) -> str:
    """Map boolean flags to an application action."""
    flagged = {name for name, on in result["categories"].items() if on}

    if flagged & BLOCK:
        return "block"
    if flagged & REVIEW:
        return "review"
    if flagged & LABEL:
        return "label"
    return "allow"

Those three sets are an example, not a recommendation from the API. The API does not rank categories, weight them or express any opinion about severity. If your product blocks profanity and labels PII, swap the names in the sets and the function keeps working. If you want per-category routing, return the flagged set alongside the action and store both.

Two habits worth adopting here. First, never log raw text next to a pii flag; the flag is a signal that the payload may contain exactly the data your log retention policy is trying to avoid. Log a hash, an ID, or the category list instead. Second, persist the action alongside the flags so that when you revisit the policy you can see what the old policy did on real traffic.

Long inputs and what they cost

The cloud API checks texts of up to 10,000 characters, and each started 3,000 characters counts as one request. That is the whole cost model, and it is worth doing the arithmetic once so nobody is surprised by a bill.

Take a 7,000-character document:

• Characters 1 to 3,000: one request
• Characters 3,001 to 6,000: one request
• Characters 6,001 to 7,000: one request, because the third 3,000-character window is started even though it is not filled
• Total: 3 requests

At $0.40 USD per 1,000 requests, that document costs $0.0012. Moderate 1,000 documents of that size and you have issued 3,000 requests, which comes to $1.20.

A text at the 10,000-character ceiling is 4 requests, or $0.0016 per text. A short comment of a few hundred characters is 1 request, or $0.0004. The step function is what matters: costs move in 3,000-character increments, not per character.

Two engineering consequences. If you are moderating long documents — articles, transcripts, generated reports — the useful slice is usually what the user actually reads first, and choosing that slice is a product decision you make in your own code. And if cost visibility matters, count requests in your own metrics rather than reconstructing them later: increment a counter every time your client issues a call.

Because there is no batch endpoint, a list of 500 comments is 500 requests. If you are backfilling an archive, that is a job with a size you can calculate in advance, which is exactly what you want before you start it.

Handling errors without guessing

Failures return a JSON body with a code and a message. The documented codes are invalid_request, unauthorized, insufficient_credits, wrong_key_type and internal_error. Branch on the code rather than trying to infer meaning from a status line, and sort them into two buckets: things you fix by changing configuration, and things you surface to a human.

class ConfigurationError(Exception):
    """The request or credentials are wrong; fix the deployment."""


class AccountError(Exception):
    """The account needs attention before requests will succeed."""


class TransientError(Exception):
    """The request may succeed if tried again."""


def raise_for_moderation_error(payload: dict) -> None:
    code = payload["code"]
    message = payload["message"]

    if code in {"invalid_request", "wrong_key_type"}:
        raise ConfigurationError(f"{code}: {message}")
    if code in {"unauthorized", "insufficient_credits"}:
        raise AccountError(f"{code}: {message}")
    if code == "internal_error":
        raise TransientError(f"{code}: {message}")

    raise RuntimeError(f"Unexpected moderation error {code}: {message}")

invalid_request and wrong_key_type are bugs in your deployment — an empty text field, an oversized payload, or a private on-premises key pointed at the cloud API. They should fail loudly in development, not silently in production. unauthorized and insufficient_credits are account-level conditions, so route them somewhere a human will see them. internal_error is transient, and the decision about how to handle it belongs to your application.

The dedicated content moderation API error codes guide goes through each code in depth, including what to check first. Read it before you build out retry or alerting logic so you are branching on documented behaviour rather than guesses.

A small, reusable Python moderation client

Route handlers should not know the endpoint URL. Wrap the call once and let the rest of the service depend on the wrapper.

import os
from dataclasses import dataclass

import requests

API_URL = "https://api.cautionlabs.com/moderation"
MAX_TEXT_LENGTH = 10_000


@dataclass
class ModerationResult:
    flagged: bool
    categories: dict

    @property
    def flagged_categories(self) -> list[str]:
        return [name for name, on in self.categories.items() if on]


class ModerationClient:
    def __init__(self, api_key: str | None = None, timeout: float = 10.0):
        self._api_key = api_key or os.environ["CAUTIONLABS_API_KEY"]
        self._timeout = timeout

    def moderate(self, text: str) -> ModerationResult:
        if not text or not text.strip():
            raise ValueError("text must be a non-empty string")
        if len(text) > MAX_TEXT_LENGTH:
            raise ValueError(f"text exceeds {MAX_TEXT_LENGTH} characters")

        response = requests.post(
            API_URL,
            headers={
                "Authorization": f"Bearer {self._api_key}",
                "Content-Type": "application/json",
            },
            json={"text": text},
            timeout=self._timeout,
        )

        payload = response.json()

        if response.status_code != 200:
            raise_for_moderation_error(payload)

        return ModerationResult(
            flagged=payload["flagged"],
            categories=payload["categories"],
        )

Four things this wrapper buys you. It validates the documented envelope before spending a request. It sets a timeout, so a slow network cannot pin a worker indefinitely. It returns a typed result, so callers get attribute access instead of dictionary lookups scattered through the codebase. And it is the single place where credentials, the base URL and error mapping live, which means swapping to the on-premises endpoint later is a one-line change.

In a Django view the wrapper slots into the existing request flow without restructuring anything:

from django.http import JsonResponse

client = ModerationClient()


def create_comment(request):
    text = request.POST.get("text", "")
    result = client.moderate(text)
    action = decide_action({"categories": result.categories, "flagged": result.flagged})

    if action == "block":
        return JsonResponse({"error": "rejected"}, status=422)

    # Save the comment, attach the flagged categories as labels,
    # or enqueue it for review depending on the action.
    return JsonResponse({"status": action})

FAQ

Can I send a list of texts in one call?

No. There is no batch endpoint. One text is sent per request, and the response covers that text only. If you are moderating many items, iterate in your own code and count the requests you issue.

Does the API return a score I can threshold?

No. The response contains twelve boolean category flags and one overall flagged boolean, which is true when any category is flagged. There are no scores, probabilities or confidence values, and there are no thresholds to tune. Every allow, label, review and block decision is made by your application code.

Does the API send webhooks?

No. It is a synchronous request and response. If your architecture is event-driven, you publish your own event after you receive the response.

Can I run the same booleans on my own network?

Yes. The on-premises model returns the same flag-based result and is reachable locally at POST /moderate, defaulting to http://localhost:8000. It is the right fit for air-gapped environments, strict data-residency requirements, or teams that want inference inside their own perimeter. The self-hosted moderation with Docker guide walks through the deployment, and the On-Premises Guide has the registry login, requirements and local endpoint details. Remember that a private on-premises key does not work with the cloud API, so keep the two code paths configured separately.

Where to go next

Open the playground and paste a real comment, a flagged message from your review queue, or a piece of LLM output into the request box. Seeing the twelve booleans come back on text you recognise is the fastest way to calibrate what your policy should do with them. Then create a Cloud API key in the dashboard, drop the wrapper above into your service, and put your policy in one function.

If you are moderating model output rather than user input, checking LLM output covers the patterns that come up in that flow. If your team is polyglot, the content moderation API in Node.js tutorial mirrors this walkthrough for your JavaScript services, so both codebases can share the same policy semantics. And when you are ready to size the rollout, the pricing page lays out the cloud request rate and the on-premises license side by side.

---

Cover image: Photo by jurvetson, CC BY 2.0 · source