akto logging integration

docs/my-website/sidebars.js

@@ -1466,4 +1466,4 @@ const learnSidebar = {
   ],
 };
 
-module.exports = { ...sidebars, ...learnSidebar };
+module.exports = { ...sidebars, ...learnSidebar };

litellm/__init__.py

@@ -148,6 +148,7 @@ _custom_logger_compatible_callbacks_literal = Literal[
     "vantage",
     "posthog",
     "levo",
+    "akto",
 ]
 cold_storage_custom_logger: Optional[_custom_logger_compatible_callbacks_literal] = None
 logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None

litellm/integrations/akto/__init__.py

@@ -0,0 +1,3 @@
+from .akto_logger import AktoLogger
+
+__all__ = ["AktoLogger"]

litellm/integrations/akto/akto_logger.py

@@ -0,0 +1,247 @@
+"""Akto logging integration for LiteLLM.
+
+Ingests LLM request/response traffic to Akto for monitoring and analysis.
+Configure via success_callback/failure_callback: ["akto"].
+"""
+
+import json
+import os
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+from fastapi import HTTPException
+
+from litellm._logging import verbose_logger
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.llms.custom_httpx.http_handler import (
+    HTTPHandler,
+    get_async_httpx_client,
+    httpxSpecialProvider,
+)
+
+HTTP_PROXY_PATH = "/api/http-proxy"
+AKTO_CONNECTOR_NAME = "litellm"
+SENSITIVE_HEADERS = {"authorization", "x-litellm-api-key", "x-api-key", "cookie"}
+
+
+class AktoLogger(CustomLogger):
+    """Logs LLM traffic to Akto for monitoring and analysis."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.validate_environment()
+        self.async_http_handler = get_async_httpx_client(
+            llm_provider=httpxSpecialProvider.LoggingCallback
+        )
+        self.sync_http_handler = HTTPHandler()
+
+        self.akto_base_url = os.environ["AKTO_DATA_INGESTION_API_BASE"].rstrip("/")
+        self.akto_api_key = os.environ["AKTO_API_KEY"]
+        self.akto_account_id = os.environ.get("AKTO_ACCOUNT_ID", "1000000")
+        self.akto_vxlan_id = os.environ.get("AKTO_VXLAN_ID", "0")
+
+    def validate_environment(self) -> None:
+        missing_keys = []
+        if os.getenv("AKTO_DATA_INGESTION_API_BASE", None) is None:
+            missing_keys.append("AKTO_DATA_INGESTION_API_BASE")
+        if os.getenv("AKTO_API_KEY", None) is None:
+            missing_keys.append("AKTO_API_KEY")
+        if len(missing_keys) > 0:
+            raise Exception("Missing keys={} in environment.".format(missing_keys))
+
+    # ── Data extraction ──
+
+    @staticmethod
+    def extract_logging_data(kwargs: dict) -> dict:
+        """Promote metadata and proxy_server_request from litellm_params to top level."""
+        litellm_params = kwargs.get("litellm_params") or {}
+        data = dict(kwargs)
+        if (
+            "proxy_server_request" not in data
+            and "proxy_server_request" in litellm_params
+        ):
+            data["proxy_server_request"] = litellm_params["proxy_server_request"]
+        if "metadata" not in data and "metadata" in litellm_params:
+            data["metadata"] = litellm_params["metadata"]
+        return data
+
+    # ── Payload builders ──
+
+    @staticmethod
+    def resolve_metadata_value(data: Optional[dict], key: str) -> Optional[str]:
+        """Look up a value from metadata or litellm_metadata."""
+        if data is None:
+            return None
+        for k in ("litellm_metadata", "metadata"):
+            container = data.get(k) or {}
+            if isinstance(container, dict):
+                val = container.get(key)
+                if val is not None:
+                    return str(val).strip()
+        return None
+
+    @staticmethod
+    def extract_request_path(data: dict) -> str:
+        """Get the API route, defaulting to /v1/chat/completions."""
+        metadata = data.get("metadata") or {}
+        if isinstance(metadata, dict):
+            return metadata.get("user_api_key_request_route") or "/v1/chat/completions"
+        return "/v1/chat/completions"
+
+    @staticmethod
+    def build_request_headers(data: dict) -> Dict[str, str]:
+        """Build request headers from proxy headers, stripping sensitive values."""
+        headers: Dict[str, str] = {"content-type": "application/json"}
+        proxy_req = data.get("proxy_server_request")
+        if isinstance(proxy_req, dict):
+            for key, val in (proxy_req.get("headers") or {}).items():
+                if key and val and str(key).lower() not in SENSITIVE_HEADERS:
+                    headers[str(key).lower()] = str(val)
+        if "host" not in headers:
+            headers["host"] = "litellm.ai"
+        return headers
+
+    @staticmethod
+    def build_tag_metadata(data: dict) -> Dict[str, str]:
+        """Build tag dict with gen-ai marker, user_id, and team_id."""
+        tag: Dict[str, str] = {"gen-ai": "Gen AI"}
+        for meta_key, tag_key in [
+            ("user_api_key_user_id", "user_id"),
+            ("user_api_key_team_id", "team_id"),
+        ]:
+            val = AktoLogger.resolve_metadata_value(data, meta_key)
+            if val:
+                tag[tag_key] = val
+        return tag
+
+    @staticmethod
+    def extract_client_ip(data: dict) -> str:
+        """Extract client IP from proxy headers."""
+        proxy_req = data.get("proxy_server_request")
+        if isinstance(proxy_req, dict):
+            headers = proxy_req.get("headers") or {}
+            ip = headers.get("x-forwarded-for") or headers.get("x-real-ip") or ""
+            if ip:
+                return ip.split(",")[0].strip()
+        return "0.0.0.0"
+
+    def build_akto_payload(
+        self, data: dict, *, status_code: int = 200, response_obj: Any = None
+    ) -> Dict[str, Any]:
+        """Build the MIRRORING payload for Akto's HTTP proxy endpoint."""
+        request_body: Dict[str, Any] = {}
+        if data.get("messages"):
+            request_body["messages"] = data["messages"]
+            request_body["model"] = data.get("model", "")
+        if data.get("tools"):
+            request_body["tools"] = data["tools"]
+
+        response_payload = json.dumps({})
+        response_headers: Dict[str, str] = {}
+        if response_obj is not None and hasattr(response_obj, "model_dump"):
+            response_payload = json.dumps(response_obj.model_dump())
+            response_headers = {"content-type": "application/json"}
+
+        tag = self.build_tag_metadata(data)
+
+        return {
+            "path": self.extract_request_path(data),
+            "requestHeaders": json.dumps(self.build_request_headers(data)),
+            "responseHeaders": json.dumps(response_headers),
+            "method": "POST",
+            "requestPayload": json.dumps(request_body),
+            "responsePayload": response_payload,
+            "ip": self.extract_client_ip(data),
+            "destIp": "127.0.0.1",
+            "time": str(int(datetime.now().timestamp() * 1000)),
+            "statusCode": str(status_code),
+            "type": "HTTP/1.1",
+            "status": str(status_code),
+            "akto_account_id": self.akto_account_id,
+            "akto_vxlan_id": self.akto_vxlan_id,
+            "is_pending": "false",
+            "source": "MIRRORING",
+            "direction": None,
+            "process_id": None,
+            "socket_id": None,
+            "daemonset_id": None,
+            "enabled_graph": None,
+            "tag": json.dumps(tag),
+            "metadata": json.dumps(tag),
+            "contextSource": "AGENTIC",
+        }
+
+    # ── HTTP ──
+
+    def request_kwargs(self, payload: dict) -> dict:
+        """Build common HTTP request kwargs for Akto API."""
+        return {
+            "url": f"{self.akto_base_url}{HTTP_PROXY_PATH}",
+            "data": json.dumps(payload),
+            "params": {"akto_connector": AKTO_CONNECTOR_NAME, "ingest_data": "true"},
+            "headers": {
+                "content-type": "application/json",
+                "Authorization": self.akto_api_key,
+            },
+        }
+
+    async def async_health_check(self) -> dict:
+        """Ping Akto's health check endpoint."""
+        try:
+            response = await self.async_http_handler.get(
+                url=f"{self.akto_base_url}/",
+                headers={"Authorization": self.akto_api_key},
+                timeout=5,
+            )
+            if response.status_code == 200:
+                return {"status": "healthy", "error_message": None}
+            return {
+                "status": "unhealthy",
+                "error_message": f"Akto returned status {response.status_code}",
+            }
+        except Exception as e:
+            return {"status": "unhealthy", "error_message": str(e)}
+
+    # ── Logging callbacks ──
+
+    def log_success_event(self, kwargs, response_obj, start_time, end_time):
+        try:
+            data = self.extract_logging_data(kwargs)
+            payload = self.build_akto_payload(data, response_obj=response_obj)
+            self.sync_http_handler.post(**self.request_kwargs(payload))
+        except Exception as e:
+            verbose_logger.error("Akto logging error: %s", e)
+
+    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+        try:
+            data = self.extract_logging_data(kwargs)
+            payload = self.build_akto_payload(data, response_obj=response_obj)
+            await self.async_http_handler.post(**self.request_kwargs(payload))
+        except Exception as e:
+            verbose_logger.error("Akto logging error: %s", e)
+
+    @staticmethod
+    def get_failure_status_code(kwargs: dict) -> int:
+        """Return the appropriate status code for a failed request."""
+        exc = kwargs.get("exception")
+        if isinstance(exc, HTTPException):
+            return exc.status_code
+        return 500
+
+    def log_failure_event(self, kwargs, response_obj, start_time, end_time):
+        try:
+            data = self.extract_logging_data(kwargs)
+            status = self.get_failure_status_code(kwargs)
+            payload = self.build_akto_payload(data, status_code=status)
+            self.sync_http_handler.post(**self.request_kwargs(payload))
+        except Exception as e:
+            verbose_logger.error("Akto logging error (failure): %s", e)
+
+    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
+        try:
+            data = self.extract_logging_data(kwargs)
+            status = self.get_failure_status_code(kwargs)
+            payload = self.build_akto_payload(data, status_code=status)
+            await self.async_http_handler.post(**self.request_kwargs(payload))
+        except Exception as e:
+            verbose_logger.error("Akto logging error (failure): %s", e)

