feat(otel): let team and key Arize callbacks choose the OTLP transport (#44492)

Co-authored-by: yassin <yassin@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/integrations/arize/arize.py

@@ -17,7 +17,7 @@ from litellm.integrations.arize._utils import ArizeOTELAttributes
 from litellm.integrations.opentelemetry import _MAX_DYNAMIC_TRACER_PROVIDERS, OpenTelemetry, OpenTelemetryConfig
 from litellm.types.integrations.arize import ArizeConfig
 from litellm.types.services import ServiceLoggerPayload
-from litellm.types.utils import StandardCallbackDynamicParams
+from litellm.types.utils import ArizeOtlpProtocol, StandardCallbackDynamicParams
 
 if TYPE_CHECKING:
     from opentelemetry.trace import Span as _Span
@@ -174,7 +174,7 @@ class ArizeLogger(OpenTelemetry):
         _utils.set_attributes(span, kwargs, response_obj, ArizeOTELAttributes)
 
     @staticmethod
-    def get_arize_config() -> ArizeConfig:
+    def get_arize_config(otlp_protocol: ArizeOtlpProtocol | None = None) -> ArizeConfig:
         """
         Helper function to get Arize configuration.
 
@@ -195,7 +195,19 @@ class ArizeLogger(OpenTelemetry):
         endpoint = None
         protocol: Protocol = "otlp_grpc"
 
-        if grpc_endpoint:
+        if otlp_protocol == "http/protobuf":
+            protocol = "otlp_http"
+            if http_endpoint:
+                endpoint = http_endpoint
+            elif grpc_endpoint:
+                base: Final = grpc_endpoint.rstrip("/")
+                endpoint = f"{base}/traces" if base.endswith("/v1") else base
+            else:
+                endpoint = "https://otlp.arize.com/v1/traces"
+        elif otlp_protocol == "grpc":
+            protocol = "otlp_grpc"
+            endpoint = grpc_endpoint or http_endpoint or "https://otlp.arize.com/v1"
+        elif grpc_endpoint:
             protocol = "otlp_grpc"
             endpoint = grpc_endpoint
         elif http_endpoint:

litellm/integrations/otel/presets/destinations.py

@@ -63,7 +63,7 @@ def _langfuse_destination(params: StandardCallbackDynamicParams) -> "_Destinatio
 def _arize_destination(params: StandardCallbackDynamicParams) -> "_Destination | None":
     from litellm.integrations.arize.arize import ArizeLogger
 
-    config: Final = ArizeLogger.get_arize_config()
+    config: Final = ArizeLogger.get_arize_config(params.get("arize_otlp_protocol"))
     return (config.endpoint, config.protocol)
 
 

litellm/litellm_core_utils/initialize_dynamic_callback_params.py

@@ -4,7 +4,12 @@ from contextlib import contextmanager
 from contextvars import ContextVar
 from typing import Any, Final
 
-from litellm.types.utils import OTEL_SPAN_SCOPES, TRUSTED_CALLBACK_VARS_FIELD, StandardCallbackDynamicParams
+from litellm.types.utils import (
+    ARIZE_OTLP_PROTOCOLS,
+    OTEL_SPAN_SCOPES,
+    TRUSTED_CALLBACK_VARS_FIELD,
+    StandardCallbackDynamicParams,
+)
 
 _CLIENT_CALLBACK_METADATA_SLOTS: Final[tuple[str, ...]] = ("litellm_metadata", "metadata")
 _inherited_message_logging_disabled: Final[ContextVar[bool]] = ContextVar(
@@ -81,6 +86,11 @@ def validate_langfuse_span_scope_value(value: str) -> None:
         raise ValueError(f"Invalid langfuse_span_scope {value!r}: must be one of {sorted(OTEL_SPAN_SCOPES)}")
 
 
+def validate_arize_otlp_protocol_value(value: str) -> None:
+    if value not in ARIZE_OTLP_PROTOCOLS:
+        raise ValueError(f"Invalid arize_otlp_protocol {value!r}: must be one of {sorted(ARIZE_OTLP_PROTOCOLS)}")
+
+
 # Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict
 _supported_callback_params: Final[tuple[str, ...]] = (
     "langfuse_public_key",

litellm/proxy/_types.py

@@ -25,6 +25,7 @@ from typing_extensions import NotRequired, ReadOnly, Required, TypedDict
 from litellm._uuid import uuid
 from litellm.constants import DEFAULT_STAGGER_WINDOW_SECONDS, MCP_STDIO_ALLOWED_COMMANDS
 from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+    validate_arize_otlp_protocol_value,
     validate_langfuse_environment_value,
     validate_langfuse_span_scope_value,
     validate_no_callback_env_reference,
@@ -2383,6 +2384,8 @@ class AddTeamCallback(LiteLLMPydanticObjectBase):
                 validate_langfuse_environment_value(callback_vars[key])
             if key == "langfuse_span_scope":
                 validate_langfuse_span_scope_value(callback_vars[key])
+            if key == "arize_otlp_protocol":
+                validate_arize_otlp_protocol_value(callback_vars[key])
         return values
 
 

litellm/proxy/common_utils/callback_config_validation.py

@@ -15,15 +15,20 @@ _NEWRELIC_VAR_PREFIX: Final = "newrelic_"
 _LANGFUSE_OTEL_CALLBACK: Final = "langfuse_otel"
 _LANGFUSE_SPAN_SCOPE_VAR: Final = "langfuse_span_scope"
 _ARIZE_CALLBACK: Final = "arize"
+_ARIZE_OTLP_PROTOCOL_VAR: Final = "arize_otlp_protocol"
 _ARIZE_SAMPLING_RATE_VARS: Final[frozenset[str]] = frozenset(
     {"arize_success_sampling_rate", "arize_error_sampling_rate"}
 )
 
 
-def callback_config_error(callback_name: str | None, callback_vars: Mapping[str, str] | None) -> str | None:
+def callback_config_error(
+    callback_name: str | None, callback_vars: Mapping[str, str] | None, callback_type: str | None = None
+) -> str | None:
     if not callback_vars:
         return None
-    arize_error: Final = _arize_sampling_rate_error(callback_name, callback_vars)
+    arize_error: Final = _arize_sampling_rate_error(callback_name, callback_vars) or _arize_otlp_protocol_error(
+        callback_name, callback_vars, callback_type
+    )
     if arize_error is not None:
         return arize_error
     langfuse_error: Final = _langfuse_environment_error(callback_vars) or _langfuse_span_scope_error(
@@ -94,7 +99,9 @@ _VAR_FAMILIES: Final[Mapping[str, str]] = MappingProxyType(
     }
 )
 
-_FAMILY_OPTION_VARS: Final[frozenset[str]] = frozenset({_LANGFUSE_SPAN_SCOPE_VAR, *_ARIZE_SAMPLING_RATE_VARS})
+_FAMILY_OPTION_VARS: Final[frozenset[str]] = frozenset(
+    {_LANGFUSE_SPAN_SCOPE_VAR, _ARIZE_OTLP_PROTOCOL_VAR, *_ARIZE_SAMPLING_RATE_VARS}
+)
 
 
 def _family_of(var: str) -> str | None:
@@ -217,7 +224,10 @@ def _logging_entry_error(entry: object) -> str | None:
     callback_name: Final = entry.get("callback_name")
     if not isinstance(callback_name, str) or not isinstance(entry.get("callback_vars"), Mapping):
         return None
-    return callback_config_error(callback_name, _entry_callback_vars(entry))
+    callback_type_raw: Final = entry.get("callback_type")
+    return callback_config_error(
+        callback_name, _entry_callback_vars(entry), callback_type_raw if isinstance(callback_type_raw, str) else None
+    )
 
 
 def _arize_sampling_rate_error(callback_name: str | None, callback_vars: Mapping[str, str]) -> str | None:
@@ -236,6 +246,30 @@ def _arize_sampling_rate_error(callback_name: str | None, callback_vars: Mapping
     return None
 
 
+def _arize_otlp_protocol_error(
+    callback_name: str | None, callback_vars: Mapping[str, str], callback_type: str | None = None
+) -> str | None:
+    value: Final = callback_vars.get(_ARIZE_OTLP_PROTOCOL_VAR)
+    if value is None:
+        return None
+    if callback_name != _ARIZE_CALLBACK:
+        return f"{_ARIZE_OTLP_PROTOCOL_VAR} applies to the {_ARIZE_CALLBACK} callback only, not {callback_name!r}"
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+    from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+        validate_arize_otlp_protocol_value,
+    )
+
+    try:
+        validate_arize_otlp_protocol_value(value)
+    except ValueError as e:
+        return str(e)
+    if callback_type == "failure":
+        return f"{_ARIZE_OTLP_PROTOCOL_VAR} needs callback_type 'success' or 'success_and_failure'; failure-only Arize callbacks export over the proxy's own Arize transport"
+    if not is_otel_v2_enabled():
+        return "Per-team Arize transport selection requires the proxy to run with LITELLM_OTEL_V2=true."
+    return None
+
+
 def _newrelic_config_error(callback_vars: Mapping[str, str]) -> str | None:
     """Per-team New Relic routing runs on the OTel v2 path only.
 

litellm/proxy/management_endpoints/team_callback_endpoints.py

@@ -61,7 +61,7 @@ def _callback_config_error(message: str) -> HTTPException:
 
 
 def _validate_team_callback(data: "AddTeamCallback") -> None:
-    error: Final = callback_config_error(data.callback_name, data.callback_vars)
+    error: Final = callback_config_error(data.callback_name, data.callback_vars, data.callback_type)
     if error is not None:
         raise _callback_config_error(error)
 

litellm/types/utils.py

@@ -3648,6 +3648,9 @@ OPENAI_RESPONSE_HEADERS: Final = [
 OtelSpanScope = Literal["full", "llm_only"]
 OTEL_SPAN_SCOPES: Final[frozenset[str]] = frozenset(get_args(OtelSpanScope))
 
+ArizeOtlpProtocol = Literal["grpc", "http/protobuf"]
+ARIZE_OTLP_PROTOCOLS: Final[frozenset[str]] = frozenset(get_args(ArizeOtlpProtocol))
+
 
 class StandardCallbackDynamicParams(TypedDict, total=False):
     # Langfuse dynamic params
@@ -3681,6 +3684,7 @@ class StandardCallbackDynamicParams(TypedDict, total=False):
     arize_space_id: str | None
     arize_success_sampling_rate: ReadOnly[float | None]
     arize_error_sampling_rate: ReadOnly[float | None]
+    arize_otlp_protocol: ReadOnly[ArizeOtlpProtocol | None]
 
     # PostHog dynamic params
     posthog_api_key: str | None

tests/unit/integrations/otel/test_otel_v2_destinations.py

@@ -3081,6 +3081,126 @@ class TestBackendEndpointParity:
         assert destination.endpoint == "https://trace.wandb.ai/otel/v1/traces"
 
 
+class TestArizeOtlpProtocol:
+    @staticmethod
+    def _arize(monkeypatch, endpoint_env, **env):
+        monkeypatch.delenv("ARIZE_ENDPOINT", raising=False)
+        monkeypatch.delenv("ARIZE_HTTP_ENDPOINT", raising=False)
+        for name, value in env.items():
+            monkeypatch.setenv(name, value)
+        return destination_for("arize", {"arize_space_id": "s", "arize_api_key": "k", **endpoint_env})
+
+    def test_http_transport_rewrites_the_grpc_endpoint_to_the_traces_path(self, monkeypatch):
+        destination = self._arize(
+            monkeypatch,
+            {"arize_otlp_protocol": "http/protobuf"},
+            ARIZE_ENDPOINT="https://arize.internal.example/v1",
+        )
+
+        assert destination.protocol == "otlp_http"
+        assert destination.endpoint == "https://arize.internal.example/v1/traces"
+
+    def test_http_transport_prefers_the_http_endpoint(self, monkeypatch):
+        destination = self._arize(
+            monkeypatch,
+            {"arize_otlp_protocol": "http/protobuf"},
+            ARIZE_ENDPOINT="https://arize.internal.example/v1",
+            ARIZE_HTTP_ENDPOINT="https://http.arize.internal.example/v1/traces",
+        )
+
+        assert destination.protocol == "otlp_http"
+        assert destination.endpoint == "https://http.arize.internal.example/v1/traces"
+
+    def test_http_transport_falls_back_to_the_arize_cloud_http_endpoint(self, monkeypatch):
+        destination = self._arize(monkeypatch, {"arize_otlp_protocol": "http/protobuf"})
+
+        assert destination.protocol == "otlp_http"
+        assert destination.endpoint == "https://otlp.arize.com/v1/traces"
+
+    def test_grpc_transport_accepts_the_http_endpoint_as_the_grpc_target(self, monkeypatch):
+        destination = self._arize(
+            monkeypatch,
+            {"arize_otlp_protocol": "grpc"},
+            ARIZE_HTTP_ENDPOINT="https://http.arize.internal.example/v1/traces",
+        )
+
+        assert destination.protocol == "otlp_grpc"
+        assert destination.endpoint == "https://http.arize.internal.example/v1/traces"
+
+    def test_no_transport_var_keeps_the_grpc_env_default(self, monkeypatch):
+        destination = self._arize(monkeypatch, {}, ARIZE_ENDPOINT="https://arize.internal.example")
+
+        assert destination.protocol == "otlp_grpc"
+        assert destination.endpoint == "https://arize.internal.example"
+
+    def test_the_http_destination_builds_an_http_exporter_on_the_traces_path(self, monkeypatch):
+        from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
+            OTLPSpanExporter as HTTPSpanExporter,
+        )
+        from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+        from litellm.integrations.otel.plumbing.providers import _destination_processor
+
+        destination = self._arize(
+            monkeypatch,
+            {"arize_otlp_protocol": "http/protobuf"},
+            ARIZE_ENDPOINT="https://arize.internal.example/v1",
+        )
+        processor = _destination_processor(destination)
+        try:
+            assert isinstance(processor, BatchSpanProcessor)
+            assert isinstance(processor.span_exporter, HTTPSpanExporter)
+            assert processor.span_exporter._endpoint == "https://arize.internal.example/v1/traces"
+        finally:
+            processor.shutdown()
+
+    def test_the_grpc_destination_builds_a_grpc_exporter(self, monkeypatch):
+        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
+            OTLPSpanExporter as GRPCSpanExporter,
+        )
+        from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+        from litellm.integrations.otel.plumbing.providers import _destination_processor
+
+        destination = self._arize(
+            monkeypatch,
+            {"arize_otlp_protocol": "grpc"},
+            ARIZE_ENDPOINT="https://arize.internal.example",
+        )
+        processor = _destination_processor(destination)
+        try:
+            assert isinstance(processor, BatchSpanProcessor)
+            assert isinstance(processor.span_exporter, GRPCSpanExporter)
+        finally:
+            processor.shutdown()
+
+    def test_a_team_metadata_protocol_reaches_the_destination(self, monkeypatch):
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        monkeypatch.setenv("ARIZE_ENDPOINT", "https://arize.internal.example/v1")
+        monkeypatch.delenv("ARIZE_HTTP_ENDPOINT", raising=False)
+        is_otel_v2_enabled.cache_clear()
+        auth = UserAPIKeyAuth(
+            team_metadata={
+                "logging": [
+                    {
+                        "callback_name": "arize",
+                        "callback_type": "success",
+                        "callback_vars": {
+                            "arize_space_id": "s",
+                            "arize_api_key": "k",
+                            "arize_otlp_protocol": "http/protobuf",
+                        },
+                    }
+                ]
+            }
+        )
+
+        destinations = resolve_tenant_otel_destinations(auth)
+
+        assert [d.protocol for d in destinations] == ["otlp_http"]
+        assert destinations[0].endpoint == "https://arize.internal.example/v1/traces"
+
+
 class TestIncompleteCredentials:
     """Half a credential set builds a non-empty but unusable header dict. Accepting it
     would suppress the operator's exporter and send the trace where it cannot land."""

