From 55e70c3b3542319ffd09cb2bf33558aee5a5a085 Mon Sep 17 00:00:00 2001
From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>
Date: Tue, 5 May 2026 09:24:13 -0500
Subject: [PATCH 1/8] feat: surface X-Logfire-Warning / X-Logfire-Error
 response headers

Adds a `requests` response hook that the SDK installs on every
Logfire-bound HTTP session (OTLP exporters, token validation, CRUD
client, variables provider, CLI). The hook reads two custom headers
the server attaches to responses:

* `X-Logfire-Warning` -> `warnings.warn(..., LogfireServerWarning)`.
  Python's default filter dedupes identical messages within a process,
  so a chatty server only warns once.
* `X-Logfire-Error` -> raises `LogfireServerError`. OTLP/variables
  paths already swallow exceptions from their HTTP calls; CRUD/CLI
  propagate the error to the user.

This gives the backend an out-of-band channel to deprecate endpoints
or signal hard failures without piggy-backing on response bodies.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 logfire/_internal/cli/__init__.py    |  4 +-
 logfire/_internal/client.py          |  2 +
 logfire/_internal/config.py          |  6 ++-
 logfire/_internal/server_response.py | 52 +++++++++++++++++++
 logfire/exceptions.py                |  8 +++
 logfire/variables/remote.py          |  3 ++
 tests/test_server_response.py        | 78 ++++++++++++++++++++++++++++
 7 files changed, 151 insertions(+), 2 deletions(-)
 create mode 100644 logfire/_internal/server_response.py
 create mode 100644 tests/test_server_response.py

diff --git a/logfire/_internal/cli/__init__.py b/logfire/_internal/cli/__init__.py
index cb49be100..7c719f1da 100644
--- a/logfire/_internal/cli/__init__.py
+++ b/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)
 
diff --git a/logfire/_internal/client.py b/logfire/_internal/client.py
index 9bd9c77de..6a4f077a9 100644
--- a/logfire/_internal/client.py
+++ b/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:
diff --git a/logfire/_internal/config.py b/logfire/_internal/config.py
index 98e73d46b..4816b4b95 100644
--- a/logfire/_internal/config.py
+++ b/logfire/_internal/config.py
@@ -110,6 +110,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 (
@@ -1148,6 +1149,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,
@@ -1472,7 +1474,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.
diff --git a/logfire/_internal/server_response.py b/logfire/_internal/server_response.py
new file mode 100644
index 000000000..7a65dcedd
--- /dev/null
+++ b/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)
diff --git a/logfire/exceptions.py b/logfire/exceptions.py
index 617fba04f..532fb2720 100644
--- a/logfire/exceptions.py
+++ b/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."""
diff --git a/logfire/variables/remote.py b/logfire/variables/remote.py
index b780a658f..1c77fa0a1 100644
--- a/logfire/variables/remote.py
+++ b/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))
diff --git a/tests/test_server_response.py b/tests/test_server_response.py
new file mode 100644
index 000000000..540253aea
--- /dev/null
+++ b/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()

From eb3078f15c81fa0f978a680a562bcf5d920fdc10 Mon Sep 17 00:00:00 2001
From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>
Date: Wed, 6 May 2026 09:23:50 -0500
Subject: [PATCH 2/8] feat: add AdvancedOptions.transport_response_hook to
 customize/opt out

Lets users intercept every Logfire API response (OTLP exports, credential
init, variables provider). Default keeps the existing X-Logfire-Warning /
X-Logfire-Error handling; pass `lambda response: None` to opt out, or
compose around `process_logfire_response_headers`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 logfire/_internal/client.py          | 25 ++++++++++---
 logfire/_internal/config.py          | 32 ++++++++++++++---
 logfire/_internal/server_response.py | 38 +++++++++++++-------
 logfire/variables/remote.py          | 17 ++++++---
 tests/test_server_response.py        | 54 ++++++++++++++++++++++++++++
 5 files changed, 140 insertions(+), 26 deletions(-)

diff --git a/logfire/_internal/client.py b/logfire/_internal/client.py
index 6a4f077a9..ac20ffbaa 100644
--- a/logfire/_internal/client.py
+++ b/logfire/_internal/client.py
@@ -10,7 +10,7 @@
 from logfire.version import VERSION
 
 from .auth import UserToken, UserTokenCollection
-from .server_response import install_logfire_response_hook
+from .server_response import TransportResponseHook, install_logfire_response_hook
 from .utils import UnexpectedResponse
 
 UA_HEADER = f'logfire/{VERSION}'
@@ -30,19 +30,29 @@ class LogfireClient:
 
     Args:
         user_token: The user token to use when authenticating against the API.
+        transport_response_hook: Optional override for the API response hook (see
+            `AdvancedOptions.transport_response_hook`).
     """
 
-    def __init__(self, user_token: UserToken) -> None:
+    def __init__(
+        self,
+        user_token: UserToken,
+        transport_response_hook: TransportResponseHook | None = None,
+    ) -> None:
         if user_token.is_expired:
             raise RuntimeError('The provided user token is expired')
         self.base_url = user_token.base_url
         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)
+        install_logfire_response_hook(self._session, transport_response_hook)
 
     @classmethod
-    def from_url(cls, base_url: str | None) -> Self:
+    def from_url(
+        cls,
+        base_url: str | None,
+        transport_response_hook: TransportResponseHook | None = None,
+    ) -> Self:
         """Create a client from the provided base URL.
 
         Args:
@@ -50,8 +60,13 @@ def from_url(cls, base_url: str | None) -> Self:
                 the user into selecting a token from the token collection (or, if only one available,
                 use it directly). The token collection will be created from the `~/.logfire/default.toml`
                 file (or an empty one if no such file exists).
+            transport_response_hook: Optional override for the API response hook (see
+                `AdvancedOptions.transport_response_hook`).
         """
-        return cls(user_token=UserTokenCollection().get_token(base_url))
+        return cls(
+            user_token=UserTokenCollection().get_token(base_url),
+            transport_response_hook=transport_response_hook,
+        )
 
     def _get_raw(self, endpoint: str, params: dict[str, Any] | None = None) -> Response:
         response = self._session.get(urljoin(self.base_url, endpoint), params=params)