litellm/integrations/callback_configs.json

@@ -454,5 +454,26 @@
       }
     },
     "description": "SQS Queue (AWS) Logging Integration"
+  },
+  {
+    "id": "akto",
+    "displayName": "Akto",
+    "logo": "akto.svg",
+    "supports_key_team_logging": false,
+    "dynamic_params": {
+      "AKTO_DATA_INGESTION_API_BASE": {
+        "type": "text",
+        "ui_name": "API Base URL",
+        "description": "Akto Data Ingestion API Base URL",
+        "required": true
+      },
+      "AKTO_API_KEY": {
+        "type": "password",
+        "ui_name": "API Key",
+        "description": "Akto API key for authentication",
+        "required": true
+      }
+    },
+    "description": "Akto Logging Integration"
   }
 ]

litellm/litellm_core_utils/litellm_logging.py

@@ -162,6 +162,7 @@ from ..integrations.s3 import S3Logger
 from ..integrations.s3_v2 import S3Logger as S3V2Logger
 from ..integrations.supabase import Supabase
 from ..integrations.traceloop import TraceloopLogger
+from ..integrations.akto.akto_logger import AktoLogger
 from .exception_mapping_utils import _get_response_headers
 from .initialize_dynamic_callback_params import (
     initialize_standard_callback_dynamic_params as _initialize_standard_callback_dynamic_params,
@@ -241,6 +242,7 @@ greenscaleLogger = None
 lunaryLogger = None
 supabaseClient = None
 deepevalLogger = None
+aktoLogger = None
 callback_list: Optional[List[str]] = []
 user_logger_fn = None
 additional_details: Optional[Dict[str, str]] = {}
@@ -2387,6 +2389,26 @@ class Logging(LiteLLMLoggingBaseClass):
                                 start_time=start_time,
                                 end_time=end_time,
                             )
