ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Akto Guardrails to LiteLLM (#23250)

Browse Source 
* akto guardrails support in litellm

* docs(guardrails): add akto to supported values in types/guardrails.py

* frontend changes + fixes

* feat(akto): update Akto guardrail integration with new configuration options and modes

* docs(akto): enhance Akto documentation and configuration descriptions for clarity

* feat(tests): add proxy server request headers to sample request data

* refactor(akto): remove optional account and VXLAN IDs; update documentation and tests

* feat(akto): add event_type parameter for enhanced observability in guardrail logging

* refactor(akto): update environment variable references

* refactor the python codes

* refactor and fix linting

* refactor(akto): remove unused event hook and clean up imports

* refactor(akto): enhance AktoGuardrail with async support and improved logging

* fix: Register DynamoAI guardrail initializer and enum entry (#23752)

* fix: Register DynamoAI guardrail initializer and enum entry

Fix the "Unsupported guardrail: dynamoai" error by:
1. Adding DYNAMOAI to SupportedGuardrailIntegrations enum
2. Implementing initialize_guardrail() and registries in dynamoai/__init__.py

The DynamoAI guardrail was added in PR #15920 but never properly registered
in the initialization system. The __init__.py was missing the
guardrail_initializer_registry and guardrail_class_registry dictionaries
that the dynamic discovery mechanism looks for at module load time.

Fixes #22773

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

* Update litellm/proxy/guardrails/guardrail_hooks/dynamoai/__init__.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* Update litellm/proxy/guardrails/guardrail_hooks/dynamoai/__init__.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* test: Add tests for DynamoAI guardrail registration

Verifies enum entry, initializer registry, class registry,
instance creation, and global registry discovery.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Haiku 4.5 <noreply@anthropic.com>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* docs: add v1.82.3 release notes and update provider_endpoints_support.json (#23816)

* Revert "docs: add v1.82.3 release notes and update provider_endpoints_support…" (#23817)

This reverts commit 966124966f.

* Refactor Akto guardrail configuration and tests; update UI description and tags

* add account and vxlan ID parameters to Akto guardrail initialization; update Akto logo format

* enhance Akto guardrail documentation and improve error handling for non-JSON responses

* address greptile issues

* fix: update payload handling to use 'data' instead of 'json' in AktoGuardrail and adjust tests accordingly

---------

Co-authored-by: Harshit Jain <48647625+Harshit28j@users.noreply.github.com>
Co-authored-by: Claude Haiku 4.5 <noreply@anthropic.com>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
Co-authored-by: Joe Reyna <joseph.reyna@gmail.com>
Co-authored-by: Krish Dholakia <krrishdholakia@gmail.com>
This commit is contained in:
Rohan 2026-03-18 03:08:04 +05:30 committed by GitHub
parent b0db75df1f
commit bed44f5fe5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 1382 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

139
docs/my-website/docs/proxy/guardrails/akto.md Normal file
View File

@ -0,0 +1,139 @@
# Akto
## Overview
[Akto](https://www.akto.io/) provides API security guardrails and data ingestion for LLM traffic.
Akto now uses a **two-entry guardrail pattern** in LiteLLM:
- `akto-validate` (`pre_call`) for request validation
- `akto-ingest` (`post_call`) for request/response ingestion
There is no `on_flagged` setting anymore.
Use these as two separate guardrails in `config.yaml`:
- `guardrail_name: "akto-validate"`
- `guardrail_name: "akto-ingest"`
## 1. Get Your Akto Credentials
Set up the Akto Guardrail API Service and grab:
- `AKTO_GUARDRAIL_API_BASE` — your Guardrail API Base URL
- `AKTO_API_KEY` — your API key
## 2. Configure in `config.yaml`
### Block + Ingest (recommended)
Use both entries below. This gives you:
- pre-call block decision
- post-call ingestion for allowed traffic
Keep these as two separate entries (`akto-validate` and `akto-ingest`).
```yaml
guardrails:
- guardrail_name: "akto-validate"
litellm_params:
guardrail: akto
mode: pre_call
akto_base_url: os.environ/AKTO_GUARDRAIL_API_BASE
akto_api_key: os.environ/AKTO_API_KEY
default_on: true
unreachable_fallback: fail_closed # optional: fail_open | fail_closed (default: fail_closed)
guardrail_timeout: 5 # optional, default: 5
akto_account_id: "1000000" # optional, env fallback: AKTO_ACCOUNT_ID
akto_vxlan_id: "0" # optional, env fallback: AKTO_VXLAN_ID
- guardrail_name: "akto-ingest"
litellm_params:
guardrail: akto
mode: post_call
akto_base_url: os.environ/AKTO_GUARDRAIL_API_BASE
akto_api_key: os.environ/AKTO_API_KEY
default_on: true
```
### Monitor-only mode
If you only want logging/ingestion and no blocking, keep only `akto-ingest`.
```yaml
guardrails:
- guardrail_name: "akto-ingest"
litellm_params:
guardrail: akto
mode: post_call
akto_base_url: os.environ/AKTO_GUARDRAIL_API_BASE
akto_api_key: os.environ/AKTO_API_KEY
default_on: true
```
## 3. Test It
```shell
curl -i http://localhost:4000/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer <your litellm key>" \
-d '{
"model": "gpt-3.5-turbo",
"messages": [
{"role": "user", "content": "Hello, how are you?"}
]
}'
```
If a request gets blocked:
```json
{
"error": {
"message": "Prompt injection detected",
"type": "None",
"param": "None",
"code": "403"
}
}
```
## 4. How It Works
**Block + Ingest mode:**
```
Request → LiteLLM → Akto guardrail check
→ Allowed → forward to LLM → ingest response
→ Blocked → ingest blocked marker → 403 error
```
**Monitor-only mode:**
```
Request → LiteLLM → forward to LLM → get response
→ Send to Akto (guardrails + ingest) → log only
```
## 5. Event behavior
| Entry | LiteLLM hook | Akto call behavior |
|------|---|---|
| `akto-validate` | `pre_call` | Awaited call with `guardrails=true`, `ingest_data=false` |
| `akto-ingest` | `post_call` | Fire-and-forget call with `guardrails=true`, `ingest_data=true` |
When blocked in `pre_call`, LiteLLM sends one fire-and-forget ingest payload with blocked metadata and returns `403`.
## 6. Parameters
| Parameter | Env Variable | Default | Description |
|-----------|-------------|---------|-------------|
| `akto_base_url` | `AKTO_GUARDRAIL_API_BASE` | *required* | Akto Guardrail API Base URL |
| `akto_api_key` | `AKTO_API_KEY` | *required* | API key (sent as `Authorization` header) |
| `akto_account_id` | `AKTO_ACCOUNT_ID` | `1000000` | Akto account id included in payload |
| `akto_vxlan_id` | `AKTO_VXLAN_ID` | `0` | Akto vxlan id included in payload |
| `unreachable_fallback` | — | `fail_closed` | `fail_open` or `fail_closed` |
| `guardrail_timeout` | — | `5` | Timeout in seconds |
| `default_on` | — | `true` (recommended) | Enables the guardrail entry by default |
## 7. Error Handling
| Scenario | `fail_closed` (default) | `fail_open` |
|----------|------------------------|-------------|
| Akto unreachable | ❌ Blocked (503) | ✅ Passes through |
| Akto returns error | ❌ Blocked (503) | ✅ Passes through |
| Guardrail says no | ❌ Blocked (403) | ❌ Blocked (403) |

1
docs/my-website/sidebars.js
View File

@ -52,6 +52,7 @@ const sidebars = {
label: "Providers",
items: [
...[
"proxy/guardrails/akto",
"proxy/guardrails/qualifire",
"proxy/guardrails/aim_security",
"proxy/guardrails/onyx_security",

37
litellm/proxy/guardrails/guardrail_hooks/akto/__init__.py Normal file
View File

@ -0,0 +1,37 @@
from typing import TYPE_CHECKING
from litellm.types.guardrails import SupportedGuardrailIntegrations
from .akto import AktoGuardrail
if TYPE_CHECKING:
from litellm.types.guardrails import Guardrail, LitellmParams
def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"):
import litellm
_akto_callback = AktoGuardrail(
akto_base_url=getattr(litellm_params, "akto_base_url", None),
akto_api_key=getattr(litellm_params, "akto_api_key", None),
akto_account_id=getattr(litellm_params, "akto_account_id", None),
akto_vxlan_id=getattr(litellm_params, "akto_vxlan_id", None),
unreachable_fallback=getattr(litellm_params, "unreachable_fallback", "fail_closed"),
guardrail_timeout=getattr(litellm_params, "guardrail_timeout", None),
guardrail_name=guardrail.get("guardrail_name", ""),
event_hook=litellm_params.mode,
default_on=litellm_params.default_on,
)
litellm.logging_callback_manager.add_litellm_callback(_akto_callback)
return _akto_callback
guardrail_initializer_registry = {
SupportedGuardrailIntegrations.AKTO.value: initialize_guardrail,
}
guardrail_class_registry = {
SupportedGuardrailIntegrations.AKTO.value: AktoGuardrail,
}

456
litellm/proxy/guardrails/guardrail_hooks/akto/akto.py Normal file
View File

@ -0,0 +1,456 @@
"""Akto guardrail integration for LiteLLM proxy.
Uses a two-config-entry pattern:
- akto-validate (pre_call): Checks request against Akto guardrails, blocks if flagged.
- akto-ingest (post_call): Sends request+response to Akto for data ingestion.
For monitor-only mode, enable only akto-ingest without akto-validate.
"""
import asyncio
import json
import os
from datetime import datetime
from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Tuple, Type
from fastapi import HTTPException
import httpx
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
)
from litellm.llms.custom_httpx.http_handler import (
get_async_httpx_client,
httpxSpecialProvider,
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import GenericGuardrailAPIInputs
if TYPE_CHECKING:
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
HTTP_PROXY_PATH = "/api/http-proxy"
AKTO_CONNECTOR_NAME = "litellm"
DEFAULT_GUARDRAIL_TIMEOUT = 5
class AktoGuardrail(CustomGuardrail):
"""LiteLLM guardrail hook that validates and ingests LLM traffic via the Akto API."""
# Maps event_hook to the input_type it should handle; mismatches are no-ops
HOOK_TO_INPUT = {"pre_call": "request", "post_call": "response"}
@staticmethod
def get_config_model() -> Type["GuardrailConfigModel"]:
"""Return the Pydantic config model for YAML-based initialization."""
from litellm.types.proxy.guardrails.guardrail_hooks.akto import (
AktoConfigModel,
)
return AktoConfigModel
def __init__(
self,
akto_base_url: Optional[str] = None,
akto_api_key: Optional[str] = None,
akto_account_id: Optional[str] = None,
akto_vxlan_id: Optional[str] = None,
unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
guardrail_timeout: Optional[int] = None,
**kwargs: Any,
) -> None:
"""Initialize the Akto guardrail.
Args:
akto_base_url: Akto API base URL. Falls back to AKTO_GUARDRAIL_API_BASE env var.
akto_api_key: Akto API key. Falls back to AKTO_API_KEY env var.
akto_account_id: Akto account ID. Falls back to AKTO_ACCOUNT_ID env var, then "1000000".
akto_vxlan_id: Akto VXLAN ID. Falls back to AKTO_VXLAN_ID env var, then "0".
unreachable_fallback: Behavior when Akto is unreachable — block or allow.
guardrail_timeout: HTTP timeout in seconds for Akto API calls.
"""
self.async_handler = get_async_httpx_client(
llm_provider=httpxSpecialProvider.GuardrailCallback,
)
self.background_tasks: set = set()
self.akto_base_url = (akto_base_url or os.environ.get("AKTO_GUARDRAIL_API_BASE", "")).rstrip("/")
if not self.akto_base_url:
raise ValueError("akto_base_url is required. Set AKTO_GUARDRAIL_API_BASE or pass it in litellm_params.")
self.akto_api_key = akto_api_key or os.environ.get("AKTO_API_KEY", "")
if not self.akto_api_key:
raise ValueError("akto_api_key is required. Set AKTO_API_KEY or pass it in litellm_params.")
self.unreachable_fallback: Literal["fail_closed", "fail_open"] = unreachable_fallback
self.guardrail_timeout = guardrail_timeout or DEFAULT_GUARDRAIL_TIMEOUT
self.akto_account_id = akto_account_id or os.environ.get("AKTO_ACCOUNT_ID", "1000000")
self.akto_vxlan_id = akto_vxlan_id or os.environ.get("AKTO_VXLAN_ID", "0")
kwargs["supported_event_hooks"] = [
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
]
super().__init__(**kwargs)
verbose_proxy_logger.debug(
"Akto guardrail initialized: base_url=%s fallback=%s",
self.akto_base_url,
self.unreachable_fallback,
)
@staticmethod
def resolve_metadata_value(request_data: Optional[dict], key: str) -> Optional[str]:
"""Look up a metadata value from litellm_metadata or metadata dicts."""
if request_data is None:
return None
for dict_key in ("litellm_metadata", "metadata"):
container = request_data.get(dict_key) or {}
if isinstance(container, dict) and container:
value = container.get(key)
if value is not None:
return str(value).strip()
return None
@staticmethod
def extract_request_path(request_data: dict) -> str:
"""Extract the API route from request metadata, defaulting to /v1/chat/completions."""
metadata = request_data.get("metadata") or {}
if not isinstance(metadata, dict):
metadata = {}
route = metadata.get("user_api_key_request_route")
return route if route else "/v1/chat/completions"
def prepare_headers(self) -> Dict[str, str]:
"""Build HTTP headers for the Akto API call."""
return {
"content-type": "application/json",
"Authorization": self.akto_api_key,
}
@staticmethod
def build_query_params(*, guardrails: bool, ingest_data: bool) -> Dict[str, str]:
"""Build query params that control Akto backend behavior (guardrail check and/or data ingestion)."""
params: Dict[str, str] = {"akto_connector": AKTO_CONNECTOR_NAME}
if guardrails:
params["guardrails"] = "true"
if ingest_data:
params["ingest_data"] = "true"
return params
@staticmethod
def build_request_headers(request_data: dict) -> Dict[str, str]:
"""Build the requestHeaders field from proxy request headers."""
headers: Dict[str, str] = {"content-type": "application/json"}
proxy_req = request_data.get("proxy_server_request", {})
if not isinstance(proxy_req, dict):
return headers
proxy_req_headers = proxy_req.get("headers")
if isinstance(proxy_req_headers, dict):
for key, val in proxy_req_headers.items():
if key and val:
headers[str(key).lower()] = str(val)
return headers
@staticmethod
def build_request_body(
inputs: GenericGuardrailAPIInputs,
request_data: Optional[dict] = None,
) -> Dict[str, Any]:
"""Build the LLM request body from guardrail inputs (messages, model, tools)."""
model = inputs.get("model", "") or ""
body: Dict[str, Any] = {"model": model}
structured = inputs.get("structured_messages")
if structured:
body["messages"] = structured
elif request_data is not None and request_data.get("messages"):
body["messages"] = request_data["messages"]
if request_data.get("model"):
body["model"] = request_data["model"]
else:
texts = inputs.get("texts", [])
body["messages"] = [{"role": "user", "content": t} for t in texts] if texts else []
tools = inputs.get("tools")
if tools:
body["tools"] = tools
elif request_data is not None and request_data.get("tools"):
body["tools"] = request_data["tools"]
tool_calls = inputs.get("tool_calls")
if tool_calls:
body["tool_calls"] = tool_calls
return body
@staticmethod
def build_response_body(
inputs: GenericGuardrailAPIInputs,
request_data: Optional[dict] = None,
) -> Dict[str, Any]:
"""Build the LLM response body, preferring the actual model response if available."""
model_response = request_data.get("response") if request_data else None
if model_response is not None and hasattr(model_response, "model_dump"):
return model_response.model_dump()
texts = inputs.get("texts", [])
if texts:
return {"choices": [{"message": {"content": t, "role": "assistant"}} for t in texts]}
return {}
@staticmethod
def build_tag_metadata(request_data: dict) -> Dict[str, str]:
"""Build tag/metadata dict with user_id and team_id for Akto tracking."""
tag: Dict[str, str] = {"gen-ai": "Gen AI"}
user_id = AktoGuardrail.resolve_metadata_value(request_data, "user_api_key_user_id")
team_id = AktoGuardrail.resolve_metadata_value(request_data, "user_api_key_team_id")
if user_id:
tag["user_id"] = user_id
if team_id:
tag["team_id"] = team_id
return tag
def build_akto_payload(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict,
*,
status_code: int = 200,
include_response: bool = False,
) -> Dict[str, Any]:
"""Build the flat MIRRORING payload sent to Akto's HTTP proxy endpoint.
All body fields use double-encoding: json.dumps({"body": json.dumps(actual_body)})
to match the canonical CLI hook format.
"""
request_path = self.extract_request_path(request_data)
request_headers = self.build_request_headers(request_data)
request_body = self.build_request_body(inputs, request_data)
tag = self.build_tag_metadata(request_data)
response_payload = json.dumps({}) # Empty body wrapper when no response yet
response_headers: Dict[str, str] = {}
if include_response:
response_body = self.build_response_body(inputs, request_data)
response_payload = json.dumps({"body": json.dumps(response_body)}) # Double-encoded
response_headers = {"content-type": "application/json"}
# Extract client IP from proxy headers
ip = ""
proxy_req = request_data.get("proxy_server_request", {})
proxy_headers = proxy_req.get("headers", {}) if isinstance(proxy_req, dict) else {}
if isinstance(proxy_headers, dict):
ip = proxy_headers.get("x-forwarded-for") or proxy_headers.get("x-real-ip") or ""
if "," in ip:
ip = ip.split(",")[0].strip()
return {
"path": request_path,
"requestHeaders": json.dumps(request_headers),
"responseHeaders": json.dumps(response_headers),
"method": "POST",
"requestPayload": json.dumps({"body": json.dumps(request_body)}), # Double-encoded
"responsePayload": response_payload,
"ip": ip,
"destIp": "127.0.0.1",
"time": str(int(datetime.now().timestamp() * 1000)),
"statusCode": str(status_code),
"type": "HTTP/1.1",
"status": str(status_code),
"akto_account_id": self.akto_account_id,
"akto_vxlan_id": self.akto_vxlan_id,
"is_pending": "false",
"source": "MIRRORING",
"direction": None,
"process_id": None,
"socket_id": None,
"daemonset_id": None,
"enabled_graph": None,
"tag": json.dumps(tag),
"metadata": json.dumps(tag),
"contextSource": "AGENTIC",
}
async def send_request(
self,
*,
guardrails: bool,
ingest_data: bool,
payload: dict,
) -> httpx.Response:
"""Send an HTTP POST to the Akto API endpoint."""
endpoint = f"{self.akto_base_url}{HTTP_PROXY_PATH}"
params = self.build_query_params(guardrails=guardrails, ingest_data=ingest_data)
headers = self.prepare_headers()
return await self.async_handler.post(
url=endpoint,
data=json.dumps(payload),
params=params,
headers=headers,
timeout=self.guardrail_timeout,
)
@staticmethod
def handle_guardrail_response(response: httpx.Response) -> Tuple[bool, str]:
"""Parse the Akto guardrail response. Returns (allowed, reason)."""
if response.status_code != 200:
verbose_proxy_logger.error("Akto returned HTTP %d", response.status_code)
raise httpx.HTTPStatusError(
f"Akto returned unexpected status {response.status_code}",
request=response.request,
response=response,
)
try:
result = response.json()
except (json.JSONDecodeError, ValueError) as e:
response_text = getattr(response, "text", "")
verbose_proxy_logger.error(
"Akto returned non-JSON body for status 200: %r",
response_text[:200],
)
raise httpx.RequestError(
"Akto returned non-JSON body",
request=response.request,
) from e
if not isinstance(result, dict):
return True, ""
data = result.get("data") or {}
if not isinstance(data, dict):
return True, ""
guardrails_result = data.get("guardrailsResult") or {}
if not isinstance(guardrails_result, dict):
return True, ""
return (
bool(guardrails_result.get("Allowed", True)),
str(guardrails_result.get("Reason", "")),
)
def handle_unreachable(
self,
inputs: GenericGuardrailAPIInputs,
error: Exception,
) -> GenericGuardrailAPIInputs:
"""Handle Akto being unreachable based on fail_open/fail_closed config."""
if self.unreachable_fallback == "fail_open":
verbose_proxy_logger.critical(
"Akto unreachable (fail-open): %s",
str(error),
exc_info=error,
)
return inputs
verbose_proxy_logger.error("Akto unreachable (fail-closed): %s", str(error))
raise HTTPException(
status_code=503,
detail="Akto guardrail service unreachable",
)
async def fire_and_forget_request(
self,
*,
guardrails: bool,
ingest_data: bool,
payload: dict,
) -> None:
"""Send a request without awaiting it in the caller. Errors are logged, not raised."""
try:
response = await self.send_request(
guardrails=guardrails,
ingest_data=ingest_data,
payload=payload,
)
if response.status_code != 200:
verbose_proxy_logger.error(
"Akto fire-and-forget returned HTTP %d",
response.status_code,
)
except Exception as e:
verbose_proxy_logger.error("Akto fire-and-forget error: %s", str(e))
@log_guardrail_information
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict,
input_type: Literal["request", "response"],
logging_obj=None,
) -> GenericGuardrailAPIInputs:
"""Main entry point called by LiteLLM's guardrail framework.
Pre_call (input_type="request"):
- Awaits guardrail check. If blocked, fires off ingest with 403 marker and raises.
Post_call (input_type="response"):
- Fire-and-forget combined guardrail + ingest call.
"""
# Skip if this hook doesn't handle the current input_type
expected = self.HOOK_TO_INPUT.get(str(self.event_hook))
if expected and expected != input_type:
return inputs
if input_type == "request":
# Pre_call: awaited guardrail check (no ingestion)
payload = self.build_akto_payload(inputs, request_data, include_response=False)
try:
response = await self.send_request(
guardrails=True,
ingest_data=False,
payload=payload,
)
allowed, reason = self.handle_guardrail_response(response)
except HTTPException:
raise
except (httpx.RequestError, httpx.HTTPStatusError) as e:
return self.handle_unreachable(
inputs=inputs,
error=e,
)
if not allowed:
# Build a blocked marker payload with 403 status and reason
blocked_payload = self.build_akto_payload(
inputs,
request_data,
include_response=False,
status_code=403,
)
blocked_payload["responsePayload"] = json.dumps(
{
"body": json.dumps({"x-blocked-by": "Akto Proxy", "reason": reason}),
}
)
blocked_payload["responseHeaders"] = json.dumps(
{"content-type": "application/json"},
)
# Fire-and-forget ingest of the blocked request, then raise 403
task = asyncio.create_task(
self.fire_and_forget_request(
guardrails=False,
ingest_data=True,
payload=blocked_payload,
)
)
self.background_tasks.add(task)
task.add_done_callback(self.background_tasks.discard)
raise HTTPException(
status_code=403,
detail=reason or "Blocked by Akto Guardrails",
)
elif input_type == "response":
# Post_call: fire-and-forget combined guardrail + ingest
payload = self.build_akto_payload(inputs, request_data, include_response=True)
task = asyncio.create_task(
self.fire_and_forget_request(
guardrails=True,
ingest_data=True,
payload=payload,
)
)
self.background_tasks.add(task)
task.add_done_callback(self.background_tasks.discard)
return inputs

32
litellm/proxy/guardrails/guardrail_hooks/dynamoai/__init__.py
View File

@ -1,3 +1,33 @@
from typing import TYPE_CHECKING
from litellm.types.guardrails import SupportedGuardrailIntegrations
from .dynamoai import DynamoAIGuardrails
__all__ = ["DynamoAIGuardrails"]
if TYPE_CHECKING:
from litellm.types.guardrails import Guardrail, LitellmParams
def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"):
import litellm
_dynamoai_callback = DynamoAIGuardrails(
api_base=litellm_params.api_base,
api_key=litellm_params.api_key,
guardrail_name=guardrail.get("guardrail_name", ""),
event_hook=litellm_params.mode,
default_on=litellm_params.default_on,
)
litellm.logging_callback_manager.add_litellm_callback(_dynamoai_callback)
return _dynamoai_callback
guardrail_initializer_registry = {
SupportedGuardrailIntegrations.DYNAMOAI.value: initialize_guardrail,
}
guardrail_class_registry = {
SupportedGuardrailIntegrations.DYNAMOAI.value: DynamoAIGuardrails,
}

8
litellm/types/guardrails.py
View File

@ -17,6 +17,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.grayswan import (
from litellm.types.proxy.guardrails.guardrail_hooks.ibm import (
IBMGuardrailsBaseConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.akto import (
AktoConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
ContentFilterCategoryConfig,
)
@ -33,7 +36,7 @@ Pydantic object defining how to set guardrails on litellm proxy
guardrails:
- guardrail_name: "bedrock-pre-guard"
litellm_params:
guardrail: bedrock # supported values: "aporia", "bedrock", "lakera", "zscaler_ai_guard"
guardrail: bedrock # supported values: "akto", "aporia", "bedrock", "lakera", "zscaler_ai_guard"
mode: "during_call"
guardrailIdentifier: ff6ujrregl1q
guardrailVersion: "DRAFT"
@ -44,6 +47,7 @@ guardrails:
class SupportedGuardrailIntegrations(Enum):
APORIA = "aporia"
BEDROCK = "bedrock"
DYNAMOAI = "dynamoai"
GUARDRAILS_AI = "guardrails_ai"
LAKERA = "lakera"
LAKERA_V2 = "lakera_v2"
@ -78,6 +82,7 @@ class SupportedGuardrailIntegrations(Enum):
SEMANTIC_GUARD = "semantic_guard"
MCP_END_USER_PERMISSION = "mcp_end_user_permission"
BLOCK_CODE_EXECUTION = "block_code_execution"
AKTO = "akto"
class Role(Enum):
@ -735,6 +740,7 @@ class LitellmParams(
NomaGuardrailConfigModel,
ToolPermissionGuardrailConfigModel,
ZscalerAIGuardConfigModel,
AktoConfigModel,
JavelinGuardrailConfigModel,
BaseLitellmParams,
EnkryptAIGuardrailConfigs,

55
litellm/types/proxy/guardrails/guardrail_hooks/akto.py Normal file
View File

@ -0,0 +1,55 @@
from typing import Optional, Literal
from pydantic import Field
from .base import GuardrailConfigModel
class AktoConfigModel(GuardrailConfigModel):
"""
Config for the Akto guardrail.
Use two separate config entries to control behaviour:
akto-validate (mode: pre_call) -> check guardrails, block if flagged
akto-ingest (mode: post_call) -> ingest request+response data
"""
akto_base_url: Optional[str] = Field(
default=None,
description="Akto Guardrail API Base URL. Env: AKTO_GUARDRAIL_API_BASE.",
json_schema_extra={
"examples": [
"http://localhost:9090",
"https://akto-ingestion.example.com",
]
},
)
akto_api_key: Optional[str] = Field(
default=None,
description="API key for Akto. Env: AKTO_API_KEY.",
)
akto_account_id: Optional[str] = Field(
default=None,
description="Akto account ID for multi-tenant deployments. Env: AKTO_ACCOUNT_ID. Default: '1000000'.",
)
akto_vxlan_id: Optional[str] = Field(
default=None,
description="Akto VXLAN ID. Env: AKTO_VXLAN_ID. Default: '0'.",
)
unreachable_fallback: Literal["fail_closed", "fail_open"] = Field(
default="fail_closed",
description="What to do when Akto is unreachable. 'fail_open' = allow, 'fail_closed' = block.",
)
guardrail_timeout: Optional[int] = Field(
default=None,
description="HTTP timeout in seconds. Default: 5.",
)
@staticmethod
def ui_friendly_name() -> str:
return "Akto"

550
tests/guardrails_tests/test_akto_guardrails.py Normal file
View File

@ -0,0 +1,550 @@
import asyncio
import json
import os
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from starlette.exceptions import HTTPException
from litellm.types.utils import GenericGuardrailAPIInputs
from litellm.proxy.guardrails.guardrail_registry import guardrail_initializer_registry, guardrail_class_registry
from litellm.proxy.guardrails.guardrail_hooks.akto.akto import AktoGuardrail
# ---------------------------------------------------------------------------
# Registry tests
# ---------------------------------------------------------------------------
def test_akto_in_guardrail_initializer_registry():
assert "akto" in guardrail_initializer_registry
def test_akto_in_guardrail_class_registry():
assert "akto" in guardrail_class_registry
assert guardrail_class_registry["akto"] is AktoGuardrail
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def akto_validate():
"""AktoGuardrail configured for pre_call (akto-validate)."""
return AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
unreachable_fallback="fail_closed",
guardrail_name="test-akto-validate",
event_hook="pre_call",
)
@pytest.fixture
def akto_ingest():
"""AktoGuardrail configured for post_call (akto-ingest)."""
return AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
unreachable_fallback="fail_open",
guardrail_name="test-akto-ingest",
event_hook="post_call",
)
@pytest.fixture
def sample_inputs() -> GenericGuardrailAPIInputs:
return GenericGuardrailAPIInputs(
texts=["Hello, how are you?"],
model="gpt-4",
)
@pytest.fixture
def sample_request_data() -> dict:
return {
"metadata": {
"user_api_key_request_route": "/v1/chat/completions",
"user_api_key": "sk-test-123",
"user_api_key_user_id": "user-1",
"user_api_key_team_id": "team-1",
},
"proxy_server_request": {
"headers": {
"x-forwarded-for": "10.0.0.1",
}
},
}
def _mock_allowed_response():
mock = MagicMock(spec=httpx.Response)
mock.status_code = 200
mock.json.return_value = {"data": {"guardrailsResult": {"Allowed": True, "Reason": ""}}}
return mock
def _mock_blocked_response(reason="Prompt injection detected"):
mock = MagicMock(spec=httpx.Response)
mock.status_code = 200
mock.json.return_value = {"data": {"guardrailsResult": {"Allowed": False, "Reason": reason}}}
return mock
# ---------------------------------------------------------------------------
# Initialization tests
# ---------------------------------------------------------------------------
def test_init_requires_akto_base_url():
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(ValueError, match="akto_base_url is required"):
AktoGuardrail(
akto_base_url="",
akto_api_key="test-token",
guardrail_name="test",
event_hook="pre_call",
)
def test_init_requires_api_key():
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(ValueError, match="akto_api_key is required"):
AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="",
guardrail_name="test",
event_hook="pre_call",
)
def test_init_from_env():
with patch.dict(
os.environ,
{
"AKTO_GUARDRAIL_API_BASE": "http://env-host:9090",
"AKTO_API_KEY": "env-token",
"AKTO_ACCOUNT_ID": "2000000",
"AKTO_VXLAN_ID": "42",
},
):
g = AktoGuardrail(guardrail_name="env-test", event_hook="post_call")
assert g.akto_base_url == "http://env-host:9090"
assert g.akto_api_key == "env-token"
assert g.guardrail_timeout == 5
assert g.akto_account_id == "2000000"
assert g.akto_vxlan_id == "42"
def test_init_defaults():
g = AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
guardrail_name="default-test",
event_hook="pre_call",
)
assert g.unreachable_fallback == "fail_closed"
assert g.guardrail_timeout == 5
assert g.akto_account_id == "1000000"
assert g.akto_vxlan_id == "0"
def test_background_tasks_per_instance():
a = AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
guardrail_name="instance-a",
event_hook="pre_call",
)
b = AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
guardrail_name="instance-b",
event_hook="post_call",
)
assert a.background_tasks is not b.background_tasks
# ---------------------------------------------------------------------------
# Payload format tests
# ---------------------------------------------------------------------------
def test_build_akto_payload_format(akto_validate, sample_inputs, sample_request_data):
payload = akto_validate.build_akto_payload(sample_inputs, sample_request_data, include_response=False)
assert payload["path"] == "/v1/chat/completions"
assert payload["method"] == "POST"
assert payload["type"] == "HTTP/1.1"
assert payload["akto_account_id"] == "1000000"
assert payload["akto_vxlan_id"] == "0"
assert payload["is_pending"] == "false"
assert payload["source"] == "MIRRORING"
assert payload["contextSource"] == "AGENTIC"
assert payload["ip"] == "10.0.0.1"
req_headers = json.loads(payload["requestHeaders"])
assert "content-type" in req_headers
req_wrapper = json.loads(payload["requestPayload"])
req_body = json.loads(req_wrapper["body"])
assert req_body["model"] == "gpt-4"
assert req_body["messages"][0]["content"] == "Hello, how are you?"
tag = json.loads(payload["tag"])
assert tag["gen-ai"] == "Gen AI"
assert payload["responsePayload"] == json.dumps({})
assert payload["time"].isdigit()
assert len(payload["time"]) >= 13
def test_build_akto_payload_with_response(akto_validate, sample_inputs, sample_request_data):
payload = akto_validate.build_akto_payload(sample_inputs, sample_request_data, include_response=True)
resp_wrapper = json.loads(payload["responsePayload"])
resp_body = json.loads(resp_wrapper["body"])
assert "choices" in resp_body
def test_build_akto_payload_custom_account_ids(sample_inputs, sample_request_data):
g = AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
akto_account_id="9999",
akto_vxlan_id="7",
guardrail_name="custom-ids-test",
event_hook="pre_call",
)
payload = g.build_akto_payload(sample_inputs, sample_request_data, include_response=False)
assert payload["akto_account_id"] == "9999"
assert payload["akto_vxlan_id"] == "7"
def test_build_query_params():
params = AktoGuardrail.build_query_params(guardrails=True, ingest_data=False)
assert params == {"akto_connector": "litellm", "guardrails": "true"}
params = AktoGuardrail.build_query_params(guardrails=False, ingest_data=True)
assert params == {"akto_connector": "litellm", "ingest_data": "true"}
params = AktoGuardrail.build_query_params(guardrails=True, ingest_data=True)
assert params == {
"akto_connector": "litellm",
"guardrails": "true",
"ingest_data": "true",
}
# ---------------------------------------------------------------------------
# Guardrail response handling
# ---------------------------------------------------------------------------
def test_handle_guardrail_response_allowed():
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 200
mock_resp.json.return_value = {"data": {"guardrailsResult": {"Allowed": True, "Reason": ""}}}
allowed, reason = AktoGuardrail.handle_guardrail_response(mock_resp)
assert allowed is True
assert reason == ""
def test_handle_guardrail_response_blocked():
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 200
mock_resp.json.return_value = {"data": {"guardrailsResult": {"Allowed": False, "Reason": "PII detected"}}}
allowed, reason = AktoGuardrail.handle_guardrail_response(mock_resp)
assert allowed is False
assert reason == "PII detected"
def test_handle_guardrail_response_missing_result():
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 200
mock_resp.json.return_value = {}
allowed, _ = AktoGuardrail.handle_guardrail_response(mock_resp)
assert allowed is True
def test_handle_guardrail_response_data_none():
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 200
mock_resp.json.return_value = {"data": None}
allowed, reason = AktoGuardrail.handle_guardrail_response(mock_resp)
assert allowed is True
assert reason == ""
def test_handle_guardrail_response_guardrails_result_not_dict():
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 200
mock_resp.json.return_value = {"data": {"guardrailsResult": "invalid"}}
allowed, reason = AktoGuardrail.handle_guardrail_response(mock_resp)
assert allowed is True
assert reason == ""
def test_handle_guardrail_response_non_dict():
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 200
mock_resp.json.return_value = "invalid"
allowed, _ = AktoGuardrail.handle_guardrail_response(mock_resp)
assert allowed is True
def test_handle_guardrail_response_error_status():
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 500
mock_resp.request = MagicMock()
with pytest.raises(httpx.HTTPStatusError):
AktoGuardrail.handle_guardrail_response(mock_resp)
def test_handle_guardrail_response_non_json_body():
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 200
mock_resp.request = MagicMock()
mock_resp.text = "<html>not json</html>"
mock_resp.json.side_effect = json.JSONDecodeError("Expecting value", "<html>", 0)
with pytest.raises(httpx.RequestError):
AktoGuardrail.handle_guardrail_response(mock_resp)
# ---------------------------------------------------------------------------
# Pre-call (akto-validate) — allowed
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_pre_call_allowed(akto_validate, sample_inputs, sample_request_data):
akto_validate.async_handler.post = AsyncMock(return_value=_mock_allowed_response())
result = await akto_validate.apply_guardrail(
inputs=sample_inputs,
request_data=sample_request_data,
input_type="request",
)
assert result == sample_inputs
akto_validate.async_handler.post.assert_called_once()
call_params = akto_validate.async_handler.post.call_args.kwargs["params"]
assert call_params.get("guardrails") == "true"
assert "ingest_data" not in call_params
# ---------------------------------------------------------------------------
# Pre-call (akto-validate) — blocked
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_pre_call_blocked(akto_validate, sample_inputs, sample_request_data):
akto_validate.async_handler.post = AsyncMock(
side_effect=[
_mock_blocked_response("PII detected"),
_mock_allowed_response(),
]
)
with pytest.raises(HTTPException) as exc_info:
await akto_validate.apply_guardrail(
inputs=sample_inputs,
request_data=sample_request_data,
input_type="request",
)
await asyncio.sleep(0)
await asyncio.sleep(0)
assert exc_info.value.status_code == 403
assert akto_validate.async_handler.post.call_count == 2
first_call_params = akto_validate.async_handler.post.call_args_list[0].kwargs["params"]
assert first_call_params.get("guardrails") == "true"
second_call_params = akto_validate.async_handler.post.call_args_list[1].kwargs["params"]
assert second_call_params.get("ingest_data") == "true"
assert "guardrails" not in second_call_params
second_payload = json.loads(akto_validate.async_handler.post.call_args_list[1].kwargs["data"])
assert second_payload["statusCode"] == "403"
resp_body = json.loads(second_payload["responsePayload"])
inner = json.loads(resp_body["body"])
assert inner["x-blocked-by"] == "Akto Proxy"
assert inner["reason"] == "PII detected"
# ---------------------------------------------------------------------------
# Pre-call (akto-validate) — response input is no-op
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_validate_response_noop(akto_validate, sample_inputs, sample_request_data):
akto_validate.async_handler.post = AsyncMock()
result = await akto_validate.apply_guardrail(
inputs=sample_inputs,
request_data=sample_request_data,
input_type="response",
)
assert result == sample_inputs
akto_validate.async_handler.post.assert_not_called()
# ---------------------------------------------------------------------------
# Post-call (akto-ingest) — combined guardrail + ingest
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_post_call_combined(akto_ingest, sample_inputs, sample_request_data):
akto_ingest.async_handler.post = AsyncMock(return_value=_mock_allowed_response())
result = await akto_ingest.apply_guardrail(
inputs=sample_inputs,
request_data=sample_request_data,
input_type="response",
)
await asyncio.sleep(0)
await asyncio.sleep(0)
assert result == sample_inputs
akto_ingest.async_handler.post.assert_called_once()
call_params = akto_ingest.async_handler.post.call_args.kwargs["params"]
assert call_params.get("guardrails") == "true"
assert call_params.get("ingest_data") == "true"
# ---------------------------------------------------------------------------
# Post-call (akto-ingest) — request input is no-op
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ingest_request_noop(akto_ingest, sample_inputs, sample_request_data):
akto_ingest.async_handler.post = AsyncMock()
result = await akto_ingest.apply_guardrail(
inputs=sample_inputs,
request_data=sample_request_data,
input_type="request",
)
assert result == sample_inputs
akto_ingest.async_handler.post.assert_not_called()
# ---------------------------------------------------------------------------
# Fail-open / fail-closed
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_fail_open_on_unreachable():
g = AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
unreachable_fallback="fail_open",
guardrail_name="fail-open-test",
event_hook="pre_call",
)
g.async_handler.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused"))
inputs = GenericGuardrailAPIInputs(texts=["test"], model="gpt-4")
result = await g.apply_guardrail(inputs=inputs, request_data={}, input_type="request")
assert result.get("texts") == ["test"]
@pytest.mark.asyncio
async def test_fail_closed_on_unreachable():
g = AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
unreachable_fallback="fail_closed",
guardrail_name="fail-closed-test",
event_hook="pre_call",
)
g.async_handler.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused"))
inputs = GenericGuardrailAPIInputs(texts=["test"], model="gpt-4")
with pytest.raises(HTTPException) as exc_info:
await g.apply_guardrail(inputs=inputs, request_data={}, input_type="request")
assert exc_info.value.status_code == 503
def test_fail_closed_generic_message():
g = AktoGuardrail(
akto_base_url="http://localhost:9090",
akto_api_key="test-token",
unreachable_fallback="fail_closed",
guardrail_name="msg-test",
event_hook="pre_call",
)
with pytest.raises(HTTPException) as exc_info:
g.handle_unreachable(
inputs=GenericGuardrailAPIInputs(texts=["test"], model="gpt-4"),
error=Exception("http://internal-host:9090/secret-path"),
)
assert "internal-host" not in exc_info.value.detail
assert exc_info.value.detail == "Akto guardrail service unreachable"
# ---------------------------------------------------------------------------
# Helper method tests
# ---------------------------------------------------------------------------
def test_extract_request_path_from_metadata():
path = AktoGuardrail.extract_request_path({"metadata": {"user_api_key_request_route": "/v1/embeddings"}})
assert path == "/v1/embeddings"
def test_extract_request_path_fallback():
path = AktoGuardrail.extract_request_path({})
assert path == "/v1/chat/completions"
def test_extract_request_path_non_dict_metadata():
path = AktoGuardrail.extract_request_path({"metadata": "invalid"})
assert path == "/v1/chat/completions"
def test_resolve_metadata_value():
assert (
AktoGuardrail.resolve_metadata_value({"metadata": {"user_api_key_user_id": "u1"}}, "user_api_key_user_id")
== "u1"
)
assert (
AktoGuardrail.resolve_metadata_value(
{"litellm_metadata": {"user_api_key_team_id": "t1"}},
"user_api_key_team_id",
)
== "t1"
)
assert AktoGuardrail.resolve_metadata_value({}, "some_key") is None
assert AktoGuardrail.resolve_metadata_value(None, "some_key") is None
def test_resolve_metadata_value_non_dict_containers():
assert (
AktoGuardrail.resolve_metadata_value(
{"metadata": "invalid", "litellm_metadata": ["bad"]},
"some_key",
)
is None
)
def test_build_tag_metadata(akto_validate, sample_request_data):
tag = akto_validate.build_tag_metadata(sample_request_data)
assert tag["gen-ai"] == "Gen AI"
assert tag["user_id"] == "user-1"
assert tag["team_id"] == "team-1"

81
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_dynamoai.py Normal file
View File

@ -0,0 +1,81 @@
"""
Tests for DynamoAI guardrail registration and initialization.
"""
import os
from unittest.mock import patch
import pytest
class TestDynamoAIGuardrailRegistration:
"""Tests for DynamoAI guardrail registration in the guardrail system."""
def test_supported_guardrail_enum_entry(self):
"""Test that DYNAMOAI is in SupportedGuardrailIntegrations enum."""
from litellm.types.guardrails import SupportedGuardrailIntegrations
assert hasattr(SupportedGuardrailIntegrations, "DYNAMOAI")
assert SupportedGuardrailIntegrations.DYNAMOAI.value == "dynamoai"
def test_initialize_guardrail_function_exists(self):
"""Test that initialize_guardrail function is properly exported."""
from litellm.proxy.guardrails.guardrail_hooks.dynamoai import (
guardrail_initializer_registry,
initialize_guardrail,
)
assert initialize_guardrail is not None
assert "dynamoai" in guardrail_initializer_registry
def test_guardrail_class_registry_exists(self):
"""Test that guardrail_class_registry is properly exported."""
from litellm.proxy.guardrails.guardrail_hooks.dynamoai import (
guardrail_class_registry,
)
from litellm.proxy.guardrails.guardrail_hooks.dynamoai.dynamoai import (
DynamoAIGuardrails,
)
assert "dynamoai" in guardrail_class_registry
assert guardrail_class_registry["dynamoai"] == DynamoAIGuardrails
def test_initialize_guardrail_creates_instance(self):
"""Test that initialize_guardrail creates a DynamoAIGuardrails instance."""
from litellm.proxy.guardrails.guardrail_hooks.dynamoai import (
initialize_guardrail,
)
from litellm.proxy.guardrails.guardrail_hooks.dynamoai.dynamoai import (
DynamoAIGuardrails,
)
from litellm.types.guardrails import LitellmParams
litellm_params = LitellmParams(
guardrail="dynamoai",
mode="pre_call",
api_key="test-key",
api_base="https://test.dynamo.ai",
)
guardrail = {
"guardrail_name": "test-dynamoai-guard",
}
with patch(
"litellm.logging_callback_manager.add_litellm_callback"
) as mock_add:
result = initialize_guardrail(litellm_params, guardrail)
assert isinstance(result, DynamoAIGuardrails)
assert result.api_key == "test-key"
assert result.api_base == "https://test.dynamo.ai"
assert result.guardrail_name == "test-dynamoai-guard"
mock_add.assert_called_once_with(result)
def test_dynamoai_in_global_registry(self):
"""Test that dynamoai is discoverable in the global guardrail registry."""
from litellm.proxy.guardrails.guardrail_registry import (
guardrail_initializer_registry,
)
assert "dynamoai" in guardrail_initializer_registry

10
ui/litellm-dashboard/public/assets/logos/akto.svg Normal file
View File

@ -0,0 +1,10 @@
<svg width="20" height="20" viewBox="0 0 20 20" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M18.0216 0C17.8264 0 17.5923 0.078125 17.3581 0.15625L1.12266 7.65625C-0.750664 8.51562 -0.126225 11.2891 1.86418 11.2891H8.73302V18.1641C8.73302 19.3359 9.7087 20 10.6454 20C11.3088 20 12.0113 19.6875 12.3626 18.9062L19.8559 2.65625C20.4022 1.36719 19.3095 0 18.0216 0ZM18.7241 2.14844L14.6743 10.625L11.2308 18.3594C11.1137 18.6328 10.9186 18.7891 10.6454 18.7891C10.4112 18.7891 9.9819 18.6328 9.9819 18.1641V10H1.87332C1.5611 10 1.24888 9.6875 1.24888 9.375C1.24888 9.0625 1.5611 8.75 1.87332 8.75L14.6743 10.625L17.8264 1.32812C17.9045 1.28906 17.9825 1.25 18.0216 1.25C18.2557 1.25 18.4899 1.40625 18.646 1.64062C18.7241 1.75781 18.8021 1.95312 18.7241 2.14844Z" fill="url(#paint0_linear_22912_90694)"/>
<defs>
<linearGradient id="paint0_linear_22912_90694" x1="20" y1="9.56158e-07" x2="3.95833" y2="16.25" gradientUnits="userSpaceOnUse">
<stop stop-color="#D500F9"/>
<stop offset="0.5" stop-color="#6200EA"/>
<stop offset="1" stop-color="#2E006D"/>
</linearGradient>
</defs>
</svg>
Side by Side

After

Width:  |  Height:  |  Size: 1.1 KiB

6
ui/litellm-dashboard/src/components/guardrails/guardrail_garden_configs.ts
View File

@ -264,4 +264,10 @@ export const GUARDRAIL_PRESETS: Record<string, GuardrailPreset> = {
mode: "pre_call",
defaultOn: false,
},
akto: {
provider: "Akto",
guardrailNameSuggestion: "Akto Guardrail",
mode: "pre_call",
defaultOn: false,
},
};

8
ui/litellm-dashboard/src/components/guardrails/guardrail_garden_data.ts
View File

@ -373,6 +373,14 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [
logo: `${ASSET_PREFIX}pillar.jpeg`,
tags: ["Monitoring", "Safety"],
},
{
id: "akto",
name: "Akto Guardrail",
description: "AI security platform from Akto.io with automatic monitoring and guardrails for AI/ML applications.",
category: "partner",
logo: `${ASSET_PREFIX}akto.svg`,
tags: ["Security", "Safety", "Monitoring"],
},
];
export const ALL_CARDS = [...LITELLM_CONTENT_FILTER_CARDS, ...PARTNER_GUARDRAIL_CARDS];

1
ui/litellm-dashboard/src/components/guardrails/guardrail_info_helpers.tsx
View File

@ -125,6 +125,7 @@ export const guardrailLogoMap: Record<string, string> = {
EnkryptAI: `${asset_logos_folder}enkrypt_ai.avif`,
"Prompt Security": `${asset_logos_folder}prompt_security.png`,
"LiteLLM Content Filter": `${asset_logos_folder}litellm_logo.jpg`,
"Akto": `${asset_logos_folder}akto.svg`,
};
export const getGuardrailLogoAndName = (guardrailValue: string): { logo: string; displayName: string } => {