split out response-header handling into its own PR

Per review (#1905#discussion_r3189043209), the X-Logfire-Warning /
X-Logfire-Error response-hook handling is logically independent from
the X-Logfire-Telemetry request header — they just happened to be
introduced together. Moved that code (plus its exceptions, install
calls, and tests) to a separate PR (#1906) so each can be reviewed
and landed on its own.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adriangb

logfire/_internal/cli/__init__.py

@@ -25,7 +25,7 @@
 from ..client import LogfireClient
 from ..config import REGIONS, LogfireCredentials, get_base_url_from_token
 from ..config_params import ParamManager
-from ..telemetry_header import TELEMETRY_HEADER_NAME, build_telemetry_header, install_logfire_response_hook
+from ..telemetry_header import TELEMETRY_HEADER_NAME, build_telemetry_header
 from ..tracer import SDKTracerProvider
 from .auth import parse_auth, parse_logout
 from .prompt import parse_prompt
@@ -435,10 +435,9 @@ def log_trace_id(response: requests.Response, context: ContextCarrier, *args: An
     else:
         with tracer.start_as_current_span('logfire._internal.cli'), requests.Session() as session:
             context = get_context()
-            session.hooks = {'response': [functools.partial(log_trace_id, context=context)]}
+            session.hooks = {'response': functools.partial(log_trace_id, context=context)}
             session.headers.update(context)
             session.headers[TELEMETRY_HEADER_NAME] = build_telemetry_header()
-            install_logfire_response_hook(session)
             namespace._session = session
             namespace.func(namespace)
 

logfire/_internal/client.py

@@ -10,11 +10,7 @@
 from logfire.version import VERSION
 
 from .auth import UserToken, UserTokenCollection
-from .telemetry_header import (
-    TELEMETRY_HEADER_NAME,
-    build_telemetry_header,
-    install_logfire_response_hook,
-)
+from .telemetry_header import TELEMETRY_HEADER_NAME, build_telemetry_header
 from .utils import UnexpectedResponse
 
 UA_HEADER = f'logfire/{VERSION}'
@@ -49,7 +45,6 @@ def __init__(self, user_token: UserToken) -> None:
                 TELEMETRY_HEADER_NAME: build_telemetry_header(),
             }
         )
-        install_logfire_response_hook(self._session)
 
     @classmethod
     def from_url(cls, base_url: str | None) -> Self:

logfire/_internal/config.py

@@ -111,11 +111,7 @@
 from .metrics import ProxyMeterProvider
 from .scrubbing import NOOP_SCRUBBER, BaseScrubber, Scrubber, ScrubbingOptions
 from .stack_info import warn_at_user_stacklevel
-from .telemetry_header import (
-    TELEMETRY_HEADER_NAME,
-    build_telemetry_header,
-    install_logfire_response_hook,
-)
+from .telemetry_header import TELEMETRY_HEADER_NAME, build_telemetry_header
 from .tracer import OPEN_SPANS, PendingSpanProcessor, ProxyTracerProvider
 from .utils import (
     SeededRandomIdGenerator,
@@ -1164,7 +1160,6 @@ def check_tokens():
                             TELEMETRY_HEADER_NAME: telemetry_header_value,
                         }
                         session = OTLPExporterHttpSession()
-                        install_logfire_response_hook(session)
                         span_exporter = BodySizeCheckingOTLPSpanExporter(
                             endpoint=urljoin(base_url, '/v1/traces'),
                             session=session,
@@ -1491,11 +1486,9 @@ def warn_if_not_initialized(self, message: str):
             )
 
     def _initialize_credentials_from_token(self, token: str) -> LogfireCredentials | None:
-        session = requests.Session()
-        install_logfire_response_hook(session)
         return LogfireCredentials.from_token(
             token,
-            session,
+            requests.Session(),
             self.advanced.generate_base_url(token),
             telemetry_header=build_telemetry_header(self),
         )

logfire/_internal/telemetry_header.py

@@ -1,39 +1,26 @@
-"""SDK <-> server out-of-band metadata exchanged via custom HTTP headers.
-
-* `X-Logfire-Telemetry` (request): non-sensitive information about the SDK and how
-  it is configured, encoded as a compact JSON object. Used by the backend to
-  answer questions like which SDK versions are still in active use, which Python
-  versions we can drop, and which configuration options users actually enable.
-  Secrets (`token`, `api_key`, `service_name`, etc.) are never included.
-* `X-Logfire-Warning` (response): an out-of-band warning the server wants the
-  user to see. Surfaced via `warnings.warn(...)`; the standard "default" filter
-  deduplicates identical messages so a chatty server only warns once.
-* `X-Logfire-Error` (response): an out-of-band error the server wants the SDK
-  to raise. Always raised — callers that want to keep working past it (the OTLP
-  pipeline, the variables provider) already swallow exceptions from their HTTP
-  calls.
+"""Build the `X-Logfire-Telemetry` request header.
+
+The header carries non-sensitive information about the SDK and how it is
+configured, encoded as a compact JSON object. The backend uses it to answer
+questions like which SDK versions are still in active use, which Python
+versions we can drop, and which configuration options users actually enable.
+Secrets (`token`, `api_key`, `service_name`, etc.) are never included.
 """
 
 from __future__ import annotations
 
 import functools
 import json
 import sys
-import warnings
 from typing import TYPE_CHECKING, Any
 
-import requests
-
-from logfire.exceptions import LogfireServerError, LogfireServerWarning
 from logfire.version import VERSION
 
 if TYPE_CHECKING:
     from .config import LogfireConfig
 
 
 TELEMETRY_HEADER_NAME = 'X-Logfire-Telemetry'
-WARNING_HEADER_NAME = 'X-Logfire-Warning'
-ERROR_HEADER_NAME = 'X-Logfire-Error'
 
 
 @functools.cache
@@ -102,28 +89,3 @@ def build_telemetry_header(config: LogfireConfig | None = None) -> str:
     if config is not None:
         pairs.update(_config_telemetry_pairs(config))
     return json.dumps(pairs, separators=(',', ':'))
-
-
-def process_logfire_response_headers(response: requests.Response, *_args: Any, **_kwargs: Any) -> requests.Response:
-    """Handle `X-Logfire-Warning` / `X-Logfire-Error` headers on a Logfire API response.
-
-    Designed to be installed as a `requests` response hook
-    (`session.hooks['response'].append(...)`).
-    """
-    warning_message = response.headers.get(WARNING_HEADER_NAME)
-    if warning_message:
-        warnings.warn(warning_message, LogfireServerWarning, stacklevel=2)
-    error_message = response.headers.get(ERROR_HEADER_NAME)
-    if error_message:
-        raise LogfireServerError(error_message)
-    return response
-
-
-def install_logfire_response_hook(session: requests.Session) -> None:
-    """Install `process_logfire_response_headers` as a response hook on `session`.
-
-    `requests.Session()` always initialises `hooks['response']` to a list, and every
-    call site here passes a freshly-built session, so we just append.
-    """
-    response_hooks: list[Any] = session.hooks.setdefault('response', [])
-    response_hooks.append(process_logfire_response_headers)

logfire/exceptions.py

@@ -3,11 +3,3 @@
 
 class LogfireConfigError(ValueError):
     """Error raised when there is a problem with the Logfire configuration."""
-
-
-class LogfireServerError(Exception):
-    """Error raised when the Logfire server returns an `X-Logfire-Error` header on a response."""
-
-
-class LogfireServerWarning(UserWarning):
-    """Warning emitted when the Logfire server returns an `X-Logfire-Warning` header on a response."""

logfire/variables/remote.py

@@ -17,11 +17,7 @@
 
 from logfire._internal.client import UA_HEADER
 from logfire._internal.config import VariablesOptions
-from logfire._internal.telemetry_header import (
-    TELEMETRY_HEADER_NAME,
-    build_telemetry_header,
-    install_logfire_response_hook,
-)
+from logfire._internal.telemetry_header import TELEMETRY_HEADER_NAME, build_telemetry_header
 from logfire._internal.utils import UnexpectedResponse
 from logfire.variables.abstract import (
     ResolvedVariable,
@@ -85,7 +81,6 @@ def __init__(self, base_url: str, token: str, options: VariablesOptions, telemet
                 TELEMETRY_HEADER_NAME: self._telemetry_header,
             }
         )
-        install_logfire_response_hook(self._session)
         self._timeout = options.timeout
         self._block_before_first_fetch = block_before_first_resolve
         self._polling_interval: timedelta = (
@@ -215,7 +210,6 @@ def _sse_listener(self):  # pragma: no cover
                             'Cache-Control': 'no-cache',
                         }
                     )
-                    install_logfire_response_hook(sse_session)
 
                     # Open streaming connection
                     response = sse_session.get(sse_url, stream=True, timeout=(10, None))

tests/test_telemetry_header.py

@@ -1,25 +1,15 @@
 from __future__ import annotations
 
 import json
-import warnings
 from typing import Any
 from unittest.mock import patch
 
-import pytest
 import requests
 import requests_mock
-from inline_snapshot import snapshot
 
 import logfire
 from logfire._internal.config import GLOBAL_CONFIG, LogfireCredentials
-from logfire._internal.telemetry_header import (
-    ERROR_HEADER_NAME,
-    TELEMETRY_HEADER_NAME,
-    WARNING_HEADER_NAME,
-    build_telemetry_header,
-    process_logfire_response_headers,
-)
-from logfire.exceptions import LogfireServerError, LogfireServerWarning
+from logfire._internal.telemetry_header import TELEMETRY_HEADER_NAME, build_telemetry_header
 from logfire.version import VERSION
 
 
@@ -112,71 +102,3 @@ def test_from_token_sends_telemetry_header():
         )
         [history] = m.request_history
         assert history.headers[TELEMETRY_HEADER_NAME] == '{"sdk_version":"1.2.3"}'
-
-
-def test_process_response_warning_header_emits_warning():
-    response = requests.Response()
-    response.headers[WARNING_HEADER_NAME] = 'The /foo/bar endpoint is deprecated, please use /bar/baz'
-    with warnings.catch_warnings(record=True) as caught:
-        warnings.simplefilter('always')
-        process_logfire_response_headers(response)
-    assert [(w.category, str(w.message)) for w in caught] == snapshot(
-        [(LogfireServerWarning, 'The /foo/bar endpoint is deprecated, please use /bar/baz')]
-    )
-
-
-def test_process_response_warning_header_dedupes():
-    """Python's default `warnings` filter should fold repeats of the same message into one entry."""
-    response = requests.Response()
-    response.headers[WARNING_HEADER_NAME] = 'a duplicated warning'
-    with warnings.catch_warnings(record=True) as caught:
-        warnings.simplefilter('default')
-        for _ in range(5):
-            process_logfire_response_headers(response)
-    messages = [str(w.message) for w in caught]
-    assert messages == ['a duplicated warning']
-
-
-def test_process_response_error_header_raises():
-    response = requests.Response()
-    response.headers[ERROR_HEADER_NAME] = 'something is wrong'
-    with pytest.raises(LogfireServerError, match='something is wrong'):
-        process_logfire_response_headers(response)
-
-
-def test_response_hook_installed_on_logfire_client():
-    from logfire._internal.auth import UserToken
-    from logfire._internal.client import LogfireClient
-
-    token = UserToken(
-        token='pylf_v1_us_xxx',
-        base_url='https://logfire-us.pydantic.dev',
-        expiration='2099-12-31T23:59:59',
-    )
-    client = LogfireClient(user_token=token)
-
-    with requests_mock.Mocker() as m:
-        m.get(
-            'https://logfire-us.pydantic.dev/v1/account/me',
-            json={'name': 'me'},
-            headers={WARNING_HEADER_NAME: 'deprecated endpoint'},
-        )
-        with warnings.catch_warnings(record=True) as caught:
-            warnings.simplefilter('always')
-            client.get_user_information()
-
-    assert any(isinstance(w.message, LogfireServerWarning) for w in caught)
-
-    with requests_mock.Mocker() as m:
-        m.get(
-            'https://logfire-us.pydantic.dev/v1/account/me',
-            json={'name': 'me'},
-            headers={ERROR_HEADER_NAME: 'no longer supported'},
-        )
-        with pytest.raises(LogfireServerError, match='no longer supported'):
-            client.get_user_information()
-
-    [history, *_] = m.request_history
-    assert TELEMETRY_HEADER_NAME in history.headers
-    pairs = _parse_header(history.headers[TELEMETRY_HEADER_NAME])
-    assert pairs['sdk_version'] == VERSION