+                    if callback == "akto" and is_sync_request:
+                        global aktoLogger
+                        if aktoLogger is None:
+                            aktoLogger = AktoLogger()
+                        if self.stream and complete_streaming_response is None:
+                            pass  # skip partial stream chunks
+                        else:
+                            if self.stream and complete_streaming_response:
+                                self.model_call_details[
+                                    "complete_response"
+                                ] = self.model_call_details.get(
+                                    "complete_streaming_response", {}
+                                )
+                                result = self.model_call_details["complete_response"]
+                            aktoLogger.log_success_event(
+                                kwargs=self.model_call_details,
+                                response_obj=result,
+                                start_time=start_time,
+                                end_time=end_time,
+                            )
                     if (
                         isinstance(callback, CustomLogger)
                         and is_sync_request
@@ -2702,6 +2724,31 @@ class Logging(LiteLLMLoggingBaseClass):
                             end_time=end_time,
                         )
 
+                if callback == "akto":
+                    global aktoLogger
+                    if aktoLogger is None:
+                        aktoLogger = AktoLogger()
+                    if self.stream is True:
+                        if (
+                            "async_complete_streaming_response"
+                            in self.model_call_details
+                        ):
+                            await aktoLogger.async_log_success_event(
+                                kwargs=self.model_call_details,
+                                response_obj=self.model_call_details[
+                                    "async_complete_streaming_response"
+                                ],
+                                start_time=start_time,
+                                end_time=end_time,
+                            )
+                    else:
+                        await aktoLogger.async_log_success_event(
+                            kwargs=self.model_call_details,
+                            response_obj=result,
+                            start_time=start_time,
+                            end_time=end_time,
+                        )
+
                 if isinstance(callback, CustomLogger):  # custom logger class
                     model_call_details: Dict = self.model_call_details
                     ##################################
