feat(client): support organization API keys with project_id

.env.template

@@ -2,6 +2,8 @@
 LANGFUSE_BASE_URL=http://localhost:3000
 LANGFUSE_PUBLIC_KEY=pk-lf-1234567890
 LANGFUSE_SECRET_KEY=sk-lf-1234567890
+# Only for organization API keys: the project to write to
+# LANGFUSE_PROJECT_ID=
 
 # Optionally for E2E tests
 OPENAI_API_KEY=

langfuse/__init__.py

@@ -41,7 +41,8 @@
 ```
 
 Configuration is via constructor args or environment variables: `LANGFUSE_PUBLIC_KEY`,
-`LANGFUSE_SECRET_KEY`, `LANGFUSE_BASE_URL` (defaults to https://cloud.langfuse.com). See `langfuse._client.environment_variables`
+`LANGFUSE_SECRET_KEY`, `LANGFUSE_BASE_URL` (defaults to https://cloud.langfuse.com), and
+`LANGFUSE_PROJECT_ID` (required with organization API keys). See `langfuse._client.environment_variables`
 for the full list.
 
 Docs: https://langfuse.com/docs — machine-readable index: https://langfuse.com/llms.txt

langfuse/_client/client.py

@@ -19,6 +19,7 @@
     Dict,
     List,
     Literal,
+    Mapping,
     Optional,
     Tuple,
     Type,
@@ -46,6 +47,7 @@
 )
 from langfuse._client.constants import (
     LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT,
+    PROJECT_ID_HEADER_NAMES,
     ObservationTypeGenerationLike,
     ObservationTypeLiteralNoEvent,
     ObservationTypeSpanLike,
@@ -56,6 +58,7 @@
     LANGFUSE_BASE_URL,
     LANGFUSE_DEBUG,
     LANGFUSE_HOST,
+    LANGFUSE_PROJECT_ID,
     LANGFUSE_PUBLIC_KEY,
     LANGFUSE_RELEASE,
     LANGFUSE_SAMPLE_RATE,
@@ -173,6 +176,34 @@ def _serialize_evaluations(evaluations: List[Evaluation]) -> List[Dict[str, Any]
     ]
 
 
+def _validate_project_id_headers(
+    project_id: Optional[str],
+    additional_headers: Optional[Dict[str, str]],
+    httpx_client: Optional[httpx.Client],
+) -> None:
+    """Raise if a user-supplied project header names a different project."""
+    if project_id is None:
+        return
+
+    sources: List[Tuple[str, Mapping[str, str]]] = []
+    if additional_headers:
+        sources.append(("additional_headers", additional_headers))
+
+    client_headers = getattr(httpx_client, "headers", None)
+    if isinstance(client_headers, Mapping):
+        sources.append(("httpx_client.headers", client_headers))
+
+    for source, headers in sources:
+        for name, value in headers.items():
+            if name.lower() in PROJECT_ID_HEADER_NAMES and value != project_id:
+                message = (
+                    f"Conflicting project IDs: project_id is '{project_id}' but "
+                    f"{source} sets '{name}: {value}'. Remove the header or make "
+                    "the values match."
+                )
+                raise ValueError(message)
+
+
 class Langfuse:
     """Main client for Langfuse tracing and platform features.
 
@@ -195,6 +226,7 @@ class Langfuse:
     Parameters:
         public_key (Optional[str]): Your Langfuse public API key. Can also be set via LANGFUSE_PUBLIC_KEY environment variable.
         secret_key (Optional[str]): Your Langfuse secret API key. Can also be set via LANGFUSE_SECRET_KEY environment variable.
+        project_id (Optional[str]): Langfuse project that receives this client's data. Required for organization API keys; omit for project API keys. Can also be set via LANGFUSE_PROJECT_ID environment variable. Use one client per project. Raises `ValueError` if a project header in `additional_headers` or `httpx_client` names a different project.
         base_url (Optional[str]): The Langfuse API base URL. Defaults to "https://cloud.langfuse.com". Can also be set via LANGFUSE_BASE_URL environment variable.
         host (Optional[str]): Deprecated. Use base_url instead. The Langfuse API host URL. Defaults to "https://cloud.langfuse.com".
         timeout (Optional[int]): Timeout in seconds for API requests. Defaults to 5 seconds.
@@ -305,6 +337,7 @@ def __init__(
         secret_key: Optional[str] = None,
         base_url: Optional[str] = None,
         host: Optional[str] = None,
+        project_id: Optional[str] = None,
         timeout: Optional[int] = None,
         httpx_client: Optional[httpx.Client] = None,
         debug: bool = False,
@@ -339,7 +372,10 @@ def __init__(
             or os.environ.get(LANGFUSE_RELEASE, None)
             or get_common_release_envs()
         )
-        self._project_id: Optional[str] = None
+        project_id = project_id or os.environ.get(LANGFUSE_PROJECT_ID) or None
+        _validate_project_id_headers(project_id, additional_headers, httpx_client)
+        # Cache read by _get_project_id(); a configured project needs no lookup
+        self._project_id: Optional[str] = project_id
         if sample_rate is None:
             sample_rate = float(os.environ.get(LANGFUSE_SAMPLE_RATE, 1.0))
         if not 0.0 <= sample_rate <= 1.0:
@@ -395,6 +431,7 @@ def __init__(
             public_key=public_key,
             secret_key=secret_key,
             base_url=self._base_url,
+            project_id=project_id,
             timeout=timeout,
             environment=self._environment,
             release=release,
@@ -2373,7 +2410,7 @@ def get_current_observation_id(self) -> Optional[str]:
         return self._get_otel_span_id(current_otel_span) if current_otel_span else None
 
     def _get_project_id(self) -> Optional[str]:
-        """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys."""
+        """Return the configured project id, or fetch and cache it for project-bound api keys. Returns None if no project id is found."""
         if not self._project_id:
             proj = self.api.projects.get()
             if not proj.data or not proj.data[0].id:

langfuse/_client/constants.py

@@ -9,6 +9,12 @@
 
 LANGFUSE_TRACER_NAME = "langfuse-sdk"
 
+PROJECT_ID_HEADER = "x-langfuse-project-id"
+"""Header naming the target project. Required with organization API keys."""
+
+# Lowercase header names the server reads as the target project
+PROJECT_ID_HEADER_NAMES = (PROJECT_ID_HEADER, "langfuse-project-id")
+
 LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT = "sdk-experiment"
 
 MASK_FALLBACK_VALUE = "<fully masked due to failed mask function>"

langfuse/_client/environment_variables.py

@@ -35,6 +35,14 @@
 Secret API key of Langfuse project
 """
 
+LANGFUSE_PROJECT_ID = "LANGFUSE_PROJECT_ID"
+"""
+.. envvar:: LANGFUSE_PROJECT_ID
+
+Langfuse project that receives the client's data. Required for organization API keys;
+leave unset for project API keys.
+"""
+
 LANGFUSE_BASE_URL = "LANGFUSE_BASE_URL"
 """
 .. envvar:: LANGFUSE_BASE_URL

langfuse/_client/get_client.py

@@ -1,32 +1,37 @@
 from contextlib import contextmanager
 from contextvars import ContextVar
-from typing import Iterator, Optional
+from typing import Iterator, List, Optional, Tuple
 
 from langfuse._client.client import Langfuse
 from langfuse._client.resource_manager import LangfuseResourceManager
 from langfuse.logger import langfuse_logger
 
-# Context variable to track the current langfuse_public_key in execution context
-_current_public_key: ContextVar[Optional[str]] = ContextVar(
-    "langfuse_public_key", default=None
+# Context variable to track the current (langfuse_public_key, langfuse_project_id)
+# in execution context
+_current_public_key: ContextVar[Optional[Tuple[Optional[str], Optional[str]]]] = (
+    ContextVar("langfuse_public_key", default=None)
 )
 
 
 @contextmanager
-def _set_current_public_key(public_key: Optional[str]) -> Iterator[None]:
-    """Context manager to set and restore the current public key in execution context.
+def _set_current_public_key(
+    public_key: Optional[str], project_id: Optional[str] = None
+) -> Iterator[None]:
+    """Context manager to set and restore the current client identity in execution context.
 
     Args:
-        public_key: The public key to set in context. If None, context is not modified.
+        public_key: The public key to set in context.
+        project_id: The project ID to set in context.
+            If both are None, context is not modified.
 
     Yields:
         None
     """
-    if public_key is None:
+    if public_key is None and project_id is None:
         yield  # Don't modify context if no key provided
         return
 
-    token = _current_public_key.set(public_key)
+    token = _current_public_key.set((public_key, project_id))
     try:
         yield
     finally:
@@ -41,6 +46,7 @@ def _create_client_from_instance(
         public_key=public_key or instance.public_key,
         secret_key=instance.secret_key,
         base_url=instance.base_url,
+        project_id=instance.project_id,
         tracing_enabled=instance.tracing_enabled,
         environment=instance.environment,
         timeout=instance.timeout,
@@ -62,29 +68,33 @@ def _create_client_from_instance(
     )
 
 
-def get_client(*, public_key: Optional[str] = None) -> Langfuse:
+def get_client(
+    *, public_key: Optional[str] = None, project_id: Optional[str] = None
+) -> Langfuse:
     """Get or create a Langfuse client instance.
 
     Returns an existing Langfuse client or creates a new one if none exists. In multi-project setups,
     providing a public_key is required. Multi-project support is experimental - see Langfuse docs.
 
     Behavior:
     - Single project: Returns existing client or creates new one
-    - Multi-project: Requires public_key to return specific client
+    - Multi-project: Requires public_key to return specific client, plus project_id when one organization key serves several projects
     - No public_key in multi-project: Returns disabled client to prevent data leakage
 
-    The function uses a singleton pattern per public_key to conserve resources and maintain state.
+    The function uses a singleton pattern per public_key and project_id to conserve resources and maintain state.
 
     Args:
         public_key (Optional[str]): Project identifier
             - With key: Returns client for that project
             - Without key: Returns single client or disabled client if multiple exist
+        project_id (Optional[str]): Selects the client for this project when one
+            organization API key serves several projects.
 
     Returns:
         Langfuse: Client instance in one of three states:
-            1. Client for specified public_key
+            1. Client for specified public_key and project_id
             2. Default client for single-project setup
-            3. Disabled client when multiple projects exist without key
+            3. Disabled client when no client or several clients match
 
     Security:
         Disables tracing when multiple projects exist without explicit key to prevent
@@ -101,16 +111,24 @@ def get_client(*, public_key: Optional[str] = None) -> Langfuse:
 
         # Without specific key in multi-project setup:
         client = get_client()  # Returns disabled client for safety
+
+        # One organization API key writing to two projects:
+        Langfuse(public_key="org_key", secret_key="...", project_id="project_a_id")
+        Langfuse(public_key="org_key", secret_key="...", project_id="project_b_id")
+
+        client_a = get_client(public_key="org_key", project_id="project_a_id")
+        client_b = get_client(public_key="org_key", project_id="project_b_id")
+        client = get_client(public_key="org_key")  # Ambiguous: disabled client
         ```
     """
     with LangfuseResourceManager._lock:
         active_instances = LangfuseResourceManager._instances
 
-        # If no explicit public_key provided, check execution context
-        if not public_key:
-            public_key = _current_public_key.get(None)
+        # If no explicit client identity provided, check execution context
+        if not public_key and not project_id:
+            public_key, project_id = _current_public_key.get(None) or (None, None)
 
-        if not public_key:
+        if not public_key and not project_id:
             if len(active_instances) == 0:
                 # No clients initialized yet, create default instance
                 return Langfuse()
@@ -135,21 +153,39 @@ def get_client(*, public_key: Optional[str] = None) -> Langfuse:
                 )
 
         else:
-            # Specific key provided, look up existing instance
-            target_instance: Optional[LangfuseResourceManager] = active_instances.get(
-                public_key, None
-            )
+            # Specific identity provided, look up matching instances
+            matching_instances: List[LangfuseResourceManager] = [
+                instance
+                for (instance_public_key, instance_project_id), instance in (
+                    active_instances.items()
+                )
+                if (not public_key or instance_public_key == public_key)
+                and (not project_id or instance_project_id == project_id)
+            ]
 
-            if target_instance is None:
+            if len(matching_instances) == 0:
                 # No instance found with this key - client not initialized properly
                 langfuse_logger.warning(
-                    "No Langfuse client with public key %s has been initialized. Skipping "
-                    "tracing for decorated function.",
+                    "No Langfuse client with public key %s and project id %s has been "
+                    "initialized. Skipping tracing for decorated function.",
+                    public_key,
+                    project_id,
+                )
+                return Langfuse(
+                    tracing_enabled=False, public_key="fake", secret_key="fake"
+                )
+
+            if len(matching_instances) > 1:
+                # Several projects share this organization key - disable tracing
+                # to prevent cross-project data leakage
+                langfuse_logger.warning(
+                    "Multiple Langfuse clients with public key %s are instantiated for "
+                    "different projects, but no 'langfuse_project_id' was passed. "
+                    "Skipping tracing to avoid cross-project leakage.",
                     public_key,
                 )
                 return Langfuse(
                     tracing_enabled=False, public_key="fake", secret_key="fake"
                 )
 
-            # target_instance is guaranteed to be not None at this point
-            return _create_client_from_instance(target_instance, public_key)
+            return _create_client_from_instance(matching_instances[0], public_key)

langfuse/_client/observe.py

@@ -189,6 +189,7 @@ def sub_process():
               - langfuse_trace_id: Explicitly set the trace ID for this function call
               - langfuse_parent_observation_id: Explicitly set the parent span ID
               - langfuse_public_key: Use a specific Langfuse project (when multiple clients exist)
+              - langfuse_project_id: Use the client for this project (when several clients share an organization API key)
             - For async functions, the decorator returns an async function wrapper.
             - For sync functions, the decorator returns a synchronous wrapper.
         """
