Last reviewed: 2026-07-31

Direct answer

Put a separate moderation request between FastAPI request validation and the generation request. CometAPI’s moderation endpoint reference documents POST /v1/moderations with required model and input fields. For one chat message, send the validated string with omni-moderation-latest, read the single item in results, and forward the message to generation only when your application policy allows it.

The sample below uses flagged as the first-pass decision. That is deliberately an application policy, not a claim that moderation replaces policy design. The OpenAI moderation guide describes flagged, category booleans, category scores, and applied input types, while cautioning that scores are signals that may require recalibration. The gate therefore avoids inventing a score threshold.

It also fails closed: if moderation times out, returns a non-success status, or produces an unexpected result count, the route does not call the generation endpoint. Users receive a stable error code and a request ID; operators receive sanitized metadata without the prompt, generated answer, credential values, or upstream response body.

import logging
import os
import time
import uuid
from typing import Any

from fastapi import FastAPI, HTTPException
from openai import APIConnectionError, APIStatusError, OpenAI
from pydantic import BaseModel, Field

app = FastAPI()
log = logging.getLogger('moderation_gate')

runtime_value = os.environ.get('COMETAPI_KEY')
gateway_url = os.environ.get('COMETAPI_BASE_URL')
if not runtime_value or not gateway_url:
    raise RuntimeError('Required CometAPI environment configuration is missing')

client = OpenAI(
    api_key=runtime_value,
    base_url=gateway_url,
    timeout=15.0,
    max_retries=0,
)


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


class ChatResponse(BaseModel):
    request_id: str
    answer: str


def elapsed_ms(started: float) -> int:
    return round((time.monotonic() - started) * 1000)


def log_event(name: str, **fields: Any) -> None:
    log.info('chat_gate', extra={'event_name': name, **fields})


@app.post('/chat', response_model=ChatResponse)
def chat(payload: ChatRequest) -> ChatResponse:
    request_id = f'req-{uuid.uuid4().hex[:12]}'
    moderation_started = time.monotonic()

    try:
        moderation = client.moderations.create(
            model='omni-moderation-latest',
            input=payload.message,
        )
        if len(moderation.results) != 1:
            raise RuntimeError('Unexpected moderation result count')
    except APIStatusError as exc:
        log_event(
            'moderation_failed',
            request_id=request_id,
            outcome='upstream_status',
            upstream_status=exc.status_code,
            duration_ms=elapsed_ms(moderation_started),
        )
        raise HTTPException(
            status_code=503,
            detail={'code': 'moderation_unavailable', 'request_id': request_id},
        ) from exc
    except APIConnectionError as exc:
        log_event(
            'moderation_failed',
            request_id=request_id,
            outcome='connection_error',
            duration_ms=elapsed_ms(moderation_started),
        )
        raise HTTPException(
            status_code=503,
            detail={'code': 'moderation_unavailable', 'request_id': request_id},
        ) from exc
    except RuntimeError as exc:
        log_event(
            'moderation_failed',
            request_id=request_id,
            outcome='contract_mismatch',
            duration_ms=elapsed_ms(moderation_started),
        )
        raise HTTPException(
            status_code=502,
            detail={'code': 'moderation_invalid_response', 'request_id': request_id},
        ) from exc

    result_data = moderation.results[0].model_dump(
        mode='json',
        by_alias=True,
    )
    category_flags = result_data.get('categories', {})
    flagged_categories = sorted(
        name for name, is_flagged in category_flags.items() if is_flagged
    )
    is_flagged = bool(result_data.get('flagged'))

    log_event(
        'moderation_complete',
        request_id=request_id,
        outcome='blocked' if is_flagged else 'allowed',
        moderation_model=moderation.model,
        input_chars=len(payload.message),
        flagged_categories=flagged_categories,
        duration_ms=elapsed_ms(moderation_started),
    )

    if is_flagged:
        raise HTTPException(
            status_code=400,
            detail={'code': 'input_not_allowed', 'request_id': request_id},
        )

    generation_started = time.monotonic()
    try:
        generation = client.responses.create(
            model='gpt-5.6-sol',
            input=payload.message,
        )
    except APIStatusError as exc:
        log_event(
            'generation_failed',
            request_id=request_id,
            outcome='upstream_status',
            upstream_status=exc.status_code,
            duration_ms=elapsed_ms(generation_started),
        )
        raise HTTPException(
            status_code=502,
            detail={'code': 'generation_failed', 'request_id': request_id},
        ) from exc
    except APIConnectionError as exc:
        log_event(
            'generation_failed',
            request_id=request_id,
            outcome='connection_error',
            duration_ms=elapsed_ms(generation_started),
        )
        raise HTTPException(
            status_code=502,
            detail={'code': 'generation_failed', 'request_id': request_id},
        ) from exc

    answer = (generation.output_text or '').strip()
    if not answer:
        log_event(
            'generation_failed',
            request_id=request_id,
            outcome='empty_output',
            duration_ms=elapsed_ms(generation_started),
        )
        raise HTTPException(
            status_code=502,
            detail={'code': 'generation_empty', 'request_id': request_id},
        )

    log_event(
        'generation_complete',
        request_id=request_id,
        outcome='success',
        output_chars=len(answer),
        duration_ms=elapsed_ms(generation_started),
    )
    return ChatResponse(request_id=request_id, answer=answer)