diff --git a/logfire/_internal/config.py b/logfire/_internal/config.py
index 4816b4b95..78dd2e7c4 100644
--- a/logfire/_internal/config.py
+++ b/logfire/_internal/config.py
@@ -110,7 +110,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 .server_response import TransportResponseHook, install_logfire_response_hook
 from .stack_info import warn_at_user_stacklevel
 from .tracer import OPEN_SPANS, PendingSpanProcessor, ProxyTracerProvider
 from .utils import (
@@ -217,6 +217,29 @@ class AdvancedOptions:
     This log and configuration is experimental and may be modified or removed.
     """
 
+    transport_response_hook: TransportResponseHook | None = None
+    """Optional callback invoked for every HTTP response received from the Logfire API.
+
+    This applies to OTLP exports, credential / project initialisation, and the remote
+    variables provider. The default surfaces `X-Logfire-Warning` and `X-Logfire-Error`
+    headers as `LogfireServerWarning` / `LogfireServerError`.
+
+    Setting this replaces the default; pass `lambda response: None` to opt out entirely,
+    or compose your own logic on top of `process_logfire_response_headers`:
+
+    ```python
+    from logfire._internal.server_response import process_logfire_response_headers
+
+    def hook(response):
+        my_metric.inc(response.status_code)
+        process_logfire_response_headers(response)
+
+    logfire.configure(advanced=AdvancedOptions(transport_response_hook=hook))
+    ```
+
+    Raise from the hook to abort the calling code path.
+    """
+
     def generate_base_url(self, token: str) -> str:
         if self.base_url is not None:
             return self.base_url
@@ -1098,7 +1121,7 @@ def add_span_processor(span_processor: SpanProcessor) -> None:
                     # If we don't have tokens or credentials from a file,
                     # try initializing a new project and writing a new creds file.
                     # note, we only do this if `send_to_logfire` is explicitly `True`, not 'if-token-present'
-                    client = LogfireClient.from_url(self.advanced.base_url)
+                    client = LogfireClient.from_url(self.advanced.base_url, self.advanced.transport_response_hook)
                     credentials = LogfireCredentials.initialize_project(client=client)
                     credentials.write_creds_file(self.data_dir)
 
@@ -1149,7 +1172,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)
+                        install_logfire_response_hook(session, self.advanced.transport_response_hook)
                         span_exporter = BodySizeCheckingOTLPSpanExporter(
                             endpoint=urljoin(base_url, '/v1/traces'),
                             session=session,
@@ -1326,6 +1349,7 @@ def fix_pid():  # pragma: no cover
                     base_url=base_url,
                     token=self.api_key,
                     options=self.variables,
+                    transport_response_hook=self.advanced.transport_response_hook,
                 )
             multi_log_processor = SynchronousMultiLogRecordProcessor()
             for processor in log_record_processors:
@@ -1475,7 +1499,7 @@ 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)
+        install_logfire_response_hook(session, self.advanced.transport_response_hook)
         return LogfireCredentials.from_token(token, session, self.advanced.generate_base_url(token))
 
     def _ensure_flush_after_aws_lambda(self):
diff --git a/logfire/_internal/server_response.py b/logfire/_internal/server_response.py
index 7a65dcedd..bb1e6b4c0 100644
--- a/logfire/_internal/server_response.py
+++ b/logfire/_internal/server_response.py
@@ -11,13 +11,15 @@
   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.
+a response hook so every Logfire-bound HTTP response is inspected. Callers can
+pass a custom `hook` to replace the default behaviour (see
+`AdvancedOptions.transport_response_hook`).
 """
 
 from __future__ import annotations
 
 import warnings
-from typing import Any
+from typing import Any, Callable
 
 import requests
 
@@ -26,27 +28,37 @@
 WARNING_HEADER_NAME = 'X-Logfire-Warning'
 ERROR_HEADER_NAME = 'X-Logfire-Error'
 
+TransportResponseHook = Callable[[requests.Response], object]
+"""Callable invoked for every Logfire API response received by the SDK.
 
-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.
+The return value is ignored; raise to abort the call.
+"""
 
-    Designed to be installed as a `requests` response hook
-    (`session.hooks['response'].append(...)`).
-    """
+
+def process_logfire_response_headers(response: requests.Response) -> None:
+    """Default transport response hook: surface `X-Logfire-Warning` / `X-Logfire-Error` headers."""
     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`.
+def install_logfire_response_hook(
+    session: requests.Session,
+    hook: TransportResponseHook | None = None,
+) -> None:
+    """Install a `requests` response hook on `session` for every Logfire API response.
 
-    `requests.Session()` always initialises `hooks['response']` to a list, and every
-    call site here passes a freshly-built session, so we just append.
+    `hook` defaults to `process_logfire_response_headers`. Pass a custom callable
+    to replace the default behaviour (e.g. opt out by passing `lambda response: None`).
     """
+    user_hook = hook if hook is not None else process_logfire_response_headers
+
+    def _hook(response: requests.Response, *_args: Any, **_kwargs: Any) -> requests.Response:
+        user_hook(response)
+        return response
+
     response_hooks: list[Any] = session.hooks.setdefault('response', [])
-    response_hooks.append(process_logfire_response_headers)
+    response_hooks.append(_hook)
diff --git a/logfire/variables/remote.py b/logfire/variables/remote.py
index 1c77fa0a1..91034a96c 100644
--- a/logfire/variables/remote.py
+++ b/logfire/variables/remote.py
@@ -17,7 +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.server_response import TransportResponseHook, install_logfire_response_hook
 from logfire._internal.utils import UnexpectedResponse
 from logfire.variables.abstract import (
     ResolvedVariable,
@@ -55,22 +55,31 @@ class LogfireRemoteVariableProvider(VariableProvider):
     The threading implementation draws heavily from opentelemetry.sdk._shared_internal.BatchProcessor.
     """
 
-    def __init__(self, base_url: str, token: str, options: VariablesOptions):
+    def __init__(
+        self,
+        base_url: str,
+        token: str,
+        options: VariablesOptions,
+        transport_response_hook: TransportResponseHook | None = None,
+    ):
         """Create a new remote variable provider.
 
         Args:
             base_url: The base URL of the Logfire API.
             token: Authentication token for the Logfire API.
             options: Options for retrieving remote variables.
+            transport_response_hook: Optional override for the API response hook
+                (see `AdvancedOptions.transport_response_hook`).
         """
         block_before_first_resolve = options.block_before_first_resolve
         polling_interval = options.polling_interval
 
         self._base_url = base_url
         self._token = token
+        self._transport_response_hook = transport_response_hook
         self._session = Session()
         self._session.headers.update({'Authorization': f'bearer {token}', 'User-Agent': UA_HEADER})
-        install_logfire_response_hook(self._session)
+        install_logfire_response_hook(self._session, transport_response_hook)
         self._timeout = options.timeout
         self._block_before_first_fetch = block_before_first_resolve
         self._polling_interval: timedelta = (
@@ -199,7 +208,7 @@ def _sse_listener(self):  # pragma: no cover
                             'Cache-Control': 'no-cache',
                         }
                     )
