---
title: "Custom Code Guardrail"
url: "/docs/proxy/guardrails/custom_code_guardrail"
canonical_url: "https://docs.litellm.ai/docs/proxy/guardrails/custom_code_guardrail"
type: "docs"
last_updated: "2026-10-06"
summary: "Write custom guardrail logic using Python-like code that runs in a sandboxed environment."
related:
  - "/docs/proxy/guardrails/crowdstrike_aidr"
  - "/docs/proxy/guardrails/custom_guardrail"
---
# Custom Code Guardrail

> Index of all LiteLLM docs: https://docs.litellm.ai/llms.txt


Write custom guardrail logic using Python-like code that runs in a sandboxed environment.

## Quick Start

### 1. Define the guardrail in config

```yaml
model_list:
    - model_name: gpt-5.6-terra
      litellm_params:
        model: gpt-5.6-terra
        api_key: os.environ/OPENAI_API_KEY

guardrails:
    - guardrail_name: block-ssn
      litellm_params:
        guardrail: custom_code
        mode: pre_call
        custom_code: |
            def apply_guardrail(inputs, request_data, input_type):
                for text in inputs["texts"]:
                    if regex_match(text, r"\d{3}-\d{2}-\d{4}"):
                        return block("SSN detected")
                return allow()
```

### 2. Start proxy

```bash
litellm --config config.yaml
```

### 3. Test

```bash
curl -X POST http://localhost:4000/chat/completions \
  -H "Authorization: Bearer $LITELLM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-5.6-terra",
    "messages": [{"role": "user", "content": "My SSN is 123-45-6789"}],
    "guardrails": ["block-ssn"]
  }'
```

