feat(langfuse): support langfuse_environment as a per-key dynamic callback param (#38264)

* feat(langfuse): support langfuse_environment as a per-key dynamic callback param

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(langfuse): type the langfuse_environment constructor param

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(langfuse): only pass environment when the SDK client supports it

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(langfuse): drop the request-body metadata test for langfuse_environment

The proxy bans request-body callback params by default (derived from
_supported_callback_params in auth_utils), so the metadata channel this
test asserted is rejected with a 401 on the proxy. The supported channel
is admin-set key/team callback_vars, with LANGFUSE_TRACING_ENVIRONMENT
as the deployment-wide fallback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(langfuse): validate langfuse_environment, avoid redundant clients, honor it in langfuse_otel

Closes the review gaps on the langfuse_environment param:

- Validate values against Langfuse's environment pattern at save time
  (/key/generate, /key/update, /team callback all 400 on e.g. 'Production'
  instead of 200-then-silently-dropping every trace server-side) and at
  logger init; non-string values are str()-coerced instead of crashing
  the SDK's regex check per event.
- Treat empty/whitespace values and values equal to the deployment-wide
  LANGFUSE_TRACING_ENVIRONMENT as non-dynamic so an environment-only
  override that changes nothing no longer mints a duplicate SDK client
  against MAX_LANGFUSE_INITIALIZED_CLIENTS.
- langfuse_otel now reads the per-key/team langfuse_environment from
  standard_callback_dynamic_params instead of only the env var.
- Advertise the param on the discovery surfaces: callback_configs.json
  (langfuse + langfuse_otel), the dashboard callback registry, and the
  /team/{team_id}/callback docstring (schema.d.ts regenerated).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* style: ruff format langfuse files

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(lint): remove duplicate test import, LIT002 dict literal, and mock-echo otel test

- drop redundant in-function import of callback_config_error (F811)
- avoid the `or {}` mutable literal in _set_langfuse_specific_attributes (LIT002)
- rewrite the dynamic-env otel test to observe span.set_attribute output
  instead of patching litellm internals (TQ002/TQ008)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: milan <milan@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: yucheng-berri <yucheng@berri.ai>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

litellm/integrations/callback_configs.json

@@ -220,6 +220,12 @@
         "ui_name": "Host URL",
         "description": "Langfuse host URL (default: https://cloud.langfuse.com)",
         "required": false
+      },
+      "langfuse_environment": {
+        "type": "text",
+        "ui_name": "Tracing Environment",
+        "description": "Langfuse tracing environment (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)",
+        "required": false
       }
     },
     "description": "Langfuse v2 Logging Integration"
@@ -247,6 +253,12 @@
         "ui_name": "Host URL",
         "description": "Langfuse host URL (default: https://cloud.langfuse.com)",
         "required": false
+      },
+      "langfuse_environment": {
+        "type": "text",
+        "ui_name": "Tracing Environment",
+        "description": "Langfuse tracing environment (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)",
+        "required": false
       }
     },
     "description": "Langfuse v3 OTEL Logging Integration"

litellm/integrations/langfuse/langfuse.py

@@ -1,5 +1,6 @@
 #### What this does ####
 #    On success, logs events to Langfuse
+import inspect
 import os
 import traceback
 from collections.abc import Callable, Iterable, Mapping
@@ -21,6 +22,9 @@ from litellm.litellm_core_utils.core_helpers import (
     reconstruct_model_name,
     safe_deep_copy,
 )
+from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+    validate_langfuse_environment_value,
+)
 from litellm.litellm_core_utils.redact_messages import redact_user_api_key_info
 from litellm.llms.custom_httpx.http_handler import _get_httpx_client
 from litellm.secret_managers.main import str_to_bool
@@ -140,6 +144,7 @@ class LangFuseLogger:
         langfuse_public_key=None,
         langfuse_secret=None,
         langfuse_host=None,
+        langfuse_environment: str | None = None,
         flush_interval=1,
         allow_env_credentials: bool = True,
     ):
@@ -159,6 +164,10 @@ class LangFuseLogger:
         if not (self.langfuse_host.startswith("http://") or self.langfuse_host.startswith("https://")):
             # add http:// if unset, assume communicating over private network - e.g. render
             self.langfuse_host = "http://" + self.langfuse_host