-                    install_logfire_response_hook(sse_session)
+                    install_logfire_response_hook(sse_session, self._transport_response_hook)
 
                     # Open streaming connection
                     response = sse_session.get(sse_url, stream=True, timeout=(10, None))
diff --git a/tests/test_server_response.py b/tests/test_server_response.py
index 540253aea..6af1ddeee 100644
--- a/tests/test_server_response.py
+++ b/tests/test_server_response.py
@@ -76,3 +76,57 @@ def test_response_hook_installed_on_logfire_client():
         )
         with pytest.raises(LogfireServerError, match='no longer supported'):
             client.get_user_information()
+
+
+def test_custom_transport_response_hook_replaces_default():
+    """A custom hook replaces the built-in header processor entirely."""
+    from logfire._internal.auth import UserToken
+    from logfire._internal.client import LogfireClient
+
+    seen: list[requests.Response] = []
+
+    def my_hook(response: requests.Response) -> None:
+        seen.append(response)
+
+    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, transport_response_hook=my_hook)
+
+    with requests_mock.Mocker() as m:
+        m.get(
+            'https://logfire-us.pydantic.dev/v1/account/me',
+            json={'name': 'me'},
+            # Both headers set: default would warn AND raise; custom hook ignores them.
+            headers={WARNING_HEADER_NAME: 'deprecated', ERROR_HEADER_NAME: 'broken'},
+        )
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            client.get_user_information()
+
+    assert len(seen) == 1
+    assert not any(isinstance(w.message, LogfireServerWarning) for w in caught)
+
+
+def test_transport_response_hook_can_opt_out():
+    """`lambda response: None` disables both warnings and errors."""
+    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, transport_response_hook=lambda response: None)
+
+    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'},
+        )
+        # No exception raised.
+        assert client.get_user_information() == {'name': 'me'}

From 278814c9f51c9f0115947a4e94813b5f88b1351d Mon Sep 17 00:00:00 2001
From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>
Date: Wed, 6 May 2026 09:31:58 -0500
Subject: [PATCH 3/8] fix: mark transport_response_hook docstring example as
 skip-run

The example references `logfire`/`my_metric` without imports, which
test_docs runs through pytest-examples and trips on.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 logfire/_internal/config.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/logfire/_internal/config.py b/logfire/_internal/config.py
index 78dd2e7c4..1bb69d5d0 100644
--- a/logfire/_internal/config.py
+++ b/logfire/_internal/config.py
@@ -227,7 +227,7 @@ class AdvancedOptions:
     Setting this replaces the default; pass `lambda response: None` to opt out entirely,
     or compose your own logic on top of `process_logfire_response_headers`:
 
-    ```python
+    ```python skip-run="true" skip-reason="needs metric/logfire setup"
     from logfire._internal.server_response import process_logfire_response_headers
 
     def hook(response):

From be7a64db331c668d5d179ac41d30f6a47884ce04 Mon Sep 17 00:00:00 2001
From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>
Date: Wed, 6 May 2026 11:29:43 -0500
Subject: [PATCH 4/8] fix: pass transport_response_hook through lazy variable
 provider init

`_lazy_init_variable_provider` was constructing `LogfireRemoteVariableProvider`
without the configured hook, so providers spun up via the lazy path bypassed
the user-supplied response handling.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 logfire/_internal/config.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/logfire/_internal/config.py b/logfire/_internal/config.py
index 1bb69d5d0..74a634ca2 100644
--- a/logfire/_internal/config.py
+++ b/logfire/_internal/config.py
@@ -1482,6 +1482,7 @@ def _lazy_init_variable_provider(self) -> VariableProvider:
                 base_url=base_url,
                 token=api_key,
                 options=options,
+                transport_response_hook=self.advanced.transport_response_hook,
             )
             self._variable_provider = provider
             provider.start(Logfire(config=self))

From 1523cab85f1eb14e3c6a74bec9080ac25d80f165 Mon Sep 17 00:00:00 2001
From: Alex Hall <alex.mojaki@gmail.com>
Date: Thu, 7 May 2026 15:22:23 +0200
Subject: [PATCH 5/8] ServerResponseCallbackHelper

---
 logfire/_internal/client.py          |  6 +--
 logfire/_internal/config.py          |  4 +-
 logfire/_internal/server_response.py | 43 +++++++--------------
 logfire/types.py                     | 58 +++++++++++++++++++++++++++-
 logfire/variables/remote.py          |  4 +-
 tests/test_server_response.py        | 33 +++++++++-------
 6 files changed, 96 insertions(+), 52 deletions(-)

diff --git a/logfire/_internal/client.py b/logfire/_internal/client.py
index ac20ffbaa..390992791 100644
--- a/logfire/_internal/client.py
+++ b/logfire/_internal/client.py
@@ -10,7 +10,7 @@
 from logfire.version import VERSION
 
 from .auth import UserToken, UserTokenCollection
-from .server_response import TransportResponseHook, install_logfire_response_hook
+from .server_response import ServerResponseCallback, install_logfire_response_hook
 from .utils import UnexpectedResponse
 
 UA_HEADER = f'logfire/{VERSION}'
@@ -37,7 +37,7 @@ class LogfireClient:
     def __init__(
         self,
         user_token: UserToken,
-        transport_response_hook: TransportResponseHook | None = None,
+        transport_response_hook: ServerResponseCallback | None = None,
     ) -> None:
         if user_token.is_expired:
             raise RuntimeError('The provided user token is expired')