Set COMETAPI_KEY through your deployment’s secret mechanism and set COMETAPI_BASE_URL through normal runtime configuration. The sample refuses to start if either setting is absent. It never places either value in a URL, log event, exception detail, or source-controlled example.

The 4,000-character request limit is a local application budget, not a documented CometAPI limit. Adjust it for your product and test it independently. Before deployment, review the CometAPI base URL and endpoint family so runtime configuration matches the current contract.

Who this is for

This pattern is for Python teams that already have, or are building, a server-side FastAPI chat route. It is especially useful when untrusted user text would otherwise flow directly into a generation model. FastAPI’s request-body documentation confirms that a Pydantic model can parse and validate JSON input before route logic runs, which gives the moderation gate a bounded, typed string to inspect.

Keep the gate server-side. A browser or mobile client can provide useful interface checks, but it cannot protect a server credential or guarantee that callers will use the intended path. The related guide on how to keep CometAPI keys out of tutorial repositories covers that boundary in more detail.

The sample handles one text input per request. CometAPI also documents string arrays and multimodal parts, but adding those formats changes result alignment, input validation, and privacy handling. Extend the route only after the single-text contract is covered by tests.

Key takeaways

  • Validate the incoming JSON body before making any upstream request.
  • Call the documented moderation endpoint separately, before generation.
  • Treat flagged and category data as inputs to an explicit application policy.
  • Do not invent a score threshold without a labeled evaluation set and a recalibration plan.
  • Fail closed when moderation is unavailable or its response shape is unexpected.
  • Return generic public errors and keep detailed operational state in sanitized logs.
  • Distinguish moderation failure from generation failure so operators know which dependency failed.

Happy-path operator workflow

  1. At startup, verify that required runtime configuration exists. A missing setting should stop the service before it accepts traffic.
  2. Submit a normal test fixture to /chat. FastAPI validates the body, then the route sends exactly one string to moderation.
  3. Confirm the moderation response contains exactly one result and that flagged is false.
  4. Confirm the generation request occurs only after that allowed decision.
  5. Verify the client receives an answer and a request ID.
  6. Verify the logs contain event name, request ID, outcome, model name, character counts, duration, and flagged category names. They must not contain the input or answer.

A sanitized happy-path log can look like this:

event_name=moderation_complete request_id=req-7f3a outcome=allowed moderation_model=omni-moderation-latest input_chars=42 flagged_categories=[] duration_ms=218
event_name=generation_complete request_id=req-7f3a outcome=success output_chars=164 duration_ms=604

Error-path operator workflow

  1. Send a policy-sensitive test fixture from an approved test set. Confirm moderation returns a flagged decision, the route returns input_not_allowed, and no generation call occurs.
  2. Simulate an upstream connection failure. Confirm the route returns moderation_unavailable and still makes no generation call.
  3. Stub an empty or extra moderation result. Confirm the route reports moderation_invalid_response rather than guessing which result applies.
  4. Allow moderation, then simulate a generation failure. Confirm the route reports generation_failed; this separates a safety-gate outage from a model-generation outage.
  5. Search the resulting logs for the test fixture. Finding the raw fixture is a logging defect that should block release.

A sanitized moderation outage record can look like this:

event_name=moderation_failed request_id=req-8c21 outcome=connection_error duration_ms=15001

Sources checked

  • The CometAPI moderation endpoint reference supports the endpoint, required fields, recommended moderation model, response shape, batch behavior, and text or multimodal input claims used here.
  • The OpenAI moderation guide supports the interpretation of flags, categories, scores, and applied input types, plus the warning that scores are policy signals that may need recalibration.
  • The FastAPI request-body guide supports the Pydantic request model, JSON parsing, validation, and generated schema behavior.
  • The CometAPI Python adapter repository defines its current contract-tested surface and documents direct interoperability through the official OpenAI client.

Contract details to verify

The CometAPI contract currently names POST /v1/moderations and requires model plus input. A single string produces a results array whose corresponding item includes flagged, category booleans, category scores, and applied input types. The documented response also includes an ID, model name, and usage data. For an array of strings, the documentation says the response has one result per input, so batch code must preserve ordering and verify the count.

