feat(guardrails): wire apply_guardrail into proxy logging callbacks

[Sameerlite]

Summary

• Route /apply_guardrail through proxy pre/post hooks and LiteLLM success/failure logging handlers so configured callbacks (Langfuse, OTEL, etc.) run on guardrail-only requests.
• Set logging call type to pass_through_endpoint with messages/response payload so Langfuse records input and output on this endpoint.
• Add regression test covering logging pipeline invocation.

Fixes LIT-3139

Test plan

• poetry run pytest tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py::test_apply_guardrail_not_found -v
• poetry run pytest tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py::test_apply_guardrail_execution_error -v
• poetry run pytest tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py::test_apply_guardrail_invokes_logging_pipeline -v

---

Note

Low Risk
Observability and hook wiring around an existing guardrail test endpoint; hook failures are isolated so API responses still return.

Overview
The /apply_guardrail endpoint now follows the same proxy request and logging path as LLM routes instead of running guardrails in isolation.

It uses ProxyBaseLLMRequestProcessing.common_processing_pre_call_logic with a new apply_guardrail route type (registered in common_request_processing), accepts the FastAPI Request, and on success runs post_call_success_hook, LiteLLM async_success_handler, and sync success_handler. Failures go through the matching failure hooks. The logging object is patched so call_type is pass_through_endpoint and messages/response are shaped for integrations like Langfuse and OTEL. Post-call guardrails still get the correct input_type via _resolve_guardrail_input_type.

Tests add a shared mock_proxy_logging_ctx fixture and assert the logging pipeline is invoked on success.

^{Reviewed by Cursor Bugbot for commit 156e0d2. Bugbot is set up for automated code reviews on this repo. Configure here.}

[codecov]

Codecov Report

❌ Patch coverage is 75.80645% with 15 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
litellm/proxy/guardrails/guardrail_endpoints.py	75.80%	15 Missing ⚠️

📢 Thoughts on this report? Let us know!

[greptile-apps]

Greptile Summary

This PR wires the /apply_guardrail endpoint into the existing proxy pre/post hook pipeline so that configured callbacks (Langfuse, OTEL, etc.) observe guardrail-only requests the same way they observe LLM calls.

• Three helpers (_resolve_guardrail_input_type, _patch_logging_obj_for_guardrail, _emit_guardrail_success_logs) encapsulate the new logging plumbing; the success/failure hook separation is correctly placed outside the core try-block so a hook failure cannot flip a successful guardrail call into a failure trace.
• common_processing_pre_call_logic is registered for a new \"apply_guardrail\" route type, and all existing tests are updated with the required fastapi_request parameter and proxy-global mocks; a new test_apply_guardrail_invokes_logging_pipeline test verifies end-to-end invocation of the logging pipeline.

Confidence Score: 4/5

Safe to merge; the hook failures are fully isolated from guardrail responses and the existing success/failure callback separation is correctly implemented.

The GUARDRAIL_REGISTRY lookup currently happens before common_processing_pre_call_logic, so a 404 for an unknown guardrail produces no LiteLLM-level span in Langfuse/OTEL even though proxy_logging_obj.post_call_failure_hook fires. This is an observable gap between how 404 errors are traced here versus on other proxy endpoints, and whether it is intentional should be confirmed before merging.

litellm/proxy/guardrails/guardrail_endpoints.py — the ordering of the registry lookup relative to common_processing_pre_call_logic determines whether not-found errors produce a full trace

Important Files Changed

Filename	Overview
litellm/proxy/guardrails/guardrail_endpoints.py	Adds pre-call processing, success/failure logging hooks, and a new helper trio to the apply_guardrail endpoint; success/failure hook separation is correct but 404 errors skip pre-call processing and produce no LiteLLM-level trace
litellm/proxy/common_request_processing.py	Single-line addition registering 'apply_guardrail' as a valid route_type in common_processing_pre_call_logic; change is minimal and correct
tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py	Three existing tests updated with proper mocks for proxy globals, plus new test_apply_guardrail_invokes_logging_pipeline asserting the full logging pipeline
tests/enterprise/litellm_enterprise/proxy/guardrails/conftest.py	New conftest providing a shared mock_proxy_logging_ctx fixture that patches all proxy-server globals required by the updated apply_guardrail endpoint
tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py	Existing tests updated to add fastapi_request=Mock() parameter and wrap with mock_proxy_logging_ctx; no logic changes to assertions
tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py	Bedrock integration test updated to add fastapi_request=Mock() parameter and mock_proxy_logging_ctx wrapper; no assertion changes

_{Reviews (5): Last reviewed commit: "fix(guardrails): use update_messages to ..." | Re-trigger Greptile}

[Sameerlite]

Fixed both issues:

P1 — Success logging is now outside the except block entirely. post_call_success_hook, async_success_handler, and success_handler only run after the guardrail execution completes cleanly. A failure in any of those hooks no longer triggers the failure path.

P2 — model_call_details["messages"] now uses request.messages when provided, falling back to a single synthetic user message only when the caller sends text only.

@greptileai

[greptile-apps]

Both fixes look correct. P1 is properly addressed — the success handlers are now entirely outside the try/except block and only reachable on a clean execution path. P2 is handled with the appropriate fallback.

One remaining edge case worth noting: if post_call_success_hook or async_success_handler raises, the exception will bubble up to the caller as an unhandled error, returning a 500 even though the guardrail itself succeeded. The response was already computed — you may want to wrap the success logging block in a try/except that logs but doesn't re-raise, similar to how other endpoints handle hook failures gracefully. Whether that's worth doing here depends on how critical logging failures should be treated vs. delivering the response to the client.

Otherwise the changes look solid. The structure is clean, the test coverage is good, and the messages fallback logic is sensible.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution.
2 out of 3 committers have signed the CLA.

✅ Sameerlite
✅ mateo-berri
❌ cursoragent
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[Sameerlite]

@greptileai update the score

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 2 potential issues.

Autofix Details

Bugbot Autofix prepared fixes for both issues found in the latest run.

• ✅ Fixed: Failure in async_failure_handler silently skips post_call_failure_hook
  • Split the failure path into two independent try/except blocks so a failure in async_failure_handler no longer prevents post_call_failure_hook from running.
• ✅ Fixed: Logged response diverges from actually returned response
  • Moved response_for_logging construction into _emit_guardrail_success_logs after post_call_success_hook so logging reflects the (possibly hook-modified) response actually returned to the caller.

Preview (9b2e2675a2)

diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py
--- a/litellm/proxy/common_request_processing.py
+++ b/litellm/proxy/common_request_processing.py
@@ -839,6 +839,7 @@
             "aget_run",
             "acancel_run",
             "adelete_run",
+            "apply_guardrail",
         ],
         version: Optional[str] = None,
         user_model: Optional[str] = None,

diff --git a/litellm/proxy/guardrails/guardrail_endpoints.py b/litellm/proxy/guardrails/guardrail_endpoints.py
--- a/litellm/proxy/guardrails/guardrail_endpoints.py
+++ b/litellm/proxy/guardrails/guardrail_endpoints.py
@@ -10,7 +10,7 @@
 from typing import Any, Dict, List, Literal, Optional, Type, TypeVar, Union, cast
 from urllib.parse import urlparse
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
 
 from litellm.proxy.common_utils.path_utils import safe_join
@@ -2187,9 +2187,88 @@
         )
 
 