@@ -51,7 +51,7 @@ def __init__(
     def from_url(
         cls,
         base_url: str | None,
-        transport_response_hook: TransportResponseHook | None = None,
+        transport_response_hook: ServerResponseCallback | None = None,
     ) -> Self:
         """Create a client from the provided base URL.
 
diff --git a/logfire/_internal/config.py b/logfire/_internal/config.py
index 74a634ca2..0052765d8 100644
--- a/logfire/_internal/config.py
+++ b/logfire/_internal/config.py
@@ -110,7 +110,7 @@
 from .logs import ProxyLoggerProvider
 from .metrics import ProxyMeterProvider
 from .scrubbing import NOOP_SCRUBBER, BaseScrubber, Scrubber, ScrubbingOptions
-from .server_response import TransportResponseHook, install_logfire_response_hook
+from .server_response import ServerResponseCallback, install_logfire_response_hook
 from .stack_info import warn_at_user_stacklevel
 from .tracer import OPEN_SPANS, PendingSpanProcessor, ProxyTracerProvider
 from .utils import (
@@ -217,7 +217,7 @@ class AdvancedOptions:
     This log and configuration is experimental and may be modified or removed.
     """
 
-    transport_response_hook: TransportResponseHook | None = None
+    transport_response_hook: ServerResponseCallback | None = None
     """Optional callback invoked for every HTTP response received from the Logfire API.
 
     This applies to OTLP exports, credential / project initialisation, and the remote
diff --git a/logfire/_internal/server_response.py b/logfire/_internal/server_response.py
index bb1e6b4c0..385a1b6ba 100644
--- a/logfire/_internal/server_response.py
+++ b/logfire/_internal/server_response.py
@@ -12,52 +12,37 @@
 
 `install_logfire_response_hook(session)` wires this into a `requests.Session` as
 a response hook so every Logfire-bound HTTP response is inspected. Callers can
-pass a custom `hook` to replace the default behaviour (see
+pass a custom `hook` to replace the default behavior (see
 `AdvancedOptions.transport_response_hook`).
 """
 
 from __future__ import annotations
 
-import warnings
-from typing import Any, Callable
+from typing import Any
 
 import requests
 
-from logfire.exceptions import LogfireServerError, LogfireServerWarning
-
-WARNING_HEADER_NAME = 'X-Logfire-Warning'
-ERROR_HEADER_NAME = 'X-Logfire-Error'
-
-TransportResponseHook = Callable[[requests.Response], object]
-"""Callable invoked for every Logfire API response received by the SDK.
-
-The return value is ignored; raise to abort the call.
-"""
-
-
-def process_logfire_response_headers(response: requests.Response) -> None:
-    """Default transport response hook: surface `X-Logfire-Warning` / `X-Logfire-Error` headers."""
-    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)
+from logfire.types import ServerResponseCallback, ServerResponseCallbackHelper
 
 
 def install_logfire_response_hook(
     session: requests.Session,
-    hook: TransportResponseHook | None = None,
+    hook: ServerResponseCallback | None = None,
 ) -> None:
     """Install a `requests` response hook on `session` for every Logfire API response.
 
-    `hook` defaults to `process_logfire_response_headers`. Pass a custom callable
-    to replace the default behaviour (e.g. opt out by passing `lambda response: None`).
+    By default, calls ServerResponseCallbackHelper.default_hook(), which emits warnings and raises errors based
+    on the presence of `X-Logfire-Warning` and `X-Logfire-Error` response headers.
+
+    Pass a custom callable to replace the default behavior (e.g. opt out by passing `lambda _: None`).
     """
-    user_hook = hook if hook is not None else process_logfire_response_headers
 
-    def _hook(response: requests.Response, *_args: Any, **_kwargs: Any) -> requests.Response:
-        user_hook(response)
+    def _hook(response: requests.Response, *args: Any, **kwargs: Any) -> requests.Response:
+        helper = ServerResponseCallbackHelper(response, args, kwargs)
+        if hook:
+            hook(helper)
+        else:
+            helper.default_hook()
         return response
 
     response_hooks: list[Any] = session.hooks.setdefault('response', [])
diff --git a/logfire/types.py b/logfire/types.py
index 1f1561700..57f72e35e 100644
--- a/logfire/types.py
+++ b/logfire/types.py
@@ -2,7 +2,9 @@
 
 from collections.abc import Callable
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
+
+import requests
 
 from logfire._internal.constants import (
     ATTRIBUTES_LOG_LEVEL_NUM_KEY,
@@ -11,8 +13,10 @@
     LevelName,
     log_level_attributes,
 )
+from logfire._internal.stack_info import warn_at_user_stacklevel
 from logfire._internal.tracer import get_parent_span
 from logfire._internal.utils import canonicalize_exception_traceback
+from logfire.exceptions import LogfireServerError, LogfireServerWarning
 
 if TYPE_CHECKING:
     from opentelemetry.sdk.trace import ReadableSpan, Span
@@ -260,3 +264,55 @@ def my_callback(helper: logfire.types.ExceptionCallbackHelper):
 
     helper.no_record_exception()
 """
+
+
+@dataclass
+class ServerResponseCallbackHelper:
+    """Helper object passed to the server response callback.
+
+    This is experimental and may change significantly in future releases.
+    """
+
+    response: requests.Response
+    """The raw HTTP response from the Logfire API."""
+
+    args: tuple[Any, ...]
+    """Positional arguments passed to the response hook by `requests`."""
+
+    kwargs: dict[str, Any]
+    """Keyword arguments passed to the response hook by `requests`."""
+
+    WARNING_HEADER_NAME = 'X-Logfire-Warning'
+    ERROR_HEADER_NAME = 'X-Logfire-Error'
+
+    @property
+    def warning_header(self) -> str | None:
+        """Value of the Logfire warning header, or `None` if not present."""
+        return self.response.headers.get(self.WARNING_HEADER_NAME)
+
+    @property
+    def error_header(self) -> str | None:
+        """Value of the Logfire error header, or `None` if not present."""
+        return self.response.headers.get(self.ERROR_HEADER_NAME)
+
+    def default_hook(self, *, check_warning: bool = True, check_error: bool = True) -> None:
+        """The default hook behavior.
+
+        If check_warning is true, check for a warning header and raise it as a LogfireServerWarning if present.
+        If check_error is true, check for an error header and raise it as a LogfireServerError if present.
+        """
+        if check_warning:
+            warning_message = self.warning_header
+            if warning_message:
+                warn_at_user_stacklevel(warning_message, LogfireServerWarning)
+        if check_error:
+            error_message = self.error_header
+            if error_message:
+                raise LogfireServerError(error_message)
+
+
+ServerResponseCallback = Callable[[ServerResponseCallbackHelper], None]
+"""Callable invoked for every Logfire API response received by the SDK.
+
+This is experimental and may change significantly in future releases.
+"""
diff --git a/logfire/variables/remote.py b/logfire/variables/remote.py
index 91034a96c..db3cf4f7c 100644
--- a/logfire/variables/remote.py
+++ b/logfire/variables/remote.py
@@ -17,7 +17,7 @@
 
 from logfire._internal.client import UA_HEADER
 from logfire._internal.config import VariablesOptions
-from logfire._internal.server_response import TransportResponseHook, install_logfire_response_hook
+from logfire._internal.server_response import ServerResponseCallback, install_logfire_response_hook
 from logfire._internal.utils import UnexpectedResponse
 from logfire.variables.abstract import (
     ResolvedVariable,
@@ -60,7 +60,7 @@ def __init__(
         base_url: str,
         token: str,
         options: VariablesOptions,
-        transport_response_hook: TransportResponseHook | None = None,
+        transport_response_hook: ServerResponseCallback | None = None,
     ):
         """Create a new remote variable provider.
 