The documented moderation model is omni-moderation-latest for text and image moderation unless a different requirement applies. This article uses only text. Do not infer support for audio, inline generation moderation, or another request parameter from a different provider’s documentation. The separate CometAPI moderation endpoint is the supported boundary used by this implementation.

The current CometAPI Python adapter identifies chat.completions.create, responses.create, and models.list as its contract-tested 0.1 surface. Moderation is not listed there. That is why the sample uses the official OpenAI client with CometAPI runtime configuration instead of implying that the adapter guarantees moderations.create. Recheck the repository before changing that client boundary.

Also verify the generation model separately. Passing moderation does not prove that the chosen generation model is available or accepts the same parameters. Use the current catalog and a controlled smoke test rather than coupling moderation success to model availability.

Finally, document the local policy decision. The sample blocks when flagged is true and does not block on a custom score threshold. If your policy uses individual scores, maintain labeled fixtures, record the policy version, and retest thresholds when the underlying moderation behavior changes.

Failure modes

Invalid local input

An absent, empty, or oversized message should fail Pydantic validation before CometAPI is called. Keep this distinct from moderation. Local validation means the request did not meet your route contract; it says nothing about whether its content is safe.

Moderation is unreachable

A transport failure or non-success upstream status leaves the route without a moderation decision. For this protected path, the sample returns a temporary service error and skips generation. A fail-open design is possible, but it should require explicit risk acceptance rather than appearing accidentally in an exception handler. Choose timeout and retry behavior for your latency budget; the related guide explains how to add client timeouts around CometAPI text requests .

The response shape does not match the request

An empty results array, or more than one result for a single input, is a contract mismatch. Never default to allowed and never index blindly. Record the mismatch outcome, alert on it, and retain the sanitized request ID for investigation.

Input is flagged

A flagged result is not an upstream failure. Return a stable application error without disclosing category scores or reproducing the input. Do not send that input to generation. Product teams should separately define user messaging, review paths, appeal handling, and any intervention required by their policy.

Generation fails after moderation succeeds

Moderation and generation are independent upstream operations. An allowed moderation result can be followed by a generation status error, connection failure, or empty output. Emit a distinct event and error code so an operator does not misdiagnose a model outage as a moderation outage.

Score-based policy drifts

The OpenAI guide notes that category-score-based policies may need recalibration as moderation behavior changes. If you adopt thresholds, test them against labeled examples and version the policy. Avoid copying a numeric threshold from an unrelated application.

Logs become a second data store

Raw prompts, generated answers, score vectors, upstream bodies, and credential values do not belong in routine gate logs. Prefer the generated request ID, event name, outcome, upstream status when present, model name, character counts, flagged category names, duration, and policy version. Apply retention and access controls appropriate to those fields.

FAQ

Does moderation replace application policy?

No. The moderation response supplies classifications and scores. Your application still decides whether to block, allow, route to review, or take another action. The sample’s use of flagged is a clear first-pass rule, not a universal policy.

Should the route log the user message for debugging?

No. Reproduce issues with approved fixtures in a controlled environment. Routine logs should identify the stage and outcome without copying user content. An application request ID is enough to correlate moderation and generation events in this example.

Should I block on a category score?

Not by default. Scores can support a calibrated policy, but the cited guidance warns that score-based policies may need recalibration. Start with documented flags, gather representative labeled fixtures, and evaluate any threshold before deployment.

Can the endpoint check several strings at once?

Yes. CometAPI documents an array of strings and one result per input. Batch code must validate that result count and preserve positional alignment. The single-message route intentionally avoids that extra state.

Can I extend this to images?

The CometAPI reference documents OpenAI-style multimodal parts and omni-moderation-latest for supported text and image checks. An image route also needs file-size controls, media validation, fetch policy, and privacy review. Do not add an image URL field to this text-only route without those controls.

Why use the official OpenAI client instead of the CometAPI adapter?

The adapter repository’s current contract-tested 0.1 surface does not list moderation. It does document direct OpenAI-client interoperability with CometAPI configuration. Using that path keeps the sample aligned with the published support boundary instead of relying on an inherited method as an implied guarantee.

What should clients receive when moderation blocks input?

Return a stable application code and request ID, as the sample does. Avoid returning detailed category scores or echoing the original text. The exact status and user message are product-policy choices and should remain consistent across clients.

Reader next step

Implement the route in a staging service, then prove three outcomes before exposing it to users: an allowed fixture reaches generation, a flagged fixture never reaches generation, and a moderation outage fails closed. Inspect logs after every test and verify that only the sanitized fields listed above are present.

Then compare the implementation with the current CometAPI moderation contract , confirm the configured endpoint family, and add alerting for moderation_failed, moderation_invalid_response, and generation_failed. Before release, run a focused local smoke test for CometAPI tutorial code and have the blocking policy reviewed by the people responsible for product safety and incident response.