## Configuration

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `guardrail` | string | ✅ | Must be `custom_code` |
| `mode` | string | ✅ | When to run: `pre_call`, `post_call`, `during_call` |
| `custom_code` | string | ✅ | Python-like code with `apply_guardrail` function |
| `default_on` | bool | ❌ | Run on all requests (default: `false`) |
| `timeout` | float | ❌ | Wall-clock limit in seconds for one run of `apply_guardrail`, module-level code included (default: `30`). A run that exceeds it fails the request instead of stalling the proxy. See [Execution timeout](#execution-timeout) |

## Writing Custom Code

### Function Signature

Your code must define an `apply_guardrail` function. It can be either sync or async:

```python
# Sync version
def apply_guardrail(inputs, request_data, input_type):
    # inputs: see table below
    # request_data: {"model": "...", "user_id": "...", "team_id": "...", "metadata": {...}}
    # input_type: "request" or "response"
    
    return allow()  # or block(), flag() or modify()

# Async version (recommended when using HTTP primitives)
async def apply_guardrail(inputs, request_data, input_type):
    response = await http_post("https://api.example.com/check", body={"text": inputs["texts"][0]})
    if response["success"] and response["body"].get("flagged"):
        return block("Content flagged")
    return allow()
```

### `inputs` Parameter

| Field | Type | Description |
|-------|------|-------------|
| `texts` | `List[str]` | Extracted text from the request/response |
| `images` | `List[str]` | Extracted images (for image guardrails) |
| `tools` | `List[dict]` | Tools sent to the LLM |
| `tool_calls` | `List[dict]` | Tool calls returned from the LLM |
| `structured_messages` | `List[dict]` | Full messages with role info (system/user/assistant) |
| `model` | `str` | The model being used |

### `request_data` Parameter

| Field | Type | Description |
|-------|------|-------------|
| `model` | `str` | Model name |
| `user_id` | `str` | User ID from API key |
| `team_id` | `str` | Team ID from API key |
| `end_user_id` | `str` | End user ID |
| `metadata` | `dict` | Request metadata |

### Return Values

| Function | Description |
|----------|-------------|
| `allow()` | Let request/response through |
| `block(reason)` | Reject with message |
| `flag(reason, metadata={})` | Let request/response through unchanged, but record a non-blocking violation |
| `modify(texts=[], images=[], tool_calls=[])` | Transform content |

### Flagging without blocking

`flag(reason, metadata={...})` is for audit-only or monitor-mode guardrails. The request or response continues exactly as it would have with `allow()`, and the guardrail records a `guardrail_flagged` entry in the request's guardrail information with the guardrail name, the configured `mode`, the `input_type` it evaluated (`request` or `response`), the `reason`, and your structured `metadata`. It works the same way for streaming and non-streaming requests.

In the Guardrails Monitor a flagged request counts as flagged rather than passed or blocked, and in Request Logs the guardrail row shows `flagged` (when a guardrail runs on both `pre_call` and `post_call`, the row shows the most severe of the two outcomes). The request-level `guardrail_status` becomes `guardrail_flagged` unless another guardrail on the same request blocked or failed, in which case that status wins.

```yaml
guardrails:
  - guardrail_name: audit-competitor-mentions
    litellm_params:
      guardrail: custom_code
      mode: [pre_call, post_call]
      default_on: true
      custom_code: |
        def apply_guardrail(inputs, request_data, input_type):
            for text in inputs["texts"]:
                if contains_any(lower(text), ["acme", "globex"]):
                    return flag(
                        "competitor mentioned",
                        metadata={"category": "competitor", "phase": input_type},
                    )
            return allow()
```

The recorded entry looks like this in `standard_logging_guardrail_information`:

```json
{
  "guardrail_name": "audit-competitor-mentions",
  "guardrail_mode": ["pre_call", "post_call"],
  "guardrail_status": "guardrail_flagged",
  "guardrail_response": {
    "action": "flag",
    "reason": "competitor mentioned",
    "input_type": "request",
    "metadata": {"category": "competitor", "phase": "request"}
  }
}
```

## Built-in Primitives

### Regex

| Function | Description |
|----------|-------------|
| `regex_match(text, pattern)` | Returns `True` if pattern found |
| `regex_replace(text, pattern, replacement)` | Replace all matches |
| `regex_find_all(text, pattern)` | Return list of matches |

### JSON

| Function | Description |
|----------|-------------|
| `json_parse(text)` | Parse JSON string, returns `None` on error |
| `json_stringify(obj)` | Convert to JSON string |
| `json_schema_valid(obj, schema)` | Validate against JSON schema |

### URL

| Function | Description |
|----------|-------------|
| `extract_urls(text)` | Extract all URLs from text |
| `is_valid_url(url)` | Check if URL is valid |
| `all_urls_valid(text)` | Check all URLs in text are valid |

### Code Detection

| Function | Description |
|----------|-------------|
| `detect_code(text)` | Returns `True` if code detected |
| `detect_code_languages(text)` | Returns list of detected languages |
| `contains_code_language(text, ["sql", "python"])` | Check for specific languages |

### Text Utilities

| Function | Description |
|----------|-------------|
| `contains(text, substring)` | Check if substring exists |
| `contains_any(text, [substr1, substr2])` | Check if any substring exists |
| `word_count(text)` | Count words |
| `char_count(text)` | Count characters |
| `lower(text)` / `upper(text)` / `trim(text)` | String transforms |

### HTTP Requests (Async)

Make async HTTP requests to external APIs for additional validation or content moderation.

| Function | Description |
|----------|-------------|
| `await http_request(url, method, headers, body, timeout)` | General async HTTP request |
| `await http_get(url, headers, timeout)` | Async GET request |
| `await http_post(url, body, headers, timeout)` | Async POST request |

**Response format:**
```python
{
    "status_code": 200,        # HTTP status code
    "body": {...},             # Response body (parsed JSON or string)
    "headers": {...},          # Response headers
    "success": True,           # True if status code is 2xx
    "error": None              # Error message if request failed
}
```

**Note:** When using HTTP primitives, define your function as `async def apply_guardrail(...)` for non-blocking execution.

#### Blocked destinations

Every URL, redirect hops included, goes through the proxy's SSRF validation before a connection is opened. Private (RFC 1918), loopback, link-local and cloud metadata addresses are refused, and the primitive returns an error response instead of raising:

```python
{
    "status_code": 0,
    "body": None,
    "headers": {},
    "success": False,
    "error": "Blocked URL: URL targets a blocked address (169.254.169.254). If this is a legitimate internal service, add the host to `user_url_allowed_hosts` in litellm_settings."
}
```

To let a guardrail call an internal service, list its host (as written in the URL, with the port when the URL carries one) under `litellm_settings`:

```yaml
litellm_settings:
  user_url_allowed_hosts:
    - moderation.corp.internal
    - 10.0.0.12:8080
```

`user_url_validation: false` turns the check off for the whole proxy. Both settings are documented in [config settings](../config_settings#litellm_settings---reference). GET requests follow redirects, validating each hop; POST, PUT, PATCH and DELETE never follow redirects, so a 3xx comes back as the response.

### Execution timeout

Each run of `apply_guardrail` is bounded by `timeout` (default 30 seconds), and so is the module-level code that runs when the guardrail loads. A sync function runs on a worker thread, so a busy loop never stalls the proxy's event loop, and every `while` test, `for` iteration, and comprehension in guardrail code checks the budget, so a loop that never yields or awaits still stops at the deadline. An async function is checked the same way, with no `await` point needed. A run that exceeds the budget fails the request with a `Custom code guardrail '<name>' exceeded its 30s execution timeout` error, and compile-time code that exceeds it fails the guardrail's load.

```yaml
guardrails:
  - guardrail_name: "slow-moderation"
    litellm_params:
      guardrail: custom_code
      mode: "pre_call"
      timeout: 10
      custom_code: |
        async def apply_guardrail(inputs, request_data, input_type):
            result = await http_post("https://moderation.example.com/check", body={"texts": inputs["texts"]})
            if result["success"] and result["body"].get("flagged"):
                return block("Flagged by moderation service")
            return allow()
```

The `timeout` argument of `http_request` bounds a single HTTP call (default 30 seconds, at most 60) and counts against the run's budget.

## Examples

### Block PII (SSN)

```python
def apply_guardrail(inputs, request_data, input_type):
    for text in inputs["texts"]:
        if regex_match(text, r"\d{3}-\d{2}-\d{4}"):
            return block("SSN detected")
    return allow()
```

### Redact Email Addresses

```python
def apply_guardrail(inputs, request_data, input_type):
    pattern = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"
    modified = []
    for text in inputs["texts"]:
        modified.append(regex_replace(text, pattern, "[EMAIL REDACTED]"))
    return modify(texts=modified)
```

### Block SQL Injection

```python
def apply_guardrail(inputs, request_data, input_type):
    if input_type != "request":
        return allow()
    for text in inputs["texts"]:
        if contains_code_language(text, ["sql"]):
            return block("SQL code not allowed")
    return allow()
```

### Validate JSON Response

```python
def apply_guardrail(inputs, request_data, input_type):
    if input_type != "response":
        return allow()
    
    schema = {
        "type": "object",
        "required": ["name", "value"]
    }
    
    for text in inputs["texts"]:
        obj = json_parse(text)
        if obj is None:
            return block("Invalid JSON response")
        if not json_schema_valid(obj, schema):
            return block("Response missing required fields")
    return allow()
```

### Check URLs in Response

```python
def apply_guardrail(inputs, request_data, input_type):
    if input_type != "response":
        return allow()
    for text in inputs["texts"]:
        if not all_urls_valid(text):
            return block("Response contains invalid URLs")
    return allow()
```

### Call External Moderation API (Async)

```python
async def apply_guardrail(inputs, request_data, input_type):
    # Call an external moderation API
    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 - decide whether to allow or block
            return allow()
        
        if response["body"].get("flagged"):
            return block(response["body"].get("reason", "Content flagged"))
    
    return allow()
```

### Combine Multiple Checks

```python
def apply_guardrail(inputs, request_data, input_type):
    modified = []
    
    for text in inputs["texts"]:
        # Redact SSN
        text = regex_replace(text, r"\d{3}-\d{2}-\d{4}", "[SSN]")
        # Redact credit cards
        text = regex_replace(text, r"\d{16}", "[CARD]")
        modified.append(text)
    
    # Block SQL in requests
    if input_type == "request":
        for text in inputs["texts"]:
            if contains_code_language(text, ["sql"]):
                return block("SQL injection blocked")
    
    return modify(texts=modified)
```

## Sandbox Restrictions

Custom code runs in a restricted environment:

- ❌ No `import` statements
- ❌ No file I/O
- ❌ No `exec()` or `eval()`
- ✅ HTTP requests via built-in `http_request`, `http_get`, `http_post` primitives, to public addresses or hosts listed in `user_url_allowed_hosts` (see [Blocked destinations](#blocked-destinations))
- ✅ Only LiteLLM-provided primitives available
- ⏱ Each run is bounded by the guardrail's `timeout` (see [Execution timeout](#execution-timeout))

## Per-Request Usage

Enable guardrail per request:

```bash
curl -X POST http://localhost:4000/chat/completions \
  -H "Authorization: Bearer $LITELLM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-5.6-terra",
    "messages": [{"role": "user", "content": "Hello"}],
    "guardrails": ["block-ssn"]
  }'
```

## Default On

Run guardrail on all requests by setting `default_on: true` on the top-level `guardrails` entry:

```yaml
guardrails:
  - guardrail_name: block-ssn
    litellm_params:
      guardrail: custom_code
      mode: pre_call
      default_on: true
      custom_code: |
        def apply_guardrail(inputs, request_data, input_type):
            ...
```

## Related pages

- [CrowdStrike AIDR](https://docs.litellm.ai/docs/proxy/guardrails/crowdstrike_aidr.md)
- [Custom Guardrail](https://docs.litellm.ai/docs/proxy/guardrails/custom_guardrail.md)