+def _resolve_guardrail_input_type(
+    active_guardrail: CustomGuardrail, input_type: str
+) -> Literal["request", "response"]:
+    """Return the effective input_type, auto-upgrading to 'response' for post_call guardrails."""
+    if input_type == "request":
+        hook = getattr(active_guardrail, "event_hook", None)
+        if hook == GuardrailEventHooks.post_call or hook == "post_call":
+            return "response"
+    return "response" if input_type == "response" else "request"
+
+
+def _patch_logging_obj_for_guardrail(
+    litellm_logging_obj: Any, request: ApplyGuardrailRequest
+) -> None:
+    """Configure the logging object so Langfuse/OTEL extract input and output correctly."""
+    litellm_logging_obj.call_type = "pass_through_endpoint"
+    litellm_logging_obj.model_call_details["call_type"] = "pass_through_endpoint"
+    litellm_logging_obj.model_call_details["messages"] = (
+        request.messages if request.messages else [{"role": "user", "content": request.text}]
+    )
+
+
+async def _emit_guardrail_success_logs(
+    proxy_logging_obj: Any,
+    litellm_logging_obj: Any,
+    data: dict,
+    user_api_key_dict: UserAPIKeyAuth,
+    response: ApplyGuardrailResponse,
+    start_time: datetime,
+) -> ApplyGuardrailResponse:
+    """Fire proxy and LiteLLM success hooks after a successful guardrail run.
+
+    Each hook is wrapped defensively so a callback failure never prevents the
+    caller from receiving the guardrail response.  Returns the (possibly
+    hook-modified) response.
+    """
+    from litellm.litellm_core_utils.thread_pool_executor import (
+        executor as thread_pool_executor,
+    )
+
+    try:
+        modified = await proxy_logging_obj.post_call_success_hook(
+            data=data,
+            user_api_key_dict=user_api_key_dict,
+            response=response,
+        )
+        if isinstance(modified, ApplyGuardrailResponse):
+            response = modified
+    except Exception:
+        verbose_proxy_logger.exception("apply_guardrail: post_call_success_hook failed")
+
+    # Build the logging payload after post_call_success_hook so that logged
+    # data matches what the caller actually receives if the hook modified
+    # the response.
+    response_for_logging = {"response": response.model_dump(exclude_none=True)}
+
+    if litellm_logging_obj is not None:
+        end_time = datetime.now(timezone.utc)
+        try:
+            await litellm_logging_obj.async_success_handler(
+                result=response_for_logging,
+                start_time=start_time,
+                end_time=end_time,
+                cache_hit=False,
+            )
+            thread_pool_executor.submit(
+                litellm_logging_obj.success_handler,
+                response_for_logging,
+                start_time,
+                end_time,
+                False,
+            )
+        except Exception:
+            verbose_proxy_logger.exception("apply_guardrail: success logging failed")
+
+    return response
+
+
 @router.post("/guardrails/apply_guardrail", response_model=ApplyGuardrailResponse)
 @router.post("/apply_guardrail", response_model=ApplyGuardrailResponse)
 async def apply_guardrail(
+    fastapi_request: Request,
     request: ApplyGuardrailRequest,
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
 ):
@@ -2198,8 +2277,29 @@
 
     This endpoint allows testing guardrails by applying them to custom text inputs.
     """
+    import traceback
+
+    from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
+    from litellm.litellm_core_utils.thread_pool_executor import (
+        executor as thread_pool_executor,
+    )
+    from litellm.proxy.proxy_server import (
+        general_settings,
+        proxy_config,
+        proxy_logging_obj,
+        version,
+    )
     from litellm.proxy.utils import handle_exception_on_proxy
 
+    data: dict = {
+        "guardrail_name": request.guardrail_name,
+        "input": [request.text],
+        "messages": request.messages or [],
+        "metadata": {"route": "/apply_guardrail"},
+    }
+    litellm_logging_obj = None
+    start_time = datetime.now(timezone.utc)
+
     try:
         active_guardrail: Optional[CustomGuardrail] = (
             GUARDRAIL_REGISTRY.get_initialized_guardrail_callback(
@@ -2212,37 +2312,70 @@
                 detail=f"Guardrail '{request.guardrail_name}' not found. Please ensure the guardrail is configured in your LiteLLM proxy.",
             )
 
-        request_data: dict = {}
-        if request.messages:
-            request_data["messages"] = request.messages
+        request_processor = ProxyBaseLLMRequestProcessing(data=data)
+        data, litellm_logging_obj = await request_processor.common_processing_pre_call_logic(
+            request=fastapi_request,
+            general_settings=general_settings,
+            user_api_key_dict=user_api_key_dict,
+            version=version,
+            proxy_logging_obj=proxy_logging_obj,
+            proxy_config=proxy_config,
+            route_type="apply_guardrail",
+        )
 
-        # Auto-detect input_type: if the caller didn't specify "response" but the
-        # guardrail only runs post_call (e.g. LLM-as-a-judge), use "response" so
-        # the test actually exercises the guardrail logic.
-        from litellm.types.guardrails import GuardrailEventHooks
+        if litellm_logging_obj is not None:
+            _patch_logging_obj_for_guardrail(litellm_logging_obj, request)
 
-        resolved_input_type = request.input_type
-        if resolved_input_type == "request":
-            hook = getattr(active_guardrail, "event_hook", None)
-            if hook == GuardrailEventHooks.post_call or hook == "post_call":
-                resolved_input_type = "response"
-
-        _input_type: Literal["request", "response"] = (
-            "response" if resolved_input_type == "response" else "request"
-        )
+        request_data: dict = {"messages": request.messages} if request.messages else {}
+        _input_type = _resolve_guardrail_input_type(active_guardrail, request.input_type)
         guardrailed_inputs = await active_guardrail.apply_guardrail(
             inputs={"texts": [request.text]},
             request_data=request_data,
             input_type=_input_type,
         )
         response_text = guardrailed_inputs.get("texts", [])
-
-        return ApplyGuardrailResponse(
+        response = ApplyGuardrailResponse(
             response_text=response_text[0] if response_text else request.text
         )
     except Exception as e:
+        try:
+            if litellm_logging_obj is not None and not isinstance(e, HTTPException):
+                await litellm_logging_obj.async_failure_handler(
+                    exception=e,
+                    traceback_exception=traceback.format_exc(),
+                )
+                thread_pool_executor.submit(
+                    litellm_logging_obj.failure_handler,
+                    e,
+                    traceback.format_exc(),
+                )
+        except Exception:
+            verbose_proxy_logger.exception(
+                "apply_guardrail: async_failure_handler failed"
+            )
+        try:
+            await proxy_logging_obj.post_call_failure_hook(
+                user_api_key_dict=user_api_key_dict,
+                original_exception=e,
+                request_data=data,
+            )
+        except Exception:
+            verbose_proxy_logger.exception(
+                "apply_guardrail: post_call_failure_hook failed"
+            )
         raise handle_exception_on_proxy(e)
 
+    # Success logging outside except so a hook error never triggers failure handlers.
+    response = await _emit_guardrail_success_logs(
+        proxy_logging_obj=proxy_logging_obj,
+        litellm_logging_obj=litellm_logging_obj,
+        data=data,
+        user_api_key_dict=user_api_key_dict,
+        response=response,
+        start_time=start_time,
+    )
+    return response
 
+
 # Usage (dashboard) endpoints: overview, detail, logs
 router.include_router(guardrails_usage_router)

diff --git a/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py b/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
--- a/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
+++ b/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
@@ -1159,7 +1159,11 @@
 
     # Call endpoint and expect ProxyException
     with pytest.raises(ProxyException) as exc_info:
-        await apply_guardrail(request=request, user_api_key_dict=mock_user_auth)
+        await apply_guardrail(
+            fastapi_request=mocker.Mock(),
+            request=request,
+            user_api_key_dict=mock_user_auth,
+        )
 
     # Verify error details
     assert str(exc_info.value.code) == "404"
@@ -1186,6 +1190,25 @@
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_registry
     )
 
+    mock_logging_obj = mocker.Mock()
+    mock_logging_obj.async_failure_handler = AsyncMock()
+    mock_logging_obj.model_call_details = {}
+    mock_processor = mocker.Mock()
+    mock_processor.common_processing_pre_call_logic = AsyncMock(
+        return_value=({"guardrail_name": "test-guardrail"}, mock_logging_obj)
+    )
+    mocker.patch(
+        "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing",
+        return_value=mock_processor,
+    )
+    mock_proxy_logging = mocker.Mock()
+    mock_proxy_logging.post_call_failure_hook = AsyncMock()
+    mocker.patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging)
+    mocker.patch("litellm.proxy.proxy_server.general_settings", {})
+    mocker.patch("litellm.proxy.proxy_server.proxy_config", mocker.Mock())
+    mocker.patch("litellm.proxy.proxy_server.version", "test")
+    mocker.patch("litellm.litellm_core_utils.thread_pool_executor.executor")
+
     # Create request
     request = ApplyGuardrailRequest(
         guardrail_name="test-guardrail", text="Test input text with forbidden content"
@@ -1196,13 +1219,71 @@
 
     # Call endpoint and expect ProxyException
     with pytest.raises(ProxyException) as exc_info:
-        await apply_guardrail(request=request, user_api_key_dict=mock_user_auth)
+        await apply_guardrail(
+            fastapi_request=mocker.Mock(),
+            request=request,
+            user_api_key_dict=mock_user_auth,
+        )
 
     # Verify error is properly handled
     assert "Bedrock guardrail failed" in str(exc_info.value.message)
 
 
 @pytest.mark.asyncio
+async def test_apply_guardrail_invokes_logging_pipeline(mocker):
+    mock_guardrail = mocker.Mock()
+    mock_guardrail.apply_guardrail = AsyncMock(return_value={"texts": ["masked"]})
+
+    mock_registry = mocker.Mock()
+    mock_registry.get_initialized_guardrail_callback.return_value = mock_guardrail
+    mocker.patch(
+        "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_registry
+    )
+
+    mock_logging_obj = mocker.Mock()
+    mock_logging_obj.async_success_handler = AsyncMock()
+    mock_logging_obj.model_call_details = {}
+    mock_processor = mocker.Mock()
+    mock_processor.common_processing_pre_call_logic = AsyncMock(
+        return_value=({"guardrail_name": "test-guardrail"}, mock_logging_obj)
+    )
+    mocker.patch(
+        "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing",
+        return_value=mock_processor,
+    )
+
+    mock_proxy_logging = mocker.Mock()
+    mock_proxy_logging.post_call_success_hook = AsyncMock()
+    mocker.patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging)
+    mocker.patch("litellm.proxy.proxy_server.general_settings", {})
+    mocker.patch("litellm.proxy.proxy_server.proxy_config", mocker.Mock())
+    mocker.patch("litellm.proxy.proxy_server.version", "test")
+    mock_executor = mocker.Mock()
+    mocker.patch(
+        "litellm.litellm_core_utils.thread_pool_executor.executor", mock_executor
+    )
+
+    request = ApplyGuardrailRequest(
+        guardrail_name="test-guardrail", text="hello@example.com"
+    )
+    response = await apply_guardrail(
+        fastapi_request=mocker.Mock(),
+        request=request,
+        user_api_key_dict=UserAPIKeyAuth(),
+    )
+
+    assert response.response_text == "masked"
+    mock_processor.common_processing_pre_call_logic.assert_awaited_once()
+    mock_proxy_logging.post_call_success_hook.assert_awaited_once()
+    mock_logging_obj.async_success_handler.assert_awaited_once()
+    assert mock_logging_obj.call_type == "pass_through_endpoint"
+    mock_executor.submit.assert_called_once()
+    assert mock_logging_obj.async_success_handler.await_args.kwargs["result"] == {
+        "response": {"response_text": "masked"}
+    }
+
+
+@pytest.mark.asyncio
 async def test_get_guardrail_info_endpoint_config_guardrail(mocker):
     """
     Test get_guardrail_info endpoint returns proper response when guardrail is found in config.

