diff --git a/litellm/__init__.py b/litellm/__init__.py
index 03d8fccb27e..d2c5ed8400c 100644
--- a/litellm/__init__.py
+++ b/litellm/__init__.py
@@ -57,6 +57,7 @@ from litellm._logging import (
     verbose_logger,
     json_logs,
     _turn_on_json,
+    ecs_logs,
     log_level,
 )
 import re
diff --git a/litellm/_logging.py b/litellm/_logging.py
index 5e02ff8de35..e8c543c4dec 100644
--- a/litellm/_logging.py
+++ b/litellm/_logging.py
@@ -7,11 +7,14 @@ import os
 import re
 import sys
 from collections.abc import Iterator
-from datetime import datetime
+from datetime import datetime, timezone
 from logging import Formatter
+from types import MappingProxyType
 from typing import Final, TextIO
 from urllib.parse import unquote
 
+from typing_extensions import ReadOnly, TypedDict
+
 import litellm
 from litellm.constants import (
     LITELLM_TRUNCATED_PAYLOAD_FIELD,
@@ -644,6 +647,7 @@ def resolve_log_level(log_level: str) -> int:
 
 
 json_logs: Final = _parse_json_logs_env(os.getenv("JSON_LOGS"))
+ecs_logs: Final = _parse_json_logs_env(os.getenv("LITELLM_ECS_LOGS"))
 # Create a handler for the logger (you may need to adapt this based on your needs)
 log_level: Final = os.getenv("LITELLM_LOG", "DEBUG")
 numeric_level: Final[int] = resolve_log_level(log_level)
@@ -804,6 +808,94 @@ class CorrelationPlainFormatter(logging.Formatter):
         return f"{formatted} [{' '.join(parts)}]"
 
 
+_ECS_RESERVED_KEYS: Final = frozenset({"@timestamp", "log", "message", "service", "ecs", "error"})
+
+
+class _ECSFile(TypedDict):
+    name: ReadOnly[str]
+    line: ReadOnly[int]
+
+
+class _ECSOrigin(TypedDict):
+    file: ReadOnly[_ECSFile]
+    function: ReadOnly[str]
+
+
+class _ECSLog(TypedDict):
+    level: ReadOnly[str]
+    logger: ReadOnly[str]
+    origin: ReadOnly[_ECSOrigin]
+
+
+class _ECSService(TypedDict):
+    name: ReadOnly[str]
+
+
+class _ECSMeta(TypedDict):
+    version: ReadOnly[str]
+
+
+class _ECSError(TypedDict):
+    type: ReadOnly[str | None]
+    message: ReadOnly[str]
+    stack_trace: ReadOnly[str]
+
+
+class _ECSRecord(TypedDict):
+    log: ReadOnly[_ECSLog]
+    message: ReadOnly[str]
+    service: ReadOnly[_ECSService]
+    ecs: ReadOnly[_ECSMeta]
+
+
+class ECSFormatter(Formatter):
+    ECS_VERSION = "8.11.0"
+
+    def __init__(self, service_name: str | None = None) -> None:
+        super().__init__()
+        self._service_name = service_name or os.getenv("LITELLM_SERVICE_NAME", "litellm")
+
+    def formatTime(self, record: logging.LogRecord, datefmt: str | None = None) -> str:
+        dt: Final = datetime.fromtimestamp(record.created, tz=timezone.utc)
+        return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}Z"
+
+    def format(self, record: logging.LogRecord) -> str:
+        ecs_fields: Final[_ECSRecord] = {
+            "log": {
+                "level": record.levelname.lower(),
+                "logger": record.name,
+                "origin": {"file": {"name": record.filename, "line": record.lineno}, "function": record.funcName},
+            },
+            "message": record.getMessage(),
+            "service": {"name": self._service_name},
+            "ecs": {"version": self.ECS_VERSION},
+        }
+        extra_fields: Final = MappingProxyType(
+            {
+                key: value
+                for key, value in record.__dict__.items()
+                if key not in _NON_EXTRA_RECORD_ATTRS and key not in _ECS_RESERVED_KEYS
+            }
+        )
+        document: Final = MappingProxyType(
+            {"@timestamp": self.formatTime(record), **ecs_fields, **self._error_fields(record), **extra_fields}
+        )
+        return safe_dumps(
+            document if _is_redacted(record) or not _ENABLE_SECRET_REDACTION else _redact_json_record(document)
+        )
+
+    def _error_fields(self, record: logging.LogRecord) -> MappingProxyType[str, _ECSError]:
+        exc_info: Final = record.exc_info
+        if exc_info is None or exc_info[1] is None:
+            return MappingProxyType({})
+        error: Final[_ECSError] = {
+            "type": exc_info[0].__name__ if exc_info[0] else None,
+            "message": str(exc_info[1]),
+            "stack_trace": record.exc_text or self.formatException(exc_info),
+        }
+        return MappingProxyType({"error": error})
+
+
 # Function to set up exception handlers for JSON logging
 def _setup_json_exception_handlers(formatter):
     # Create a handler with JSON formatting for exceptions