diff --git a/tests/test_server_response.py b/tests/test_server_response.py
index 6af1ddeee..c38e6cb2b 100644
--- a/tests/test_server_response.py
+++ b/tests/test_server_response.py
@@ -8,19 +8,19 @@
 from inline_snapshot import snapshot
 
 from logfire._internal.server_response import (
-    ERROR_HEADER_NAME,
-    WARNING_HEADER_NAME,
-    process_logfire_response_headers,
+    ServerResponseCallbackHelper,
 )
 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'
+    response.headers[ServerResponseCallbackHelper.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)
+        ServerResponseCallbackHelper(response, (), {}).default_hook()
     assert [(w.category, str(w.message)) for w in caught] == snapshot(
         [(LogfireServerWarning, 'The /foo/bar endpoint is deprecated, please use /bar/baz')]
     )
@@ -29,20 +29,20 @@ def test_process_response_warning_header_emits_warning():
 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'
+    response.headers[ServerResponseCallbackHelper.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)
+            ServerResponseCallbackHelper(response, (), {}).default_hook()
     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'
+    response.headers[ServerResponseCallbackHelper.ERROR_HEADER_NAME] = 'something is wrong'
     with pytest.raises(LogfireServerError, match='something is wrong'):
-        process_logfire_response_headers(response)
+        ServerResponseCallbackHelper(response, (), {}).default_hook()
 
 
 def test_response_hook_installed_on_logfire_client():
@@ -60,7 +60,7 @@ def test_response_hook_installed_on_logfire_client():
         m.get(
             'https://logfire-us.pydantic.dev/v1/account/me',
             json={'name': 'me'},
-            headers={WARNING_HEADER_NAME: 'deprecated endpoint'},
+            headers={ServerResponseCallbackHelper.WARNING_HEADER_NAME: 'deprecated endpoint'},
         )
         with warnings.catch_warnings(record=True) as caught:
             warnings.simplefilter('always')
@@ -72,7 +72,7 @@ def test_response_hook_installed_on_logfire_client():
         m.get(
             'https://logfire-us.pydantic.dev/v1/account/me',
             json={'name': 'me'},
-            headers={ERROR_HEADER_NAME: 'no longer supported'},
+            headers={ServerResponseCallbackHelper.ERROR_HEADER_NAME: 'no longer supported'},
         )
         with pytest.raises(LogfireServerError, match='no longer supported'):
             client.get_user_information()
@@ -85,8 +85,8 @@ def test_custom_transport_response_hook_replaces_default():
 
     seen: list[requests.Response] = []
 
-    def my_hook(response: requests.Response) -> None:
-        seen.append(response)
+    def my_hook(helper: ServerResponseCallbackHelper) -> None:
+        seen.append(helper.response)
 
     token = UserToken(
         token='pylf_v1_us_xxx',
@@ -100,7 +100,10 @@ def my_hook(response: requests.Response) -> None:
             'https://logfire-us.pydantic.dev/v1/account/me',
             json={'name': 'me'},
             # Both headers set: default would warn AND raise; custom hook ignores them.
-            headers={WARNING_HEADER_NAME: 'deprecated', ERROR_HEADER_NAME: 'broken'},
+            headers={
+                ServerResponseCallbackHelper.WARNING_HEADER_NAME: 'deprecated',
+                ServerResponseCallbackHelper.ERROR_HEADER_NAME: 'broken',
+            },
         )
         with warnings.catch_warnings(record=True) as caught:
             warnings.simplefilter('always')
@@ -126,7 +129,7 @@ def test_transport_response_hook_can_opt_out():
         m.get(
             'https://logfire-us.pydantic.dev/v1/account/me',
             json={'name': 'me'},
-            headers={ERROR_HEADER_NAME: 'no longer supported'},
+            headers={ServerResponseCallbackHelper.ERROR_HEADER_NAME: 'no longer supported'},
         )
         # No exception raised.
         assert client.get_user_information() == {'name': 'me'}

From 90dc841b749a6415a04b3fd94e4b902d0a101bc1 Mon Sep 17 00:00:00 2001
From: Alex Hall <alex.mojaki@gmail.com>
Date: Thu, 7 May 2026 15:24:51 +0200
Subject: [PATCH 6/8] server_response_hook

---
 logfire/_internal/client.py          | 16 ++++++++--------
 logfire/_internal/config.py          | 14 +++++++-------
 logfire/_internal/server_response.py |  2 +-
 logfire/variables/remote.py          | 12 ++++++------
 tests/test_server_response.py        | 12 +++++-------
 5 files changed, 27 insertions(+), 29 deletions(-)

diff --git a/logfire/_internal/client.py b/logfire/_internal/client.py
index 390992791..a66a35b57 100644
--- a/logfire/_internal/client.py
+++ b/logfire/_internal/client.py
@@ -30,14 +30,14 @@ class LogfireClient:
 
     Args:
         user_token: The user token to use when authenticating against the API.
-        transport_response_hook: Optional override for the API response hook (see
-            `AdvancedOptions.transport_response_hook`).
+        server_response_hook: Optional override for the API response hook (see
+            `AdvancedOptions.server_response_hook`).
     """
 
     def __init__(
         self,
         user_token: UserToken,
-        transport_response_hook: ServerResponseCallback | None = None,
+        server_response_hook: ServerResponseCallback | None = None,
     ) -> None:
         if user_token.is_expired:
             raise RuntimeError('The provided user token is expired')
@@ -45,13 +45,13 @@ def __init__(
         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, transport_response_hook)
+        install_logfire_response_hook(self._session, server_response_hook)
 
     @classmethod
     def from_url(
         cls,
         base_url: str | None,
-        transport_response_hook: ServerResponseCallback | None = None,
+        server_response_hook: ServerResponseCallback | None = None,
     ) -> Self:
         """Create a client from the provided base URL.
 