_{You can send follow-ups to the cloud agent here.}

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Sync handler silently skipped when async handler fails
  • Split each sync and async handler call into its own try/except in both _emit_guardrail_success_logs and the failure path of apply_guardrail so a failure in one channel no longer prevents the other from running.

Preview (2a400e80e0)

diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py
--- a/litellm/proxy/common_request_processing.py
+++ b/litellm/proxy/common_request_processing.py
@@ -839,6 +839,7 @@
             "aget_run",
             "acancel_run",
             "adelete_run",
+            "apply_guardrail",
         ],
         version: Optional[str] = None,
         user_model: Optional[str] = None,

diff --git a/litellm/proxy/guardrails/guardrail_endpoints.py b/litellm/proxy/guardrails/guardrail_endpoints.py
--- a/litellm/proxy/guardrails/guardrail_endpoints.py
+++ b/litellm/proxy/guardrails/guardrail_endpoints.py
@@ -10,7 +10,7 @@
 from typing import Any, Dict, List, Literal, Optional, Type, TypeVar, Union, cast
 from urllib.parse import urlparse
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
 
 from litellm.proxy.common_utils.path_utils import safe_join
@@ -2187,9 +2187,97 @@
         )
 
 
+def _resolve_guardrail_input_type(
+    active_guardrail: CustomGuardrail, input_type: str
+) -> Literal["request", "response"]:
+    """Return the effective input_type, auto-upgrading to 'response' for post_call guardrails."""
+    if input_type == "request":
+        hook = getattr(active_guardrail, "event_hook", None)
+        if hook == GuardrailEventHooks.post_call or hook == "post_call":
+            return "response"
+    return "response" if input_type == "response" else "request"
+
+
+def _patch_logging_obj_for_guardrail(
+    litellm_logging_obj: Any, request: ApplyGuardrailRequest
+) -> None:
+    """Configure the logging object so Langfuse/OTEL extract input and output correctly."""
+    litellm_logging_obj.call_type = "pass_through_endpoint"
+    litellm_logging_obj.model_call_details["call_type"] = "pass_through_endpoint"
+    litellm_logging_obj.model_call_details["messages"] = (
+        request.messages
+        if request.messages
+        else [{"role": "user", "content": request.text}]
+    )
+
+
+async def _emit_guardrail_success_logs(
+    proxy_logging_obj: Any,
+    litellm_logging_obj: Any,
+    data: dict,
+    user_api_key_dict: UserAPIKeyAuth,
+    response: ApplyGuardrailResponse,
+    start_time: datetime,
+) -> ApplyGuardrailResponse:
+    """Fire proxy and LiteLLM success hooks after a successful guardrail run.
+
+    Each hook is wrapped defensively so a callback failure never prevents the
+    caller from receiving the guardrail response.  Returns the (possibly
+    hook-modified) response.
+    """
+    from litellm.litellm_core_utils.thread_pool_executor import (
+        executor as thread_pool_executor,
+    )
+
+    try:
+        modified = await proxy_logging_obj.post_call_success_hook(
+            data=data,
+            user_api_key_dict=user_api_key_dict,
+            response=response,
+        )
+        if isinstance(modified, ApplyGuardrailResponse):
+            response = modified
+    except Exception:
+        verbose_proxy_logger.exception("apply_guardrail: post_call_success_hook failed")
+
+    # Build the logging payload after post_call_success_hook so that logged
+    # data matches what the caller actually receives if the hook modified
+    # the response.
+    response_for_logging = {"response": response.model_dump(exclude_none=True)}
+
+    if litellm_logging_obj is not None:
+        end_time = datetime.now(timezone.utc)
+        try:
+            await litellm_logging_obj.async_success_handler(
+                result=response_for_logging,
+                start_time=start_time,
+                end_time=end_time,
+                cache_hit=False,
+            )
+        except Exception:
+            verbose_proxy_logger.exception(
+                "apply_guardrail: async_success_handler failed"
+            )
+        try:
+            thread_pool_executor.submit(
+                litellm_logging_obj.success_handler,
+                response_for_logging,
+                start_time,
+                end_time,
+                False,
+            )
+        except Exception:
+            verbose_proxy_logger.exception(
+                "apply_guardrail: success_handler submit failed"
+            )
+
+    return response
+
+
 @router.post("/guardrails/apply_guardrail", response_model=ApplyGuardrailResponse)
 @router.post("/apply_guardrail", response_model=ApplyGuardrailResponse)
 async def apply_guardrail(
+    fastapi_request: Request,
     request: ApplyGuardrailRequest,
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
 ):
