feat: surface X-Logfire-Warning / X-Logfire-Error response headers

logfire/_internal/cli/__init__.py

@@ -25,6 +25,7 @@
 from ..client import LogfireClient
 from ..config import REGIONS, LogfireCredentials, get_base_url_from_token
 from ..config_params import ParamManager
+from ..server_response import install_logfire_response_hook
 from ..tracer import SDKTracerProvider
 from .auth import parse_auth, parse_logout
 from .prompt import parse_prompt
@@ -434,8 +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)
+            install_logfire_response_hook(session)
             namespace._session = session
             namespace.func(namespace)
 

logfire/_internal/client.py

@@ -10,6 +10,7 @@
 from logfire.version import VERSION
 
 from .auth import UserToken, UserTokenCollection
+from .server_response import install_logfire_response_hook
 from .utils import UnexpectedResponse
 
 UA_HEADER = f'logfire/{VERSION}'
@@ -38,6 +39,7 @@ def __init__(self, user_token: UserToken) -> None:
         self._token = user_token.token
         self._session = Session()
         self._session.headers.update({'Authorization': self._token, 'User-Agent': UA_HEADER})
+        install_logfire_response_hook(self._session)
 
     @classmethod
     def from_url(cls, base_url: str | None) -> Self:

logfire/_internal/config.py

@@ -106,6 +106,7 @@
 from .logs import ProxyLoggerProvider
 from .metrics import ProxyMeterProvider
 from .scrubbing import NOOP_SCRUBBER, BaseScrubber, Scrubber, ScrubbingOptions
+from .server_response import install_logfire_response_hook
 from .stack_info import warn_at_user_stacklevel
 from .tracer import OPEN_SPANS, PendingSpanProcessor, ProxyTracerProvider
 from .utils import (
@@ -1129,6 +1130,7 @@ def check_tokens():
                         base_url = self.advanced.generate_base_url(token)
                         headers = {'User-Agent': f'logfire/{VERSION}', 'Authorization': token}
                         session = OTLPExporterHttpSession()
+                        install_logfire_response_hook(session)
                         span_exporter = BodySizeCheckingOTLPSpanExporter(
                             endpoint=urljoin(base_url, '/v1/traces'),
                             session=session,
@@ -1453,7 +1455,9 @@ def warn_if_not_initialized(self, message: str):
             )
 
     def _initialize_credentials_from_token(self, token: str) -> LogfireCredentials | None:
-        return LogfireCredentials.from_token(token, requests.Session(), self.advanced.generate_base_url(token))
+        session = requests.Session()
+        install_logfire_response_hook(session)
+        return LogfireCredentials.from_token(token, session, self.advanced.generate_base_url(token))
 
     def _ensure_flush_after_aws_lambda(self):
         """Ensure that `force_flush` is called after an AWS Lambda invocation.

logfire/_internal/server_response.py

@@ -0,0 +1,52 @@
+"""Surface out-of-band signals the Logfire backend wants every SDK request to know about.
+
+The server attaches custom headers to API responses:
+
+* `X-Logfire-Warning`: an out-of-band warning the server wants the user to see.
+  Surfaced via `warnings.warn(..., LogfireServerWarning)`. Python's standard
+  "default" filter dedupes identical messages, so a chatty server only warns once.
+* `X-Logfire-Error`: an out-of-band error the server wants the SDK to raise.
+  Always raised as `LogfireServerError`. Callers that want to keep working past
+  it (the OTLP pipeline, the variables provider) already swallow exceptions from
+  their HTTP calls; CRUD/CLI propagate the error to the user.
+
+`install_logfire_response_hook(session)` wires this into a `requests.Session` as
+a response hook so every Logfire-bound HTTP response is inspected.
+"""
+
+from __future__ import annotations
+
+import warnings
+from typing import Any
+
+import requests
+
+from logfire.exceptions import LogfireServerError, LogfireServerWarning
+
+WARNING_HEADER_NAME = 'X-Logfire-Warning'
+ERROR_HEADER_NAME = 'X-Logfire-Error'
+
+
+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,3 +3,11 @@
 
 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,6 +17,7 @@
 
 from logfire._internal.client import UA_HEADER
 from logfire._internal.config import VariablesOptions
+from logfire._internal.server_response import install_logfire_response_hook
 from logfire._internal.utils import UnexpectedResponse
 from logfire.variables.abstract import (
     ResolvedVariable,
@@ -69,6 +70,7 @@ def __init__(self, base_url: str, token: str, options: VariablesOptions):
         self._token = token
         self._session = Session()
         self._session.headers.update({'Authorization': f'bearer {token}', 'User-Agent': UA_HEADER})
+        install_logfire_response_hook(self._session)
         self._timeout = options.timeout
         self._block_before_first_fetch = block_before_first_resolve
         self._polling_interval: timedelta = (
@@ -197,6 +199,7 @@ 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_server_response.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import warnings
+
+import pytest
+import requests
+import requests_mock
+from inline_snapshot import snapshot
+
+from logfire._internal.server_response import (
+    ERROR_HEADER_NAME,
+    WARNING_HEADER_NAME,
+    process_logfire_response_headers,
+)
+from logfire.exceptions import LogfireServerError, LogfireServerWarning
+
+
+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()