@@ -855,7 +947,10 @@ def _setup_json_exception_handlers(formatter):
 
 
 # Create a formatter and set it for the handler
-if json_logs:
+if ecs_logs:
+    handler.setFormatter(ECSFormatter())
+    _setup_json_exception_handlers(ECSFormatter())
+elif json_logs:
     handler.setFormatter(JsonFormatter())
     _setup_json_exception_handlers(JsonFormatter())
 else:
@@ -997,7 +1092,7 @@ def _get_uvicorn_json_log_config():
     This ensures that uvicorn's access logs, error logs, and all application logs
     are formatted as JSON when json_logs is enabled.
     """
-    json_formatter_class: Final = "litellm._logging.JsonFormatter"
+    formatter_class: Final = "litellm._logging.ECSFormatter" if ecs_logs else "litellm._logging.JsonFormatter"
 
     # Use the module-level log_level variable for consistency
     uvicorn_log_level: Final = log_level.upper()
@@ -1007,13 +1102,13 @@ def _get_uvicorn_json_log_config():
         "disable_existing_loggers": False,
         "formatters": {
             "json": {
-                "()": json_formatter_class,
+                "()": formatter_class,
             },
             "default": {
-                "()": json_formatter_class,
+                "()": formatter_class,
             },
             "access": {
-                "()": json_formatter_class,
+                "()": formatter_class,
             },
         },
         "handlers": {
@@ -1064,6 +1159,14 @@ def _turn_on_json():
     _setup_json_exception_handlers(JsonFormatter())
 
 
+def _turn_on_ecs():
+    handler: Final = LevelRoutingStreamHandler()
+    handler.setLevel(numeric_level)
+    handler.setFormatter(ECSFormatter())
+    _initialize_loggers_with_handler(handler)
+    _setup_json_exception_handlers(ECSFormatter())
+
+
 def _turn_on_debug():
     verbose_logger.setLevel(level=logging.DEBUG)  # set package log to debug
     verbose_router_logger.setLevel(level=logging.DEBUG)  # set router logs to debug
diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py
index b2c22880f5a..b8a7db3b64b 100644
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/litellm/litellm_core_utils/litellm_logging.py
@@ -1493,7 +1493,7 @@ class Logging(LiteLLMLoggingBaseClass):
         Prints the RAW curl command sent from LiteLLM
         """
         if _is_debugging_on() or self.litellm_request_debug:
-            if litellm.json_logs:
+            if litellm.json_logs or litellm.ecs_logs:
                 masked_headers: Final = self._get_masked_headers(headers or {})
                 masked_api_base: Final = self._get_masked_api_base(str(api_base or ""))
                 if self.litellm_request_debug:
