feat: Add experimental async transport (port of PR #4572) (#5646)

Add an experimental async transport using httpcore's async backend,
enabled via `_experiments={"transport_async": True}`.

This is a manual port of PR #4572 (originally merged into `potel-base`)
onto the current `master` branch.

## Key changes

- **transport.py**: Refactor `BaseHttpTransport` into
`HttpTransportCore`
  (shared base) + `BaseHttpTransport` (sync) + `AsyncHttpTransport`
  (async, conditional on `httpcore[asyncio]`). Extract shared helpers:
  `_handle_request_error`, `_handle_response`, `_update_headers`,
  `_prepare_envelope`. Update `make_transport()` to detect the
  `transport_async` experiment.

- **worker.py**: Add `Worker` ABC base class and `AsyncWorker`
  implementation using `asyncio.Queue` / `asyncio.Task`.

- **client.py**: Add `close_async()` / `flush_async()` with
async-vs-sync
transport detection. Extract `_close_components()` /
`_flush_components()`.

- **api.py**: Expose `flush_async()` as a public API.

- **integrations/asyncio.py**: Patch `loop.close` to flush pending
events
  before shutdown. Skip span wrapping for internal Sentry tasks.

- **utils.py**: Add `is_internal_task()` / `mark_sentry_task_internal()`
  via ContextVar for async task filtering.

- **setup.py**: Add `"asyncio"` extras_require
(`httpcore[asyncio]==1.*`).

- **config.py / tox.ini**: Widen anyio to `>=3,<5` for httpx and
FastAPI.

## Notes

- `tox.ini` was manually edited (the generation script requires a
  free-threaded Python interpreter). A full regeneration should be done
  before merge.

Refs: GH-4568

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Ivana Kellyer <ivana.kellyer@sentry.io>

Author: 3 people

scripts/populate_tox/config.py

@@ -122,7 +122,7 @@
                 "pytest-asyncio",
                 "python-multipart",
                 "requests",
-                "anyio<4",
+                "anyio>=3,<5",
                 "jinja2",
             ],
             # There's an incompatibility between FastAPI's TestClient, which is
@@ -133,6 +133,7 @@
             # FastAPI versions we use older httpx which still supports the
             # deprecated argument.
             "<0.110.1": ["httpx<0.28.0"],
+            "<0.80": ["anyio<4"],
             "py3.6": ["aiocontextvars"],
         },
     },