@@ -3552,7 +3599,7 @@ def set_callbacks(callback_list, function_id=None):  # noqa: PLR0915
     """
     Globally sets the callback client
     """
-    global sentry_sdk_instance, capture_exception, add_breadcrumb, slack_app, alerts_channel, traceloopLogger, athinaLogger, heliconeLogger, supabaseClient, lunaryLogger, promptLayerLogger, langFuseLogger, customLogger, weightsBiasesLogger, logfireLogger, dynamoLogger, s3Logger, dataDogLogger, prometheusLogger, greenscaleLogger, openMeterLogger, deepevalLogger
+    global sentry_sdk_instance, capture_exception, add_breadcrumb, slack_app, alerts_channel, traceloopLogger, athinaLogger, heliconeLogger, supabaseClient, lunaryLogger, promptLayerLogger, langFuseLogger, customLogger, weightsBiasesLogger, logfireLogger, dynamoLogger, s3Logger, dataDogLogger, prometheusLogger, greenscaleLogger, openMeterLogger, deepevalLogger, aktoLogger
 
     try:
         for callback in callback_list:
@@ -3642,6 +3689,8 @@ def set_callbacks(callback_list, function_id=None):  # noqa: PLR0915
             elif callback == "greenscale":
                 greenscaleLogger = GreenscaleLogger()
                 print_verbose("Initialized Greenscale Logger")
+            elif callback == "akto":
+                aktoLogger = AktoLogger()
             elif callable(callback):
                 customLogger = CustomLogger()
     except Exception as e:

litellm/litellm_core_utils/logging_callback_manager.py

@@ -77,9 +77,9 @@ class LoggingCallbackManager:
         """
         Add a success callback to `litellm.success_callback`.
         Auto-routes async callbacks to litellm._async_success_callback.
-        Special-cases 'dynamodb' and 'openmeter' as async callbacks.
+        Special-cases 'dynamodb', 'openmeter', and 'akto' as async callbacks.
         """
-        if isinstance(callback, str) and callback in ("dynamodb", "openmeter"):
+        if isinstance(callback, str) and callback in ("dynamodb", "openmeter", "akto"):
             self._safe_add_callback_to_list(
                 callback=callback, parent_list=litellm._async_success_callback
             )

litellm/proxy/_types.py

@@ -3215,6 +3215,15 @@ class AllCallbacks(LiteLLMPydanticObjectBase):
         ui_callback_name="Traceloop",
     )
 
+    akto: CallbackOnUI = CallbackOnUI(
+        litellm_callback_name="akto",
+        litellm_callback_params=[
+            "AKTO_DATA_INGESTION_API_BASE",
+            "AKTO_API_KEY",
+        ],
+        ui_callback_name="Akto",
+    )
+
 
 class SpendLogsMetadata(TypedDict):
     """

litellm/proxy/health_endpoints/_health_endpoints.py

@@ -161,6 +161,7 @@ services = Union[
         "generic_api",
         "arize",
         "sqs",
+        "akto",
     ],
     str,
 ]
@@ -235,6 +236,7 @@ async def health_services_endpoint(  # noqa: PLR0915
             "generic_api",
             "arize",
             "sqs",
+            "akto",
         ]:
             raise HTTPException(
                 status_code=400,
@@ -338,7 +340,20 @@ async def health_services_endpoint(  # noqa: PLR0915
                 "status": "success",
                 "message": "Mock LLM request made - check langfuse.",
             }
+        elif service == "akto":
+            from litellm.integrations.akto.akto_logger import AktoLogger
 
+            akto_logger = AktoLogger()
+            response = await akto_logger.async_health_check()
+            return {
+                "status": response["status"],
+                "message": (
+                    response["error_message"]
+                    if response["status"] == "unhealthy"
+                    else "Akto is healthy"
+                ),
+            }
+        
         if service == "webhook":
             user_info = CallInfo(
                 token=user_api_key_dict.token or "",