@@ -286,10 +287,13 @@ async def async_wrapper(*args: Tuple[Any], **kwargs: Dict[str, Any]) -> Any:
                 else None
             )
             public_key = cast(str, kwargs.pop("langfuse_public_key", None))
+            project_id = cast(Optional[str], kwargs.pop("langfuse_project_id", None))
 
-            # Set public key in execution context for nested decorated functions
-            with _set_current_public_key(public_key):
-                langfuse_client = get_client(public_key=public_key)
+            # Set client identity in execution context for nested decorated functions
+            with _set_current_public_key(public_key, project_id):
+                langfuse_client = get_client(
+                    public_key=public_key, project_id=project_id
+                )
                 context_manager: Optional[
                     Union[
                         _AgnosticContextManager[LangfuseGeneration],
@@ -377,10 +381,13 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
                 else None
             )
             public_key = kwargs.pop("langfuse_public_key", None)
+            project_id = kwargs.pop("langfuse_project_id", None)
 
-            # Set public key in execution context for nested decorated functions
-            with _set_current_public_key(public_key):
-                langfuse_client = get_client(public_key=public_key)
+            # Set client identity in execution context for nested decorated functions
+            with _set_current_public_key(public_key, project_id):
+                langfuse_client = get_client(
+                    public_key=public_key, project_id=project_id
+                )
                 context_manager: Optional[
                     Union[
                         _AgnosticContextManager[LangfuseGeneration],