@@ -2198,8 +2286,29 @@
 
     This endpoint allows testing guardrails by applying them to custom text inputs.
     """
+    import traceback
+
+    from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
+    from litellm.litellm_core_utils.thread_pool_executor import (
+        executor as thread_pool_executor,
+    )
+    from litellm.proxy.proxy_server import (
+        general_settings,
+        proxy_config,
+        proxy_logging_obj,
+        version,
+    )
     from litellm.proxy.utils import handle_exception_on_proxy
 
+    data: dict = {
+        "guardrail_name": request.guardrail_name,
+        "input": [request.text],
+        "messages": request.messages or [],
+        "metadata": {"route": "/apply_guardrail"},
+    }
+    litellm_logging_obj = None
+    start_time = datetime.now(timezone.utc)
+
     try:
         active_guardrail: Optional[CustomGuardrail] = (
             GUARDRAIL_REGISTRY.get_initialized_guardrail_callback(
@@ -2212,23 +2321,25 @@
                 detail=f"Guardrail '{request.guardrail_name}' not found. Please ensure the guardrail is configured in your LiteLLM proxy.",
             )
 
-        request_data: dict = {}
-        if request.messages:
-            request_data["messages"] = request.messages
+        request_processor = ProxyBaseLLMRequestProcessing(data=data)
+        data, litellm_logging_obj = (
+            await request_processor.common_processing_pre_call_logic(
+                request=fastapi_request,
+                general_settings=general_settings,
+                user_api_key_dict=user_api_key_dict,
+                version=version,
+                proxy_logging_obj=proxy_logging_obj,
+                proxy_config=proxy_config,
+                route_type="apply_guardrail",
+            )
+        )
 
-        # Auto-detect input_type: if the caller didn't specify "response" but the
-        # guardrail only runs post_call (e.g. LLM-as-a-judge), use "response" so
-        # the test actually exercises the guardrail logic.
-        from litellm.types.guardrails import GuardrailEventHooks
+        if litellm_logging_obj is not None:
+            _patch_logging_obj_for_guardrail(litellm_logging_obj, request)
 
-        resolved_input_type = request.input_type
-        if resolved_input_type == "request":
-            hook = getattr(active_guardrail, "event_hook", None)
-            if hook == GuardrailEventHooks.post_call or hook == "post_call":
-                resolved_input_type = "response"
-
-        _input_type: Literal["request", "response"] = (
-            "response" if resolved_input_type == "response" else "request"
+        request_data: dict = {"messages": request.messages} if request.messages else {}
+        _input_type = _resolve_guardrail_input_type(
+            active_guardrail, request.input_type
         )
         guardrailed_inputs = await active_guardrail.apply_guardrail(
             inputs={"texts": [request.text]},
@@ -2236,13 +2347,55 @@
             input_type=_input_type,
         )
         response_text = guardrailed_inputs.get("texts", [])
-
-        return ApplyGuardrailResponse(
+        response = ApplyGuardrailResponse(
             response_text=response_text[0] if response_text else request.text
         )
     except Exception as e:
+        if litellm_logging_obj is not None and not isinstance(e, HTTPException):
+            try:
+                await litellm_logging_obj.async_failure_handler(
+                    exception=e,
+                    traceback_exception=traceback.format_exc(),
+                )
+            except Exception:
+                verbose_proxy_logger.exception(
+                    "apply_guardrail: async_failure_handler failed"
+                )
+            try:
+                thread_pool_executor.submit(
+                    litellm_logging_obj.failure_handler,
+                    e,
+                    traceback.format_exc(),
+                )
+            except Exception:
+                verbose_proxy_logger.exception(
+                    "apply_guardrail: failure_handler submit failed"
+                )
+        try:
+            transformed_exception = await proxy_logging_obj.post_call_failure_hook(
+                user_api_key_dict=user_api_key_dict,
+                original_exception=e,
+                request_data=data,
+            )
+            if transformed_exception is not None:
+                e = transformed_exception
+        except Exception:
+            verbose_proxy_logger.exception(
+                "apply_guardrail: post_call_failure_hook failed"
+            )
         raise handle_exception_on_proxy(e)
 
+    # Success logging outside except so a hook error never triggers failure handlers.
+    response = await _emit_guardrail_success_logs(
+        proxy_logging_obj=proxy_logging_obj,
+        litellm_logging_obj=litellm_logging_obj,
+        data=data,
+        user_api_key_dict=user_api_key_dict,
+        response=response,
+        start_time=start_time,
+    )
+    return response
 
+
 # Usage (dashboard) endpoints: overview, detail, logs
 router.include_router(guardrails_usage_router)

diff --git a/tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py b/tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py
--- a/tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py
+++ b/tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py
@@ -4,7 +4,8 @@
 
 import os
 import sys
-from unittest.mock import AsyncMock, Mock, patch
+from contextlib import contextmanager
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
 
@@ -17,6 +18,38 @@
 from litellm.types.guardrails import ApplyGuardrailRequest, ApplyGuardrailResponse
 
 
+@contextmanager
+def _mock_proxy_logging():
+    """Patch the proxy-server globals that apply_guardrail imports at call time."""
+    mock_proxy_logging = MagicMock()
+    mock_proxy_logging.post_call_success_hook = AsyncMock(return_value=None)
+    mock_proxy_logging.post_call_failure_hook = AsyncMock(return_value=None)
+    mock_logging_obj = MagicMock()
+    mock_logging_obj.async_success_handler = AsyncMock(return_value=None)
+    mock_logging_obj.async_failure_handler = AsyncMock(return_value=None)
+    mock_logging_obj.success_handler = MagicMock(return_value=None)
+    mock_logging_obj.failure_handler = MagicMock(return_value=None)
+    mock_logging_obj.model_call_details = {}
+
+    with patch(
+        "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing"
+    ) as mock_proc_cls, patch(
+        "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging
+    ), patch(
+        "litellm.proxy.proxy_server.general_settings", {}
+    ), patch(
+        "litellm.proxy.proxy_server.proxy_config", MagicMock()
+    ), patch(
+        "litellm.proxy.proxy_server.version", "0.0.0"
+    ):
+        mock_proc = MagicMock()
+        mock_proc.common_processing_pre_call_logic = AsyncMock(
+            return_value=({}, mock_logging_obj)
+        )
+        mock_proc_cls.return_value = mock_proc
+        yield mock_proxy_logging
+
+
 @pytest.mark.asyncio
 async def test_apply_guardrail_endpoint_returns_correct_response():
     """Test that apply_guardrail endpoint returns ApplyGuardrailResponse object"""
@@ -25,7 +58,7 @@
     # Mock the guardrail registry
     with patch(
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    ) as mock_registry, _mock_proxy_logging():
         # Create a mock guardrail
         mock_guardrail = Mock(spec=CustomGuardrail)
         # Apply guardrail returns GenericGuardrailAPIInputs (dict with texts key)
@@ -49,7 +82,9 @@
 
         # Call the endpoint
         response = await apply_guardrail(
-            request=request, user_api_key_dict=user_api_key_dict
+            fastapi_request=Mock(),
+            request=request,
+            user_api_key_dict=user_api_key_dict,
         )
 
         # Verify the response is of the correct type
@@ -73,7 +108,7 @@
     # Mock the guardrail registry to return None
     with patch(
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    ) as mock_registry, _mock_proxy_logging():
         mock_registry.get_initialized_guardrail_callback.return_value = None
 
         # Create the request
@@ -86,7 +121,11 @@
 
         # Verify exception is raised
         with pytest.raises(ProxyException) as exc_info:
-            await apply_guardrail(request=request, user_api_key_dict=user_api_key_dict)
+            await apply_guardrail(
+                fastapi_request=Mock(),
+                request=request,
+                user_api_key_dict=user_api_key_dict,
+            )
 
         assert "non-existent-guardrail" in exc_info.value.message
         assert "not found" in exc_info.value.message
@@ -100,7 +139,7 @@
     # Mock the guardrail registry
     with patch(
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    ) as mock_registry, _mock_proxy_logging():
         # Create a mock guardrail that simulates Presidio behavior
         mock_guardrail = Mock(spec=CustomGuardrail)
         # Simulate masking PII entities - returns GenericGuardrailAPIInputs (dict with texts key)
@@ -124,7 +163,9 @@
 
         # Call the endpoint
         response = await apply_guardrail(
-            request=request, user_api_key_dict=user_api_key_dict
+            fastapi_request=Mock(),
+            request=request,
+            user_api_key_dict=user_api_key_dict,
         )
 
         # Verify the response is of the correct type
@@ -145,7 +186,7 @@
     # Mock the guardrail registry
     with patch(
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    ) as mock_registry, _mock_proxy_logging():
         # Create a mock guardrail
         mock_guardrail = Mock(spec=CustomGuardrail)
         # Returns GenericGuardrailAPIInputs (dict with texts key)
@@ -166,7 +207,9 @@
 
         # Call the endpoint
         response = await apply_guardrail(
-            request=request, user_api_key_dict=user_api_key_dict
+            fastapi_request=Mock(),
+            request=request,
+            user_api_key_dict=user_api_key_dict,
         )
 
         # Verify the response is of the correct type

diff --git a/tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py b/tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py
--- a/tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py
+++ b/tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py
@@ -4,7 +4,8 @@
 
 import os
 import sys
-from unittest.mock import AsyncMock, patch
+from contextlib import contextmanager
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
 
@@ -16,6 +17,38 @@
 from litellm.types.guardrails import ApplyGuardrailRequest, ApplyGuardrailResponse
 
 
+@contextmanager
+def _mock_proxy_logging():
+    """Patch the proxy-server globals that apply_guardrail imports at call time."""
+    mock_proxy_logging = MagicMock()
+    mock_proxy_logging.post_call_success_hook = AsyncMock(return_value=None)
+    mock_proxy_logging.post_call_failure_hook = AsyncMock(return_value=None)
+    mock_logging_obj = MagicMock()
+    mock_logging_obj.async_success_handler = AsyncMock(return_value=None)
+    mock_logging_obj.async_failure_handler = AsyncMock(return_value=None)
+    mock_logging_obj.success_handler = MagicMock(return_value=None)
+    mock_logging_obj.failure_handler = MagicMock(return_value=None)
+    mock_logging_obj.model_call_details = {}
+
+    with patch(
+        "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing"
+    ) as mock_proc_cls, patch(
+        "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging
+    ), patch(
+        "litellm.proxy.proxy_server.general_settings", {}
+    ), patch(
+        "litellm.proxy.proxy_server.proxy_config", MagicMock()
+    ), patch(
+        "litellm.proxy.proxy_server.version", "0.0.0"
+    ):
+        mock_proc = MagicMock()
+        mock_proc.common_processing_pre_call_logic = AsyncMock(
+            return_value=({}, mock_logging_obj)
+        )
+        mock_proc_cls.return_value = mock_proc
+        yield mock_proxy_logging
+
+
 @pytest.mark.asyncio
 async def test_bedrock_apply_guardrail_success():
     """Test that Bedrock guardrail apply_guardrail method works correctly"""
@@ -167,7 +200,7 @@
     # Mock the guardrail registry
     with patch(
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    ) as mock_registry, _mock_proxy_logging():
         # Mock the make_bedrock_api_request method
         with patch.object(
             guardrail, "make_bedrock_api_request", new_callable=AsyncMock
@@ -194,7 +227,9 @@
 
             # Call the endpoint
             response = await apply_guardrail(
-                request=request, user_api_key_dict=user_api_key_dict
+                fastapi_request=Mock(),
+                request=request,
+                user_api_key_dict=user_api_key_dict,
             )
 
             # Verify the response

diff --git a/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py b/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
--- a/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
+++ b/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
@@ -1159,7 +1159,11 @@
 
     # Call endpoint and expect ProxyException
     with pytest.raises(ProxyException) as exc_info:
-        await apply_guardrail(request=request, user_api_key_dict=mock_user_auth)
+        await apply_guardrail(
+            fastapi_request=mocker.Mock(),
+            request=request,
+            user_api_key_dict=mock_user_auth,
+        )
 
     # Verify error details
     assert str(exc_info.value.code) == "404"
@@ -1186,6 +1190,25 @@
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_registry
     )
 
+    mock_logging_obj = mocker.Mock()
+    mock_logging_obj.async_failure_handler = AsyncMock()
+    mock_logging_obj.model_call_details = {}
+    mock_processor = mocker.Mock()
+    mock_processor.common_processing_pre_call_logic = AsyncMock(
+        return_value=({"guardrail_name": "test-guardrail"}, mock_logging_obj)
+    )
+    mocker.patch(
+        "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing",
+        return_value=mock_processor,
+    )
+    mock_proxy_logging = mocker.Mock()
+    mock_proxy_logging.post_call_failure_hook = AsyncMock()
+    mocker.patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging)
+    mocker.patch("litellm.proxy.proxy_server.general_settings", {})
+    mocker.patch("litellm.proxy.proxy_server.proxy_config", mocker.Mock())
+    mocker.patch("litellm.proxy.proxy_server.version", "test")
+    mocker.patch("litellm.litellm_core_utils.thread_pool_executor.executor")
+
     # Create request
     request = ApplyGuardrailRequest(
         guardrail_name="test-guardrail", text="Test input text with forbidden content"
@@ -1196,13 +1219,71 @@
 
     # Call endpoint and expect ProxyException
     with pytest.raises(ProxyException) as exc_info:
-        await apply_guardrail(request=request, user_api_key_dict=mock_user_auth)
+        await apply_guardrail(
+            fastapi_request=mocker.Mock(),
+            request=request,
+            user_api_key_dict=mock_user_auth,
+        )
 
     # Verify error is properly handled
     assert "Bedrock guardrail failed" in str(exc_info.value.message)
 
 
 @pytest.mark.asyncio
+async def test_apply_guardrail_invokes_logging_pipeline(mocker):
+    mock_guardrail = mocker.Mock()
+    mock_guardrail.apply_guardrail = AsyncMock(return_value={"texts": ["masked"]})
+
+    mock_registry = mocker.Mock()
+    mock_registry.get_initialized_guardrail_callback.return_value = mock_guardrail
+    mocker.patch(
+        "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_registry
+    )
+
+    mock_logging_obj = mocker.Mock()
+    mock_logging_obj.async_success_handler = AsyncMock()
+    mock_logging_obj.model_call_details = {}
+    mock_processor = mocker.Mock()
+    mock_processor.common_processing_pre_call_logic = AsyncMock(
+        return_value=({"guardrail_name": "test-guardrail"}, mock_logging_obj)
+    )
+    mocker.patch(
+        "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing",
+        return_value=mock_processor,
+    )
+
+    mock_proxy_logging = mocker.Mock()
+    mock_proxy_logging.post_call_success_hook = AsyncMock()
+    mocker.patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging)
+    mocker.patch("litellm.proxy.proxy_server.general_settings", {})
+    mocker.patch("litellm.proxy.proxy_server.proxy_config", mocker.Mock())
+    mocker.patch("litellm.proxy.proxy_server.version", "test")
+    mock_executor = mocker.Mock()
+    mocker.patch(
+        "litellm.litellm_core_utils.thread_pool_executor.executor", mock_executor
+    )
+
+    request = ApplyGuardrailRequest(
+        guardrail_name="test-guardrail", text="hello@example.com"
+    )
+    response = await apply_guardrail(
+        fastapi_request=mocker.Mock(),
+        request=request,
+        user_api_key_dict=UserAPIKeyAuth(),
+    )
+
+    assert response.response_text == "masked"
+    mock_processor.common_processing_pre_call_logic.assert_awaited_once()
+    mock_proxy_logging.post_call_success_hook.assert_awaited_once()
+    mock_logging_obj.async_success_handler.assert_awaited_once()
+    assert mock_logging_obj.call_type == "pass_through_endpoint"
+    mock_executor.submit.assert_called_once()
+    assert mock_logging_obj.async_success_handler.await_args.kwargs["result"] == {
+        "response": {"response_text": "masked"}
+    }
+
+
+@pytest.mark.asyncio
 async def test_get_guardrail_info_endpoint_config_guardrail(mocker):
     """
     Test get_guardrail_info endpoint returns proper response when guardrail is found in config.