@@ -60,12 +60,12 @@ def from_url(
                 the user into selecting a token from the token collection (or, if only one available,
                 use it directly). The token collection will be created from the `~/.logfire/default.toml`
                 file (or an empty one if no such file exists).
-            transport_response_hook: Optional override for the API response hook (see
-                `AdvancedOptions.transport_response_hook`).
+            server_response_hook: Optional override for the API response hook (see
+                `AdvancedOptions.server_response_hook`).
         """
         return cls(
             user_token=UserTokenCollection().get_token(base_url),
-            transport_response_hook=transport_response_hook,
+            server_response_hook=server_response_hook,
         )
 
     def _get_raw(self, endpoint: str, params: dict[str, Any] | None = None) -> Response:
diff --git a/logfire/_internal/config.py b/logfire/_internal/config.py
index 0052765d8..699ff5900 100644
--- a/logfire/_internal/config.py
+++ b/logfire/_internal/config.py
@@ -217,7 +217,7 @@ class AdvancedOptions:
     This log and configuration is experimental and may be modified or removed.
     """
 
-    transport_response_hook: ServerResponseCallback | None = None
+    server_response_hook: ServerResponseCallback | None = None
     """Optional callback invoked for every HTTP response received from the Logfire API.
 
     This applies to OTLP exports, credential / project initialisation, and the remote
@@ -234,7 +234,7 @@ def hook(response):
         my_metric.inc(response.status_code)
         process_logfire_response_headers(response)
 
-    logfire.configure(advanced=AdvancedOptions(transport_response_hook=hook))
+    logfire.configure(advanced=AdvancedOptions(server_response_hook=hook))
     ```
 
     Raise from the hook to abort the calling code path.
@@ -1121,7 +1121,7 @@ def add_span_processor(span_processor: SpanProcessor) -> None:
                     # If we don't have tokens or credentials from a file,
                     # try initializing a new project and writing a new creds file.
                     # note, we only do this if `send_to_logfire` is explicitly `True`, not 'if-token-present'
-                    client = LogfireClient.from_url(self.advanced.base_url, self.advanced.transport_response_hook)
+                    client = LogfireClient.from_url(self.advanced.base_url, self.advanced.server_response_hook)
                     credentials = LogfireCredentials.initialize_project(client=client)
                     credentials.write_creds_file(self.data_dir)
 
@@ -1172,7 +1172,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, self.advanced.transport_response_hook)
+                        install_logfire_response_hook(session, self.advanced.server_response_hook)
                         span_exporter = BodySizeCheckingOTLPSpanExporter(
                             endpoint=urljoin(base_url, '/v1/traces'),
                             session=session,
@@ -1349,7 +1349,7 @@ def fix_pid():  # pragma: no cover
                     base_url=base_url,
                     token=self.api_key,
                     options=self.variables,
-                    transport_response_hook=self.advanced.transport_response_hook,
+                    server_response_hook=self.advanced.server_response_hook,
                 )
             multi_log_processor = SynchronousMultiLogRecordProcessor()
             for processor in log_record_processors:
@@ -1482,7 +1482,7 @@ def _lazy_init_variable_provider(self) -> VariableProvider:
                 base_url=base_url,
                 token=api_key,
                 options=options,
-                transport_response_hook=self.advanced.transport_response_hook,
+                server_response_hook=self.advanced.server_response_hook,
             )
             self._variable_provider = provider
             provider.start(Logfire(config=self))
@@ -1500,7 +1500,7 @@ 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, self.advanced.transport_response_hook)
+        install_logfire_response_hook(session, self.advanced.server_response_hook)
         return LogfireCredentials.from_token(token, session, self.advanced.generate_base_url(token))
 
     def _ensure_flush_after_aws_lambda(self):
diff --git a/logfire/_internal/server_response.py b/logfire/_internal/server_response.py
index 385a1b6ba..c58a529ca 100644
--- a/logfire/_internal/server_response.py
+++ b/logfire/_internal/server_response.py
@@ -13,7 +13,7 @@
 `install_logfire_response_hook(session)` wires this into a `requests.Session` as
 a response hook so every Logfire-bound HTTP response is inspected. Callers can
 pass a custom `hook` to replace the default behavior (see
-`AdvancedOptions.transport_response_hook`).
+`AdvancedOptions.server_response_hook`).
 """
 
 from __future__ import annotations
diff --git a/logfire/variables/remote.py b/logfire/variables/remote.py
index db3cf4f7c..0de792837 100644
--- a/logfire/variables/remote.py
+++ b/logfire/variables/remote.py
@@ -60,7 +60,7 @@ def __init__(
         base_url: str,
         token: str,
         options: VariablesOptions,
-        transport_response_hook: ServerResponseCallback | None = None,
+        server_response_hook: ServerResponseCallback | None = None,
     ):
         """Create a new remote variable provider.
 
@@ -68,18 +68,18 @@ def __init__(
             base_url: The base URL of the Logfire API.
             token: Authentication token for the Logfire API.
             options: Options for retrieving remote variables.
-            transport_response_hook: Optional override for the API response hook
-                (see `AdvancedOptions.transport_response_hook`).
+            server_response_hook: Optional override for the API response hook
+                (see `AdvancedOptions.server_response_hook`).
         """
         block_before_first_resolve = options.block_before_first_resolve
         polling_interval = options.polling_interval
 
         self._base_url = base_url
         self._token = token
-        self._transport_response_hook = transport_response_hook
+        self._server_response_hook = server_response_hook
         self._session = Session()
         self._session.headers.update({'Authorization': f'bearer {token}', 'User-Agent': UA_HEADER})
-        install_logfire_response_hook(self._session, transport_response_hook)
+        install_logfire_response_hook(self._session, server_response_hook)
         self._timeout = options.timeout
         self._block_before_first_fetch = block_before_first_resolve
         self._polling_interval: timedelta = (
@@ -208,7 +208,7 @@ def _sse_listener(self):  # pragma: no cover
                             'Cache-Control': 'no-cache',
                         }
                     )
-                    install_logfire_response_hook(sse_session, self._transport_response_hook)
+                    install_logfire_response_hook(sse_session, self._server_response_hook)
 
                     # Open streaming connection
                     response = sse_session.get(sse_url, stream=True, timeout=(10, None))
diff --git a/tests/test_server_response.py b/tests/test_server_response.py
index c38e6cb2b..3a7eaaf2e 100644
--- a/tests/test_server_response.py
+++ b/tests/test_server_response.py
@@ -7,10 +7,8 @@
 import requests_mock
 from inline_snapshot import snapshot
 
-from logfire._internal.server_response import (
-    ServerResponseCallbackHelper,
-)
 from logfire.exceptions import LogfireServerError, LogfireServerWarning