tests/unit/proxy/common_utils/test_callback_config_validation.py

@@ -121,6 +121,137 @@ def test_arize_sampling_rate_rejected_on_non_arize_callback():
     assert callback_config_error("arize", {"arize_success_sampling_rate": "0.5"}) is None
 
 
+class TestArizeOtlpProtocol:
+    def test_protocol_on_a_non_arize_callback_is_rejected(self, monkeypatch):
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+        is_otel_v2_enabled.cache_clear()
+        try:
+            for callback_name in ["langfuse", "langfuse_otel", "datadog", None]:
+                error = callback_config_error(callback_name, {"arize_otlp_protocol": "grpc"})
+                assert error is not None and "applies to the arize callback only" in error
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_unknown_protocols_are_rejected(self, monkeypatch):
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+        is_otel_v2_enabled.cache_clear()
+        try:
+            for bad in ["otlp_http", "HTTP/PROTOBUF", "http_json", "", "None"]:
+                error = callback_config_error("arize", {"arize_otlp_protocol": bad})
+                assert error is not None and "arize_otlp_protocol" in error and "http/protobuf" in error
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_protocol_is_rejected_while_otel_v2_is_off(self, monkeypatch):
+        monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+        is_otel_v2_enabled.cache_clear()
+        try:
+            error = callback_config_error("arize", {"arize_otlp_protocol": "http/protobuf"})
+            assert error is not None and "LITELLM_OTEL_V2" in error
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_both_protocols_are_accepted_with_otel_v2_on(self, monkeypatch):
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+        is_otel_v2_enabled.cache_clear()
+        try:
+            assert callback_config_error("arize", {"arize_otlp_protocol": "grpc"}) is None
+            assert callback_config_error("arize", {"arize_otlp_protocol": "http/protobuf"}) is None
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_protocol_is_not_a_family_credential(self):
+        stored = [{"arize_api_key": "k1"}]
+        assert cross_entry_family_error({"arize_otlp_protocol": "grpc"}, stored) is None
+        assert cross_entry_family_error({"arize_otlp_protocol": "http/protobuf"}, stored) is None
+
+    def test_protocol_on_a_key_logging_entry_is_validated(self, monkeypatch):
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+        is_otel_v2_enabled.cache_clear()
+        try:
+            invalid_metadata = {
+                "logging": [
+                    {
+                        "callback_name": "arize",
+                        "callback_type": "success",
+                        "callback_vars": {"arize_otlp_protocol": "carrier-pigeon"},
+                    }
+                ]
+            }
+            error = logging_metadata_config_error(invalid_metadata)
+            assert error is not None and "arize_otlp_protocol" in error
+
+            valid_metadata = {
+                "logging": [
+                    {
+                        "callback_name": "arize",
+                        "callback_type": "success",
+                        "callback_vars": {"arize_otlp_protocol": "http/protobuf"},
+                    }
+                ]
+            }
+            assert logging_metadata_config_error(valid_metadata) is None
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_failure_only_arize_callbacks_reject_the_protocol(self, monkeypatch):
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+        is_otel_v2_enabled.cache_clear()
+        try:
+            error = callback_config_error("arize", {"arize_otlp_protocol": "http/protobuf"}, "failure")
+            assert error is not None and "arize_otlp_protocol" in error and "failure" in error
+            assert callback_config_error("arize", {"arize_otlp_protocol": "http/protobuf"}, "success") is None
+            assert (
+                callback_config_error("arize", {"arize_otlp_protocol": "http/protobuf"}, "success_and_failure") is None
+            )
+            assert callback_config_error("arize", {"arize_otlp_protocol": "http/protobuf"}) is None
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_failure_only_key_logging_entry_rejects_the_protocol(self, monkeypatch):
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+
+        is_otel_v2_enabled.cache_clear()
+        try:
+            failure_metadata = {
+                "logging": [
+                    {
+                        "callback_name": "arize",
+                        "callback_type": "failure",
+                        "callback_vars": {"arize_otlp_protocol": "http/protobuf"},
+                    }
+                ]
+            }
+            error = logging_metadata_config_error(failure_metadata)
+            assert error is not None and "arize_otlp_protocol" in error
+
+            success_metadata = {
+                "logging": [
+                    {
+                        "callback_name": "arize",
+                        "callback_type": "success",
+                        "callback_vars": {"arize_otlp_protocol": "http/protobuf"},
+                    }
+                ]
+            }
+            assert logging_metadata_config_error(success_metadata) is None
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+
 def test_arize_sampling_rates_are_not_family_credentials():
     """The rates choose what the Arize family exports, not where it sends, so an
     entry that repeats or adds a rate next to a stored Arize entry is not the

tests/unit/proxy/management_endpoints/test_callback_management_endpoints.py

@@ -361,6 +361,70 @@ class TestNewRelicTeamCallbackValidation:
             is_otel_v2_enabled.cache_clear()
 
 
+class TestArizeOtlpProtocolValidation:
+    def test_valid_protocols_are_accepted(self):
+        from litellm.proxy._types import AddTeamCallback
+
+        for protocol in ("grpc", "http/protobuf"):
+            data = AddTeamCallback(
+                callback_name="arize",
+                callback_type="success",
+                callback_vars={"arize_api_key": "k", "arize_space_id": "s", "arize_otlp_protocol": protocol},
+            )
+            assert data.callback_vars["arize_otlp_protocol"] == protocol
+
+    def test_an_invalid_protocol_is_rejected_at_model_validation(self):
+        from pydantic import ValidationError
+
+        from litellm.proxy._types import AddTeamCallback
+
+        with pytest.raises(ValidationError, match="arize_otlp_protocol"):
+            AddTeamCallback(
+                callback_name="arize",
+                callback_type="success",
+                callback_vars={"arize_otlp_protocol": "otlp_grpc"},
+            )
+
+    def test_a_failure_only_team_callback_rejects_the_protocol(self, monkeypatch):
+        from fastapi import HTTPException
+
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+        from litellm.proxy._types import AddTeamCallback
+        from litellm.proxy.management_endpoints.team_callback_endpoints import _validate_team_callback
+
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        is_otel_v2_enabled.cache_clear()
+        try:
+            with pytest.raises(HTTPException) as exc:
+                _validate_team_callback(
+                    AddTeamCallback(
+                        callback_name="arize",
+                        callback_type="failure",
+                        callback_vars={
+                            "arize_api_key": "k",
+                            "arize_space_id": "s",
+                            "arize_otlp_protocol": "http/protobuf",
+                        },
+                    )
+                )
+            assert exc.value.status_code == 400
+            assert "arize_otlp_protocol" in str(exc.value.detail)
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_the_raw_otlp_env_var_name_stays_rejected(self):
+        from pydantic import ValidationError
+
+        from litellm.proxy._types import AddTeamCallback
+
+        with pytest.raises(ValidationError, match="Invalid callback variable"):
+            AddTeamCallback(
+                callback_name="arize",
+                callback_type="success",
+                callback_vars={"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf"},
+            )
+
+
 class TestNewRelicKeyLoggingValidation:
     """Key-level logging is written through key metadata, not /team/callback."""
 

tests/unit/proxy/test_credential_slot_registry.py

@@ -89,6 +89,7 @@ CALLBACK_PARAM_CLASSIFICATION: Final[Mapping[str, Classification]] = MappingProx
         "arize_space_id": NotSecret("space identifier"),
         "arize_success_sampling_rate": NotSecret("sampling rate"),
         "arize_error_sampling_rate": NotSecret("sampling rate"),
+        "arize_otlp_protocol": NotSecret("transport protocol enum"),
         "posthog_api_key": Unplanted(),
         "posthog_api_url": NotSecret("sink endpoint URL"),
         "wandb_api_key": Unplanted(),

tests/unit/proxy/test_litellm_pre_call_utils.py

@@ -8590,3 +8590,38 @@ def test_signoz_callback_vars_are_scoped_to_the_signoz_callback():
         team_callback_settings_obj=None,
     )
     assert under_other.callback_vars == {"langfuse_host": "https://cloud.langfuse.com"}
+
+
+def test_arize_otlp_protocol_on_a_key_logging_entry_reaches_the_destination(monkeypatch):
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+    from litellm.proxy.litellm_pre_call_utils import resolve_tenant_otel_destinations
+
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    monkeypatch.setenv("ARIZE_ENDPOINT", "https://arize.internal.example/v1")
+    monkeypatch.delenv("ARIZE_HTTP_ENDPOINT", raising=False)
+    is_otel_v2_enabled.cache_clear()
+    try:
+        auth = UserAPIKeyAuth(
+            api_key="hashed-key",
+            metadata={
+                "logging": [
+                    {
+                        "callback_name": "arize",
+                        "callback_type": "success",
+                        "callback_vars": {
+                            "arize_space_id": "s",
+                            "arize_api_key": "k",
+                            "arize_otlp_protocol": "http/protobuf",
+                        },
+                    }
+                ]
+            },
+            team_metadata={},
+        )
+
+        destinations = resolve_tenant_otel_destinations(auth)
+
+        assert [d.protocol for d in destinations] == ["otlp_http"]
+        assert destinations[0].endpoint == "https://arize.internal.example/v1/traces"
+    finally:
+        is_otel_v2_enabled.cache_clear()

ui/litellm-dashboard/src/components/callback_info_helpers.tsx

@@ -20,6 +20,7 @@ interface CallbackConfig {
   supports_key_team_logging: boolean;
   dynamic_params: Record<string, "text" | "password" | "select" | "upload" | "number">;
   dynamic_param_options?: Record<string, readonly string[]>;
+  success_event_params?: string[];
   description: string;
 }
 
@@ -34,7 +35,12 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [
       arize_space_id: "password",
       arize_success_sampling_rate: "number",
       arize_error_sampling_rate: "number",
+      arize_otlp_protocol: "select",
     },
+    dynamic_param_options: {
+      arize_otlp_protocol: ["grpc", "http/protobuf"],
+    },
+    success_event_params: ["arize_otlp_protocol"],
     description: "Arize Logging Integration",
   },
   {

ui/litellm-dashboard/src/components/team/LoggingSettings.test.tsx

@@ -239,6 +239,94 @@ describe("LoggingSettings", () => {
     ]);
   });
 
+  it("offers the Arize OTLP protocol as a pick between grpc and http/protobuf rather than free text", async () => {
+    const user = userEvent.setup({ delay: null });
+    const mockOnChange = vi.fn();
+    const initialValue = [
+      {
+        callback_name: "arize",
+        callback_type: "success",
+        callback_vars: {},
+      },
+    ];
+
+    renderWithProviders(<LoggingSettings value={initialValue} onChange={mockOnChange} />);
+
+    expect(screen.queryByPlaceholderText("os.environ/ARIZE_OTLP_PROTOCOL")).not.toBeInTheDocument();
+    await user.click(screen.getByRole("combobox", { name: "arize otlp protocol" }));
+    expect((await screen.findAllByRole("option")).map((option) => option.textContent)).toEqual([
+      "grpc",
+      "http/protobuf",
+    ]);
+    await user.click(screen.getByRole("option", { name: "http/protobuf" }));
+
+    expect(mockOnChange).toHaveBeenCalledWith([
+      expect.objectContaining({ callback_vars: expect.objectContaining({ arize_otlp_protocol: "http/protobuf" }) }),
+    ]);
+  });
+
+  it("hides the Arize OTLP protocol picker on a failure-only Arize callback", () => {
+    const mockOnChange = vi.fn();
+    const initialValue = [
+      {
+        callback_name: "arize",
+        callback_type: "failure",
+        callback_vars: {},
+      },
+    ];
+
+    renderWithProviders(<LoggingSettings value={initialValue} onChange={mockOnChange} />);
+
+    expect(screen.queryByRole("combobox", { name: "arize otlp protocol" })).not.toBeInTheDocument();
+    expect(screen.getByPlaceholderText("os.environ/ARIZE_API_KEY")).toBeInTheDocument();
+    expect(screen.getByPlaceholderText("os.environ/ARIZE_SPACE_ID")).toBeInTheDocument();
+  });
+
+  it("drops the Arize OTLP protocol from callback_vars when the event type switches to failure", async () => {
+    const user = userEvent.setup({ delay: null });
+    const mockOnChange = vi.fn();
+    const initialValue = [
+      {
+        callback_name: "arize",
+        callback_type: "success",
+        callback_vars: { arize_space_id: "s", arize_otlp_protocol: "http/protobuf" },
+      },
+    ];
+
+    renderWithProviders(<LoggingSettings value={initialValue} onChange={mockOnChange} />);
+
+    await user.click(screen.getByRole("combobox", { name: "Event Type" }));
+    await user.click(await screen.findByRole("option", { name: "Failure Only" }));
+
+    const lastCall = mockOnChange.mock.calls[mockOnChange.mock.calls.length - 1];
+    expect(lastCall[0][0].callback_type).toBe("failure");
+    expect(lastCall[0][0].callback_vars).toEqual({ arize_space_id: "s" });
+  });
+
+  it("keeps the Arize OTLP protocol in callback_vars when the event type switches to success_and_failure", async () => {
+    const user = userEvent.setup({ delay: null });
+    const mockOnChange = vi.fn();
+    const initialValue = [
+      {
+        callback_name: "arize",
+        callback_type: "success",
+        callback_vars: { arize_space_id: "s", arize_otlp_protocol: "http/protobuf" },
+      },
+    ];
+
+    renderWithProviders(<LoggingSettings value={initialValue} onChange={mockOnChange} />);
+
+    await user.click(screen.getByRole("combobox", { name: "Event Type" }));
+    await user.click(await screen.findByRole("option", { name: "Success & Failure" }));
+
+    const lastCall = mockOnChange.mock.calls[mockOnChange.mock.calls.length - 1];
+    expect(lastCall[0][0].callback_type).toBe("success_and_failure");
+    expect(lastCall[0][0].callback_vars).toEqual({
+      arize_space_id: "s",
+      arize_otlp_protocol: "http/protobuf",
+    });
+  });
+
   it("renders sampling rate inputs for the Arize callback and records changes", () => {
     const mockOnChange = vi.fn();
 

ui/litellm-dashboard/src/components/team/LoggingSettings.tsx

@@ -114,6 +114,21 @@ const LoggingSettings: React.FC<LoggingSettingsProps> = ({
         [field]: callbackValue,
         callback_vars: {},
       };
+    } else if (field === "callback_type" && newValue === "failure") {
+      const callbackDisplayName = Object.entries(callback_map).find(
+        ([_, value]) => value === updatedConfigs[index].callback_name,
+      )?.[0];
+      const successOnlyParams = new Set(
+        (callbackDisplayName && callbackInfo[callbackDisplayName]?.success_event_params) || [],
+      );
+      const callbackVars = Object.fromEntries(
+        Object.entries(updatedConfigs[index].callback_vars).filter(([key]) => !successOnlyParams.has(key)),
+      );
+      updatedConfigs[index] = {
+        ...updatedConfigs[index],
+        [field]: newValue,
+        callback_vars: callbackVars,
+      };
     } else {
       updatedConfigs[index] = {
         ...updatedConfigs[index],
@@ -196,6 +211,10 @@ const LoggingSettings: React.FC<LoggingSettingsProps> = ({
 
     const dynamicParams = callbackInfo[callbackDisplayName]?.dynamic_params || {};
     const paramOptions = callbackInfo[callbackDisplayName]?.dynamic_param_options || {};
+    const successOnlyParams =
+      config.callback_type === "failure"
+        ? new Set(callbackInfo[callbackDisplayName]?.success_event_params ?? [])
+        : new Set<string>();
 
     if (Object.keys(dynamicParams).length === 0) return null;
 
@@ -208,22 +227,24 @@ const LoggingSettings: React.FC<LoggingSettingsProps> = ({
           <span className="text-sm font-medium text-foreground">Integration Parameters</span>
         </div>
         <div className="grid grid-cols-1 gap-4">
-          {Object.entries(dynamicParams).map(([paramName, paramType]) => (
-            <div key={paramName} className="space-y-2">
-              <label className="text-sm font-medium text-foreground capitalize flex items-center space-x-1">
-                <span>{paramName.replace(/_/g, " ")}</span>
-                {paramType === "password" && <Badge variant="secondary">Sensitive</Badge>}
-                {paramType === "number" && <Badge variant="secondary">Number</Badge>}
-              </label>
-              {paramType === "number" && (
-                <span className="text-xs text-muted-foreground">Value must be between 0 and 1</span>
-              )}
-              {renderParamControl(config, configIndex, paramName, {
-                type: paramType,
-                options: paramType === "select" ? paramOptions[paramName] || [] : [],
-              })}
-            </div>
-          ))}
+          {Object.entries(dynamicParams)
+            .filter(([paramName]) => !successOnlyParams.has(paramName))
+            .map(([paramName, paramType]) => (
+              <div key={paramName} className="space-y-2">
+                <label className="text-sm font-medium text-foreground capitalize flex items-center space-x-1">
+                  <span>{paramName.replace(/_/g, " ")}</span>
+                  {paramType === "password" && <Badge variant="secondary">Sensitive</Badge>}
+                  {paramType === "number" && <Badge variant="secondary">Number</Badge>}
+                </label>
+                {paramType === "number" && (
+                  <span className="text-xs text-muted-foreground">Value must be between 0 and 1</span>
+                )}
+                {renderParamControl(config, configIndex, paramName, {
+                  type: paramType,
+                  options: paramType === "select" ? paramOptions[paramName] || [] : [],
+                })}
+              </div>
+            ))}
         </div>
       </div>
     );