_{You can send follow-ups to the cloud agent here.}

[Sameerlite]

@greptileai

[mateo-berri]

@greptileai

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Logging object messages attribute not updated, causing inconsistency
  • Replaced the direct write to model_call_details['messages'] with a call to the existing update_messages() API, which updates both self.messages and self.model_call_details['messages'] in sync.

Preview (156e0d2c1a)

diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py
--- a/litellm/proxy/common_request_processing.py
+++ b/litellm/proxy/common_request_processing.py
@@ -839,6 +839,7 @@
             "aget_run",
             "acancel_run",
             "adelete_run",
+            "apply_guardrail",
         ],
         version: Optional[str] = None,
         user_model: Optional[str] = None,

diff --git a/litellm/proxy/guardrails/guardrail_endpoints.py b/litellm/proxy/guardrails/guardrail_endpoints.py
--- a/litellm/proxy/guardrails/guardrail_endpoints.py
+++ b/litellm/proxy/guardrails/guardrail_endpoints.py
@@ -10,7 +10,7 @@
 from typing import Any, Dict, List, Literal, Optional, Type, TypeVar, Union, cast
 from urllib.parse import urlparse
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
 
 from litellm.proxy.common_utils.path_utils import safe_join
@@ -2187,9 +2187,97 @@
         )
 
 
