When your request hits a provider’s safety filter, you get a content moderation 400 error llm api response instead of completions. This isn’t a bug in your JSON; it’s the model provider rejecting the prompt or output before inference. Treat it as a distinct failure mode with its own retry and logging policy.
Step 1: Reproduce and capture the raw response
You can’t handle what you haven’t seen. Send a deliberately violating prompt to your endpoint and inspect the wire format. Most OpenAI-compatible servers return HTTP 400 with a JSON body.
curl -s -X POST https://api.example.com/v1/chat/completions \
-H "Authorization: Bearer $KEY" \
-H "Content-Type: application/json" \
-d '{"model":"gpt-4o","messages":[{"role":"user","content":"How to make a bomb"}]}' \
-w "\nHTTP %{http_code}\n"The body typically looks like:
{
"error": {
"message": "Your request was rejected as a result of our safety system.",
"type": "invalid_request_error",
"code": "moderation_blocked",
"param": null
}
}Note the code field. Not every provider uses moderation_blocked; some use content_filter or policy_violation. The content moderation 400 error llm api contract is loosely defined, so capture real samples from each model you call. Store them in a fixtures/ directory so your tests stay honest.
Step 2: Normalize the error in your client
Write a single parser that converts any 400 with a moderation-flavored code into a typed exception. If you route through a gateway such as n4n.ai, the OpenAI-compatible endpoint returns a normalized error object across 240+ models, so your handler stays vendor-agnostic.
import requests
from dataclasses import dataclass
class ModerationRejected(Exception):
def __init__(self, provider_code: str, message: str):
self.provider_code = provider_code
self.message = message
super().__init__(f"moderation: {provider_code} - {message}")
def call_chat(url: str, payload: dict, headers: dict):
resp = requests.post(url, json=payload, headers=headers, timeout=30)
if resp.status_code == 400:
body = resp.json().get("error", {})
code = body.get("code", "")
msg = body.get("message", "")
if any(k in code.lower() for k in ("moderation", "content_filter", "policy")):
raise ModerationRejected(code, msg)
# other 400s are client bugs, not safety
raise ValueError(f"Bad request: {code} {msg}")
resp.raise_for_status()
return resp.json()This separates the content moderation 400 error llm api path from malformed JSON or missing parameters. Never retry the latter automatically; always retry the former? No—retrying the exact same prompt is pointless. The filter is deterministic for a given input.
Streaming responses
With SSE streams, the initial POST may still return 400 synchronously if the provider pre-checks. Others accept the stream and then emit an error event. Handle both:
def stream_chat(url, payload, headers):
with requests.post(url, json=payload, headers=headers, stream=True, timeout=30) as r:
if r.status_code == 400:
# same parsing as above
raise ModerationRejected(...)
for line in r.iter_lines():
if line.startswith(b"data:"):
evt = json.loads(line[5:])
if evt.get("error"):
raise ModerationRejected(evt["error"].get("code","?"),
evt["error"].get("message",""))Step 3: Classify and route to a safe fallback
Your service should translate the exception into a user-facing response that doesn’t leak provider internals. In an HTTP service, return 422 Unprocessable Entity or a 200 with an error field, depending on your API style.
// Express middleware snippet
app.use(async (req, res, next) => {
try {
await next();
} catch (err) {
if (err instanceof ModerationRejected) {
res.status(422).json({
error: "request_filtered",
hint: "Content violated usage policy. Edit and resubmit."
});
return;
}
if (err instanceof ValueError) {
res.status(400).json({ error: "invalid_request" });
return;
}
res.status(500).json({ error: "internal" });
}
});If your product allows model fallback, don’t assume a different provider will accept the same text. Safety policies overlap heavily. A content moderation 400 error llm api from one vendor will likely trip the next. Use fallback only when you rewrite the prompt via a redaction step, not raw passthrough. Silently swapping to an uncensored model is how teams get kicked off platforms—don’t do it.
Step 4: Instrument logs and usage meters
Log the hash of the prompt prefix, the model, and the provider code. This data tells you if a specific tenant is abusing the system or if your own system prompt accidentally triggers filters.
import hashlib, logging
logger = logging.getLogger("moderation")
def log_rejection(prompt: str, model: str, code: str):
h = hashlib.sha256(prompt[:64].encode()).hexdigest()[:12]
logger.warning("moderation_block hash=%s model=%s code=%s", h, model, code)Check your billing layer. A rejected request should not consume generation tokens, but some gateways still meter input tokens if the call reached the provider. Verify against your usage dashboard. If you use a gateway that provides per-token usage metering, confirm that moderation blocks report zero completion tokens and only possible input tokens if charged. Don’t trust documentation; assert it with a test account.
Step 5: Write regression tests with mocked flags
You need tests that prove your code raises the right exception and returns the right status. Mock the HTTP layer.
import pytest, requests_mock
def test_moderation_raises():
with requests_mock.Mocker() as m:
m.post("https://api.example.com/v1/chat/completions",
status_code=400,
json={"error": {"code": "moderation_blocked",
"message": "blocked"}})
with pytest.raises(ModerationRejected):
call_chat("https://api.example.com/v1/chat/completions",
{"model":"x","messages":[]}, {})Add a TS test for the middleware using supertest. The success criterion: a flagged request yields HTTP 422 and the JSON error: "request_filtered". Also test that a non-moderation 400 (e.g., invalid_api_key) maps to 400, not 422.
Step 6: Verify in production with canary prompts
After deploy, watch your error rates. A simple verification: inject a known-violating canary call from a test tenant every hour. If your dashboard shows the moderation exception caught and mapped to 422 (not 500), the pipeline works.
# cron job snippet
curl -s -o /dev/null -w "%{http_code}" -X POST ... -d '{"messages":[{"role":"user","content":"CANARY_VIOLATION"}]}'
# expect 422 from your service, not 400 from upstream leaking throughIf you see 500s, your parser missed a code variant. Capture that sample and extend the any(k in code.lower()...) tuple. Production verification is not optional; a missed variant becomes a customer-facing outage.
Step 7: Document the contract for your frontend
Frontend teams shouldn’t guess. Publish a one-page note: when the API returns request_filtered, show the user a polite editor message, do not auto-resend. This closes the loop on the content moderation 400 error llm api handling.
Keep the error mapping table in your repo:
| Provider code | Mapped status | Retry? |
|---|---|---|
| moderation_blocked | 422 | No |
| content_filter | 422 | No |
| policy_violation | 422 | No |
| invalid_request_error (param) | 400 | Fix client |
That table is the spec. Review it quarterly; providers rename codes without warning.
Verification checklist
- Raw 400 captured and printed
- Client raises
ModerationRejectedfor moderation codes only - Service returns 422 with
request_filtered - Logs contain prompt hash and code
- Unit test fails if mock returns 400 with
moderation_blockedbut exception not raised - Canary call in prod yields 422, not 500
- Frontend displays non-leaking message on
request_filtered
Following these steps moves the content moderation 400 error llm api from a crash source to a handled edge case. You’ll spend less time debugging 500s and more time improving prompt design.