Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_endpoints.py: 55%
862 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2CRUD ENDPOINTS FOR GUARDRAILS
3"""
5import concurrent.futures
6import inspect
7import json
8import os
9from collections.abc import Awaitable, Callable, Mapping, Sequence
10from datetime import datetime, timezone
11from types import MappingProxyType, UnionType
12from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, TypeVar, Union, cast, get_args, get_origin
13from urllib.parse import urlparse
15from fastapi import APIRouter, Depends, HTTPException, Request
16from pydantic import BaseModel, ValidationError
18from litellm._logging import verbose_proxy_logger
19from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
20from litellm.integrations.custom_guardrail import CustomGuardrail
21from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
22from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
23from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
24from litellm.proxy.common_utils.path_utils import safe_join
25from litellm.proxy.guardrails.guardrail_hooks.custom_code.sandbox import (
26 build_sandbox_globals,
27 compile_sandboxed,
28)
29from litellm.proxy.guardrails.guardrail_registry import GuardrailRegistry
30from litellm.proxy.guardrails.usage_endpoints import router as guardrails_usage_router
31from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
32from litellm.repositories.prisma_protocols import TableActions
33from litellm.repositories.table_repositories import GuardrailsRepository
34from litellm.types.guardrails import (
35 PII_ENTITY_CATEGORIES_MAP,
36 ApplyGuardrailRequest,
37 ApplyGuardrailResponse,
38 BaseLitellmParams,
39 BedrockGuardrailConfigModel,
40 BedrockGuardrailStreamingParams,
41 Guardrail,
42 GuardrailEventHooks,
43 GuardrailInfoResponse,
44 GuardrailUIAddGuardrailSettings,
45 LakeraV2GuardrailConfigModel,
46 ListGuardrailsResponse,
47 LitellmParams,
48 PatchGuardrailRequest,
49 PiiAction,
50 PiiEntityType,
51 PresidioPresidioConfigModelUserInterface,
52 SupportedGuardrailIntegrations,
53 ToolPermissionGuardrailConfigModel,
54)
55from litellm.types.proxy.guardrails.guardrail_hooks.hide_secrets import (
56 HideSecretsGuardrailConfigModel,
57)
59if TYPE_CHECKING: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true
60 from types import CodeType
62 from prisma.models import LiteLLM_GuardrailsTable
63 from pydantic.fields import FieldInfo
65 from litellm.proxy.utils import PrismaClient
67#### GUARDRAILS ENDPOINTS ####
69router: Final = APIRouter()
70GUARDRAIL_REGISTRY: Final = GuardrailRegistry()
73def _as_str_object_mapping(mapping: Mapping[str, object]) -> Mapping[str, object]:
74 return mapping
77def _guardrails_table(prisma_client: "PrismaClient") -> "TableActions[LiteLLM_GuardrailsTable]":
78 return GuardrailsRepository(prisma_client).table
81async def _create_guardrail_row(prisma_client: "PrismaClient", data: Mapping[str, object]) -> "LiteLLM_GuardrailsTable":
82 row: Final = await _guardrails_table(prisma_client).create(data=data)
83 return row
86async def _delete_guardrail_row(prisma_client: "PrismaClient", where: Mapping[str, object]) -> None:
87 await _guardrails_table(prisma_client).delete(where=where)
90async def _find_team_guardrail_rows(
91 prisma_client: "PrismaClient", where: Mapping[str, object]
92) -> "Sequence[LiteLLM_GuardrailsTable]":
93 rows: Final = await _guardrails_table(prisma_client).find_many(
94 where=where,
95 order={"created_at": "desc"},
96 )
97 return rows
100def _get_guardrails_list_response(
101 guardrails_config: list[dict],
102) -> ListGuardrailsResponse:
103 """
104 Helper function to get the guardrails list response
105 """
106 from litellm.litellm_core_utils.litellm_logging import _get_masked_values
108 guardrail_configs: Final[list[GuardrailInfoResponse]] = []
109 for guardrail in guardrails_config: 109 ↛ 110line 109 didn't jump to line 110 because the loop on line 109 never started
110 litellm_params = guardrail.get("litellm_params") or {}
111 masked_params = _get_masked_values(
112 litellm_params,
113 unmasked_length=4,
114 number_of_asterisks=4,
115 )
116 guardrail_configs.append(
117 GuardrailInfoResponse(
118 guardrail_id=guardrail.get("guardrail_id"),
119 guardrail_name=guardrail.get("guardrail_name"),
120 litellm_params=masked_params,
121 guardrail_info=guardrail.get("guardrail_info"),
122 )
123 )
124 return ListGuardrailsResponse(guardrails=guardrail_configs)
127@router.get(
128 "/guardrails/list",
129 tags=["Guardrails"],
130 dependencies=[Depends(user_api_key_auth)],
131 response_model=ListGuardrailsResponse,
132)
133async def list_guardrails():
134 """
135 List the guardrails that are available on the proxy server
137 👉 [Guardrail docs](https://docs.litellm.ai/docs/proxy/guardrails/quick_start)
139 Example Request:
140 ```bash
141 curl -X GET "http://localhost:4000/guardrails/list" -H "Authorization: Bearer <your_api_key>"
142 ```
144 Example Response:
145 ```json
146 {
147 "guardrails": [
148 {
149 "guardrail_name": "bedrock-pre-guard",
150 "guardrail_info": {
151 "params": [
152 {
153 "name": "toxicity_score",
154 "type": "float",
155 "description": "Score between 0-1 indicating content toxicity level"
156 },
157 {
158 "name": "pii_detection",
159 "type": "boolean"
160 }
161 ]
162 }
163 }
164 ]
165 }
166 ```
167 """
168 from litellm.proxy.proxy_server import proxy_config
170 config: Final = proxy_config.config
172 _guardrails_config: Final = cast(list[dict] | None, config.get("guardrails"))
174 if _guardrails_config is None: 174 ↛ 177line 174 didn't jump to line 177 because the condition on line 174 was always true
175 return _get_guardrails_list_response([])
177 return _get_guardrails_list_response(_guardrails_config)
180@router.get(
181 "/v2/guardrails/list",
182 tags=["Guardrails"],
183 response_model=ListGuardrailsResponse,
184)
185async def list_guardrails_v2(
186 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
187):
188 """
189 List the guardrails that are available in the database using GuardrailRegistry
191 👉 [Guardrail docs](https://docs.litellm.ai/docs/proxy/guardrails/quick_start)
193 Example Request:
194 ```bash
195 curl -X GET "http://localhost:4000/v2/guardrails/list" -H "Authorization: Bearer <your_api_key>"
196 ```
198 Example Response:
199 ```json
200 {
201 "guardrails": [
202 {
203 "guardrail_id": "123e4567-e89b-12d3-a456-426614174000",
204 "guardrail_name": "my-bedrock-guard",
205 "litellm_params": {
206 "guardrail": "bedrock",
207 "mode": "pre_call",
208 "guardrailIdentifier": "ff6ujrregl1q",
209 "guardrailVersion": "DRAFT",
210 "default_on": true
211 },
212 "guardrail_info": {
213 "description": "Bedrock content moderation guardrail"
214 }
215 }
216 ]
217 }
218 ```
219 """
220 from litellm.litellm_core_utils.litellm_logging import _get_masked_values
221 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER
222 from litellm.proxy.proxy_server import prisma_client
224 is_admin: Final = _user_has_admin_view(user_api_key_dict)
226 try:
227 guardrails = (
228 await GUARDRAIL_REGISTRY.get_all_guardrails_from_db(prisma_client=prisma_client)
229 if prisma_client is not None
230 else []
231 )
233 excluded_guardrail_ids: Final[set] = set()
234 if not is_admin: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 caller_team_ids: Final = await _get_user_team_ids(user_api_key_dict)
236 allowed: Final[list[Guardrail]] = []
237 for g in guardrails:
238 g_team_id = g.get("team_id")
239 if g_team_id is None or g_team_id in caller_team_ids:
240 allowed.append(g)
241 else:
242 gid = g.get("guardrail_id")
243 if gid:
244 excluded_guardrail_ids.add(gid)
245 guardrails = allowed
247 guardrail_configs: Final[list[GuardrailInfoResponse]] = []
248 seen_guardrail_ids: Final[set] = excluded_guardrail_ids.copy()
249 for guardrail in guardrails: 249 ↛ 250line 249 didn't jump to line 250 because the loop on line 249 never started
250 litellm_params: LitellmParams | dict | None = guardrail.get("litellm_params")
251 litellm_params_dict = (
252 litellm_params.model_dump(exclude_none=True)
253 if isinstance(litellm_params, LitellmParams)
254 else litellm_params
255 ) or {}
256 masked_litellm_params_dict = _get_masked_values(
257 litellm_params_dict,
258 unmasked_length=4,
259 number_of_asterisks=4,
260 )
261 masked_litellm_params = (
262 BaseLitellmParams(**masked_litellm_params_dict) if masked_litellm_params_dict else None
263 )
264 guardrail_configs.append(
265 GuardrailInfoResponse(
266 guardrail_id=guardrail.get("guardrail_id"),
267 guardrail_name=guardrail.get("guardrail_name"),
268 litellm_params=masked_litellm_params,
269 guardrail_info=guardrail.get("guardrail_info"),
270 created_at=guardrail.get("created_at"),
271 updated_at=guardrail.get("updated_at"),
272 guardrail_definition_location="db",
273 )
274 )
275 seen_guardrail_ids.add(guardrail.get("guardrail_id"))
277 # get guardrails initialized on litellm config.yaml
278 in_memory_guardrails: Final = IN_MEMORY_GUARDRAIL_HANDLER.list_in_memory_guardrails()
279 for guardrail in in_memory_guardrails: 279 ↛ 280line 279 didn't jump to line 280 because the loop on line 279 never started
280 gid = guardrail.get("guardrail_id")
281 if gid in seen_guardrail_ids:
282 continue
283 # Skip stale DB-backed entries — the DB row was deleted (likely by
284 # another pod) and reconciliation hasn't fired yet on this pod.
285 if gid is not None and IN_MEMORY_GUARDRAIL_HANDLER.get_source(gid) == "db":
286 continue
287 if not is_admin:
288 g_team_id = guardrail.get("team_id")
289 if g_team_id is not None and g_team_id not in caller_team_ids:
290 continue
291 in_memory_litellm_params_raw = guardrail.get("litellm_params")
292 in_memory_litellm_params_dict = (
293 in_memory_litellm_params_raw.model_dump(exclude_none=True)
294 if isinstance(in_memory_litellm_params_raw, LitellmParams)
295 else in_memory_litellm_params_raw
296 ) or {}
297 masked_in_memory_litellm_params = _get_masked_values(
298 in_memory_litellm_params_dict,
299 unmasked_length=4,
300 number_of_asterisks=4,
301 )
302 masked_in_memory_litellm_params_typed = (
303 BaseLitellmParams(**masked_in_memory_litellm_params) if masked_in_memory_litellm_params else None
304 )
305 guardrail_configs.append(
306 GuardrailInfoResponse(
307 guardrail_id=guardrail.get("guardrail_id"),
308 guardrail_name=guardrail.get("guardrail_name"),
309 litellm_params=masked_in_memory_litellm_params_typed,
310 guardrail_info=dict(guardrail.get("guardrail_info") or {}),
311 guardrail_definition_location="config",
312 )
313 )
314 seen_guardrail_ids.add(gid)
316 return ListGuardrailsResponse(guardrails=guardrail_configs)
317 except Exception as e:
318 verbose_proxy_logger.exception("Error getting guardrails from db: %s", e)
319 raise HTTPException(status_code=500, detail=str(e))
322class CreateGuardrailRequest(BaseModel):
323 guardrail: Guardrail
326@router.post(
327 "/guardrails",
328 tags=["Guardrails"],
329)
330async def create_guardrail(
331 request: CreateGuardrailRequest,
332 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
333):
334 """
335 Create a new guardrail
337 👉 [Guardrail docs](https://docs.litellm.ai/docs/proxy/guardrails/quick_start)
339 Example Request:
340 ```bash
341 curl -X POST "http://localhost:4000/guardrails" \\
342 -H "Authorization: Bearer <your_api_key>" \\
343 -H "Content-Type: application/json" \\
344 -d '{
345 "guardrail": {
346 "guardrail_name": "my-bedrock-guard",
347 "litellm_params": {
348 "guardrail": "bedrock",
349 "mode": "pre_call",
350 "guardrailIdentifier": "ff6ujrregl1q",
351 "guardrailVersion": "DRAFT",
352 "default_on": true
353 },
354 "guardrail_info": {
355 "description": "Bedrock content moderation guardrail"
356 }
357 }
358 }'
359 ```
361 Example Response:
362 ```json
363 {
364 "guardrail_id": "123e4567-e89b-12d3-a456-426614174000",
365 "guardrail_name": "my-bedrock-guard",
366 "litellm_params": {
367 "guardrail": "bedrock",
368 "mode": "pre_call",
369 "guardrailIdentifier": "ff6ujrregl1q",
370 "guardrailVersion": "DRAFT",
371 "default_on": true
372 },
373 "guardrail_info": {
374 "description": "Bedrock content moderation guardrail"
375 },
376 "created_at": "2023-11-09T12:34:56.789Z",
377 "updated_at": "2023-11-09T12:34:56.789Z"
378 }
379 ```
380 """
381 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER
382 from litellm.proxy.proxy_server import prisma_client
384 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 384 ↛ 385line 384 didn't jump to line 385 because the condition on line 384 was never true
385 raise HTTPException(
386 status_code=403,
387 detail="Admin access required to manage guardrails",
388 )
390 if prisma_client is None: 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true
391 raise HTTPException(status_code=500, detail="Prisma client not initialized")
393 try:
394 result = await GUARDRAIL_REGISTRY.add_guardrail_to_db(guardrail=request.guardrail, prisma_client=prisma_client)
396 guardrail_name: Final = result.get("guardrail_name", "Unknown")
397 guardrail_id: Final = result.get("guardrail_id", "Unknown")
399 try:
400 IN_MEMORY_GUARDRAIL_HANDLER.initialize_guardrail(guardrail=cast(Guardrail, result), source="db")
401 verbose_proxy_logger.info(
402 "Immediate sync: Successfully initialized guardrail '%s' (ID: %s)", guardrail_name, guardrail_id
403 )
404 except (ValueError, TypeError) as init_error:
405 # Configuration error — roll back the DB write so the guardrail isn't orphaned
406 if prisma_client is not None: 406 ↛ 411line 406 didn't jump to line 411 because the condition on line 406 was always true
407 try:
408 await _delete_guardrail_row(prisma_client, where={"guardrail_id": guardrail_id})
409 except Exception as rollback_err:
410 verbose_proxy_logger.warning("Rollback failed for guardrail '%s': %s", guardrail_id, rollback_err)
411 raise HTTPException(
412 status_code=400,
413 detail=f"Guardrail configuration error: {init_error}",
414 )
415 except Exception as init_error:
416 verbose_proxy_logger.warning(
417 "Immediate sync: Failed to initialize guardrail '%s' (ID: %s) in memory: %s",
418 guardrail_name,
419 guardrail_id,
420 init_error,
421 )
423 return result
424 except Exception as e:
425 verbose_proxy_logger.exception("Error adding guardrail to db: %s", e)
426 raise HTTPException(status_code=500, detail=str(e))
429class UpdateGuardrailRequest(BaseModel):
430 guardrail: Guardrail
433@router.put(
434 "/guardrails/{guardrail_id}",
435 tags=["Guardrails"],
436)
437async def update_guardrail(
438 guardrail_id: str,
439 request: UpdateGuardrailRequest,
440 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
441):
442 """
443 Update an existing guardrail
445 👉 [Guardrail docs](https://docs.litellm.ai/docs/proxy/guardrails/quick_start)
447 Example Request:
448 ```bash
449 curl -X PUT "http://localhost:4000/guardrails/123e4567-e89b-12d3-a456-426614174000" \\
450 -H "Authorization: Bearer <your_api_key>" \\
451 -H "Content-Type: application/json" \\
452 -d '{
453 "guardrail": {
454 "guardrail_name": "updated-bedrock-guard",
455 "litellm_params": {
456 "guardrail": "bedrock",
457 "mode": "pre_call",
458 "guardrailIdentifier": "ff6ujrregl1q",
459 "guardrailVersion": "1.0",
460 "default_on": true
461 },
462 "guardrail_info": {
463 "description": "Updated Bedrock content moderation guardrail"
464 }
465 }
466 }'
467 ```
469 Example Response:
470 ```json
471 {
472 "guardrail_id": "123e4567-e89b-12d3-a456-426614174000",
473 "guardrail_name": "updated-bedrock-guard",
474 "litellm_params": {
475 "guardrail": "bedrock",
476 "mode": "pre_call",
477 "guardrailIdentifier": "ff6ujrregl1q",
478 "guardrailVersion": "1.0",
479 "default_on": true
480 },
481 "guardrail_info": {
482 "description": "Updated Bedrock content moderation guardrail"
483 },
484 "created_at": "2023-11-09T12:34:56.789Z",
485 "updated_at": "2023-11-09T13:45:12.345Z"
486 }
487 ```
488 """
489 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER
490 from litellm.proxy.proxy_server import prisma_client
492 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 492 ↛ 493line 492 didn't jump to line 493 because the condition on line 492 was never true
493 raise HTTPException(
494 status_code=403,
495 detail="Admin access required to manage guardrails",
496 )
498 if prisma_client is None: 498 ↛ 499line 498 didn't jump to line 499 because the condition on line 498 was never true
499 raise HTTPException(status_code=500, detail="Prisma client not initialized")
501 try:
502 # Check if guardrail exists
503 existing_guardrail: Final = await GUARDRAIL_REGISTRY.get_guardrail_by_id_from_db(
504 guardrail_id=guardrail_id, prisma_client=prisma_client
505 )
507 if existing_guardrail is None: 507 ↛ 510line 507 didn't jump to line 510 because the condition on line 507 was always true
508 raise HTTPException(status_code=404, detail=f"Guardrail with ID {guardrail_id} not found")
510 result: Final = _as_str_object_mapping(
511 await GUARDRAIL_REGISTRY.update_guardrail_in_db(
512 guardrail_id=guardrail_id,
513 guardrail=request.guardrail,
514 prisma_client=prisma_client,
515 )
516 )
518 guardrail_name: Final = result.get("guardrail_name", "Unknown")
520 try:
521 IN_MEMORY_GUARDRAIL_HANDLER.sync_guardrail_from_db(guardrail=cast(Guardrail, result))
522 verbose_proxy_logger.info(
523 "Immediate sync: Successfully updated guardrail '%s' (ID: %s)", guardrail_name, guardrail_id
524 )
525 except (ValueError, TypeError) as update_error:
526 # The new config is invalid (a raising guardrail __init__):
527 # reinitialize_guardrail already restored the previous live instance, but
528 # update_guardrail_in_db above already persisted the rejected config to
529 # the DB. Roll that back too, so the DB and the live guardrail never
530 # disagree about what's actually enforcing, and surface the rejection to
531 # the caller instead of a misleading 200.
532 await GUARDRAIL_REGISTRY.update_guardrail_in_db(
533 guardrail_id=guardrail_id,
534 guardrail=existing_guardrail,
535 prisma_client=prisma_client,
536 )
537 raise HTTPException(
538 status_code=422,
539 detail=f"Invalid guardrail configuration, update rejected: {update_error}",
540 ) from update_error
541 except Exception as update_error:
542 verbose_proxy_logger.warning(
543 "Immediate sync: Failed to update '%s' (ID: %s) in memory: %s",
544 guardrail_name,
545 guardrail_id,
546 update_error,
547 )
549 return result
550 except HTTPException as e:
551 raise e
552 except Exception as e:
553 raise HTTPException(status_code=500, detail=str(e))
556@router.delete(
557 "/guardrails/{guardrail_id}",
558 tags=["Guardrails"],
559)
560async def delete_guardrail(
561 guardrail_id: str,
562 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
563):
564 """
565 Delete a guardrail
567 👉 [Guardrail docs](https://docs.litellm.ai/docs/proxy/guardrails/quick_start)
569 Example Request:
570 ```bash
571 curl -X DELETE "http://localhost:4000/guardrails/123e4567-e89b-12d3-a456-426614174000" \\
572 -H "Authorization: Bearer <your_api_key>"
573 ```
575 Example Response:
576 ```json
577 {
578 "message": "Guardrail 123e4567-e89b-12d3-a456-426614174000 deleted successfully"
579 }
580 ```
581 """
582 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER
583 from litellm.proxy.proxy_server import prisma_client
585 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 585 ↛ 586line 585 didn't jump to line 586 because the condition on line 585 was never true
586 raise HTTPException(
587 status_code=403,
588 detail="Admin access required to manage guardrails",
589 )
591 if prisma_client is None: 591 ↛ 592line 591 didn't jump to line 592 because the condition on line 591 was never true
592 raise HTTPException(status_code=500, detail="Prisma client not initialized")
594 try:
595 # Check if guardrail exists
596 existing_guardrail: Final = await GUARDRAIL_REGISTRY.get_guardrail_by_id_from_db(
597 guardrail_id=guardrail_id, prisma_client=prisma_client
598 )
600 if existing_guardrail is None: 600 ↛ 603line 600 didn't jump to line 603 because the condition on line 600 was always true
601 raise HTTPException(status_code=404, detail=f"Guardrail with ID {guardrail_id} not found")
603 result: Final = await GUARDRAIL_REGISTRY.delete_guardrail_from_db(
604 guardrail_id=guardrail_id, prisma_client=prisma_client
605 )
607 guardrail_name: Final = result.get("guardrail_name", "Unknown")
609 try:
610 IN_MEMORY_GUARDRAIL_HANDLER.delete_in_memory_guardrail(
611 guardrail_id=guardrail_id,
612 )
613 verbose_proxy_logger.info(
614 "Immediate sync: Successfully removed guardrail '%s' (ID: %s) from memory", guardrail_name, guardrail_id
615 )
616 except Exception as delete_error:
617 verbose_proxy_logger.warning(
618 "Immediate sync: Failed to remove guardrail '%s' (ID: %s) from memory: %s",
619 guardrail_name,
620 guardrail_id,
621 delete_error,
622 )
624 return result
625 except HTTPException as e:
626 raise e
627 except Exception as e:
628 raise HTTPException(status_code=500, detail=str(e))
631# --- Team guardrail registration (Generic Guardrail API spec) ---
633GENERIC_GUARDRAIL_API: Final = "generic_guardrail_api"
636class RegisterGuardrailRequest(BaseModel):
637 """Request body for POST /guardrails/register. Follows Generic Guardrail API config."""
639 guardrail_name: str
640 litellm_params: dict[str, object] # guardrail, mode, api_base required; api_key, headers, etc. optional
641 guardrail_info: dict[str, object] | None = None
642 team_id: str | None = None
644 def get_litellm_params_dict(self) -> dict[str, Any]:
645 return dict(self.litellm_params)
648class RegisterGuardrailResponse(BaseModel):
649 guardrail_id: str
650 guardrail_name: str
651 status: str
652 submitted_at: datetime | None = None
655class GuardrailSubmissionSummary(BaseModel):
656 total: int
657 pending_review: int
658 active: int
659 rejected: int
662class GuardrailSubmissionItem(BaseModel):
663 guardrail_id: str
664 guardrail_name: str
665 status: str # pending_review | active | rejected
666 team_id: str | None = None
667 team_guardrail: bool = (
668 False # True when submitted via team (team_id set); use to distinguish team vs regular guardrails
669 )
670 litellm_params: dict[str, object] | None = None
671 guardrail_info: dict[str, object] | None = None
672 submitted_by_user_id: str | None = None
673 submitted_by_email: str | None = None
674 submitted_at: datetime | None = None
675 reviewed_at: datetime | None = None
676 created_at: datetime | None = None
677 updated_at: datetime | None = None
680class ListGuardrailSubmissionsResponse(BaseModel):
681 submissions: list[GuardrailSubmissionItem]
682 summary: GuardrailSubmissionSummary
685@router.post(
686 "/guardrails/register",
687 tags=["Guardrails"],
688 response_model=RegisterGuardrailResponse,
689)
690async def register_guardrail(
691 request: RegisterGuardrailRequest,
692 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
693):
694 """
695 Register a guardrail for onboarding (team submission).
697 Accepts a guardrail config in the
698 [Generic Guardrail API](https://docs.litellm.ai/docs/adding_provider/generic_guardrail_api) format.
699 The submission is stored with status `pending_review` until an admin approves it.
700 """
701 from litellm.proxy.proxy_server import prisma_client
703 if prisma_client is None: 703 ↛ 704line 703 didn't jump to line 704 because the condition on line 703 was never true
704 raise HTTPException(status_code=500, detail="Prisma client not initialized")
706 # Resolve team_id: prefer request body, fall back to API key's team
707 team_id: Final = request.team_id or user_api_key_dict.team_id
708 if not team_id:
709 raise HTTPException(
710 status_code=400,
711 detail="team_id is required. Provide it in the request body or use a team-scoped API key.",
712 )
714 # Validate team membership for non-admin users when team differs from key
715 is_admin: Final = user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
716 if not is_admin and team_id != user_api_key_dict.team_id: 716 ↛ 717line 716 didn't jump to line 717 because the condition on line 716 was never true
717 user_team_ids: Final = await _get_user_team_ids(user_api_key_dict)
718 if team_id not in user_team_ids:
719 raise HTTPException(
720 status_code=403,
721 detail=f"You are not a member of team {team_id!r}",
722 )
724 params: Final = request.get_litellm_params_dict()
725 if params.get("guardrail") != GENERIC_GUARDRAIL_API: 725 ↛ 730line 725 didn't jump to line 730 because the condition on line 725 was always true
726 raise HTTPException(
727 status_code=400,
728 detail=f"Only guardrails with litellm_params.guardrail={GENERIC_GUARDRAIL_API!r} are accepted for registration",
729 )
730 api_base: Final = params.get("api_base")
731 if not api_base:
732 raise HTTPException(
733 status_code=400,
734 detail="litellm_params.api_base is required for generic_guardrail_api",
735 )
736 parsed: Final = urlparse(api_base)
737 if parsed.scheme not in ("http", "https"):
738 raise HTTPException(
739 status_code=400,
740 detail="litellm_params.api_base must use http or https scheme",
741 )
742 if not parsed.hostname:
743 raise HTTPException(
744 status_code=400,
745 detail="litellm_params.api_base must contain a valid hostname",
746 )
747 mode: Final = params.get("mode")
748 if mode is None:
749 raise HTTPException(
750 status_code=400,
751 detail="litellm_params.mode is required (e.g. pre_call, post_call)",
752 )
754 try:
755 existing = await _guardrails_table(prisma_client).find_unique(where={"guardrail_name": request.guardrail_name})
756 if existing is not None:
757 raise HTTPException(
758 status_code=400,
759 detail=f"Guardrail with name {request.guardrail_name!r} already exists",
760 )
761 except HTTPException:
762 raise
763 except Exception as e:
764 verbose_proxy_logger.exception("Error checking guardrail name uniqueness: %s", e)
765 raise HTTPException(status_code=500, detail=str(e))
767 now: Final = datetime.now(timezone.utc)
768 litellm_params_str: Final = safe_dumps(params)
769 guardrail_info: Final = dict(request.guardrail_info or {})
770 guardrail_info["submitted_by_user_id"] = user_api_key_dict.user_id
771 guardrail_info["submitted_by_email"] = user_api_key_dict.user_email
772 guardrail_info["team_guardrail"] = True # Mark as team submission for filtering/display
773 guardrail_info_str: Final = safe_dumps(guardrail_info)
775 try:
776 created: Final = await _create_guardrail_row(
777 prisma_client,
778 data={
779 "guardrail_name": request.guardrail_name,
780 "litellm_params": litellm_params_str,
781 "guardrail_info": guardrail_info_str,
782 "status": "pending_review",
783 "team_id": team_id,
784 "submitted_at": now,
785 "created_at": now,
786 "updated_at": now,
787 },
788 )
789 return RegisterGuardrailResponse(
790 guardrail_id=created.guardrail_id,
791 guardrail_name=created.guardrail_name,
792 status=created.status,
793 submitted_at=created.submitted_at,
794 )
795 except Exception as e:
796 verbose_proxy_logger.exception("Error registering guardrail: %s", e)
797 raise HTTPException(status_code=500, detail=str(e))
800def _parse_json_field(value: object) -> dict[str, Any] | None:
801 if value is None:
802 return None
803 if isinstance(value, dict):
804 return value
805 if isinstance(value, str):
806 try:
807 return json.loads(value)
808 except Exception:
809 return None
810 return None
813async def _get_user_team_ids(user_api_key_dict: UserAPIKeyAuth) -> list[str]:
814 """Return the list of team_ids the caller belongs to (empty list if none)."""
815 from litellm.proxy.auth.auth_checks import get_user_object
816 from litellm.proxy.proxy_server import (
817 prisma_client,
818 proxy_logging_obj,
819 user_api_key_cache,
820 )
822 if not user_api_key_dict.user_id or prisma_client is None:
823 return []
824 user_obj: Final = await get_user_object(
825 user_id=user_api_key_dict.user_id,
826 prisma_client=prisma_client,
827 user_api_key_cache=user_api_key_cache,
828 user_id_upsert=False,
829 parent_otel_span=user_api_key_dict.parent_otel_span,
830 proxy_logging_obj=proxy_logging_obj,
831 )
832 if user_obj is None or not user_obj.teams:
833 return []
834 return [t for t in user_obj.teams if t]
837def _row_to_submission_item(row: "LiteLLM_GuardrailsTable") -> GuardrailSubmissionItem:
838 from litellm.litellm_core_utils.litellm_logging import _get_masked_values
840 guardrail_info: Final = _parse_json_field(row.guardrail_info) or {}
841 team_guardrail: Final = row.team_id is not None
842 raw_params: Final = _parse_json_field(row.litellm_params) or {}
843 masked_params: Final = _get_masked_values(raw_params, unmasked_length=4, number_of_asterisks=4)
844 return GuardrailSubmissionItem(
845 guardrail_id=row.guardrail_id,
846 guardrail_name=row.guardrail_name,
847 status=row.status or "active",
848 team_id=row.team_id,
849 team_guardrail=team_guardrail,
850 litellm_params=masked_params,
851 guardrail_info=guardrail_info,
852 submitted_by_user_id=guardrail_info.get("submitted_by_user_id"),
853 submitted_by_email=guardrail_info.get("submitted_by_email"),
854 submitted_at=getattr(row, "submitted_at", None),
855 reviewed_at=getattr(row, "reviewed_at", None),
856 created_at=row.created_at,
857 updated_at=row.updated_at,
858 )
861@router.get(
862 "/guardrails/submissions",
863 tags=["Guardrails"],
864 response_model=ListGuardrailSubmissionsResponse,
865)
866async def list_guardrail_submissions(
867 status: str | None = None,
868 team_id: str | None = None,
869 search: str | None = None,
870 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
871):
872 """
873 List team guardrail submissions. Returns only guardrails with a team_id.
875 Admins see all submissions. Non-admin users see submissions for teams they are
876 a member of.
878 Status values: pending_review (team-registered, awaiting approval), active (approved), rejected.
880 Optional filters:
881 - status: pending_review | active | rejected
882 - team_id: filter by specific team (non-admins must be a member of that team)
883 - search: name/description
884 """
885 from litellm.proxy.proxy_server import prisma_client
887 if prisma_client is None: 887 ↛ 888line 887 didn't jump to line 888 because the condition on line 887 was never true
888 raise HTTPException(status_code=500, detail="Prisma client not initialized")
890 # Admin Viewer follows the read-parity rule: see all submissions like a
891 # Proxy Admin would (no writes — registration / approval still gated
892 # elsewhere by their own per-action checks).
893 is_admin: Final = _user_has_admin_view(user_api_key_dict)
894 visible_team_ids: list[str] | None = None
895 if not is_admin: 895 ↛ 896line 895 didn't jump to line 896 because the condition on line 895 was never true
896 visible_team_ids = await _get_user_team_ids(user_api_key_dict)
897 if team_id is not None and team_id not in visible_team_ids:
898 raise HTTPException(
899 status_code=403,
900 detail=f"You are not a member of team {team_id!r}",
901 )
903 try:
904 where_clause: Final[dict[str, object]] = {"team_id": {"not": None}}
905 if visible_team_ids is not None: 905 ↛ 906line 905 didn't jump to line 906 because the condition on line 905 was never true
906 if not visible_team_ids:
907 # Non-admin with no team memberships: nothing visible.
908 return ListGuardrailSubmissionsResponse(
909 submissions=[],
910 summary=GuardrailSubmissionSummary(total=0, pending_review=0, active=0, rejected=0),
911 )
912 where_clause["team_id"] = {"in": visible_team_ids}
914 # Single query: fetch team guardrails visible to the caller
915 all_team_rows: Final = await _find_team_guardrail_rows(prisma_client, where_clause)
917 # Derive summary counts from the full result set
918 total: Final = len(all_team_rows)
919 pending_review: Final = sum(1 for r in all_team_rows if (r.status or "active") == "pending_review")
920 active_count: Final = sum(1 for r in all_team_rows if (r.status or "active") == "active")
921 rejected: Final = sum(1 for r in all_team_rows if (r.status or "active") == "rejected")
923 # Apply filters to get the submissions list
924 rows = all_team_rows
925 if status:
926 rows = [r for r in rows if r.status == status]
927 if team_id:
928 rows = [r for r in rows if r.team_id == team_id]
929 if search:
930 search_lower: Final = search.lower()
931 rows = [
932 r
933 for r in rows
934 if search_lower in (r.guardrail_name or "").lower()
935 or (
936 isinstance(r.guardrail_info, dict)
937 and search_lower in str((r.guardrail_info or {}).get("description", "")).lower()
938 )
939 or (isinstance(r.guardrail_info, str) and search_lower in r.guardrail_info.lower())
940 ]
942 items: Final = [_row_to_submission_item(r) for r in rows]
943 return ListGuardrailSubmissionsResponse(
944 submissions=items,
945 summary=GuardrailSubmissionSummary(
946 total=total,
947 pending_review=pending_review,
948 active=active_count,
949 rejected=rejected,
950 ),
951 )
952 except Exception as e:
953 verbose_proxy_logger.exception("Error listing guardrail submissions: %s", e)
954 raise HTTPException(status_code=500, detail=str(e))
957@router.get(
958 "/guardrails/submissions/{guardrail_id}",
959 tags=["Guardrails"],
960 response_model=GuardrailSubmissionItem,
961)
962async def get_guardrail_submission(
963 guardrail_id: str,
964 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
965):
966 """Get a single guardrail submission by id. Non-admins may only access submissions for teams they belong to."""
967 from litellm.proxy.proxy_server import prisma_client
969 if prisma_client is None: 969 ↛ 970line 969 didn't jump to line 970 because the condition on line 969 was never true
970 raise HTTPException(status_code=500, detail="Prisma client not initialized")
972 is_admin: Final = _user_has_admin_view(user_api_key_dict)
974 try:
975 row: Final = await _guardrails_table(prisma_client).find_unique(where={"guardrail_id": guardrail_id})
976 if row is None: 976 ↛ 978line 976 didn't jump to line 978 because the condition on line 976 was always true
977 raise HTTPException(status_code=404, detail="Guardrail submission not found")
978 if not is_admin:
979 visible_team_ids: Final = await _get_user_team_ids(user_api_key_dict)
980 if row.team_id is None or row.team_id not in visible_team_ids:
981 raise HTTPException(
982 status_code=403,
983 detail="You are not a member of the team that owns this submission",
984 )
985 return _row_to_submission_item(row)
986 except HTTPException:
987 raise
988 except Exception as e:
989 verbose_proxy_logger.exception("Error getting guardrail submission: %s", e)
990 raise HTTPException(status_code=500, detail=str(e))
993@router.post(
994 "/guardrails/submissions/{guardrail_id}/approve",
995 tags=["Guardrails"],
996)
997async def approve_guardrail_submission(
998 guardrail_id: str,
999 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1000):
1001 """Approve a pending guardrail submission: set status to active and initialize in memory (admin only)."""
1002 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER
1003 from litellm.proxy.proxy_server import prisma_client
1005 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 1005 ↛ 1006line 1005 didn't jump to line 1006 because the condition on line 1005 was never true
1006 raise HTTPException(status_code=403, detail="Admin access required")
1008 if prisma_client is None: 1008 ↛ 1009line 1008 didn't jump to line 1009 because the condition on line 1008 was never true
1009 raise HTTPException(status_code=500, detail="Prisma client not initialized")
1011 try:
1012 row: Final = await _guardrails_table(prisma_client).find_unique(where={"guardrail_id": guardrail_id})
1013 if row is None: 1013 ↛ 1015line 1013 didn't jump to line 1015 because the condition on line 1013 was always true
1014 raise HTTPException(status_code=404, detail="Guardrail submission not found")
1015 if row.status != "pending_review":
1016 raise HTTPException(
1017 status_code=400,
1018 detail=f"Guardrail is not pending review (status={row.status})",
1019 )
1021 now: Final = datetime.now(timezone.utc)
1022 await _guardrails_table(prisma_client).update(
1023 where={"guardrail_id": guardrail_id},
1024 data={"status": "active", "reviewed_at": now, "updated_at": now},
1025 )
1027 litellm_params: Final = _parse_json_field(row.litellm_params)
1028 guardrail_info: Final = _parse_json_field(row.guardrail_info)
1029 if not litellm_params:
1030 raise HTTPException(
1031 status_code=500,
1032 detail="Guardrail litellm_params is missing or invalid",
1033 )
1034 guardrail_dict: Final = {
1035 "guardrail_id": row.guardrail_id,
1036 "guardrail_name": row.guardrail_name,
1037 "litellm_params": litellm_params,
1038 "guardrail_info": guardrail_info or {},
1039 "team_id": row.team_id,
1040 }
1041 try:
1042 IN_MEMORY_GUARDRAIL_HANDLER.initialize_guardrail(guardrail=cast(Guardrail, guardrail_dict), source="db")
1043 verbose_proxy_logger.info(
1044 "Approved guardrail %s (ID: %s) and initialized in memory",
1045 row.guardrail_name,
1046 guardrail_id,
1047 )
1048 except Exception as init_err:
1049 verbose_proxy_logger.warning(
1050 "Failed to initialize approved guardrail %s in memory: %s",
1051 guardrail_id,
1052 init_err,
1053 )
1054 return {
1055 "guardrail_id": guardrail_id,
1056 "status": "active",
1057 "message": "Guardrail approved",
1058 "warning": f"Guardrail was marked active but failed to initialize in memory: {init_err}. "
1059 "It will be picked up on the next sync cycle.",
1060 }
1062 return {
1063 "guardrail_id": guardrail_id,
1064 "status": "active",
1065 "message": "Guardrail approved",
1066 }
1067 except HTTPException:
1068 raise
1069 except Exception as e:
1070 verbose_proxy_logger.exception("Error approving guardrail submission: %s", e)
1071 raise HTTPException(status_code=500, detail=str(e))
1074@router.post(
1075 "/guardrails/submissions/{guardrail_id}/reject",
1076 tags=["Guardrails"],
1077)
1078async def reject_guardrail_submission(
1079 guardrail_id: str,
1080 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1081):
1082 """Reject a guardrail submission (admin only)."""
1083 from litellm.proxy.proxy_server import prisma_client
1085 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 1085 ↛ 1086line 1085 didn't jump to line 1086 because the condition on line 1085 was never true
1086 raise HTTPException(status_code=403, detail="Admin access required")
1088 if prisma_client is None: 1088 ↛ 1089line 1088 didn't jump to line 1089 because the condition on line 1088 was never true
1089 raise HTTPException(status_code=500, detail="Prisma client not initialized")
1091 try:
1092 row: Final = await _guardrails_table(prisma_client).find_unique(where={"guardrail_id": guardrail_id})
1093 if row is None: 1093 ↛ 1095line 1093 didn't jump to line 1095 because the condition on line 1093 was always true
1094 raise HTTPException(status_code=404, detail="Guardrail submission not found")
1095 if row.status != "pending_review":
1096 raise HTTPException(
1097 status_code=400,
1098 detail=f"Guardrail is not pending review (status={row.status})",
1099 )
1101 now: Final = datetime.now(timezone.utc)
1102 await _guardrails_table(prisma_client).update(
1103 where={"guardrail_id": guardrail_id},
1104 data={"status": "rejected", "reviewed_at": now, "updated_at": now},
1105 )
1106 return {
1107 "guardrail_id": guardrail_id,
1108 "status": "rejected",
1109 "message": "Guardrail rejected",
1110 }
1111 except HTTPException:
1112 raise
1113 except Exception as e:
1114 verbose_proxy_logger.exception("Error rejecting guardrail submission: %s", e)
1115 raise HTTPException(status_code=500, detail=str(e))
1118@router.patch(
1119 "/guardrails/{guardrail_id}",
1120 tags=["Guardrails"],
1121)
1122async def patch_guardrail(
1123 guardrail_id: str,
1124 request: PatchGuardrailRequest,
1125 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1126):
1127 """
1128 Partially update an existing guardrail
1130 👉 [Guardrail docs](https://docs.litellm.ai/docs/proxy/guardrails/quick_start)
1132 This endpoint allows updating specific fields of a guardrail without sending the entire object.
1133 Only the following fields can be updated:
1134 - guardrail_name: The name of the guardrail
1135 - default_on: Whether the guardrail is enabled by default
1136 - guardrail_info: Additional information about the guardrail
1138 Example Request:
1139 ```bash
1140 curl -X PATCH "http://localhost:4000/guardrails/123e4567-e89b-12d3-a456-426614174000" \\
1141 -H "Authorization: Bearer <your_api_key>" \\
1142 -H "Content-Type: application/json" \\
1143 -d '{
1144 "guardrail_name": "updated-name",
1145 "default_on": true,
1146 "guardrail_info": {
1147 "description": "Updated description"
1148 }
1149 }'
1150 ```
1152 Example Response:
1153 ```json
1154 {
1155 "guardrail_id": "123e4567-e89b-12d3-a456-426614174000",
1156 "guardrail_name": "updated-name",
1157 "litellm_params": {
1158 "guardrail": "bedrock",
1159 "mode": "pre_call",
1160 "guardrailIdentifier": "ff6ujrregl1q",
1161 "guardrailVersion": "DRAFT",
1162 "default_on": true
1163 },
1164 "guardrail_info": {
1165 "description": "Updated description"
1166 },
1167 "created_at": "2023-11-09T12:34:56.789Z",
1168 "updated_at": "2023-11-09T14:22:33.456Z"
1169 }
1170 ```
1171 """
1172 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER
1173 from litellm.proxy.proxy_server import prisma_client
1175 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 1175 ↛ 1176line 1175 didn't jump to line 1176 because the condition on line 1175 was never true
1176 raise HTTPException(
1177 status_code=403,
1178 detail="Admin access required to manage guardrails",
1179 )
1181 if prisma_client is None: 1181 ↛ 1182line 1181 didn't jump to line 1182 because the condition on line 1181 was never true
1182 raise HTTPException(status_code=500, detail="Prisma client not initialized")
1184 try:
1185 # Check if guardrail exists and get current data
1186 existing_guardrail: Final = await GUARDRAIL_REGISTRY.get_guardrail_by_id_from_db(
1187 guardrail_id=guardrail_id, prisma_client=prisma_client
1188 )
1190 if existing_guardrail is None: 1190 ↛ 1194line 1190 didn't jump to line 1194 because the condition on line 1190 was always true
1191 raise HTTPException(status_code=404, detail=f"Guardrail with ID {guardrail_id} not found")
1193 # Create updated guardrail object
1194 guardrail_name = (
1195 request.guardrail_name if request.guardrail_name is not None else existing_guardrail.get("guardrail_name")
1196 )
1198 # Update litellm_params if default_on is provided or pii_entities_config is provided
1199 existing_litellm_params: Final = _as_str_object_mapping(dict(existing_guardrail.get("litellm_params", {})))
1200 litellm_params = LitellmParams(**existing_litellm_params)
1201 if request.litellm_params is not None:
1202 requested_litellm_params: Final = request.litellm_params.model_dump(exclude_unset=True)
1203 litellm_params_dict: Final = litellm_params.model_dump(exclude_unset=True)
1204 litellm_params_dict.update(requested_litellm_params)
1205 merged_litellm_params: Final = _as_str_object_mapping(litellm_params_dict)
1206 try:
1207 litellm_params = LitellmParams(**merged_litellm_params)
1208 except ValidationError as validation_error:
1209 raise HTTPException(
1210 status_code=422,
1211 detail=f"Invalid guardrail configuration, update rejected: {validation_error}",
1212 ) from validation_error
1214 # Update guardrail_info if provided
1215 guardrail_info: Final = (
1216 request.guardrail_info
1217 if request.guardrail_info is not None
1218 else existing_guardrail.get("guardrail_info", {})
1219 )
1221 # Create the guardrail object
1222 guardrail: Final = Guardrail(
1223 guardrail_id=guardrail_id,
1224 guardrail_name=guardrail_name or "",
1225 litellm_params=litellm_params,
1226 guardrail_info=guardrail_info,
1227 )
1228 result: Final = _as_str_object_mapping(
1229 await GUARDRAIL_REGISTRY.update_guardrail_in_db(
1230 guardrail_id=guardrail_id,
1231 guardrail=guardrail,
1232 prisma_client=prisma_client,
1233 )
1234 )
1236 guardrail_name = result.get("guardrail_name", "Unknown")
1238 try:
1239 IN_MEMORY_GUARDRAIL_HANDLER.sync_guardrail_from_db(
1240 guardrail=guardrail,
1241 )
1242 verbose_proxy_logger.info(
1243 "Immediate sync: Successfully updated guardrail '%s' (ID: %s)", guardrail_name, guardrail_id
1244 )
1245 except (ValueError, TypeError) as update_error:
1246 # The new config is invalid (e.g. an unsupported on_flagged combination):
1247 # reinitialize_guardrail already restored the previous live instance, but
1248 # update_guardrail_in_db above already persisted the rejected config to
1249 # the DB. Roll that back too, so the DB and the live guardrail never
1250 # disagree about what's actually enforcing, and surface the rejection to
1251 # the caller instead of a misleading 200.
1252 await GUARDRAIL_REGISTRY.update_guardrail_in_db(
1253 guardrail_id=guardrail_id,
1254 guardrail=Guardrail(
1255 guardrail_id=guardrail_id,
1256 guardrail_name=existing_guardrail.get("guardrail_name") or "",
1257 litellm_params=LitellmParams(**existing_litellm_params),
1258 guardrail_info=existing_guardrail.get(
1259 "guardrail_info",
1260 {}, # mutable-ok: Guardrail's own constructor takes a plain dict
1261 ),
1262 ),
1263 prisma_client=prisma_client,
1264 )
1265 raise HTTPException(
1266 status_code=422,
1267 detail=f"Invalid guardrail configuration, update rejected: {update_error}",
1268 ) from update_error
1269 except Exception as update_error:
1270 verbose_proxy_logger.warning(
1271 "Immediate sync: Failed to update '%s' (ID: %s) in memory: %s",
1272 guardrail_name,
1273 guardrail_id,
1274 update_error,
1275 )
1277 return result
1278 except HTTPException as e:
1279 raise e
1280 except Exception as e:
1281 verbose_proxy_logger.exception("Error updating guardrail: %s", e)
1282 raise HTTPException(status_code=500, detail=str(e))
1285@router.get(
1286 "/guardrails/{guardrail_id}",
1287 tags=["Guardrails"],
1288 dependencies=[Depends(user_api_key_auth)],
1289)
1290@router.get(
1291 "/guardrails/{guardrail_id}/info",
1292 tags=["Guardrails"],
1293 dependencies=[Depends(user_api_key_auth)],
1294)
1295async def get_guardrail_info(guardrail_id: str):
1296 """
1297 Get detailed information about a specific guardrail by ID
1299 👉 [Guardrail docs](https://docs.litellm.ai/docs/proxy/guardrails/quick_start)
1301 Example Request:
1302 ```bash
1303 curl -X GET "http://localhost:4000/guardrails/123e4567-e89b-12d3-a456-426614174000/info" \\
1304 -H "Authorization: Bearer <your_api_key>"
1305 ```
1307 Example Response:
1308 ```json
1309 {
1310 "guardrail_id": "123e4567-e89b-12d3-a456-426614174000",
1311 "guardrail_name": "my-bedrock-guard",
1312 "litellm_params": {
1313 "guardrail": "bedrock",
1314 "mode": "pre_call",
1315 "guardrailIdentifier": "ff6ujrregl1q",
1316 "guardrailVersion": "DRAFT",
1317 "default_on": true
1318 },
1319 "guardrail_info": {
1320 "description": "Bedrock content moderation guardrail"
1321 },
1322 "created_at": "2023-11-09T12:34:56.789Z",
1323 "updated_at": "2023-11-09T12:34:56.789Z"
1324 }
1325 ```
1326 """
1328 from litellm.litellm_core_utils.litellm_logging import _get_masked_values
1329 from litellm.proxy.guardrails.guardrail_registry import IN_MEMORY_GUARDRAIL_HANDLER
1330 from litellm.proxy.proxy_server import prisma_client
1331 from litellm.types.guardrails import GUARDRAIL_DEFINITION_LOCATION
1333 try:
1334 guardrail_definition_location: GUARDRAIL_DEFINITION_LOCATION = GUARDRAIL_DEFINITION_LOCATION.DB
1335 result = (
1336 await GUARDRAIL_REGISTRY.get_guardrail_by_id_from_db(guardrail_id=guardrail_id, prisma_client=prisma_client)
1337 if prisma_client is not None
1338 else None
1339 )
1340 if result is None: 1340 ↛ 1349line 1340 didn't jump to line 1349 because the condition on line 1340 was always true
1341 in_memory: Final = IN_MEMORY_GUARDRAIL_HANDLER.get_guardrail_by_id(guardrail_id=guardrail_id)
1342 # Only return config-loaded entries here. A DB-backed entry that's
1343 # missing from the DB is stale (deleted on another pod, awaiting
1344 # reconciliation on this one) and must surface as 404.
1345 if in_memory is not None and IN_MEMORY_GUARDRAIL_HANDLER.get_source(guardrail_id) == "config": 1345 ↛ 1346line 1345 didn't jump to line 1346 because the condition on line 1345 was never true
1346 result = in_memory
1347 guardrail_definition_location = GUARDRAIL_DEFINITION_LOCATION.CONFIG
1349 if result is None: 1349 ↛ 1352line 1349 didn't jump to line 1352 because the condition on line 1349 was always true
1350 raise HTTPException(status_code=404, detail=f"Guardrail with ID {guardrail_id} not found")
1352 litellm_params: Final[LitellmParams | dict | None] = result.get("litellm_params")
1353 result_litellm_params_dict: Final = (
1354 litellm_params.model_dump(exclude_none=True)
1355 if isinstance(litellm_params, LitellmParams)
1356 else litellm_params
1357 ) or {}
1358 masked_litellm_params_dict: Final = _get_masked_values(
1359 result_litellm_params_dict,
1360 unmasked_length=4,
1361 number_of_asterisks=4,
1362 )
1363 masked_litellm_params = BaseLitellmParams(**masked_litellm_params_dict) if masked_litellm_params_dict else None
1365 return GuardrailInfoResponse(
1366 guardrail_id=result.get("guardrail_id"),
1367 guardrail_name=result.get("guardrail_name"),
1368 litellm_params=masked_litellm_params,
1369 guardrail_info=dict(result.get("guardrail_info") or {}),
1370 created_at=result.get("created_at"),
1371 updated_at=result.get("updated_at"),
1372 guardrail_definition_location=guardrail_definition_location,
1373 )
1374 except HTTPException as e:
1375 raise e
1376 except Exception as e:
1377 raise HTTPException(status_code=500, detail=str(e))
1380@router.get(
1381 "/guardrails/ui/add_guardrail_settings",
1382 tags=["Guardrails"],
1383 dependencies=[Depends(user_api_key_auth)],
1384)
1385async def get_guardrail_ui_settings():
1386 """
1387 Get the UI settings for the guardrails
1389 Returns:
1390 - Supported entities for guardrails
1391 - Supported modes for guardrails
1392 - PII entity categories for UI organization
1393 - Content filter settings (patterns and categories)
1394 """
1395 from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.patterns import (
1396 PATTERN_CATEGORIES,
1397 get_available_content_categories,
1398 get_pattern_metadata,
1399 )
1400 from litellm.proxy.guardrails.guardrail_registry import guardrail_class_registry
1402 category_maps: Final = [
1403 {
1404 "category": category.value,
1405 "entities": [entity.value for entity in entities],
1406 }
1407 for category, entities in PII_ENTITY_CATEGORIES_MAP.items()
1408 ]
1410 supported_modes_by_provider: Final = {
1411 provider: [hook.value for hook in hooks]
1412 for provider, guardrail_class in guardrail_class_registry.items()
1413 if (hooks := guardrail_class.get_supported_event_hooks()) is not None
1414 } | MappingProxyType(
1415 # hide-secrets lives in the enterprise package, not in the registry
1416 # above; it only runs on pre_call.
1417 {SupportedGuardrailIntegrations.HIDE_SECRETS.value: [GuardrailEventHooks.pre_call.value]}
1418 )
1420 return GuardrailUIAddGuardrailSettings(
1421 supported_entities=[entity.value for entity in PiiEntityType],
1422 supported_actions=[action.value for action in PiiAction],
1423 supported_modes=[mode.value for mode in GuardrailEventHooks],
1424 supported_modes_by_provider=supported_modes_by_provider,
1425 pii_entity_categories=category_maps,
1426 content_filter_settings={
1427 "prebuilt_patterns": get_pattern_metadata(),
1428 "pattern_categories": list(PATTERN_CATEGORIES.keys()),
1429 "supported_actions": ["BLOCK", "MASK"],
1430 "content_categories": get_available_content_categories(),
1431 },
1432 )
1435@router.get(
1436 "/guardrails/ui/category_yaml/{category_name}",
1437 tags=["Guardrails"],
1438 dependencies=[Depends(user_api_key_auth)],
1439)
1440async def get_category_yaml(category_name: str):
1441 """
1442 Get the YAML or JSON content for a specific content filter category.
1444 Args:
1445 category_name: The name of the category (e.g., "bias_gender", "harmful_self_harm")
1447 Returns:
1448 The raw YAML or JSON content of the category file with file type indicator
1449 """
1450 # Get the categories directory path
1451 categories_dir: Final = os.path.join(
1452 os.path.dirname(__file__),
1453 "guardrail_hooks",
1454 "litellm_content_filter",
1455 "categories",
1456 )
1458 # Try to find the file with either .yaml or .json extension
1459 try:
1460 yaml_path: Final = safe_join(categories_dir, f"{category_name}.yaml")
1461 json_path: Final = safe_join(categories_dir, f"{category_name}.json")
1462 except ValueError:
1463 raise HTTPException(status_code=400, detail="Invalid category name")
1465 category_file_path = None
1466 file_type = None
1468 if os.path.exists(yaml_path): 1468 ↛ 1469line 1468 didn't jump to line 1469 because the condition on line 1468 was never true
1469 category_file_path = yaml_path
1470 file_type = "yaml"
1471 elif os.path.exists(json_path): 1471 ↛ 1472line 1471 didn't jump to line 1472 because the condition on line 1471 was never true
1472 category_file_path = json_path
1473 file_type = "json"
1474 else:
1475 raise HTTPException(
1476 status_code=404,
1477 detail=f"Category file not found: {category_name} (tried .yaml and .json)",
1478 )
1480 try:
1481 # Read and return the raw content
1482 with open(category_file_path, "r") as f:
1483 content: Final = f.read()
1485 return {
1486 "category_name": category_name,
1487 "yaml_content": content, # Keep key name for backwards compatibility
1488 "file_type": file_type,
1489 }
1490 except Exception as e:
1491 raise HTTPException(status_code=500, detail=f"Error reading category file: {e}")
1494@router.get(
1495 "/guardrails/ui/major_airlines",
1496 tags=["Guardrails"],
1497 dependencies=[Depends(user_api_key_auth)],
1498)
1499async def get_major_airlines():
1500 """
1501 Get the major airlines list from IATA (competitor intent, airline type).
1502 Returns airline id, match variants (pipe-separated), and tags.
1503 """
1504 airlines_path: Final = os.path.join(
1505 os.path.dirname(__file__),
1506 "guardrail_hooks",
1507 "litellm_content_filter",
1508 "competitor_intent",
1509 "major_airlines.json",
1510 )
1511 if not os.path.exists(airlines_path): 1511 ↛ 1512line 1511 didn't jump to line 1512 because the condition on line 1511 was never true
1512 raise HTTPException(
1513 status_code=404,
1514 detail="major_airlines.json not found",
1515 )
1516 try:
1517 with open(airlines_path, "r", encoding="utf-8") as f:
1518 import json
1520 airlines: Final = json.load(f)
1521 return {"airlines": airlines}
1522 except Exception as e:
1523 raise HTTPException(status_code=500, detail=f"Error reading major_airlines.json: {e}") from e
1526@router.post(
1527 "/guardrails/validate_blocked_words_file",
1528 tags=["Guardrails"],
1529 dependencies=[Depends(user_api_key_auth)],
1530)
1531async def validate_blocked_words_file(request: dict[str, str]):
1532 """
1533 Validate a blocked_words YAML file content.
1535 Args:
1536 request: Dictionary with 'file_content' key containing the YAML string
1538 Returns:
1539 Dictionary with 'valid' boolean and either 'message'/'errors' depending on result
1541 Example Request:
1542 ```json
1543 {
1544 "file_content": "blocked_words:\\n - keyword: \\"test\\"\\n action: \\"BLOCK\\""
1545 }
1546 ```
1548 Example Success Response:
1549 ```json
1550 {
1551 "valid": true,
1552 "message": "Valid YAML file with 2 blocked words"
1553 }
1554 ```
1556 Example Error Response:
1557 ```json
1558 {
1559 "valid": false,
1560 "errors": ["Entry 0: missing 'action' field"]
1561 }
1562 ```
1563 """
1564 import yaml
1566 try:
1567 file_content: Final = request.get("file_content", "")
1568 if not file_content: 1568 ↛ 1571line 1568 didn't jump to line 1571 because the condition on line 1568 was always true
1569 return {"valid": False, "error": "No file content provided"}
1571 data: Final = yaml.safe_load(file_content)
1573 if not isinstance(data, dict) or "blocked_words" not in data:
1574 return {
1575 "valid": False,
1576 "error": "Invalid format: file must contain 'blocked_words' key with a list",
1577 }
1579 blocked_words_list: Final = data["blocked_words"]
1580 if not isinstance(blocked_words_list, list):
1581 return {"valid": False, "error": "'blocked_words' must be a list"}
1583 # Validate each entry
1584 errors: Final = []
1585 for idx, word_data in enumerate(blocked_words_list):
1586 if not isinstance(word_data, dict):
1587 errors.append(f"Entry {idx}: must be an object")
1588 continue
1590 if "keyword" not in word_data:
1591 errors.append(f"Entry {idx}: missing 'keyword' field")
1592 elif not isinstance(word_data["keyword"], str):
1593 errors.append(f"Entry {idx}: 'keyword' must be a string")
1595 if "action" not in word_data:
1596 errors.append(f"Entry {idx}: missing 'action' field")
1597 elif word_data["action"] not in ["BLOCK", "MASK"]:
1598 errors.append(f"Entry {idx}: action must be 'BLOCK' or 'MASK', got '{word_data['action']}'")
1600 if "description" in word_data and not isinstance(word_data["description"], str):
1601 errors.append(f"Entry {idx}: 'description' must be a string")
1603 if errors:
1604 return {"valid": False, "errors": errors}
1606 return {
1607 "valid": True,
1608 "message": f"Valid YAML file with {len(blocked_words_list)} blocked word(s)",
1609 }
1610 except yaml.YAMLError as e:
1611 return {"valid": False, "error": f"Invalid YAML syntax: {e}"}
1612 except Exception as e:
1613 verbose_proxy_logger.exception("Error validating blocked words file")
1614 return {"valid": False, "error": f"Validation error: {e}"}
1617def _dunder_origin(annotation: object) -> object:
1618 origin: Final[object] = getattr(annotation, "__origin__", None)
1619 return origin
1622def _dunder_name(annotation: object) -> object:
1623 name: Final[object] = getattr(annotation, "__name__", None)
1624 return name
1627def _dunder_args(annotation: object) -> tuple[object, ...]:
1628 args: Final[tuple[object, ...]] = getattr(annotation, "__args__", ())
1629 return args
1632def _get_field_type_from_annotation(field_annotation: object) -> str:
1633 """
1634 Convert a Python type annotation to a UI-friendly type string
1635 """
1636 # Handle Union types (like Optional[T])
1637 if get_origin(field_annotation) is Union or get_origin(field_annotation) is UnionType: 1637 ↛ 1639line 1637 didn't jump to line 1639 because the condition on line 1637 was never true
1638 # For Optional[T], get the non-None type
1639 args: Final[tuple[object, ...]] = get_args(field_annotation)
1640 non_none_args: Final = [arg for arg in args if arg is not type(None)]
1641 if non_none_args:
1642 field_annotation = non_none_args[0]
1644 # Handle List types
1645 if hasattr(field_annotation, "__origin__") and _dunder_origin(field_annotation) is list:
1646 return "array"
1648 # Handle Dict types
1649 if hasattr(field_annotation, "__origin__") and _dunder_origin(field_annotation) is dict:
1650 return "dict"
1652 # Handle Literal types
1653 if hasattr(field_annotation, "__origin__") and hasattr(field_annotation, "__args__"):
1654 # Check for Literal types (Python 3.8+)
1655 origin: Final = _dunder_origin(field_annotation)
1656 if hasattr(origin, "__name__") and _dunder_name(origin) == "Literal": 1656 ↛ 1660line 1656 didn't jump to line 1660 because the condition on line 1656 was always true
1657 return "select" # For dropdown/select inputs
1659 # Handle basic types
1660 if field_annotation is str:
1661 return "string"
1662 elif field_annotation is int or field_annotation is float:
1663 return "number"
1664 elif field_annotation is bool:
1665 return "boolean"
1666 elif field_annotation is dict:
1667 return "object"
1668 elif field_annotation is list: 1668 ↛ 1669line 1668 didn't jump to line 1669 because the condition on line 1668 was never true
1669 return "array"
1671 # Default to string for unknown types
1672 return "string"
1675def _extract_literal_values(annotation: object) -> Sequence[object]:
1676 """
1677 Extract literal values from a Literal type annotation
1678 """
1679 if hasattr(annotation, "__origin__") and hasattr(annotation, "__args__"):
1680 origin: Final = _dunder_origin(annotation)
1681 if hasattr(origin, "__name__") and _dunder_name(origin) == "Literal": 1681 ↛ 1683line 1681 didn't jump to line 1683 because the condition on line 1681 was always true
1682 return list(_dunder_args(annotation))
1683 return []
1686def _get_dict_key_options(field_annotation: object) -> Sequence[object] | None:
1687 """
1688 Extract key options from Dict[Literal[...], T] types
1689 """
1690 if ( 1690 ↛ 1699line 1690 didn't jump to line 1699 because the condition on line 1690 was always true
1691 hasattr(field_annotation, "__origin__")
1692 and _dunder_origin(field_annotation) is dict
1693 and hasattr(field_annotation, "__args__")
1694 ):
1695 args: Final = _dunder_args(field_annotation)
1696 if len(args) >= 2: 1696 ↛ 1699line 1696 didn't jump to line 1699 because the condition on line 1696 was always true
1697 key_type: Final = args[0]
1698 return _extract_literal_values(key_type)
1699 return None
1702def _get_dict_value_type(field_annotation: object) -> str:
1703 """
1704 Get the value type from Dict[K, V] types
1705 """
1706 if ( 1706 ↛ 1715line 1706 didn't jump to line 1715 because the condition on line 1706 was always true
1707 hasattr(field_annotation, "__origin__")
1708 and _dunder_origin(field_annotation) is dict
1709 and hasattr(field_annotation, "__args__")
1710 ):
1711 args: Final = _dunder_args(field_annotation)
1712 if len(args) >= 2: 1712 ↛ 1715line 1712 didn't jump to line 1715 because the condition on line 1712 was always true
1713 value_type: Final = args[1]
1714 return _get_field_type_from_annotation(value_type)
1715 return "string"
1718def _get_list_element_options(field_annotation: object) -> Sequence[object] | None:
1719 """
1720 Extract element options from List[Literal[...]] types
1721 """
1722 if ( 1722 ↛ 1731line 1722 didn't jump to line 1731 because the condition on line 1722 was always true
1723 hasattr(field_annotation, "__origin__")
1724 and _dunder_origin(field_annotation) is list
1725 and hasattr(field_annotation, "__args__")
1726 ):
1727 args: Final = _dunder_args(field_annotation)
1728 if len(args) >= 1: 1728 ↛ 1731line 1728 didn't jump to line 1731 because the condition on line 1728 was always true
1729 element_type: Final = args[0]
1730 return _extract_literal_values(element_type)
1731 return None
1734def _should_skip_optional_params(field_name: str, field_annotation: object) -> bool:
1735 """Check if optional_params field should be skipped (not meaningfully overridden)."""
1736 if field_name != "optional_params":
1737 return False
1739 if field_annotation is None: 1739 ↛ 1740line 1739 didn't jump to line 1740 because the condition on line 1739 was never true
1740 return True
1742 # Check if the annotation is still a generic TypeVar (not specialized)
1743 if isinstance(field_annotation, TypeVar) or ( 1743 ↛ 1746line 1743 didn't jump to line 1746 because the condition on line 1743 was never true
1744 hasattr(field_annotation, "__origin__") and _dunder_origin(field_annotation) is TypeVar
1745 ):
1746 return True
1748 # Also skip if it's a generic type that wasn't specialized
1749 if hasattr(field_annotation, "__name__") and _dunder_name(field_annotation) in ( 1749 ↛ 1753line 1749 didn't jump to line 1753 because the condition on line 1749 was never true
1750 "T",
1751 "TypeVar",
1752 ):
1753 return True
1755 # Handle Optional[T] where T is still a TypeVar
1756 if hasattr(field_annotation, "__args__"):
1757 non_none_args: Final = [arg for arg in _dunder_args(field_annotation) if arg is not type(None)]
1758 if non_none_args and isinstance(non_none_args[0], TypeVar):
1759 return True
1761 return False
1764def _unwrap_optional_type(field_annotation: object) -> object:
1765 """Unwrap Optional types to get the actual type."""
1766 if get_origin(field_annotation) is Union or get_origin(field_annotation) is UnionType:
1767 # For Optional[BaseModel], get the non-None type
1768 args: Final[tuple[object, ...]] = get_args(field_annotation)
1769 non_none_args: Final = [arg for arg in args if arg is not type(None)]
1770 if non_none_args: 1770 ↛ 1772line 1770 didn't jump to line 1772 because the condition on line 1770 was always true
1771 return non_none_args[0]
1772 return field_annotation
1775def _build_field_dict(
1776 field: "FieldInfo",
1777 field_annotation: object,
1778 description: str,
1779 required: bool,
1780) -> dict[str, object]:
1781 """Build field dictionary for non-nested fields."""
1782 # Determine the field type from annotation
1783 field_type = _get_field_type_from_annotation(field_annotation)
1785 # Check for custom UI type override
1786 field_json_schema_extra: Final[Mapping[str, object]] = getattr(field, "json_schema_extra", {})
1787 if field_json_schema_extra and "ui_type" in field_json_schema_extra:
1788 ui_type: Final = field_json_schema_extra["ui_type"]
1789 field_type = getattr(ui_type, "value", ui_type)
1790 elif field_json_schema_extra and "type" in field_json_schema_extra: 1790 ↛ 1791line 1790 didn't jump to line 1791 because the condition on line 1790 was never true
1791 field_type = field_json_schema_extra["type"]
1793 # Add the field to the dictionary
1794 field_dict: Final = {
1795 "description": description,
1796 "required": required,
1797 "type": field_type,
1798 }
1800 # Extract options from type annotations
1801 if field_type == "dict":
1802 # For Dict[Literal[...], T] types, extract key options
1803 dict_key_options: Final = _get_dict_key_options(field_annotation)
1804 if dict_key_options:
1805 field_dict["dict_key_options"] = dict_key_options
1807 # Extract value type for the dict values
1808 dict_value_type: Final = _get_dict_value_type(field_annotation)
1809 field_dict["dict_value_type"] = dict_value_type
1811 elif field_type == "array":
1812 # For List[Literal[...]] types, extract element options
1813 list_element_options: Final = _get_list_element_options(field_annotation)
1814 if list_element_options:
1815 field_dict["options"] = list_element_options
1816 field_dict["type"] = "multiselect"
1818 # Add options if they exist in json_schema_extra (this takes precedence)
1819 if field_json_schema_extra and "options" in field_json_schema_extra:
1820 field_dict["options"] = field_json_schema_extra["options"]
1821 elif field_type == "select":
1822 # For Literal types, populate options so the UI can render a dropdown
1823 literal_options: Final = _extract_literal_values(field_annotation)
1824 if literal_options: 1824 ↛ 1828line 1824 didn't jump to line 1828 because the condition on line 1824 was always true
1825 field_dict["options"] = literal_options
1827 # Add default value if it exists
1828 field_default: Final[object] = getattr(field, "default", None)
1829 if field_default is not None and field_default is not ...:
1830 field_dict["default_value"] = field_default
1832 # Copy min, max, step from json_schema_extra for number/percentage inputs
1833 if field_json_schema_extra:
1834 for key in ("min", "max", "step", "default_value"):
1835 if key in field_json_schema_extra:
1836 field_dict[key] = field_json_schema_extra[key]
1838 return field_dict
1841def _extract_fields_recursive(
1842 model: type[BaseModel],
1843 depth: int = 0,
1844) -> dict[str, object]:
1845 # Check if we've exceeded the maximum recursion depth
1846 if depth > DEFAULT_MAX_RECURSE_DEPTH: 1846 ↛ 1847line 1846 didn't jump to line 1847 because the condition on line 1846 was never true
1847 raise HTTPException(
1848 status_code=400,
1849 detail=f"Max depth of {DEFAULT_MAX_RECURSE_DEPTH} exceeded while processing model fields. Please check the model structure for excessive nesting.",
1850 )
1852 fields: Final = {}
1854 for field_name, field in model.model_fields.items():
1855 field_annotation = field.annotation
1857 # Skip optional_params if it's not meaningfully overridden
1858 if _should_skip_optional_params(field_name=field_name, field_annotation=field_annotation):
1859 continue
1861 # Handle Optional types and get the actual type
1862 if field_annotation is None: 1862 ↛ 1863line 1862 didn't jump to line 1863 because the condition on line 1862 was never true
1863 continue
1865 field_annotation = _unwrap_optional_type(field_annotation=field_annotation)
1867 # Get field metadata
1868 description = field.description or field_name
1869 required = field.is_required()
1871 # Check if this is a BaseModel subclass
1872 is_basemodel_subclass = (
1873 inspect.isclass(field_annotation)
1874 and issubclass(field_annotation, BaseModel)
1875 and field_annotation is not BaseModel
1876 )
1878 if is_basemodel_subclass:
1879 # Recursively get fields from the nested model
1880 nested_fields = _extract_fields_recursive(cast(type[BaseModel], field_annotation), depth + 1)
1881 fields[field_name] = {
1882 "description": description,
1883 "required": required,
1884 "type": "nested",
1885 "fields": nested_fields,
1886 }
1887 else:
1888 fields[field_name] = _build_field_dict(
1889 field=field,
1890 field_annotation=field_annotation,
1891 description=description,
1892 required=required,
1893 )
1895 return fields
1898def _get_fields_from_model(model_class: type[BaseModel]) -> dict[str, object]:
1899 """
1900 Get the fields from a Pydantic model as a nested dictionary structure
1901 """
1903 return _extract_fields_recursive(model_class, depth=0)
1906@router.get(
1907 "/guardrails/ui/provider_specific_params",
1908 tags=["Guardrails"],
1909 dependencies=[Depends(user_api_key_auth)],
1910)
1911async def get_provider_specific_params():
1912 """
1913 Get provider-specific parameters for different guardrail types.
1915 Returns a dictionary mapping guardrail providers to their specific parameters,
1916 including parameter names, descriptions, and whether they are required.
1918 Example Response:
1919 ```json
1920 {
1921 "bedrock": {
1922 "guardrailIdentifier": {
1923 "description": "The ID of your guardrail on Bedrock",
1924 "required": true,
1925 "type": null
1926 },
1927 "guardrailVersion": {
1928 "description": "The version of your Bedrock guardrail (e.g., DRAFT or version number)",
1929 "required": true,
1930 "type": null
1931 }
1932 },
1933 "azure_content_safety_text_moderation": {
1934 "api_key": {
1935 "description": "API key for the Azure Content Safety Text Moderation guardrail",
1936 "required": false,
1937 "type": null
1938 },
1939 "optional_params": {
1940 "description": "Optional parameters for the Azure Content Safety Text Moderation guardrail",
1941 "required": true,
1942 "type": "nested",
1943 "fields": {
1944 "severity_threshold": {
1945 "description": "Severity threshold for the Azure Content Safety Text Moderation guardrail across all categories",
1946 "required": false,
1947 "type": null
1948 },
1949 "categories": {
1950 "description": "Categories to scan for the Azure Content Safety Text Moderation guardrail",
1951 "required": false,
1952 "type": "multiselect",
1953 "options": ["Hate", "SelfHarm", "Sexual", "Violence"],
1954 "default_value": None
1955 }
1956 }
1957 }
1958 }
1959 }
1960 ```
1961 """
1962 # Get fields from the models
1963 bedrock_fields: Final = {
1964 **_get_fields_from_model(BedrockGuardrailConfigModel),
1965 **_get_fields_from_model(BedrockGuardrailStreamingParams),
1966 }
1967 presidio_fields: Final = _get_fields_from_model(PresidioPresidioConfigModelUserInterface)
1968 lakera_v2_fields: Final = _get_fields_from_model(LakeraV2GuardrailConfigModel)
1969 tool_permission_fields: Final = _get_fields_from_model(ToolPermissionGuardrailConfigModel)
1971 tool_permission_fields["ui_friendly_name"] = ToolPermissionGuardrailConfigModel.ui_friendly_name()
1973 # hide-secrets lives in the enterprise package, not in the registry loop below.
1974 hide_secrets_fields: Final = _get_fields_from_model(HideSecretsGuardrailConfigModel)
1976 hide_secrets_fields["ui_friendly_name"] = HideSecretsGuardrailConfigModel.ui_friendly_name()
1978 # Return the provider-specific parameters
1979 provider_params: Final = {
1980 SupportedGuardrailIntegrations.BEDROCK.value: bedrock_fields,
1981 SupportedGuardrailIntegrations.PRESIDIO.value: presidio_fields,
1982 SupportedGuardrailIntegrations.LAKERA_V2.value: lakera_v2_fields,
1983 SupportedGuardrailIntegrations.TOOL_PERMISSION.value: tool_permission_fields,
1984 SupportedGuardrailIntegrations.HIDE_SECRETS.value: hide_secrets_fields,
1985 }
1987 ### get the config model for the guardrail - go through the registry and get the config model for the guardrail
1988 from litellm.proxy.guardrails.guardrail_registry import guardrail_class_registry
1990 for guardrail_name, guardrail_class in guardrail_class_registry.items():
1991 guardrail_config_model = guardrail_class.get_config_model()
1993 if guardrail_config_model:
1994 fields = _get_fields_from_model(guardrail_config_model)
1995 ui_friendly_name = guardrail_config_model.ui_friendly_name()
1996 fields["ui_friendly_name"] = ui_friendly_name
1997 provider_params[guardrail_name] = fields
1999 return provider_params
2002class TestCustomCodeGuardrailRequest(BaseModel):
2003 """Request model for testing custom code guardrails."""
2005 custom_code: str
2006 """The Python-like code containing the apply_guardrail function."""
2008 test_input: dict[str, object]
2009 """The test input to pass to the guardrail. Should contain 'texts', optionally 'images', 'tools', etc."""
2011 input_type: str = "request"
2012 """Whether this is a 'request' or 'response' input type."""
2014 request_data: dict[str, object] | None = None
2015 """Optional mock request_data (model, user_id, team_id, metadata, etc.)."""
2018class TestCustomCodeGuardrailResponse(BaseModel):
2019 """Response model for testing custom code guardrails."""
2021 success: bool
2022 """Whether the test executed successfully (no errors)."""
2024 result: dict[str, object] | None = None
2025 """The guardrail result: action (allow/block/modify), reason, modified_texts, etc."""
2027 error: str | None = None
2028 """Error message if execution failed."""
2030 error_type: str | None = None
2031 """Type of error: 'compilation' or 'execution'."""
2034@router.post(
2035 "/guardrails/test_custom_code",
2036 tags=["Guardrails"],
2037 response_model=TestCustomCodeGuardrailResponse,
2038)
2039async def test_custom_code_guardrail(
2040 request: TestCustomCodeGuardrailRequest,
2041 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2042):
2043 """
2044 Test custom code guardrail logic without creating a guardrail.
2046 This endpoint allows admins to experiment with custom code guardrails by:
2047 1. Compiling the provided code in a sandbox
2048 2. Executing the apply_guardrail function with test input
2049 3. Returning the result (allow/block/modify)
2051 👉 [Custom Code Guardrail docs](https://docs.litellm.ai/docs/proxy/guardrails/custom_code_guardrail)
2053 Example Request:
2054 ```bash
2055 curl -X POST "http://localhost:4000/guardrails/test_custom_code" \\
2056 -H "Authorization: Bearer <your_api_key>" \\
2057 -H "Content-Type: application/json" \\
2058 -d '{
2059 "custom_code": "def apply_guardrail(inputs, request_data, input_type):\\n for text in inputs[\\"texts\\"]:\\n if regex_match(text, r\\"\\\\d{3}-\\\\d{2}-\\\\d{4}\\"):\\n return block(\\"SSN detected\\")\\n return allow()",
2060 "test_input": {
2061 "texts": ["My SSN is 123-45-6789"]
2062 },
2063 "input_type": "request"
2064 }'
2065 ```
2067 Example Success Response (blocked):
2068 ```json
2069 {
2070 "success": true,
2071 "result": {
2072 "action": "block",
2073 "reason": "SSN detected"
2074 },
2075 "error": null,
2076 "error_type": null
2077 }
2078 ```
2080 Example Success Response (allowed):
2081 ```json
2082 {
2083 "success": true,
2084 "result": {
2085 "action": "allow"
2086 },
2087 "error": null,
2088 "error_type": null
2089 }
2090 ```
2092 Example Success Response (modified):
2093 ```json
2094 {
2095 "success": true,
2096 "result": {
2097 "action": "modify",
2098 "texts": ["My SSN is [REDACTED]"]
2099 },
2100 "error": null,
2101 "error_type": null
2102 }
2103 ```
2105 Example Error Response (compilation error):
2106 ```json
2107 {
2108 "success": false,
2109 "result": null,
2110 "error": "Syntax error in custom code: invalid syntax (<guardrail>, line 1)",
2111 "error_type": "compilation"
2112 }
2113 ```
2114 """
2116 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 2116 ↛ 2117line 2116 didn't jump to line 2117 because the condition on line 2116 was never true
2117 raise HTTPException(
2118 status_code=403,
2119 detail="Admin access required to test custom code guardrails",
2120 )
2122 EXECUTION_TIMEOUT_SECONDS: Final = 5
2124 try:
2125 exec_globals: Final = build_sandbox_globals()
2127 try:
2128 compiled: Final[CodeType] = compile_sandboxed(request.custom_code)
2129 exec(compiled, exec_globals) # noqa: S102
2130 except SyntaxError as e:
2131 return TestCustomCodeGuardrailResponse(
2132 success=False,
2133 error=f"Syntax error in custom code: {e}",
2134 error_type="compilation",
2135 )
2136 except Exception as e:
2137 return TestCustomCodeGuardrailResponse(
2138 success=False,
2139 error=f"Failed to compile custom code: {e}",
2140 error_type="compilation",
2141 )
2143 # Step 2: Verify apply_guardrail function exists
2144 if "apply_guardrail" not in exec_globals: 2144 ↛ 2152line 2144 didn't jump to line 2152 because the condition on line 2144 was always true
2145 return TestCustomCodeGuardrailResponse(
2146 success=False,
2147 error="Custom code must define an 'apply_guardrail' function. "
2148 "Expected signature: apply_guardrail(inputs, request_data, input_type)",
2149 error_type="compilation",
2150 )
2152 apply_fn: Final[object] = exec_globals["apply_guardrail"]
2153 if not callable(apply_fn):
2154 return TestCustomCodeGuardrailResponse(
2155 success=False,
2156 error="'apply_guardrail' must be a callable function",
2157 error_type="compilation",
2158 )
2160 # Step 3: Prepare test inputs
2161 test_inputs: Final = request.test_input
2162 if "texts" not in test_inputs:
2163 test_inputs["texts"] = []
2165 # Prepare mock request_data
2166 mock_request_data: Final = request.request_data or {}
2167 safe_request_data: Final = {
2168 "model": mock_request_data.get("model", "test-model"),
2169 "user_id": mock_request_data.get("user_id"),
2170 "team_id": mock_request_data.get("team_id"),
2171 "end_user_id": mock_request_data.get("end_user_id"),
2172 "metadata": mock_request_data.get("metadata", {}),
2173 }
2175 # Step 4: Execute the function with timeout protection
2177 def execute_guardrail() -> object:
2178 return apply_fn(test_inputs, safe_request_data, request.input_type)
2180 try:
2181 with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
2182 future: Final = executor.submit(execute_guardrail)
2183 try:
2184 result: Final = future.result(timeout=EXECUTION_TIMEOUT_SECONDS)
2185 except concurrent.futures.TimeoutError:
2186 return TestCustomCodeGuardrailResponse(
2187 success=False,
2188 error=f"Execution timeout: code took longer than {EXECUTION_TIMEOUT_SECONDS} seconds",
2189 error_type="execution",
2190 )
2191 except Exception as e:
2192 return TestCustomCodeGuardrailResponse(
2193 success=False,
2194 error=f"Execution error: {e}",
2195 error_type="execution",
2196 )
2198 # Step 5: Validate and return result
2199 if not isinstance(result, dict):
2200 return TestCustomCodeGuardrailResponse(
2201 success=True,
2202 result={
2203 "action": "allow",
2204 "warning": f"Expected dict result, got {type(result).__name__}. Treating as allow.",
2205 },
2206 )
2208 return TestCustomCodeGuardrailResponse(
2209 success=True,
2210 result=result,
2211 )
2213 except Exception as e:
2214 verbose_proxy_logger.exception("Error testing custom code guardrail: %s", e)
2215 return TestCustomCodeGuardrailResponse(
2216 success=False,
2217 error=f"Unexpected error: {e}",
2218 error_type="execution",
2219 )
2222def _resolve_guardrail_input_type(active_guardrail: CustomGuardrail, input_type: str) -> Literal["request", "response"]:
2223 """Return the effective input_type, auto-upgrading to 'response' for post_call guardrails."""
2224 if input_type == "request":
2225 hook: Final = getattr(active_guardrail, "event_hook", None)
2226 if hook == GuardrailEventHooks.post_call or hook == "post_call":
2227 return "response"
2228 return "response" if input_type == "response" else "request"
2231class _GuardrailLoggingObj(Protocol):
2232 call_type: str
2233 model_call_details: dict[str, object]
2235 @property
2236 def update_messages(self) -> "Callable[..., object]": ... 2236 ↛ exitline 2236 didn't return from function 'update_messages' because
2238 @property
2239 def async_success_handler(self) -> "Callable[..., Awaitable[object]]": ... 2239 ↛ exitline 2239 didn't return from function 'async_success_handler' because
2241 @property
2242 def success_handler(self) -> "Callable[..., object]": ... 2242 ↛ exitline 2242 didn't return from function 'success_handler' because
2245class _GuardrailProxyLogging(Protocol):
2246 @property
2247 def post_call_success_hook(self) -> "Callable[..., Awaitable[object]]": ... 2247 ↛ exitline 2247 didn't return from function 'post_call_success_hook' because
2250def _patch_logging_obj_for_guardrail(litellm_logging_obj: _GuardrailLoggingObj, request: ApplyGuardrailRequest) -> None:
2251 """Configure the logging object so Langfuse/OTEL extract input and output correctly."""
2252 litellm_logging_obj.call_type = "pass_through_endpoint"
2253 litellm_logging_obj.model_call_details["call_type"] = "pass_through_endpoint"
2254 litellm_logging_obj.update_messages(
2255 request.messages if request.messages else [{"role": "user", "content": request.text}]
2256 )
2259async def _emit_guardrail_success_logs(
2260 proxy_logging_obj: _GuardrailProxyLogging,
2261 litellm_logging_obj: _GuardrailLoggingObj | None,
2262 data: dict,
2263 user_api_key_dict: UserAPIKeyAuth,
2264 response: ApplyGuardrailResponse,
2265 start_time: datetime,
2266) -> ApplyGuardrailResponse:
2267 """Fire proxy and LiteLLM success hooks after a successful guardrail run.
2269 Each hook is wrapped defensively so a callback failure never prevents the
2270 caller from receiving the guardrail response. Returns the (possibly
2271 hook-modified) response.
2272 """
2273 from litellm.litellm_core_utils.thread_pool_executor import (
2274 executor as thread_pool_executor,
2275 )
2277 try:
2278 modified: Final = await proxy_logging_obj.post_call_success_hook(
2279 data=data,
2280 user_api_key_dict=user_api_key_dict,
2281 response=response,
2282 )
2283 if isinstance(modified, ApplyGuardrailResponse):
2284 response = modified
2285 except Exception:
2286 verbose_proxy_logger.exception("apply_guardrail: post_call_success_hook failed")
2288 # Build the logging payload after post_call_success_hook so that logged
2289 # data matches what the caller actually receives if the hook modified
2290 # the response.
2291 response_for_logging: Final = {"response": response.model_dump(exclude_none=True)}
2293 if litellm_logging_obj is not None:
2294 end_time: Final = datetime.now(timezone.utc)
2295 try:
2296 await litellm_logging_obj.async_success_handler(
2297 result=response_for_logging,
2298 start_time=start_time,
2299 end_time=end_time,
2300 cache_hit=False,
2301 )
2302 except Exception:
2303 verbose_proxy_logger.exception("apply_guardrail: async_success_handler failed")
2304 try:
2305 thread_pool_executor.submit(
2306 litellm_logging_obj.success_handler,
2307 response_for_logging,
2308 start_time,
2309 end_time,
2310 False,
2311 )
2312 except Exception:
2313 verbose_proxy_logger.exception("apply_guardrail: success_handler submit failed")
2315 return response
2318@router.post("/guardrails/apply_guardrail", response_model=ApplyGuardrailResponse)
2319@router.post("/apply_guardrail", response_model=ApplyGuardrailResponse)
2320async def apply_guardrail(
2321 fastapi_request: Request,
2322 request: ApplyGuardrailRequest,
2323 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2324):
2325 """
2326 Apply a guardrail to text input and return the processed result.
2328 This endpoint allows testing guardrails by applying them to custom text inputs.
2329 """
2330 import traceback
2332 from litellm.litellm_core_utils.thread_pool_executor import (
2333 executor as thread_pool_executor,
2334 )
2335 from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
2336 from litellm.proxy.proxy_server import (
2337 general_settings,
2338 proxy_config,
2339 proxy_logging_obj,
2340 version,
2341 )
2342 from litellm.proxy.utils import handle_exception_on_proxy
2344 data: dict = {
2345 "guardrail_name": request.guardrail_name,
2346 "input": [request.text],
2347 "messages": request.messages or [],
2348 "metadata": {"route": "/apply_guardrail"},
2349 }
2350 litellm_logging_obj = None
2351 start_time: Final = datetime.now(timezone.utc)
2353 from litellm.proxy.common_utils.registry_read_through import (
2354 get_initialized_guardrail_with_read_through,
2355 )
2357 try:
2358 active_guardrail: Final[CustomGuardrail | None] = await get_initialized_guardrail_with_read_through(
2359 guardrail_name=request.guardrail_name
2360 )
2361 if active_guardrail is None: 2361 ↛ 2367line 2361 didn't jump to line 2367 because the condition on line 2361 was always true
2362 raise HTTPException(
2363 status_code=404,
2364 detail=f"Guardrail '{request.guardrail_name}' not found. Please ensure the guardrail is configured in your LiteLLM proxy.",
2365 )
2367 request_processor: Final = ProxyBaseLLMRequestProcessing(data=data)
2368 (
2369 data,
2370 litellm_logging_obj,
2371 ) = await request_processor.common_processing_pre_call_logic(
2372 request=fastapi_request,
2373 general_settings=general_settings,
2374 user_api_key_dict=user_api_key_dict,
2375 version=version,
2376 proxy_logging_obj=proxy_logging_obj,
2377 proxy_config=proxy_config,
2378 route_type="apply_guardrail",
2379 )
2381 if litellm_logging_obj is not None:
2382 _patch_logging_obj_for_guardrail(litellm_logging_obj, request)
2384 request_data: Final[dict] = {
2385 **({"messages": request.messages} if request.messages is not None else {}),
2386 **({"metadata": request.metadata} if request.metadata is not None else {}),
2387 }
2388 _input_type: Final = _resolve_guardrail_input_type(active_guardrail, request.input_type)
2389 guardrailed_inputs: Final = await active_guardrail.apply_guardrail(
2390 inputs={"texts": [request.text]},
2391 request_data=request_data,
2392 input_type=_input_type,
2393 )
2394 response_text: Final = guardrailed_inputs.get("texts", [])
2395 response = ApplyGuardrailResponse(response_text=response_text[0] if response_text else request.text)
2396 except Exception as e:
2397 if litellm_logging_obj is not None and not isinstance(e, HTTPException): 2397 ↛ 2398line 2397 didn't jump to line 2398 because the condition on line 2397 was never true
2398 try:
2399 await litellm_logging_obj.async_failure_handler(
2400 exception=e,
2401 traceback_exception=traceback.format_exc(),
2402 )
2403 except Exception:
2404 verbose_proxy_logger.exception("apply_guardrail: async_failure_handler failed")
2405 try:
2406 thread_pool_executor.submit(
2407 litellm_logging_obj.failure_handler,
2408 e,
2409 traceback.format_exc(),
2410 )
2411 except Exception:
2412 verbose_proxy_logger.exception("apply_guardrail: failure_handler submit failed")
2413 try:
2414 transformed_exception: Final = await proxy_logging_obj.post_call_failure_hook(
2415 user_api_key_dict=user_api_key_dict,
2416 original_exception=e,
2417 request_data=data,
2418 )
2419 if isinstance(transformed_exception, Exception): 2419 ↛ 2420line 2419 didn't jump to line 2420 because the condition on line 2419 was never true
2420 e = transformed_exception
2421 except Exception:
2422 verbose_proxy_logger.exception("apply_guardrail: post_call_failure_hook failed")
2423 raise handle_exception_on_proxy(e)
2425 # Success logging outside except so a hook error never triggers failure handlers.
2426 response = await _emit_guardrail_success_logs(
2427 proxy_logging_obj=proxy_logging_obj,
2428 litellm_logging_obj=litellm_logging_obj,
2429 data=data,
2430 user_api_key_dict=user_api_key_dict,
2431 response=response,
2432 start_time=start_time,
2433 )
2434 return response
2437# Usage (dashboard) endpoints: overview, detail, logs
2438router.include_router(guardrails_usage_router)