+def _resolve_guardrail_input_type(
+    active_guardrail: CustomGuardrail, input_type: str
+) -> Literal["request", "response"]:
+    """Return the effective input_type, auto-upgrading to 'response' for post_call guardrails."""
+    if input_type == "request":
+        hook = getattr(active_guardrail, "event_hook", None)
+        if hook == GuardrailEventHooks.post_call or hook == "post_call":
+            return "response"
+    return "response" if input_type == "response" else "request"
+
+
+def _patch_logging_obj_for_guardrail(
+    litellm_logging_obj: Any, request: ApplyGuardrailRequest
+) -> None:
+    """Configure the logging object so Langfuse/OTEL extract input and output correctly."""
+    litellm_logging_obj.call_type = "pass_through_endpoint"
+    litellm_logging_obj.model_call_details["call_type"] = "pass_through_endpoint"
+    litellm_logging_obj.update_messages(
+        request.messages
+        if request.messages
+        else [{"role": "user", "content": request.text}]
+    )
+
+
+async def _emit_guardrail_success_logs(
+    proxy_logging_obj: Any,
+    litellm_logging_obj: Any,
+    data: dict,
+    user_api_key_dict: UserAPIKeyAuth,
+    response: ApplyGuardrailResponse,
+    start_time: datetime,
+) -> ApplyGuardrailResponse:
+    """Fire proxy and LiteLLM success hooks after a successful guardrail run.
+
+    Each hook is wrapped defensively so a callback failure never prevents the
+    caller from receiving the guardrail response.  Returns the (possibly
+    hook-modified) response.
+    """
+    from litellm.litellm_core_utils.thread_pool_executor import (
+        executor as thread_pool_executor,
+    )
+
+    try:
+        modified = await proxy_logging_obj.post_call_success_hook(
+            data=data,
+            user_api_key_dict=user_api_key_dict,
+            response=response,
+        )
+        if isinstance(modified, ApplyGuardrailResponse):
+            response = modified
+    except Exception:
+        verbose_proxy_logger.exception("apply_guardrail: post_call_success_hook failed")
+
+    # Build the logging payload after post_call_success_hook so that logged
+    # data matches what the caller actually receives if the hook modified
+    # the response.
+    response_for_logging = {"response": response.model_dump(exclude_none=True)}
+
+    if litellm_logging_obj is not None:
+        end_time = datetime.now(timezone.utc)
+        try:
+            await litellm_logging_obj.async_success_handler(
+                result=response_for_logging,
+                start_time=start_time,
+                end_time=end_time,
+                cache_hit=False,
+            )
+        except Exception:
+            verbose_proxy_logger.exception(
+                "apply_guardrail: async_success_handler failed"
+            )
+        try:
+            thread_pool_executor.submit(
+                litellm_logging_obj.success_handler,
+                response_for_logging,
+                start_time,
+                end_time,
+                False,
+            )
+        except Exception:
+            verbose_proxy_logger.exception(
+                "apply_guardrail: success_handler submit failed"
+            )
+
+    return response
+
+
 @router.post("/guardrails/apply_guardrail", response_model=ApplyGuardrailResponse)
 @router.post("/apply_guardrail", response_model=ApplyGuardrailResponse)
 async def apply_guardrail(
+    fastapi_request: Request,
     request: ApplyGuardrailRequest,
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
 ):