+        _env_override: Final = str(langfuse_environment).strip() if langfuse_environment is not None else None
+        self.langfuse_environment = _env_override or os.getenv("LANGFUSE_TRACING_ENVIRONMENT")
+        if self.langfuse_environment:
+            validate_langfuse_environment_value(self.langfuse_environment)
         self.langfuse_release = os.getenv("LANGFUSE_RELEASE")
         self.langfuse_debug = os.getenv("LANGFUSE_DEBUG")
         self.langfuse_flush_interval = LangFuseLogger._get_langfuse_flush_interval(flush_interval)
@@ -182,6 +191,8 @@ class LangFuseLogger:
         }
         self.langfuse_sdk_version: str = langfuse.version.__version__
 
+        if "environment" in inspect.signature(Langfuse.__init__).parameters:
+            parameters["environment"] = self.langfuse_environment
         if Version(self.langfuse_sdk_version) >= Version("2.6.0"):
             parameters["sdk_integration"] = "litellm"
         self.Langfuse: Langfuse = self.safe_init_langfuse_client(parameters)

litellm/integrations/langfuse/langfuse_handler.py

@@ -1,3 +1,5 @@
+import os
+
 """
 This file contains the LangFuseHandler class
 
@@ -108,6 +110,7 @@ class LangFuseHandler:
             langfuse_public_key=credentials.get("langfuse_public_key"),
             langfuse_secret=credentials.get("langfuse_secret") or credentials.get("langfuse_secret_key"),
             langfuse_host=credentials.get("langfuse_host"),
+            langfuse_environment=credentials.get("langfuse_environment"),
             allow_env_credentials=credentials.get("langfuse_host") is None,
         )
         in_memory_dynamic_logger_cache.set_cache(
@@ -135,8 +138,29 @@ class LangFuseHandler:
             or standard_callback_dynamic_params.get("langfuse_secret_key"),
             langfuse_public_key=standard_callback_dynamic_params.get("langfuse_public_key"),
             langfuse_host=standard_callback_dynamic_params.get("langfuse_host"),
+            langfuse_environment=LangFuseHandler._meaningful_dynamic_environment(standard_callback_dynamic_params),
         )
 
+    @staticmethod
+    def _meaningful_dynamic_environment(
+        standard_callback_dynamic_params: StandardCallbackDynamicParams,
+    ) -> str | None:
+        """Return the per-request environment only when it changes behavior.
+
+        Empty/whitespace values and values equal to the deployment-wide
+        LANGFUSE_TRACING_ENVIRONMENT fallback are treated as absent so an
+        environment-only override that matches the default does not mint a
+        duplicate SDK client (each client costs threads and counts against
+        MAX_LANGFUSE_INITIALIZED_CLIENTS).
+        """
+        raw = standard_callback_dynamic_params.get("langfuse_environment")
+        if raw is None:
+            return None
+        value = str(raw).strip()
+        if not value or value == os.getenv("LANGFUSE_TRACING_ENVIRONMENT"):
+            return None
+        return value
+
     @staticmethod
     def _dynamic_langfuse_credentials_are_passed(
         standard_callback_dynamic_params: StandardCallbackDynamicParams,
@@ -153,6 +177,7 @@ class LangFuseHandler:
             or standard_callback_dynamic_params.get("langfuse_public_key") is not None
             or standard_callback_dynamic_params.get("langfuse_secret") is not None
             or standard_callback_dynamic_params.get("langfuse_secret_key") is not None
+            or LangFuseHandler._meaningful_dynamic_environment(standard_callback_dynamic_params) is not None
         ):
             return True
         return False

litellm/integrations/langfuse/langfuse_otel.py

@@ -231,7 +231,10 @@ class LangfuseOtelLogger(OpenTelemetry):
         from litellm.integrations.arize._utils import safe_set_attribute
         from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
 
-        langfuse_environment: Final = os.environ.get("LANGFUSE_TRACING_ENVIRONMENT")
+        dynamic_params: Final = kwargs.get("standard_callback_dynamic_params")
+        langfuse_environment: Final = (
+            dynamic_params.get("langfuse_environment") if dynamic_params else None
+        ) or os.environ.get("LANGFUSE_TRACING_ENVIRONMENT")
         if langfuse_environment:
             safe_set_attribute(
                 span,

litellm/litellm_core_utils/initialize_dynamic_callback_params.py

@@ -1,3 +1,4 @@
+import re
 from collections.abc import Iterator, Mapping
 from typing import Any, Final
 
@@ -45,12 +46,29 @@ def validate_no_callback_env_reference(param: str, value: object, *, source: str
         _raise_env_reference_error(param, source=source)
 
 
+# Langfuse rejects events whose environment does not match this pattern
+# (lowercase alphanumerics, hyphens, underscores; no "langfuse" prefix).
+# Validating here fails fast at config/init time instead of silently
+# dropping every trace server-side.
+LANGFUSE_ENVIRONMENT_PATTERN: Final = r"^(?!langfuse)[a-z0-9-_]+$"
+
+
+def validate_langfuse_environment_value(value: str) -> None:
+    if not re.match(LANGFUSE_ENVIRONMENT_PATTERN, value):
+        raise ValueError(
+            f"Invalid langfuse_environment {value!r}: must be lowercase "
+            "alphanumerics/hyphens/underscores and must not start with "
+            f"'langfuse' (pattern {LANGFUSE_ENVIRONMENT_PATTERN})"
+        )
+
+
 # Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict
 _supported_callback_params: Final[tuple[str, ...]] = (
     "langfuse_public_key",
     "langfuse_secret",
     "langfuse_secret_key",
     "langfuse_host",
+    "langfuse_environment",
     "langfuse_prompt_version",
     "langsmith_api_key",
     "langsmith_project",

litellm/proxy/_types.py

@@ -20,6 +20,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_langfuse_environment_value,
     validate_no_callback_env_reference,
 )
 from litellm.types.integrations.compression_interception import (
@@ -2027,6 +2028,8 @@ class AddTeamCallback(LiteLLMPydanticObjectBase):
                 raise ValueError(f"Invalid callback variable: {key}. Must be one of {valid_keys}")
             callback_vars[key] = str(value)
             validate_no_callback_env_reference(key, callback_vars[key], source="key/team callback metadata")
+            if key == "langfuse_environment":
+                validate_langfuse_environment_value(callback_vars[key])
         return values
 
 

litellm/proxy/common_utils/callback_config_validation.py

@@ -14,11 +14,36 @@ _NEWRELIC_VAR_PREFIX: Final = "newrelic_"
 
 
 def callback_config_error(callback_name: str | None, callback_vars: Mapping[str, str] | None) -> str | None:
-    if callback_name != _NEWRELIC_CALLBACK or not callback_vars:
+    if not callback_vars:
+        return None
+    env_error: Final = _langfuse_environment_error(callback_vars)
+    if env_error is not None:
+        return env_error
+    if callback_name != _NEWRELIC_CALLBACK:
         return None
     return _newrelic_config_error(callback_vars)
 
 
+def _langfuse_environment_error(callback_vars: Mapping[str, str]) -> str | None:
+    """Reject langfuse_environment values Langfuse ingestion would drop.
+
+    Accepting an invalid value here would 200 the config write and then
+    silently lose every trace for that key/team at request time.
+    """
+    value: Final = callback_vars.get("langfuse_environment")
+    if value is None:
+        return None
+    from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+        validate_langfuse_environment_value,
+    )
+
+    try:
+        validate_langfuse_environment_value(value)
+    except ValueError as e:
+        return str(e)
+    return None
+
+
 def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None:
     """Validate every ``logging`` entry of a team/key metadata payload."""
     if not metadata:

litellm/proxy/management_endpoints/team_callback_endpoints.py

@@ -262,6 +262,7 @@ async def add_team_callbacks(
         - langfuse_secret_key: The secret key for the Langfuse callback
         - langfuse_secret: The secret for the Langfuse callback
         - langfuse_host: The host for the Langfuse callback
+        - langfuse_environment: The tracing environment for the Langfuse callback (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)
         - gcs_bucket_name: The name of the GCS bucket
         - gcs_path_service_account: The path to the GCS service account
         - langsmith_api_key: The API key for the Langsmith callback

litellm/types/integrations/langfuse.py

@@ -1,10 +1,11 @@
-from typing_extensions import TypedDict
+from typing_extensions import ReadOnly, TypedDict
 
 
 class LangfuseLoggingConfig(TypedDict):
     langfuse_secret: str | None
     langfuse_public_key: str | None
     langfuse_host: str | None
+    langfuse_environment: ReadOnly[str | None]
 
 
 class LangfuseUsageDetails(TypedDict):

litellm/types/utils.py

@@ -3278,6 +3278,7 @@ class StandardCallbackDynamicParams(TypedDict, total=False):
     langfuse_secret: str | None
     langfuse_secret_key: str | None
     langfuse_host: str | None
+    langfuse_environment: ReadOnly[str | None]
 
     # Langfuse prompt version
     langfuse_prompt_version: int | None

tests/test_litellm/integrations/test_langfuse.py

@@ -1179,6 +1179,14 @@ def test_max_langfuse_clients_limit():
 class _RecordingLangfuse:
     last_parameters: Optional[dict] = None
 
+    def __init__(self, environment=None, **parameters):
+        type(self).last_parameters = {"environment": environment, **parameters}
+        self.client = MagicMock()
+
+
+class _RecordingLangfuseWithoutEnvironment:
+    last_parameters: Optional[dict] = None
+
     def __init__(self, **parameters):
         type(self).last_parameters = parameters
         self.client = MagicMock()
@@ -1195,6 +1203,62 @@ def _build_langfuse_logger(monkeypatch) -> LangFuseLogger:
         )
 
 
+def test_langfuse_environment_is_passed_to_sdk_client(monkeypatch):
+    monkeypatch.setenv("LANGFUSE_MOCK", "false")
+    monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False)
+    monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
+    with patch("langfuse.Langfuse", _RecordingLangfuse):
+        logger = LangFuseLogger(
+            langfuse_public_key="pk-env",
+            langfuse_secret="sk-env",
+            langfuse_host="https://test.langfuse.com",
+            langfuse_environment="staging",
+        )
+    assert logger.langfuse_environment == "staging"
+    assert _RecordingLangfuse.last_parameters["environment"] == "staging"
+
+
+def test_langfuse_environment_falls_back_to_deployment_env_var(monkeypatch):
+    monkeypatch.setenv("LANGFUSE_MOCK", "false")
+    monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", "deployment-wide")
+    monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
+    with patch("langfuse.Langfuse", _RecordingLangfuse):
+        logger = LangFuseLogger(
+            langfuse_public_key="pk-env",
+            langfuse_secret="sk-env",
+            langfuse_host="https://test.langfuse.com",
+        )
+    assert logger.langfuse_environment == "deployment-wide"
+    assert _RecordingLangfuse.last_parameters["environment"] == "deployment-wide"
+
+
+def test_langfuse_environment_omitted_for_old_sdk_versions(monkeypatch):
+    monkeypatch.setenv("LANGFUSE_MOCK", "false")
+    monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
+    with patch("langfuse.Langfuse", _RecordingLangfuseWithoutEnvironment):
+        LangFuseLogger(
+            langfuse_public_key="pk-env",
+            langfuse_secret="sk-env",
+            langfuse_host="https://test.langfuse.com",
+            langfuse_environment="staging",
+        )
+    assert "environment" not in _RecordingLangfuseWithoutEnvironment.last_parameters
+
+
+def test_dynamic_langfuse_environment_triggers_dynamic_logger():
+    from litellm.integrations.langfuse.langfuse_handler import LangFuseHandler
+    from litellm.types.utils import StandardCallbackDynamicParams
+
+    params = StandardCallbackDynamicParams(langfuse_environment="team-a-env")
+
+    assert LangFuseHandler._dynamic_langfuse_credentials_are_passed(params) is True
+
+    config = LangFuseHandler.get_dynamic_langfuse_logging_config(
+        standard_callback_dynamic_params=params
+    )
+    assert config["langfuse_environment"] == "team-a-env"
+
+
 def test_langfuse_sdk_client_survives_httpx_cache_eviction(monkeypatch):
     import gc
     import weakref
@@ -1408,3 +1472,52 @@ def test_update_trace_keys_matches_whole_keys_not_substrings():
     )
 
     assert "input" not in trace_params
+
+
+def test_langfuse_environment_is_coerced_and_validated(monkeypatch):
+    monkeypatch.setenv("LANGFUSE_MOCK", "false")
+    monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False)
+    monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
+    with patch("langfuse.Langfuse", _RecordingLangfuse):
+        logger = LangFuseLogger(
+            langfuse_public_key="pk-env",
+            langfuse_secret="sk-env",
+            langfuse_host="https://test.langfuse.com",
+            langfuse_environment=123,  # non-string: must coerce, not crash
+        )
+    assert logger.langfuse_environment == "123"
+
+    with pytest.raises(ValueError, match="langfuse_environment"):
+        LangFuseLogger(
+            langfuse_public_key="pk-env",
+            langfuse_secret="sk-env",
+            langfuse_host="https://test.langfuse.com",
+            langfuse_environment="Production",
+        )
+
+
+def test_langfuse_empty_environment_falls_back_and_is_not_dynamic(monkeypatch):
+    from litellm.integrations.langfuse.langfuse_handler import LangFuseHandler
+    from litellm.types.utils import StandardCallbackDynamicParams
+
+    monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", "production")
+
+    # '' falls back to the deployment env var at init
+    monkeypatch.setenv("LANGFUSE_MOCK", "false")
+    monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
+    with patch("langfuse.Langfuse", _RecordingLangfuse):
+        logger = LangFuseLogger(
+            langfuse_public_key="pk-env",
+            langfuse_secret="sk-env",
+            langfuse_host="https://test.langfuse.com",
+            langfuse_environment="",
+        )
+    assert logger.langfuse_environment == "production"
+
+    # env-only params that add nothing do not select a dynamic logger
+    for redundant in ["", "  ", "production"]:
+        params = StandardCallbackDynamicParams(langfuse_environment=redundant)
+        assert LangFuseHandler._dynamic_langfuse_credentials_are_passed(params) is False
+
+    params = StandardCallbackDynamicParams(langfuse_environment="team-a-prod")
+    assert LangFuseHandler._dynamic_langfuse_credentials_are_passed(params) is True

tests/test_litellm/integrations/test_langfuse_otel.py

@@ -137,6 +137,32 @@ class TestLangfuseOtelIntegration:
                     mock_span, "langfuse.environment", test_env
                 )
 
+    def test_set_langfuse_environment_attribute_prefers_dynamic_param(self):
+        """Per-key/team langfuse_environment beats the deployment env var."""
+
+        class _RecordingSpan:
+            def __init__(self):
+                self.attributes = {}
+
+            def set_attribute(self, key, value):
+                self.attributes[key] = value
+
+        span = _RecordingSpan()
+        mock_kwargs = {
+            "standard_callback_dynamic_params": {
+                "langfuse_environment": "team-a-env"
+            }
+        }
+
+        with patch.dict(
+            os.environ, {"LANGFUSE_TRACING_ENVIRONMENT": "deployment-wide"}
+        ):
+            LangfuseOtelLogger._set_langfuse_specific_attributes(
+                span, mock_kwargs, {}
+            )
+
+        assert span.attributes["langfuse.environment"] == "team-a-env"
+
     def test_extract_langfuse_metadata_basic(self):
         """Ensure metadata is correctly pulled from litellm_params."""
         metadata_in = {"generation_name": "my-gen", "custom": "data"}

tests/test_litellm/litellm_core_utils/test_initialize_dynamic_callback_params.py

@@ -233,3 +233,18 @@ def test_trusted_vars_overlay_uses_shared_parser_semantics():
     )
 
     assert params.get("newrelic_api_key") == "12345"
+
+
+def test_validate_langfuse_environment_value():
+    import pytest
+
+    from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+        validate_langfuse_environment_value,
+    )
+
+    validate_langfuse_environment_value("team-a-prod")
+    validate_langfuse_environment_value("staging_2")
+
+    for bad in ["Production", "langfuse-eu", "", "team a"]:
+        with pytest.raises(ValueError, match="langfuse_environment"):
+            validate_langfuse_environment_value(bad)

tests/test_litellm/proxy/common_utils/test_callback_config_validation.py

@@ -0,0 +1,12 @@
+from litellm.proxy.common_utils.callback_config_validation import (
+    callback_config_error,
+)
+
+
+def test_callback_config_error_rejects_invalid_langfuse_environment():
+    for callback in ["langfuse", "langfuse_otel"]:
+        error = callback_config_error(callback, {"langfuse_environment": "Production"})
+        assert error is not None and "langfuse_environment" in error
+
+    assert callback_config_error("langfuse", {"langfuse_environment": "team-a-prod"}) is None
+    assert callback_config_error("langfuse", {"langfuse_public_key": "pk"}) is None

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

@@ -109,6 +109,7 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [
       langfuse_public_key: "text",
       langfuse_secret_key: "password",
       langfuse_host: "text",
+      langfuse_environment: "text",
     },
     description: "Langfuse v2 Logging Integration",
   },
@@ -121,6 +122,7 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [
       langfuse_public_key: "text",
       langfuse_secret_key: "password",
       langfuse_host: "text",
+      langfuse_environment: "text",
     },
     description: "Langfuse v3 OTEL Logging Integration",
   },

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -14836,6 +14836,7 @@ export interface paths {
          *         - langfuse_secret_key: The secret key for the Langfuse callback
          *         - langfuse_secret: The secret for the Langfuse callback
          *         - langfuse_host: The host for the Langfuse callback
+         *         - langfuse_environment: The tracing environment for the Langfuse callback (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)
          *         - gcs_bucket_name: The name of the GCS bucket
          *         - gcs_path_service_account: The path to the GCS service account
          *         - langsmith_api_key: The API key for the Langsmith callback