+from logfire.types import ServerResponseCallbackHelper
 
 
 def test_process_response_warning_header_emits_warning():
@@ -78,7 +76,7 @@ def test_response_hook_installed_on_logfire_client():
             client.get_user_information()
 
 
-def test_custom_transport_response_hook_replaces_default():
+def test_custom_server_response_hook_replaces_default():
     """A custom hook replaces the built-in header processor entirely."""
     from logfire._internal.auth import UserToken
     from logfire._internal.client import LogfireClient
@@ -93,7 +91,7 @@ def my_hook(helper: ServerResponseCallbackHelper) -> None:
         base_url='https://logfire-us.pydantic.dev',
         expiration='2099-12-31T23:59:59',
     )
-    client = LogfireClient(user_token=token, transport_response_hook=my_hook)
+    client = LogfireClient(user_token=token, server_response_hook=my_hook)
 
     with requests_mock.Mocker() as m:
         m.get(
@@ -113,7 +111,7 @@ def my_hook(helper: ServerResponseCallbackHelper) -> None:
     assert not any(isinstance(w.message, LogfireServerWarning) for w in caught)
 
 
-def test_transport_response_hook_can_opt_out():
+def test_server_response_hook_can_opt_out():
     """`lambda response: None` disables both warnings and errors."""
     from logfire._internal.auth import UserToken
     from logfire._internal.client import LogfireClient
@@ -123,7 +121,7 @@ def test_transport_response_hook_can_opt_out():
         base_url='https://logfire-us.pydantic.dev',
         expiration='2099-12-31T23:59:59',
     )