@@ -171,7 +172,8 @@
     "httpx": {
         "package": "httpx",
         "deps": {
-            "*": ["anyio<4.0.0"],
+            "*": ["anyio>=3,<5"],
+            "<0.24": ["anyio<4"],
             ">=0.16,<0.17": ["pytest-httpx==0.10.0"],
             ">=0.17,<0.19": ["pytest-httpx==0.12.0"],
             ">=0.19,<0.21": ["pytest-httpx==0.14.0"],

scripts/populate_tox/releases.jsonl

@@ -78,14 +78,17 @@ deps =
 
     linters: -r requirements-linting.txt
     linters: werkzeug<2.3.0
+    linters: httpcore[asyncio]
 
     mypy: -r requirements-linting.txt
     mypy: werkzeug<2.3.0
+    mypy: httpcore[asyncio]
     ruff: -r requirements-linting.txt
 
     # === Common ===
     py3.8-common: hypothesis
     common: pytest-asyncio
+    common: httpcore[asyncio]
     # See https://github.com/pytest-dev/pytest/issues/9621
     # and https://github.com/pytest-dev/pytest-forked/issues/67
     # for justification of the upper bound on pytest

scripts/populate_tox/tox.jinja

@@ -25,6 +25,7 @@
     "configure_scope",
     "continue_trace",
     "flush",
+    "flush_async",
     "get_baggage",
     "get_client",
     "get_global_scope",

sentry_sdk/__init__.py

@@ -60,6 +60,7 @@ def overload(x: "T") -> "T":
     "configure_scope",
     "continue_trace",
     "flush",
+    "flush_async",
     "get_baggage",
     "get_client",
     "get_global_scope",
@@ -351,6 +352,14 @@ def flush(
     return get_client().flush(timeout=timeout, callback=callback)
 
 
+@clientmethod
+async def flush_async(
+    timeout: "Optional[float]" = None,
+    callback: "Optional[Callable[[int, float], None]]" = None,
+) -> None:
+    return await get_client().flush_async(timeout=timeout, callback=callback)
+
+
 @scopemethod
 def start_span(
     **kwargs: "Any",

sentry_sdk/api.py

@@ -31,7 +31,11 @@
 from sentry_sdk.serializer import serialize
 from sentry_sdk.tracing import trace
 from sentry_sdk.tracing_utils import has_span_streaming_enabled
-from sentry_sdk.transport import BaseHttpTransport, make_transport
+from sentry_sdk.transport import (
+    HttpTransportCore,
+    make_transport,
+    AsyncHttpTransport,
+)
 from sentry_sdk.consts import (
     SPANDATA,
     DEFAULT_MAX_VALUE_LENGTH,
@@ -251,6 +255,12 @@ def close(self, *args: "Any", **kwargs: "Any") -> None:
     def flush(self, *args: "Any", **kwargs: "Any") -> None:
         return None
 
+    async def close_async(self, *args: "Any", **kwargs: "Any") -> None:
+        return None
+
+    async def flush_async(self, *args: "Any", **kwargs: "Any") -> None:
+        return None
+
     def __enter__(self) -> "BaseClient":
         return self
 
@@ -472,7 +482,7 @@ def _record_lost_event(
             or self.metrics_batcher
             or self.span_batcher
             or has_profiling_enabled(self.options)
-            or isinstance(self.transport, BaseHttpTransport)
+            or isinstance(self.transport, HttpTransportCore)
         ):
             # If we have anything on that could spawn a background thread, we
             # need to check if it's safe to use them.
@@ -999,6 +1009,32 @@ def get_integration(
 
         return self.integrations.get(integration_name)
 
+    def _has_async_transport(self) -> bool:
+        """Check if the current transport is async."""
+        return isinstance(self.transport, AsyncHttpTransport)
+
+    @property
+    def _batchers(self) -> "tuple[Any, ...]":
+        return tuple(
+            b
+            for b in (self.log_batcher, self.metrics_batcher, self.span_batcher)
+            if b is not None
+        )
+
+    def _close_components(self) -> None:
+        """Kill all client components in the correct order."""
+        self.session_flusher.kill()
+        for b in self._batchers:
+            b.kill()
+        if self.monitor:
+            self.monitor.kill()
+
+    def _flush_components(self) -> None:
+        """Flush all client components."""
+        self.session_flusher.flush()
+        for b in self._batchers:
+            b.flush()
+
     def close(
         self,
         timeout: "Optional[float]" = None,
@@ -1009,19 +1045,40 @@ def close(
         semantics as :py:meth:`Client.flush`.
         """
         if self.transport is not None:
-            self.flush(timeout=timeout, callback=callback)
-            self.session_flusher.kill()
-            if self.log_batcher is not None:
-                self.log_batcher.kill()
-            if self.metrics_batcher is not None:
-                self.metrics_batcher.kill()
-            if self.span_batcher is not None:
-                self.span_batcher.kill()
-            if self.monitor:
-                self.monitor.kill()
+            if self._has_async_transport():
+                warnings.warn(
+                    "close() used with AsyncHttpTransport. Use close_async() instead.",
+                    stacklevel=2,
+                )
+                self._flush_components()
+            else:
+                self.flush(timeout=timeout, callback=callback)
+            self._close_components()
             self.transport.kill()
             self.transport = None
 
+    async def close_async(
+        self,
+        timeout: "Optional[float]" = None,
+        callback: "Optional[Callable[[int, float], None]]" = None,
+    ) -> None:
+        """
+        Asynchronously close the client and shut down the transport. Arguments have the same
+        semantics as :py:meth:`Client.flush_async`.
+        """
+        if self.transport is not None:
+            if not self._has_async_transport():
+                logger.debug(
+                    "close_async() used with non-async transport, aborting. Please use close() instead."
+                )
+                return
+            await self.flush_async(timeout=timeout, callback=callback)
+            self._close_components()
+            kill_task = self.transport.kill()  # type: ignore
+            if kill_task is not None:
+                await kill_task
+            self.transport = None
+
     def flush(
         self,
         timeout: "Optional[float]" = None,
@@ -1035,23 +1092,55 @@ def flush(
         :param callback: Is invoked with the number of pending events and the configured timeout.
         """
         if self.transport is not None:
+            if self._has_async_transport():
+                warnings.warn(
+                    "flush() used with AsyncHttpTransport. Use flush_async() instead.",
+                    stacklevel=2,
+                )
+                return
             if timeout is None:
                 timeout = self.options["shutdown_timeout"]
-            self.session_flusher.flush()
-            if self.log_batcher is not None:
-                self.log_batcher.flush()
-            if self.metrics_batcher is not None:
-                self.metrics_batcher.flush()
-            if self.span_batcher is not None:
-                self.span_batcher.flush()
+            self._flush_components()
+
             self.transport.flush(timeout=timeout, callback=callback)
 
+    async def flush_async(
+        self,
+        timeout: "Optional[float]" = None,
+        callback: "Optional[Callable[[int, float], None]]" = None,
+    ) -> None:
+        """
+        Asynchronously wait for the current events to be sent.
+
+        :param timeout: Wait for at most `timeout` seconds. If no `timeout` is provided, the `shutdown_timeout` option value is used.
+
+        :param callback: Is invoked with the number of pending events and the configured timeout.
+        """
+        if self.transport is not None:
+            if not self._has_async_transport():
+                logger.debug(
+                    "flush_async() used with non-async transport, aborting. Please use flush() instead."
+                )
+                return
+            if timeout is None:
+                timeout = self.options["shutdown_timeout"]
+            self._flush_components()
+            flush_task = self.transport.flush(timeout=timeout, callback=callback)  # type: ignore
+            if flush_task is not None:
+                await flush_task
+
     def __enter__(self) -> "_Client":
         return self
 
     def __exit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
         self.close()
 
+    async def __aenter__(self) -> "_Client":
+        return self
+
+    async def __aexit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
+        await self.close_async()
+
 
 from typing import TYPE_CHECKING
 

sentry_sdk/client.py

@@ -78,6 +78,7 @@ class CompressionAlgo(Enum):
             "transport_compression_algo": Optional[CompressionAlgo],
             "transport_num_pools": Optional[int],
             "transport_http2": Optional[bool],
+            "transport_async": Optional[bool],
             "enable_logs": Optional[bool],
             "before_send_log": Optional[Callable[[Log, Hint], Optional[Log]]],
             "enable_metrics": Optional[bool],