fix(custom_code_guardrail.py): add http support for custom code guardrails

allows users to call external guardrails on litellm with minimal code changes (no custom handlers)

Test guardrail integrations more easily
This commit is contained in:
Krrish Dholakia 2026-02-05 10:11:05 -08:00 • committed by Shin
parent a60122265b
commit 0ef51255dd
2 changed files with 29 additions and 10 deletions

View file

@ -29,16 +29,14 @@ Example custom code (async with HTTP):
"""
import asyncio
import inspect
import threading
from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Type, cast
from fastapi import HTTPException
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
)
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
from litellm.types.utils import GenericGuardrailAPIInputs
@ -118,9 +116,6 @@ class CustomCodeGuardrail(CustomGuardrail):
GuardrailEventHooks.pre_call,
GuardrailEventHooks.during_call,
GuardrailEventHooks.post_call,
GuardrailEventHooks.pre_mcp_call,
GuardrailEventHooks.during_mcp_call,
GuardrailEventHooks.logging_only,
]
super().__init__(
@ -182,7 +177,6 @@ class CustomCodeGuardrail(CustomGuardrail):
self._compile_error = f"Failed to compile custom code: {e}"
raise CustomCodeCompilationError(self._compile_error) from e
@log_guardrail_information
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
@ -216,7 +210,6 @@ class CustomCodeGuardrail(CustomGuardrail):
HTTPException: If content is blocked
CustomCodeExecutionError: If execution fails
"""
if self._compiled_function is None:
if self._compile_error:
raise CustomCodeExecutionError(

View file

@ -19,7 +19,7 @@ const { TextArea } = Input;
const CODE_TEMPLATES = {
empty: {
name: "Empty Template",
code: `def apply_guardrail(inputs, request_data, input_type):
code: `async def apply_guardrail(inputs, request_data, input_type):
# inputs: {texts, images, tools, tool_calls, structured_messages, model}
# request_data: {model, user_id, team_id, end_user_id, metadata}
# input_type: "request" or "response"
@ -68,6 +68,27 @@ const CODE_TEMPLATES = {
return block("Response missing required fields")
return allow()`,
},
externalAPI: {
name: "External API Check (async)",
code: `async def apply_guardrail(inputs, request_data, input_type):
# Call an external moderation API (async for non-blocking)
for text in inputs["texts"]:
response = await http_post(
"https://api.example.com/moderate",
body={"text": text, "user_id": request_data["user_id"]},
headers={"Authorization": "Bearer YOUR_API_KEY"},
timeout=10
)
if not response["success"]:
# API call failed, allow by default or block
return allow()
if response["body"].get("flagged"):
return block(response["body"].get("reason", "Content flagged"))
return allow()`,
},
};
// Available primitives organized by category
@ -77,6 +98,11 @@ const PRIMITIVES = {
{ name: "block(reason)", desc: "Reject with message" },
{ name: "modify(texts=[], images=[], tool_calls=[])", desc: "Transform content" },
],
"HTTP Requests (async)": [
{ name: "await http_request(url, method, headers, body)", desc: "Make async HTTP request" },
{ name: "await http_get(url, headers)", desc: "Async GET request" },
{ name: "await http_post(url, body, headers)", desc: "Async POST request" },
],
"Regex Functions": [
{ name: "regex_match(text, pattern)", desc: "Returns True if pattern found" },
{ name: "regex_replace(text, pattern, replacement)", desc: "Replace all matches" },