-    client = LogfireClient(user_token=token, transport_response_hook=lambda response: None)
+    client = LogfireClient(user_token=token, server_response_hook=lambda response: None)
 
     with requests_mock.Mocker() as m:
         m.get(

From 0d44d79239a921a06a74607a35319b6d4089d7a1 Mon Sep 17 00:00:00 2001
From: Alex Hall <alex.mojaki@gmail.com>
Date: Thu, 7 May 2026 15:27:00 +0200
Subject: [PATCH 7/8] docstring

---
 logfire/_internal/config.py | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/logfire/_internal/config.py b/logfire/_internal/config.py
index 699ff5900..1f473503d 100644
--- a/logfire/_internal/config.py
+++ b/logfire/_internal/config.py
@@ -220,19 +220,22 @@ class AdvancedOptions:
     server_response_hook: ServerResponseCallback | None = None
     """Optional callback invoked for every HTTP response received from the Logfire API.
 
+    This is experimental and may be modified or removed.
+
     This applies to OTLP exports, credential / project initialisation, and the remote
     variables provider. The default surfaces `X-Logfire-Warning` and `X-Logfire-Error`
     headers as `LogfireServerWarning` / `LogfireServerError`.
 
-    Setting this replaces the default; pass `lambda response: None` to opt out entirely,
-    or compose your own logic on top of `process_logfire_response_headers`:
+    Setting this replaces the default; pass `lambda response: None` to opt out entirely.
+
+    Example usage:
 
     ```python skip-run="true" skip-reason="needs metric/logfire setup"
-    from logfire._internal.server_response import process_logfire_response_headers
+    from logfire.types import ServerResponseCallbackHelper
 
-    def hook(response):
-        my_metric.inc(response.status_code)
-        process_logfire_response_headers(response)
+    def hook(helper: ServerResponseCallbackHelper):
+        my_metric.inc(response.response.status_code)
+        helper.default_hook()  # call this to keep the default behavior of raising/logging based on headers
 
     logfire.configure(advanced=AdvancedOptions(server_response_hook=hook))
     ```

From ee581a9223d5504c0e914be6a07b9e728d2fde30 Mon Sep 17 00:00:00 2001
From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>
Date: Thu, 7 May 2026 08:37:22 -0500
Subject: [PATCH 8/8] remove server-side error header, keep warning

Drop LogfireServerError and X-Logfire-Error handling; default hook now only
emits LogfireServerWarning from X-Logfire-Warning. Also fixes cubic review
comments: explicit None check on the hook callback, and fix the docstring
example to use helper.response.status_code.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 logfire/_internal/config.py          |  8 +++----
 logfire/_internal/server_response.py | 19 ++++++---------
 logfire/exceptions.py                |  4 ----
 logfire/types.py                     | 27 +++++----------------
 tests/test_server_response.py        | 36 +++++++---------------------
 5 files changed, 26 insertions(+), 68 deletions(-)

diff --git a/logfire/_internal/config.py b/logfire/_internal/config.py
index 1f473503d..05c55488c 100644
--- a/logfire/_internal/config.py
+++ b/logfire/_internal/config.py
@@ -223,8 +223,8 @@ class AdvancedOptions:
     This is experimental and may be modified or removed.
 
     This applies to OTLP exports, credential / project initialisation, and the remote
-    variables provider. The default surfaces `X-Logfire-Warning` and `X-Logfire-Error`
-    headers as `LogfireServerWarning` / `LogfireServerError`.
+    variables provider. The default surfaces the `X-Logfire-Warning` header as a
+    `LogfireServerWarning`.
 
     Setting this replaces the default; pass `lambda response: None` to opt out entirely.
 
@@ -234,8 +234,8 @@ class AdvancedOptions:
     from logfire.types import ServerResponseCallbackHelper
 
     def hook(helper: ServerResponseCallbackHelper):
-        my_metric.inc(response.response.status_code)
-        helper.default_hook()  # call this to keep the default behavior of raising/logging based on headers
+        my_metric.inc(helper.response.status_code)
+        helper.default_hook()  # call this to keep the default warning behavior
 
     logfire.configure(advanced=AdvancedOptions(server_response_hook=hook))
     ```
diff --git a/logfire/_internal/server_response.py b/logfire/_internal/server_response.py
index c58a529ca..3001bdadc 100644
--- a/logfire/_internal/server_response.py
+++ b/logfire/_internal/server_response.py
@@ -1,14 +1,9 @@
 """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.
+The server attaches the `X-Logfire-Warning` header to API responses to signal an
+out-of-band warning the server wants the user to see. It is surfaced via
+`warnings.warn(..., LogfireServerWarning)`. Python's standard "default" filter
+dedupes identical messages, so a chatty server only warns once.
 
 `install_logfire_response_hook(session)` wires this into a `requests.Session` as
 a response hook so every Logfire-bound HTTP response is inspected. Callers can
@@ -31,15 +26,15 @@ def install_logfire_response_hook(
 ) -> None:
     """Install a `requests` response hook on `session` for every Logfire API response.
 
-    By default, calls ServerResponseCallbackHelper.default_hook(), which emits warnings and raises errors based
-    on the presence of `X-Logfire-Warning` and `X-Logfire-Error` response headers.
+    By default, calls `ServerResponseCallbackHelper.default_hook()`, which emits a warning
+    if the `X-Logfire-Warning` response header is present.
 
     Pass a custom callable to replace the default behavior (e.g. opt out by passing `lambda _: None`).
     """
 
     def _hook(response: requests.Response, *args: Any, **kwargs: Any) -> requests.Response:
         helper = ServerResponseCallbackHelper(response, args, kwargs)
-        if hook:
+        if hook is not None:
             hook(helper)
         else:
             helper.default_hook()
diff --git a/logfire/exceptions.py b/logfire/exceptions.py
index 532fb2720..4a3ac54fc 100644
--- a/logfire/exceptions.py
+++ b/logfire/exceptions.py
@@ -5,9 +5,5 @@ 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."""
diff --git a/logfire/types.py b/logfire/types.py
index 57f72e35e..75b61ff75 100644
--- a/logfire/types.py
+++ b/logfire/types.py
@@ -16,7 +16,7 @@
 from logfire._internal.stack_info import warn_at_user_stacklevel
 from logfire._internal.tracer import get_parent_span
 from logfire._internal.utils import canonicalize_exception_traceback
-from logfire.exceptions import LogfireServerError, LogfireServerWarning
+from logfire.exceptions import LogfireServerWarning
 
 if TYPE_CHECKING:
     from opentelemetry.sdk.trace import ReadableSpan, Span
@@ -283,32 +283,17 @@ class ServerResponseCallbackHelper:
     """Keyword arguments passed to the response hook by `requests`."""
 
     WARNING_HEADER_NAME = 'X-Logfire-Warning'
-    ERROR_HEADER_NAME = 'X-Logfire-Error'
 
     @property
     def warning_header(self) -> str | None:
         """Value of the Logfire warning header, or `None` if not present."""
         return self.response.headers.get(self.WARNING_HEADER_NAME)
 
-    @property
-    def error_header(self) -> str | None:
-        """Value of the Logfire error header, or `None` if not present."""
-        return self.response.headers.get(self.ERROR_HEADER_NAME)
-
-    def default_hook(self, *, check_warning: bool = True, check_error: bool = True) -> None:
-        """The default hook behavior.
-
-        If check_warning is true, check for a warning header and raise it as a LogfireServerWarning if present.
-        If check_error is true, check for an error header and raise it as a LogfireServerError if present.
-        """
-        if check_warning:
-            warning_message = self.warning_header
-            if warning_message:
-                warn_at_user_stacklevel(warning_message, LogfireServerWarning)
-        if check_error:
-            error_message = self.error_header
-            if error_message:
-                raise LogfireServerError(error_message)
+    def default_hook(self) -> None:
+        """The default hook behavior: emit a `LogfireServerWarning` if the warning header is present."""
+        warning_message = self.warning_header
+        if warning_message:
+            warn_at_user_stacklevel(warning_message, LogfireServerWarning)
 
 
 ServerResponseCallback = Callable[[ServerResponseCallbackHelper], None]
diff --git a/tests/test_server_response.py b/tests/test_server_response.py
index 3a7eaaf2e..503a0f41c 100644
--- a/tests/test_server_response.py
+++ b/tests/test_server_response.py
@@ -2,12 +2,11 @@
 
 import warnings
 
-import pytest
 import requests
 import requests_mock
 from inline_snapshot import snapshot
 
-from logfire.exceptions import LogfireServerError, LogfireServerWarning
+from logfire.exceptions import LogfireServerWarning
 from logfire.types import ServerResponseCallbackHelper
 
 
@@ -36,13 +35,6 @@ def test_process_response_warning_header_dedupes():
     assert messages == ['a duplicated warning']
 
 
-def test_process_response_error_header_raises():
-    response = requests.Response()
-    response.headers[ServerResponseCallbackHelper.ERROR_HEADER_NAME] = 'something is wrong'
-    with pytest.raises(LogfireServerError, match='something is wrong'):
-        ServerResponseCallbackHelper(response, (), {}).default_hook()
-
-
 def test_response_hook_installed_on_logfire_client():
     from logfire._internal.auth import UserToken
     from logfire._internal.client import LogfireClient
@@ -66,15 +58,6 @@ def test_response_hook_installed_on_logfire_client():
 
     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={ServerResponseCallbackHelper.ERROR_HEADER_NAME: 'no longer supported'},
-        )
-        with pytest.raises(LogfireServerError, match='no longer supported'):
-            client.get_user_information()
-
 
 def test_custom_server_response_hook_replaces_default():
     """A custom hook replaces the built-in header processor entirely."""
@@ -97,11 +80,7 @@ def my_hook(helper: ServerResponseCallbackHelper) -> None:
         m.get(
             'https://logfire-us.pydantic.dev/v1/account/me',
             json={'name': 'me'},
-            # Both headers set: default would warn AND raise; custom hook ignores them.
-            headers={
-                ServerResponseCallbackHelper.WARNING_HEADER_NAME: 'deprecated',
-                ServerResponseCallbackHelper.ERROR_HEADER_NAME: 'broken',
-            },
+            headers={ServerResponseCallbackHelper.WARNING_HEADER_NAME: 'deprecated'},
         )
         with warnings.catch_warnings(record=True) as caught:
             warnings.simplefilter('always')
@@ -112,7 +91,7 @@ def my_hook(helper: ServerResponseCallbackHelper) -> None:
 
 
 def test_server_response_hook_can_opt_out():
-    """`lambda response: None` disables both warnings and errors."""
+    """`lambda response: None` disables the default warning behavior."""
     from logfire._internal.auth import UserToken
     from logfire._internal.client import LogfireClient
 
@@ -127,7 +106,10 @@ def test_server_response_hook_can_opt_out():
         m.get(
             'https://logfire-us.pydantic.dev/v1/account/me',
             json={'name': 'me'},
-            headers={ServerResponseCallbackHelper.ERROR_HEADER_NAME: 'no longer supported'},
+            headers={ServerResponseCallbackHelper.WARNING_HEADER_NAME: 'deprecated'},
         )
-        # No exception raised.
-        assert client.get_user_information() == {'name': 'me'}
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            assert client.get_user_information() == {'name': 'me'}
+
+    assert not any(isinstance(w.message, LogfireServerWarning) for w in caught)