@@ -2198,8 +2286,29 @@
 
     This endpoint allows testing guardrails by applying them to custom text inputs.
     """
+    import traceback
+
+    from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
+    from litellm.litellm_core_utils.thread_pool_executor import (
+        executor as thread_pool_executor,
+    )
+    from litellm.proxy.proxy_server import (
+        general_settings,
+        proxy_config,
+        proxy_logging_obj,
+        version,
+    )
     from litellm.proxy.utils import handle_exception_on_proxy
 
+    data: dict = {
+        "guardrail_name": request.guardrail_name,
+        "input": [request.text],
+        "messages": request.messages or [],
+        "metadata": {"route": "/apply_guardrail"},
+    }
+    litellm_logging_obj = None
+    start_time = datetime.now(timezone.utc)
+
     try:
         active_guardrail: Optional[CustomGuardrail] = (
             GUARDRAIL_REGISTRY.get_initialized_guardrail_callback(
@@ -2212,23 +2321,25 @@
                 detail=f"Guardrail '{request.guardrail_name}' not found. Please ensure the guardrail is configured in your LiteLLM proxy.",
             )
 
-        request_data: dict = {}
-        if request.messages:
-            request_data["messages"] = request.messages
+        request_processor = ProxyBaseLLMRequestProcessing(data=data)
+        data, litellm_logging_obj = (
+            await request_processor.common_processing_pre_call_logic(
+                request=fastapi_request,
+                general_settings=general_settings,
+                user_api_key_dict=user_api_key_dict,
+                version=version,
+                proxy_logging_obj=proxy_logging_obj,
+                proxy_config=proxy_config,
+                route_type="apply_guardrail",
+            )
+        )
 
-        # Auto-detect input_type: if the caller didn't specify "response" but the
-        # guardrail only runs post_call (e.g. LLM-as-a-judge), use "response" so
-        # the test actually exercises the guardrail logic.
-        from litellm.types.guardrails import GuardrailEventHooks
+        if litellm_logging_obj is not None:
+            _patch_logging_obj_for_guardrail(litellm_logging_obj, request)
 
-        resolved_input_type = request.input_type
-        if resolved_input_type == "request":
-            hook = getattr(active_guardrail, "event_hook", None)
-            if hook == GuardrailEventHooks.post_call or hook == "post_call":
-                resolved_input_type = "response"
-
-        _input_type: Literal["request", "response"] = (
-            "response" if resolved_input_type == "response" else "request"
+        request_data: dict = {"messages": request.messages} if request.messages else {}
+        _input_type = _resolve_guardrail_input_type(
+            active_guardrail, request.input_type
         )
         guardrailed_inputs = await active_guardrail.apply_guardrail(
             inputs={"texts": [request.text]},
@@ -2236,13 +2347,55 @@
             input_type=_input_type,
         )
         response_text = guardrailed_inputs.get("texts", [])
-
-        return ApplyGuardrailResponse(
+        response = ApplyGuardrailResponse(
             response_text=response_text[0] if response_text else request.text
         )
     except Exception as e:
+        if litellm_logging_obj is not None and not isinstance(e, HTTPException):
+            try:
+                await litellm_logging_obj.async_failure_handler(
+                    exception=e,
+                    traceback_exception=traceback.format_exc(),
+                )
+            except Exception:
+                verbose_proxy_logger.exception(
+                    "apply_guardrail: async_failure_handler failed"
+                )
+            try:
+                thread_pool_executor.submit(
+                    litellm_logging_obj.failure_handler,
+                    e,
+                    traceback.format_exc(),
+                )
+            except Exception:
+                verbose_proxy_logger.exception(
+                    "apply_guardrail: failure_handler submit failed"
+                )
+        try:
+            transformed_exception = await proxy_logging_obj.post_call_failure_hook(
+                user_api_key_dict=user_api_key_dict,
+                original_exception=e,
+                request_data=data,
+            )
+            if isinstance(transformed_exception, Exception):
+                e = transformed_exception
+        except Exception:
+            verbose_proxy_logger.exception(
+                "apply_guardrail: post_call_failure_hook failed"
+            )
         raise handle_exception_on_proxy(e)
 
+    # Success logging outside except so a hook error never triggers failure handlers.
+    response = await _emit_guardrail_success_logs(
+        proxy_logging_obj=proxy_logging_obj,
+        litellm_logging_obj=litellm_logging_obj,
+        data=data,
+        user_api_key_dict=user_api_key_dict,
+        response=response,
+        start_time=start_time,
+    )
+    return response
 
+
 # Usage (dashboard) endpoints: overview, detail, logs
 router.include_router(guardrails_usage_router)

diff --git a/tests/enterprise/litellm_enterprise/proxy/guardrails/conftest.py b/tests/enterprise/litellm_enterprise/proxy/guardrails/conftest.py
new file mode 100644
--- /dev/null
+++ b/tests/enterprise/litellm_enterprise/proxy/guardrails/conftest.py
@@ -1,0 +1,42 @@
+"""Shared fixtures for guardrail apply_guardrail tests."""
+
+from contextlib import contextmanager
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+@contextmanager
+def _mock_proxy_logging():
+    """Patch the proxy-server globals that apply_guardrail imports at call time."""
+    mock_proxy_logging = MagicMock()
+    mock_proxy_logging.post_call_success_hook = AsyncMock(return_value=None)
+    mock_proxy_logging.post_call_failure_hook = AsyncMock(return_value=None)
+    mock_logging_obj = MagicMock()
+    mock_logging_obj.async_success_handler = AsyncMock(return_value=None)
+    mock_logging_obj.async_failure_handler = AsyncMock(return_value=None)
+    mock_logging_obj.success_handler = MagicMock(return_value=None)
+    mock_logging_obj.failure_handler = MagicMock(return_value=None)
+    mock_logging_obj.model_call_details = {}
+
+    with (
+        patch(
+            "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing"
+        ) as mock_proc_cls,
+        patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging),
+        patch("litellm.proxy.proxy_server.general_settings", {}),
+        patch("litellm.proxy.proxy_server.proxy_config", MagicMock()),
+        patch("litellm.proxy.proxy_server.version", "0.0.0"),
+    ):
+        mock_proc = MagicMock()
+        mock_proc.common_processing_pre_call_logic = AsyncMock(
+            return_value=({}, mock_logging_obj)
+        )
+        mock_proc_cls.return_value = mock_proc
+        yield mock_proxy_logging
+
+
+@pytest.fixture
+def mock_proxy_logging_ctx():
+    """Return the proxy-logging context manager factory for use as `with ctx():`."""
+    return _mock_proxy_logging

diff --git a/tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py b/tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py
--- a/tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py
+++ b/tests/enterprise/litellm_enterprise/proxy/guardrails/test_apply_guardrail_endpoint.py
@@ -18,14 +18,19 @@
 
 
 @pytest.mark.asyncio
-async def test_apply_guardrail_endpoint_returns_correct_response():
+async def test_apply_guardrail_endpoint_returns_correct_response(
+    mock_proxy_logging_ctx,
+):
     """Test that apply_guardrail endpoint returns ApplyGuardrailResponse object"""
     from litellm.proxy.guardrails.guardrail_endpoints import apply_guardrail
 
     # Mock the guardrail registry
-    with patch(
-        "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    with (
+        patch(
+            "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
+        ) as mock_registry,
+        mock_proxy_logging_ctx(),
+    ):
         # Create a mock guardrail
         mock_guardrail = Mock(spec=CustomGuardrail)
         # Apply guardrail returns GenericGuardrailAPIInputs (dict with texts key)
@@ -49,7 +54,9 @@
 
         # Call the endpoint
         response = await apply_guardrail(
-            request=request, user_api_key_dict=user_api_key_dict
+            fastapi_request=Mock(),
+            request=request,
+            user_api_key_dict=user_api_key_dict,
         )
 
         # Verify the response is of the correct type
@@ -65,15 +72,18 @@
 
 
 @pytest.mark.asyncio
-async def test_apply_guardrail_endpoint_guardrail_not_found():
+async def test_apply_guardrail_endpoint_guardrail_not_found(mock_proxy_logging_ctx):
     """Test that apply_guardrail endpoint raises exception when guardrail not found"""
     from litellm.proxy._types import ProxyException
     from litellm.proxy.guardrails.guardrail_endpoints import apply_guardrail
 
     # Mock the guardrail registry to return None
-    with patch(
-        "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    with (
+        patch(
+            "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
+        ) as mock_registry,
+        mock_proxy_logging_ctx(),
+    ):
         mock_registry.get_initialized_guardrail_callback.return_value = None
 
         # Create the request
@@ -86,26 +96,35 @@
 
         # Verify exception is raised
         with pytest.raises(ProxyException) as exc_info:
-            await apply_guardrail(request=request, user_api_key_dict=user_api_key_dict)
+            await apply_guardrail(
+                fastapi_request=Mock(),
+                request=request,
+                user_api_key_dict=user_api_key_dict,
+            )
 
         assert "non-existent-guardrail" in exc_info.value.message
         assert "not found" in exc_info.value.message
 
 
 @pytest.mark.asyncio
-async def test_apply_guardrail_endpoint_with_presidio_guardrail():
+async def test_apply_guardrail_endpoint_with_presidio_guardrail(mock_proxy_logging_ctx):
     """Test apply_guardrail endpoint with a Presidio-like guardrail"""
     from litellm.proxy.guardrails.guardrail_endpoints import apply_guardrail
 
     # Mock the guardrail registry
-    with patch(
-        "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    with (
+        patch(
+            "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
+        ) as mock_registry,
+        mock_proxy_logging_ctx(),
+    ):
         # Create a mock guardrail that simulates Presidio behavior
         mock_guardrail = Mock(spec=CustomGuardrail)
         # Simulate masking PII entities - returns GenericGuardrailAPIInputs (dict with texts key)
         mock_guardrail.apply_guardrail = AsyncMock(
-            return_value={"texts": ["My name is [PERSON] and my email is [EMAIL_ADDRESS]"]}
+            return_value={
+                "texts": ["My name is [PERSON] and my email is [EMAIL_ADDRESS]"]
+            }
         )
 
         # Configure the registry to return our mock guardrail
@@ -124,7 +143,9 @@
 
         # Call the endpoint
         response = await apply_guardrail(
-            request=request, user_api_key_dict=user_api_key_dict
+            fastapi_request=Mock(),
+            request=request,
+            user_api_key_dict=user_api_key_dict,
         )
 
         # Verify the response is of the correct type
@@ -138,14 +159,17 @@
 
 
 @pytest.mark.asyncio
-async def test_apply_guardrail_endpoint_without_optional_params():
+async def test_apply_guardrail_endpoint_without_optional_params(mock_proxy_logging_ctx):
     """Test apply_guardrail endpoint without optional language and entities parameters"""
     from litellm.proxy.guardrails.guardrail_endpoints import apply_guardrail
 
     # Mock the guardrail registry
-    with patch(
-        "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    with (
+        patch(
+            "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
+        ) as mock_registry,
+        mock_proxy_logging_ctx(),
+    ):
         # Create a mock guardrail
         mock_guardrail = Mock(spec=CustomGuardrail)
         # Returns GenericGuardrailAPIInputs (dict with texts key)
@@ -166,7 +190,9 @@
 
         # Call the endpoint
         response = await apply_guardrail(
-            request=request, user_api_key_dict=user_api_key_dict
+            fastapi_request=Mock(),
+            request=request,
+            user_api_key_dict=user_api_key_dict,
         )
 
         # Verify the response is of the correct type

diff --git a/tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py b/tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py
--- a/tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py
+++ b/tests/enterprise/litellm_enterprise/proxy/guardrails/test_bedrock_apply_guardrail.py
@@ -4,7 +4,7 @@
 
 import os
 import sys
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 
@@ -153,7 +153,7 @@
 
 
 @pytest.mark.asyncio
-async def test_bedrock_apply_guardrail_endpoint_integration():
+async def test_bedrock_apply_guardrail_endpoint_integration(mock_proxy_logging_ctx):
     """Test the full endpoint integration with Bedrock guardrail"""
     from litellm.proxy.guardrails.guardrail_endpoints import apply_guardrail
 
@@ -165,9 +165,12 @@
     )
 
     # Mock the guardrail registry
-    with patch(
-        "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
-    ) as mock_registry:
+    with (
+        patch(
+            "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY"
+        ) as mock_registry,
+        mock_proxy_logging_ctx(),
+    ):
         # Mock the make_bedrock_api_request method
         with patch.object(
             guardrail, "make_bedrock_api_request", new_callable=AsyncMock
@@ -194,7 +197,9 @@
 
             # Call the endpoint
             response = await apply_guardrail(
-                request=request, user_api_key_dict=user_api_key_dict
+                fastapi_request=Mock(),
+                request=request,
+                user_api_key_dict=user_api_key_dict,
             )
 
             # Verify the response

diff --git a/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py b/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
--- a/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
+++ b/tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py
@@ -1149,6 +1149,13 @@
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_registry
     )
 
+    mock_proxy_logging = mocker.Mock()
+    mock_proxy_logging.post_call_failure_hook = AsyncMock()
+    mocker.patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging)
+    mocker.patch("litellm.proxy.proxy_server.general_settings", {})
+    mocker.patch("litellm.proxy.proxy_server.proxy_config", mocker.Mock())
+    mocker.patch("litellm.proxy.proxy_server.version", "test")
+
     # Create request
     request = ApplyGuardrailRequest(
         guardrail_name="non-existent-guardrail", text="Test input text"
@@ -1159,7 +1166,11 @@
 
     # Call endpoint and expect ProxyException
     with pytest.raises(ProxyException) as exc_info:
-        await apply_guardrail(request=request, user_api_key_dict=mock_user_auth)
+        await apply_guardrail(
+            fastapi_request=mocker.Mock(),
+            request=request,
+            user_api_key_dict=mock_user_auth,
+        )
 
     # Verify error details
     assert str(exc_info.value.code) == "404"
@@ -1186,6 +1197,25 @@
         "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_registry
     )
 
+    mock_logging_obj = mocker.Mock()
+    mock_logging_obj.async_failure_handler = AsyncMock()
+    mock_logging_obj.model_call_details = {}
+    mock_processor = mocker.Mock()
+    mock_processor.common_processing_pre_call_logic = AsyncMock(
+        return_value=({"guardrail_name": "test-guardrail"}, mock_logging_obj)
+    )
+    mocker.patch(
+        "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing",
+        return_value=mock_processor,
+    )
+    mock_proxy_logging = mocker.Mock()
+    mock_proxy_logging.post_call_failure_hook = AsyncMock()
+    mocker.patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging)
+    mocker.patch("litellm.proxy.proxy_server.general_settings", {})
+    mocker.patch("litellm.proxy.proxy_server.proxy_config", mocker.Mock())
+    mocker.patch("litellm.proxy.proxy_server.version", "test")
+    mocker.patch("litellm.litellm_core_utils.thread_pool_executor.executor")
+
     # Create request
     request = ApplyGuardrailRequest(
         guardrail_name="test-guardrail", text="Test input text with forbidden content"
@@ -1196,13 +1226,71 @@
 
     # Call endpoint and expect ProxyException
     with pytest.raises(ProxyException) as exc_info:
-        await apply_guardrail(request=request, user_api_key_dict=mock_user_auth)
+        await apply_guardrail(
+            fastapi_request=mocker.Mock(),
+            request=request,
+            user_api_key_dict=mock_user_auth,
+        )
 
     # Verify error is properly handled
     assert "Bedrock guardrail failed" in str(exc_info.value.message)
 
 
 @pytest.mark.asyncio
+async def test_apply_guardrail_invokes_logging_pipeline(mocker):
+    mock_guardrail = mocker.Mock()
+    mock_guardrail.apply_guardrail = AsyncMock(return_value={"texts": ["masked"]})
+
+    mock_registry = mocker.Mock()
+    mock_registry.get_initialized_guardrail_callback.return_value = mock_guardrail
+    mocker.patch(
+        "litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_registry
+    )
+
+    mock_logging_obj = mocker.Mock()
+    mock_logging_obj.async_success_handler = AsyncMock()
+    mock_logging_obj.model_call_details = {}
+    mock_processor = mocker.Mock()
+    mock_processor.common_processing_pre_call_logic = AsyncMock(
+        return_value=({"guardrail_name": "test-guardrail"}, mock_logging_obj)
+    )
+    mocker.patch(
+        "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing",
+        return_value=mock_processor,
+    )
+
+    mock_proxy_logging = mocker.Mock()
+    mock_proxy_logging.post_call_success_hook = AsyncMock()
+    mocker.patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging)
+    mocker.patch("litellm.proxy.proxy_server.general_settings", {})
+    mocker.patch("litellm.proxy.proxy_server.proxy_config", mocker.Mock())
+    mocker.patch("litellm.proxy.proxy_server.version", "test")
+    mock_executor = mocker.Mock()
+    mocker.patch(
+        "litellm.litellm_core_utils.thread_pool_executor.executor", mock_executor
+    )
+
+    request = ApplyGuardrailRequest(
+        guardrail_name="test-guardrail", text="hello@example.com"
+    )
+    response = await apply_guardrail(
+        fastapi_request=mocker.Mock(),
+        request=request,
+        user_api_key_dict=UserAPIKeyAuth(),
+    )
+
+    assert response.response_text == "masked"
+    mock_processor.common_processing_pre_call_logic.assert_awaited_once()
+    mock_proxy_logging.post_call_success_hook.assert_awaited_once()
+    mock_logging_obj.async_success_handler.assert_awaited_once()
+    assert mock_logging_obj.call_type == "pass_through_endpoint"
+    mock_executor.submit.assert_called_once()
+    assert mock_logging_obj.async_success_handler.await_args.kwargs["result"] == {
+        "response": {"response_text": "masked"}
+    }
+
+
+@pytest.mark.asyncio
 async def test_get_guardrail_info_endpoint_config_guardrail(mocker):
     """
     Test get_guardrail_info endpoint returns proper response when guardrail is found in config.

_{You can send follow-ups to the cloud agent here.}

^{Reviewed by Cursor Bugbot for commit 8fd6f71. Configure here.}

[mateo-berri]

@greptileai

[mateo-berri]

On the registry-lookup ordering observation: the current placement is intentional.

1. The cheap GUARDRAIL_REGISTRY.get_initialized_guardrail_callback check is deliberately ahead of common_processing_pre_call_logic so a misconfigured guardrail name does not trigger pre-call hooks (which run any other configured guardrails on the input text and consume rate-limit budget). For a 404 misconfiguration we want a fast, side-effect-free response.

2. The not isinstance(e, HTTPException) guard inside the except block (present since the first commit on this PR) already determines trace behavior independently of ordering: HTTPException-class results are the user-facing 4xx response and are intentionally not emitted as LiteLLM-level failure spans. Reordering the lookup would not change this — litellm_logging_obj.async_failure_handler is still skipped for HTTPException either way.

3. proxy_logging_obj.post_call_failure_hook continues to fire for the 404 path, so proxy-level observability (alerting, etc.) still sees the failure. That's consistent with how other proxy endpoints distinguish 4xx client errors from upstream LLM failures.

No code change needed.

[mateo-berri]

LGTM; thanks!