diff --git a/litellm/litellm_core_utils/safe_json_dumps.py b/litellm/litellm_core_utils/safe_json_dumps.py
index 63242a580e7..cdb1f4584b3 100644
--- a/litellm/litellm_core_utils/safe_json_dumps.py
+++ b/litellm/litellm_core_utils/safe_json_dumps.py
@@ -1,5 +1,6 @@
 import json
 from collections.abc import Callable
+from types import MappingProxyType
 from typing import Any, Final
 
 from pydantic import BaseModel
@@ -49,7 +50,7 @@ def safe_json_structure(
             return "CircularReference Detected"
         seen.add(id(obj))
         result: dict | list | tuple | set | str
-        if isinstance(obj, dict):
+        if isinstance(obj, (dict, MappingProxyType)):
             result = {}
             for k, v in obj.items():
                 if isinstance(k, (str)):
diff --git a/litellm/proxy/proxy_cli.py b/litellm/proxy/proxy_cli.py
index 08dfc79def7..271ec9de521 100644
--- a/litellm/proxy/proxy_cli.py
+++ b/litellm/proxy/proxy_cli.py
@@ -309,8 +309,7 @@ class ProxyInitializationHelpers:
         if log_config is not None:
             print(f"Using log_config: {log_config}")
             uvicorn_args["log_config"] = log_config
-        elif litellm.json_logs:
-            # Use JSON log config for uvicorn to ensure all logs (including exceptions) are JSON
+        elif litellm.ecs_logs or litellm.json_logs:
             uvicorn_args["log_config"] = _get_uvicorn_json_log_config()
         elif litellm_log := os.environ.get("LITELLM_LOG"):
             uvicorn_args["log_level"] = resolve_log_level(litellm_log)
diff --git a/litellm/sitecustomize.py b/litellm/sitecustomize.py
new file mode 100644
index 00000000000..de967781cd4
--- /dev/null
+++ b/litellm/sitecustomize.py
@@ -0,0 +1,9 @@
+import os
+
+if os.environ.get("LITELLM_ECS_LOGS", "").lower() == "true":
+    try:
+        from litellm._logging import _turn_on_ecs  # pyright: ignore[reportPrivateUsage]  # public entry point
+
+        _turn_on_ecs()
+    except Exception:
+        pass
diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py
index 9cdbd2a5d31..82ad60991ad 100644
--- a/tests/unit/litellm_core_utils/test_litellm_logging.py
+++ b/tests/unit/litellm_core_utils/test_litellm_logging.py
@@ -9373,3 +9373,32 @@ def test_signoz_dispatch_requires_an_endpoint(monkeypatch):
         logging_module._in_memory_loggers.clear()
         monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
         is_otel_v2_enabled.cache_clear()
+
+
+def test_ecs_logs_keep_the_request_body_out_of_the_debug_log(logging_obj, monkeypatch):
+    import io
+    import logging
+
+    from litellm._logging import verbose_logger
+
+    monkeypatch.setattr(litellm, "json_logs", False)
+    monkeypatch.setattr(litellm, "ecs_logs", True)
+    logging_obj.litellm_request_debug = True
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+    verbose_logger.addHandler(handler)
+    try:
+        logging_obj._print_llm_call_debugging_log(
+            api_base="https://api.example.com/v1",
+            headers={},
+            additional_args={
+                "api_base": "https://api.example.com/v1",
+                "headers": {},
+                "complete_input_dict": {"messages": [{"role": "user", "content": "my-private-prompt-text"}]},
+            },
+        )
+    finally:
+        verbose_logger.removeHandler(handler)
+
+    assert "POST Request Sent from LiteLLM" in stream.getvalue()
+    assert "my-private-prompt-text" not in stream.getvalue()
diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py
index 5cdecc31575..3a878608343 100644
--- a/tests/unit/test_logging.py
+++ b/tests/unit/test_logging.py
@@ -2,6 +2,7 @@ import ast
 import asyncio
 import base64
 import dataclasses
+import importlib
 import json
 import logging
 import re
@@ -25,6 +26,7 @@ from litellm._logging import (
     CorrelationContextFilter,
     CorrelationPlainFormatter,
     DiagnosticProcessingFilter,
+    ECSFormatter,
     JsonFormatter,
     LevelRoutingStreamHandler,
     SecretRedactionFilter,
@@ -34,6 +36,7 @@ from litellm._logging import (
     _parse_json_logs_env,
     _plain_log_format,
     _stdout_truncation_marker,
+    _turn_on_ecs,
     _turn_on_json,
     format_base64_size,
     session_id_var,
@@ -1766,3 +1769,223 @@ def test_diagnostic_filter_redacts_a_non_string_message_object(monkeypatch, nati
     assert DiagnosticProcessingFilter().filter(record) is True
 
     assert secret not in record.getMessage()
+
+
+def test_ecs_formatter_required_fields():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.INFO,
+        pathname="proxy_server.py",
+        lineno=42,
+        msg="test message",
+        args=(),
+        exc_info=None,
+    )
+    obj = json.loads(formatter.format(record))
+
+    assert obj["message"] == "test message"
+    assert "@timestamp" in obj
+    assert obj["log"]["level"] == "info"
+    assert obj["log"]["logger"] == "LiteLLM"
+    assert obj["log"]["origin"]["file"]["name"] == "proxy_server.py"
+    assert obj["log"]["origin"]["file"]["line"] == 42
+    assert obj["service"]["name"] == "litellm"
+    assert obj["ecs"]["version"] == "8.11.0"
+
+
+def test_ecs_formatter_timestamp_is_utc_iso8601_with_ms():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.DEBUG,
+        pathname="",
+        lineno=0,
+        msg="ts test",
+        args=(),
+        exc_info=None,
+    )
+    obj = json.loads(formatter.format(record))
+    assert re.match(
+        r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$", obj["@timestamp"]
+    ), f"Non-ECS timestamp: {obj['@timestamp']!r}"
+
+
+def test_ecs_formatter_log_level_is_lowercase():
+    formatter = ECSFormatter()
+    for level, expected in [
+        (logging.DEBUG, "debug"),
+        (logging.INFO, "info"),
+        (logging.WARNING, "warning"),
+        (logging.ERROR, "error"),
+        (logging.CRITICAL, "critical"),
+    ]:
+        record = logging.LogRecord(
+            name="LiteLLM",
+            level=level,
+            pathname="",
+            lineno=0,
+            msg="test",
+            args=(),
+            exc_info=None,
+        )
+        obj = json.loads(formatter.format(record))
+        assert obj["log"]["level"] == expected
+
+
+def test_ecs_formatter_error_fields_on_exception():
+    formatter = ECSFormatter()
+    try:
+        raise ValueError("something broke")
+    except ValueError:
+        exc_info = sys.exc_info()
+
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.ERROR,
+        pathname="",
+        lineno=0,
+        msg="error occurred",
+        args=(),
+        exc_info=exc_info,
+    )
+    record.exc_text = formatter.formatException(exc_info)
+    obj = json.loads(formatter.format(record))
+
+    assert "error" in obj
+    assert obj["error"]["type"] == "ValueError"
+    assert obj["error"]["message"] == "something broke"
+    assert "stack_trace" in obj["error"]
+    assert "ValueError" in obj["error"]["stack_trace"]
+
+
+def test_ecs_formatter_extra_fields_passthrough():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.DEBUG,
+        pathname="",
+        lineno=0,
+        msg="request received",
+        args=(),
+        exc_info=None,
+    )
+    record.api_base = "https://api.openai.com"
+    record.model = "gpt-4"
+    obj = json.loads(formatter.format(record))
+
+    assert obj["api_base"] == "https://api.openai.com"
+    assert obj["model"] == "gpt-4"
+
+
+def test_ecs_formatter_redacts_a_credential_in_a_structured_extra_value():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.DEBUG,
+        pathname="",
+        lineno=0,
+        msg="request sent",
+        args=(),
+        exc_info=None,
+    )
+    record.litellm_params = {"api_key": "sk-1234567890abcdefghijklmnopqrstuvwxyz"}
+    obj = json.loads(formatter.format(record))
+
+    assert "sk-1234567890abcdefghijklmnopqrstuvwxyz" not in json.dumps(obj)
+    assert "REDACTED" in obj["litellm_params"]["api_key"]
+
+
+def test_ecs_formatter_no_ecs_reserved_key_collision():
+    formatter = ECSFormatter()
+    record = logging.LogRecord(
+        name="LiteLLM",
+        level=logging.INFO,
+        pathname="",
+        lineno=0,
+        msg="test",
+        args=(),
+        exc_info=None,
+    )
+    record.message = "injected"
+    obj = json.loads(formatter.format(record))
+    assert obj["message"] == "test"
+
+
+def test_ecs_mode_emits_one_record_per_logger(capfd):
+    _turn_on_ecs()
+    for lg in (verbose_logger, verbose_router_logger, verbose_proxy_logger):
+        lg.setLevel(logging.INFO)
+
+    verbose_logger.info("first info")
+    verbose_router_logger.info("second info from router")
+    verbose_proxy_logger.info("third info from proxy")
+
+    out, err = capfd.readouterr()
+    assert [raw for raw in err.splitlines() if raw.strip()] == []
+    lines = [raw for raw in out.splitlines() if raw.strip()]
+
+    assert len(lines) == 3, f"got {len(lines)} lines, want 3: {lines!r}"
+    for line in lines:
+        obj = json.loads(line)
+        assert "@timestamp" in obj
+        assert obj["log"]["level"] == "info"
+        assert obj["ecs"]["version"] == "8.11.0"
+        assert obj["service"]["name"] == "litellm"
+        assert "litellm_redacted" not in obj
+
+
+def test_get_uvicorn_json_log_config_uses_ecs_formatter_when_ecs_logs_enabled(monkeypatch):
+    import litellm._logging as litellm_logging
+
+    monkeypatch.setattr(litellm_logging, "ecs_logs", True)
+    log_config = _get_uvicorn_json_log_config()
+
+    for formatter_config in log_config["formatters"].values():
+        assert formatter_config["()"] == "litellm._logging.ECSFormatter"
+
+
+def _run_sitecustomize_hook():
+    return importlib.reload(importlib.import_module("litellm.sitecustomize"))
+
+
+def _loggers_are_on_ecs() -> bool:
+    return any(isinstance(handler.formatter, ECSFormatter) for handler in verbose_logger.handlers)
+
+
+def test_sitecustomize_hook_turns_on_ecs_when_the_env_var_is_set(monkeypatch):
+    plain = logging.StreamHandler()
+    plain.setFormatter(JsonFormatter())
+    _initialize_loggers_with_handler(plain)
+    monkeypatch.setenv("LITELLM_ECS_LOGS", "true")
+
+    _run_sitecustomize_hook()
+
+    assert _loggers_are_on_ecs()
+
+
+def test_sitecustomize_hook_leaves_logging_alone_when_the_env_var_is_unset(monkeypatch):
+    plain = logging.StreamHandler()
+    plain.setFormatter(JsonFormatter())
+    _initialize_loggers_with_handler(plain)
+    monkeypatch.delenv("LITELLM_ECS_LOGS", raising=False)
+
+    _run_sitecustomize_hook()
+
+    assert not _loggers_are_on_ecs()
+
+
+def test_sitecustomize_hook_never_breaks_interpreter_startup(monkeypatch):
+
+    def _raise(*_args, **_kwargs):
+        raise RuntimeError("litellm is half-installed")
+
+    plain = logging.StreamHandler()
+    plain.setFormatter(JsonFormatter())
+    _initialize_loggers_with_handler(plain)
+    monkeypatch.setenv("LITELLM_ECS_LOGS", "true")
+    monkeypatch.setattr(litellm._logging, "_turn_on_ecs", _raise)
+
+    _run_sitecustomize_hook()
+
+    assert not _loggers_are_on_ecs()