diff --git a/docs/my-website/sidebars.js b/docs/my-website/sidebars.js index 6446e227d99..850a850b9ed 100644 --- a/docs/my-website/sidebars.js +++ b/docs/my-website/sidebars.js @@ -1466,4 +1466,4 @@ const learnSidebar = { ], }; -module.exports = { ...sidebars, ...learnSidebar }; +module.exports = { ...sidebars, ...learnSidebar }; \ No newline at end of file diff --git a/litellm/__init__.py b/litellm/__init__.py index e45d926e8db..a85d73dedb1 100644 --- a/litellm/__init__.py +++ b/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 diff --git a/litellm/integrations/akto/__init__.py b/litellm/integrations/akto/__init__.py new file mode 100644 index 00000000000..e1cc8062de4 --- /dev/null +++ b/litellm/integrations/akto/__init__.py @@ -0,0 +1,3 @@ +from .akto_logger import AktoLogger + +__all__ = ["AktoLogger"] diff --git a/litellm/integrations/akto/akto_logger.py b/litellm/integrations/akto/akto_logger.py new file mode 100644 index 00000000000..8c67631cc53 --- /dev/null +++ b/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) diff --git a/litellm/integrations/callback_configs.json b/litellm/integrations/callback_configs.json index c2b0c4ddce9..8f0bcb0a5c7 100644 --- a/litellm/integrations/callback_configs.json +++ b/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" } ] diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 5323f692b80..6ff6148ac06 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/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: diff --git a/litellm/litellm_core_utils/logging_callback_manager.py b/litellm/litellm_core_utils/logging_callback_manager.py index c5c150274cc..4a22cabfb5f 100644 --- a/litellm/litellm_core_utils/logging_callback_manager.py +++ b/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 ) diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 91a953c217e..eb5942bc57b 100644 --- a/litellm/proxy/_types.py +++ b/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): """ diff --git a/litellm/proxy/health_endpoints/_health_endpoints.py b/litellm/proxy/health_endpoints/_health_endpoints.py index ef9436f2d8c..bbe9e7ec77c 100644 --- a/litellm/proxy/health_endpoints/_health_endpoints.py +++ b/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 "",