[Feature] Add Audit Log Export to External Callbacks

litellm/__init__.py

@@ -97,6 +97,7 @@
 success_callback: List[CALLBACK_TYPES] = []
 failure_callback: List[CALLBACK_TYPES] = []
 service_callback: List[CALLBACK_TYPES] = []
+audit_log_callbacks: List[CALLBACK_TYPES] = []
 # logging_callback_manager is lazy-loaded via __getattr__
 _custom_logger_compatible_callbacks_literal = Literal[
     "lago",

litellm/integrations/custom_logger.py

@@ -27,6 +27,7 @@
     LLMResponseTypes,
     ModelResponse,
     ModelResponseStream,
+    StandardAuditLogPayload,
     StandardCallbackDynamicParams,
     StandardLoggingPayload,
 )
@@ -177,6 +178,10 @@ async def async_log_success_event(self, kwargs, response_obj, start_time, end_ti
     async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
         pass
 
+    async def async_log_audit_log_event(self, audit_log: "StandardAuditLogPayload"):
+        """Called when an audit log is created. Override in subclasses to handle."""
+        pass
+
     #### PROMPT MANAGEMENT HOOKS ####
 
     async def async_get_chat_completion_prompt(

litellm/integrations/s3_v2.py

@@ -22,7 +22,7 @@
     httpxSpecialProvider,
 )
 from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
-from litellm.types.utils import StandardLoggingPayload
+from litellm.types.utils import StandardAuditLogPayload, StandardLoggingPayload
 
 from .custom_batch_logger import CustomBatchLogger
 
@@ -248,6 +248,38 @@ async def async_log_failure_event(self, kwargs, response_obj, start_time, end_ti
         )
         pass
 
+    async def async_log_audit_log_event(
+        self, audit_log: StandardAuditLogPayload
+    ) -> None:
+        """Batch audit logs and upload to S3 under audit_logs/ prefix."""
+        try:
+            from datetime import timezone
+
+            now = datetime.now(timezone.utc)
+            audit_log_id = audit_log.get("id", "unknown")
+
+            s3_path = cast(Optional[str], self.s3_path) or ""
+            s3_path = s3_path.rstrip("/") + "/" if s3_path else ""
+
+            s3_object_key = (
+                f"{s3_path}audit_logs/"
+                f"{now.strftime('%Y-%m-%d')}/"
+                f"{now.strftime('%H-%M-%S')}_{audit_log_id}.json"
+            )
+
+            element = s3BatchLoggingElement(
+                payload=dict(audit_log),
+                s3_object_key=s3_object_key,
+                s3_object_download_filename=f"audit-{audit_log_id}.json",
+            )
+
+            self.log_queue.append(element)
+
+            if len(self.log_queue) >= self.batch_size:
+                await self.flush_queue()
+        except Exception as e:
+            verbose_logger.exception("S3 audit log error: %s", e)
+
     async def _async_log_event_base(self, kwargs, response_obj, start_time, end_time):
         try:
             verbose_logger.debug(

litellm/proxy/management_helpers/audit_logs.py

@@ -2,19 +2,111 @@
 Functions to create audit logs for LiteLLM Proxy
 """
 
+import asyncio
 import json
-from litellm._uuid import uuid
 from datetime import datetime, timezone
+from typing import Dict
 
 import litellm
 from litellm._logging import verbose_proxy_logger
+from litellm._uuid import uuid
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.proxy._types import (
     AUDIT_ACTIONS,
     LiteLLM_AuditLogs,
     LitellmTableNames,
     Optional,
     UserAPIKeyAuth,
 )
+from litellm.types.utils import StandardAuditLogPayload
+
+_audit_log_callback_cache: Dict[str, CustomLogger] = {}
+
+
+def _resolve_audit_log_callback(name: str) -> Optional[CustomLogger]:
+    """Resolve a string callback name to a CustomLogger instance, with caching."""
+    if name in _audit_log_callback_cache:
+        return _audit_log_callback_cache[name]
+
+    from litellm.litellm_core_utils.litellm_logging import (
+        _init_custom_logger_compatible_class,
+    )
+
+    instance = _init_custom_logger_compatible_class(
+        logging_integration=name,  # type: ignore
+        internal_usage_cache=None,
+        llm_router=None,
+    )
+
+    if instance is not None:
+        _audit_log_callback_cache[name] = instance
+    return instance
+
+
+def _build_audit_log_payload(
+    request_data: LiteLLM_AuditLogs,
+) -> StandardAuditLogPayload:
+    """Convert LiteLLM_AuditLogs to StandardAuditLogPayload for callback dispatch."""
+    updated_at = ""
+    if request_data.updated_at is not None:
+        updated_at = request_data.updated_at.isoformat()
+
+    table_name = request_data.table_name
+    if isinstance(table_name, LitellmTableNames):
+        table_name = table_name.value
+
+    return StandardAuditLogPayload(
+        id=request_data.id,
+        updated_at=updated_at,
+        changed_by=request_data.changed_by or "",
+        changed_by_api_key=request_data.changed_by_api_key or "",
+        action=request_data.action,
+        table_name=str(table_name),
+        object_id=request_data.object_id,
+        before_value=request_data.before_value,
+        updated_values=request_data.updated_values,
+    )
+
+
+def _audit_log_task_done_callback(task: asyncio.Task) -> None:
+    """Log exceptions from audit log callback tasks so they don't slip through silently."""
+    try:
+        exc = task.exception()
+    except asyncio.CancelledError:
+        return
+    if exc is not None:
+        verbose_proxy_logger.error(
+            "Audit log callback task failed: %s", exc, exc_info=exc
+        )
+
+
+async def _dispatch_audit_log_to_callbacks(
+    request_data: LiteLLM_AuditLogs,
+) -> None:
+    """Dispatch audit log to all registered audit_log_callbacks."""
+    if not litellm.audit_log_callbacks:
+        return
+
+    payload = _build_audit_log_payload(request_data)
+
+    for callback in litellm.audit_log_callbacks:
+        try:
+            resolved = callback
+            if isinstance(callback, str):
+                resolved = _resolve_audit_log_callback(callback)
+                if resolved is None:
+                    verbose_proxy_logger.warning(
+                        "Could not resolve audit log callback: %s", callback
+                    )
+                    continue
+
+            if isinstance(resolved, CustomLogger):
+                task = asyncio.create_task(resolved.async_log_audit_log_event(payload))
+                task.add_done_callback(_audit_log_task_done_callback)
+        except Exception as e:
+            verbose_proxy_logger.error(
+                "Failed dispatching audit log to callback: %s", e
+            )
 
 
 async def create_object_audit_log(
@@ -81,9 +173,6 @@ async def create_audit_log_for_update(request_data: LiteLLM_AuditLogs):
     if premium_user is not True:
         return
 
-    if prisma_client is None:
-        raise Exception("prisma_client is None, no DB connected")
-
     verbose_proxy_logger.debug("creating audit log for %s", request_data)
 
     if isinstance(request_data.updated_values, dict):
@@ -92,6 +181,15 @@ async def create_audit_log_for_update(request_data: LiteLLM_AuditLogs):
     if isinstance(request_data.before_value, dict):
         request_data.before_value = json.dumps(request_data.before_value)
 
+    # Dispatch to external audit log callbacks regardless of DB availability
+    await _dispatch_audit_log_to_callbacks(request_data)
+
+    if prisma_client is None:
+        verbose_proxy_logger.error(
+            "prisma_client is None, cannot write audit log to DB"
+        )
+        return
+
     _request_data = request_data.model_dump(exclude_none=True)
 
     try:
@@ -103,5 +201,3 @@ async def create_audit_log_for_update(request_data: LiteLLM_AuditLogs):
     except Exception as e:
         # [Non-Blocking Exception. Do not allow blocking LLM API call]
         verbose_proxy_logger.error(f"Failed Creating audit log {e}")
-
-    return

litellm/proxy/proxy_server.py

@@ -2961,6 +2961,29 @@ async def load_config(  # noqa: PLR0915
                     print(  # noqa
                         f"{blue_color_code} Initialized Failure Callbacks - {litellm.failure_callback} {reset_color_code}"
                     )  # noqa
+                elif key == "audit_log_callbacks":
+                    litellm.audit_log_callbacks = []
+
+                    for callback in value:
+                        if "." in callback:
+                            litellm.audit_log_callbacks.append(
+                                get_instance_fn(value=callback)
+                            )
+                        else:
+                            litellm.audit_log_callbacks.append(callback)
+
+                    _store_audit_logs = litellm_settings.get(
+                        "store_audit_logs", litellm.store_audit_logs
+                    )
+                    if _store_audit_logs:
+                        print(  # noqa
+                            f"{blue_color_code} Initialized Audit Log Callbacks - {litellm.audit_log_callbacks} {reset_color_code}"
+                        )  # noqa
+                    else:
+                        verbose_proxy_logger.warning(
+                            "'audit_log_callbacks' is configured but 'store_audit_logs' is not enabled. "
+                            "Audit log callbacks will not fire until 'store_audit_logs: true' is added to litellm_settings."
+                        )
                 elif key == "cache_params":
                     # this is set in the cache branch
                     # see usage here: https://docs.litellm.ai/docs/proxy/caching

litellm/types/utils.py

@@ -2804,6 +2804,20 @@ class StandardLoggingPayloadStatusFields(TypedDict, total=False):
     """
 
 
+class StandardAuditLogPayload(TypedDict):
+    """Payload for audit log events dispatched to external callbacks."""
+
+    id: str
+    updated_at: str  # ISO-8601
+    changed_by: str
+    changed_by_api_key: str
+    action: str  # "created" | "updated" | "deleted" | "blocked" | "rotated"
+    table_name: str
+    object_id: str
+    before_value: Optional[str]
+    updated_values: Optional[str]
+
+
 class StandardLoggingPayload(TypedDict):
     id: str
     trace_id: str  # Trace multiple LLM calls belonging to same overall request (e.g. fallbacks/retries)