diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md
index ca894f1e..cf647260 100644
--- a/sdk/CHANGELOG.md
+++ b/sdk/CHANGELOG.md
@@ -20,6 +20,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   the retry stream. An explicit `max_records` is kept. This means one
   `XREADGROUP` per entry: setups that relied on the unbounded read for
   throughput with small, fast messages can set `max_records` explicitly.
+- `RedisTransport` now dials its background clients (the PEL reclaimer, the group
+  monitor, and the delete-on-ack client) with the broker's own connection settings,
+  instead of a hand-maintained whitelist of forwarded kwargs. The whitelist silently
+  dropped everything outside it, so against a server that needs authentication the
+  background clients connected unauthenticated and were rejected
+  (`AuthenticationError: HELLO must be called with the client already authenticated`)
+  even while the broker connected fine — and `db=` / `ssl=` / `client_name` diverged
+  too (e.g. the reclaimer scanned db 0 while the broker ran on db 3). Now `username`,
+  `password`, `db`, `ssl`, `credential_provider` (ACL users and rotating tokens such
+  as Azure Managed Redis / Microsoft Entra ID) and the resilience kwargs all match the
+  broker, including when a pre-built `broker=` is supplied.
+  Requires **faststream >= 0.7** — on 0.6.0 `RedisBroker(url, credential_provider=…)`
+  (and `password=`) raises `TypeError`.
+- The group monitor no longer dies on a transient connection or auth failure
+  (`AuthenticationError` is a `ConnectionError`, not a `ResponseError`, so it escaped
+  the loop's handler): a failed cycle is logged and retried next interval, by when
+  redis-py has reconnected and the credential provider has re-authenticated. A dying
+  monitor previously left `disconnect()` re-raising and skipping reclaimer/broker/
+  delete-client shutdown.
+- The PEL reclaimer no longer leaks a Redis client on each `connect()`: a second
+  `connect()` on a shared transport reuses the existing client instead of replacing
+  it without closing the old one.
 
 ## [0.6.0] - 2026-09-25
 
diff --git a/sdk/eggai/transport/pending_reclaimer.py b/sdk/eggai/transport/pending_reclaimer.py
index db696752..f52f8e23 100644
--- a/sdk/eggai/transport/pending_reclaimer.py
+++ b/sdk/eggai/transport/pending_reclaimer.py
@@ -289,23 +289,21 @@ class PendingReclaimerManager:
 
     def __init__(
         self,
-        redis_url: str,
         connection_kwargs: dict[str, Any] | None = None,
         delete_on_ack_disabled: set[str] | None = None,
     ):
-        self._redis_url = redis_url
         # Stream keys where delete_on_ack was turned off at runtime (another
         # consumer group appeared). Shared with, and mutated by, the transport's
         # group monitor; _ack() falls back to a plain XACK for these.
         self._delete_on_ack_disabled: set[str] = (
             delete_on_ack_disabled if delete_on_ack_disabled is not None else set()
         )
-        # Connection-resilience settings (socket_timeout, socket_keepalive,
-        # health_check_interval, retry_on_timeout, …) forwarded from the
-        # transport so this independent client recovers from a silently dropped
-        # connection the same way the broker does. Without them a blocking read
-        # against a half-dead socket (e.g. cloud Redis failover) hangs forever
-        # with no socket timeout to break it.
+        # The broker's own connection dict (host/port/db/username/password/ssl/
+        # credential_provider plus resilience kwargs), forwarded from the transport so
+        # this independent client reaches the same server and database and authenticates
+        # the same way the broker does — and recovers from a silently dropped connection
+        # (without a socket timeout a blocking read against a half-dead socket, e.g. a
+        # cloud Redis failover, would hang forever).
         self._connection_kwargs: dict[str, Any] = connection_kwargs or {}
         self._redis_client: aioredis.Redis | None = None
         self._configs: dict[tuple[str, str, str], ReclaimerConfig] = {}
@@ -336,10 +334,15 @@ async def start(self) -> None:
         # decode_responses=False: field values are kept as raw bytes so that
         # FastStream's binary-encoded __data__ field is passed through unchanged.
         # decode_responses is pinned here and must not be overridden by callers.
-        self._redis_client = aioredis.from_url(
-            self._redis_url,
-            **{**self._connection_kwargs, "decode_responses": False},
-        )
+        # Guarded so a second connect() on a shared transport reuses the client rather
+        # than leaking the previous one (the client is closed only in stop()). A
+        # ConnectionPool is used (not Redis(**kwargs)) because the broker's dict can
+        # carry connection_class/path for rediss:// and unix://.
+        if self._redis_client is None:
+            pool = aioredis.ConnectionPool(
+                **{**self._connection_kwargs, "decode_responses": False}
+            )
+            self._redis_client = aioredis.Redis.from_pool(pool)
         self._running = True
         for key, config in self._configs.items():
             if key in self._tasks and not self._tasks[key].done():
diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py
index a2c6550e..06ff69d1 100644
--- a/sdk/eggai/transport/redis.py
+++ b/sdk/eggai/transport/redis.py
@@ -31,30 +31,6 @@
 # the stream and own their own PEL entries — the competing-consumers pattern.
 _CONSUMER_INSTANCE = f"{socket.gethostname()}-{os.getpid()}"
 
-# Connection settings that, when passed to the broker, must also be forwarded to
-# the transport's other long-lived background clients — the PEL reclaimer and the
-# consumer-group monitor — so every connection recovers from a silently dropped
-# socket the same way. Kept to redis-py connection/resilience kwargs that are valid
-# for ``aioredis.from_url`` — broker/FastStream-only kwargs (decoder, middlewares,
-# asyncapi_*, …) are intentionally excluded. ``decode_responses`` is omitted on
-# purpose: each background client pins it itself (False for the reclaimer's binary
-# passthrough, True for the monitor's string commands).
-_BACKGROUND_CLIENT_CONNECTION_KEYS = (
-    "socket_timeout",
-    "socket_connect_timeout",
-    "socket_keepalive",
-    "socket_keepalive_options",
-    "health_check_interval",
-    "retry_on_timeout",
-    "retry_on_error",
-    "max_connections",
-    "ssl_keyfile",
-    "ssl_certfile",
-    "ssl_cert_reqs",
-    "ssl_ca_certs",
-    "ssl_check_hostname",
-)
-
 
 @dataclass(frozen=True)
 class _StreamGroupInfo:
@@ -173,13 +149,14 @@ def __init__(
             self.broker = broker
         else:
             self.broker = RedisBroker(url, log_level=logging.INFO, **kwargs)
-        self._redis_url = url
-        # Forward connection-resilience kwargs to the transport's background clients
-        # (PEL reclaimer, group monitor) so they don't hang on a silently dropped
-        # connection while the broker recovers.
-        self._connection_kwargs: dict[str, Any] = {
-            k: kwargs[k] for k in _BACKGROUND_CLIENT_CONNECTION_KEYS if k in kwargs
-        }
+        # The exact connection settings the broker dials with (host/port/db/username/
+        # password/ssl/credential_provider/resilience kwargs), read from the broker so
+        # the transport's own long-lived clients — the PEL reclaimer, the group monitor,
+        # and the delete-on-ack client — reach the same server and database and
+        # authenticate the same way. Taking the broker's dict (instead of re-deriving a
+        # whitelist from kwargs) also covers the ``broker=`` case and stops auth-only
+        # params like ``password`` and ``credential_provider`` from being silently dropped.
+        self._connection_kwargs: dict[str, Any] = dict(self.broker._connection_kwargs)
         self._max_len = max_len
         self._retry_max_len = retry_max_len
         self._dlq_max_len = dlq_max_len
@@ -203,6 +180,19 @@ def __init__(
         # group appeared; shared with the reclaimer. Plain XACK from then on.
         self._delete_on_ack_disabled: set[str] = set()
 
+    def _open_background_client(self, *, decode_responses: bool) -> aioredis.Redis:
+        """Open a long-lived background client (reclaimer / monitor / delete-on-ack)
+        that dials the same server as the broker. ``self._connection_kwargs`` is the
+        broker's own connection dict, so host/port/db/username/password/ssl/
+        credential_provider all match. A ConnectionPool (rather than ``Redis(**kwargs)``)
+        is used because the broker's dict can carry ``connection_class``/``path`` for
+        ``rediss://`` and ``unix://``, which the ``Redis`` constructor rejects.
+        ``decode_responses`` is pinned per client and must not be overridden by it."""
+        pool = aioredis.ConnectionPool(
+            **{**self._connection_kwargs, "decode_responses": decode_responses}
+        )
+        return aioredis.Redis.from_pool(pool)
+
     async def connect(self):
         """
         Establishes a connection to the Redis server by starting the RedisBroker instance.
@@ -229,9 +219,7 @@ async def connect(self):
         # an earlier connect().
         opened_delete_client = False
         if self._delete_on_ack_groups and self._delete_client is None:
-            self._delete_client = aioredis.from_url(
-                self._redis_url, **self._connection_kwargs
-            )
+            self._delete_client = self._open_background_client(decode_responses=False)
             opened_delete_client = True
         try:
             await self._create_groups()
@@ -867,10 +855,7 @@ async def _create_groups(self) -> None:
         use ">" from a group that starts where the caller asked."""
         if not self._stream_subscriptions:
             return
-        client = aioredis.from_url(
-            self._redis_url,
-            **{**self._connection_kwargs, "decode_responses": True},
-        )
+        client = self._open_background_client(decode_responses=True)
         try:
             for info in list(self._stream_subscriptions):
                 try:
@@ -925,43 +910,53 @@ async def _monitor_stream_groups(self) -> None:
         group_create_id with MKSTREAM when the stream itself is gone.
         """
         # decode_responses=True (string commands here) is pinned and must not be
-        # overridden; the forwarded resilience kwargs (socket_timeout, keepalive, …)
-        # keep this client from hanging on a half-dead socket during the very
-        # failover this monitor exists to recover from.
-        client = aioredis.from_url(
-            self._redis_url,
-            **{**self._connection_kwargs, "decode_responses": True},
-        )
+        # overridden; the broker's connection kwargs (socket_timeout, keepalive,
+        # credentials, …) keep this client authenticated and from hanging on a
+        # half-dead socket during the very failover this monitor exists to recover from.
+        client = self._open_background_client(decode_responses=True)
         try:
             while self._running:
                 await asyncio.sleep(self._group_monitor_interval_s)
-                # Iterate a snapshot: a concurrent subscribe() on a shared
-                # transport may mutate the set between awaits below.
-                for info in list(self._stream_subscriptions):
-                    try:
-                        stream_exists = await client.exists(info.stream_key)
-                        create_id = "0" if stream_exists else info.group_create_id
-                        await client.xgroup_create(
-                            name=info.stream_key,
-                            groupname=info.group,
-                            id=create_id,
-                            mkstream=True,
-                        )
-                        logger.info(
-                            "Recreated missing consumer group %s on stream %s (id=%s)",
-                            info.group,
-                            info.stream_key,
-                            create_id,
-                        )
-                    except ResponseError as e:
-                        if "BUSYGROUP" not in str(e):
-                            logger.warning(
-                                "Failed to ensure consumer group %s on %s: %s",
+                try:
+                    # Iterate a snapshot: a concurrent subscribe() on a shared
+                    # transport may mutate the set between awaits below.
+                    for info in list(self._stream_subscriptions):
+                        try:
+                            stream_exists = await client.exists(info.stream_key)
+                            create_id = "0" if stream_exists else info.group_create_id
+                            await client.xgroup_create(
+                                name=info.stream_key,
+                                groupname=info.group,
+                                id=create_id,
+                                mkstream=True,
+                            )
+                            logger.info(
+                                "Recreated missing consumer group %s on stream %s (id=%s)",
                                 info.group,
                                 info.stream_key,
-                                e,
+                                create_id,
                             )
-                await self._check_delete_on_ack_groups(client)
+                        except ResponseError as e:
+                            if "BUSYGROUP" not in str(e):
+                                logger.warning(
+                                    "Failed to ensure consumer group %s on %s: %s",
+                                    info.group,
+                                    info.stream_key,
+                                    e,
+                                )
+                    await self._check_delete_on_ack_groups(client)
+                except Exception as e:
+                    # A dropped/half-dead socket or a rotated auth token must not kill
+                    # the monitor task (AuthenticationError is a ConnectionError, not a
+                    # ResponseError, so it escapes the inner handler above). Log and
+                    # retry next interval, by when redis-py has reconnected and the
+                    # credential provider has re-authenticated. CancelledError is a
+                    # BaseException and still propagates to the outer handler.
+                    logger.warning(
+                        "Consumer-group monitor cycle failed; retrying in %.1fs: %s",
+                        self._group_monitor_interval_s,
+                        e,
+                    )
         except asyncio.CancelledError:
             pass
         finally:
@@ -1097,7 +1092,6 @@ def _setup_reclaimer(
     ) -> tuple[str, str, str]:
         if self._reclaimer_manager is None:
             self._reclaimer_manager = PendingReclaimerManager(
-                self._redis_url,
                 connection_kwargs=self._connection_kwargs,
                 delete_on_ack_disabled=self._delete_on_ack_disabled,
             )
diff --git a/sdk/poetry.lock b/sdk/poetry.lock
index a0ecb282..6dc14959 100644
--- a/sdk/poetry.lock
+++ b/sdk/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 2.4.3 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand.
 
 [[package]]
 name = "a2a-sdk"
@@ -1662,7 +1662,6 @@ files = [
     {file = "jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67"},
     {file = "jinja2-3.1.6.tar.gz", hash = "sha256:0137fb05990d35f1275a587e9aee6d56da821fc83491a0fb838183be43f66d6d"},
 ]
-markers = {main = "extra == \"cli\""}
 
 [package.dependencies]
 MarkupSafe = ">=2.0"
@@ -1717,7 +1716,7 @@ files = [
 
 [package.dependencies]
 attrs = ">=22.2.0"
-jsonschema-specifications = ">=2023.3.6"
+jsonschema-specifications = ">=2023.03.6"
 referencing = ">=0.28.4"
 rpds-py = ">=0.25.0"
 
@@ -2065,7 +2064,6 @@ files = [
     {file = "markupsafe-3.0.3-cp39-cp39-win_arm64.whl", hash = "sha256:38664109c14ffc9e7437e86b4dceb442b0096dfe3541d7864d9cbe1da4cf36c8"},
     {file = "markupsafe-3.0.3.tar.gz", hash = "sha256:722695808f4b6457b320fdc131280796bdceb04ab50fe1795cd540799ebe1698"},
 ]
-markers = {main = "extra == \"cli\""}
 
 [[package]]
 name = "mcp"
@@ -4190,4 +4188,4 @@ otel = ["opentelemetry-api", "opentelemetry-exporter-otlp-proto-grpc", "opentele
 [metadata]
 lock-version = "2.1"
 python-versions = "^3.10"
-content-hash = "59e37b5cd75c9dc95bc18b0a99e481f5d52edbc58d6cbfac6bec4c2d9f696c9a"
+content-hash = "906bdc375aa13665155773f26c15b340e5bd45258e02bb17ce4c57d1ba883fb1"
diff --git a/sdk/pyproject.toml b/sdk/pyproject.toml
index b4595900..c219e078 100644
--- a/sdk/pyproject.toml
+++ b/sdk/pyproject.toml
@@ -34,7 +34,7 @@ classifiers = [
 [tool.poetry.dependencies]
 python = "^3.10"
 pydantic = "^2.11.4"
-faststream = {extras = ["kafka", "redis"], version = ">=0.6,<0.8"}
+faststream = {extras = ["kafka", "redis"], version = ">=0.7,<0.8"}
 click = {version = "^8.0.0", optional = true}
 jinja2 = {version = "^3.1.6", optional = true}
 a2a-sdk = {version = "^0.3.0", optional = true}
diff --git a/sdk/tests/test_redis.py b/sdk/tests/test_redis.py
index d7e3d61d..b3e93e84 100644
--- a/sdk/tests/test_redis.py
+++ b/sdk/tests/test_redis.py
@@ -1018,7 +1018,7 @@ async def test_reclaimer_manager_start_stop_cycles():
         ReclaimerConfig,
     )
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     assert manager._redis_client is None
 
     manager.add(
@@ -1050,7 +1050,7 @@ async def test_reclaimer_manager_stop_without_start():
     """Calling stop() before start() must not raise."""
     from eggai.transport.pending_reclaimer import PendingReclaimerManager
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     await manager.stop()  # should not raise
 
 
@@ -1396,7 +1396,7 @@ async def test_reclaimer_nogroup_recovery():
     await redis_client.xgroup_create(stream_name, group_name, id="$", mkstream=True)
     await redis_client.delete(stream_name)
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     manager.add(
         ReclaimerConfig(
             stream=stream_name,
@@ -1680,17 +1680,19 @@ async def test_connection_kwargs_propagate_to_reclaimer():
         socket_keepalive=True,
         health_check_interval=30,
         retry_on_timeout=True,
-        # A broker/FastStream-only kwarg that must NOT leak into the reclaimer's
-        # aioredis.from_url call.
+        # A broker/FastStream-only kwarg that is not a connection kwarg and must NOT
+        # reach the background clients.
         graceful_timeout=20.0,
     )
 
-    assert transport._connection_kwargs == {
-        "socket_timeout": 10.0,
-        "socket_keepalive": True,
-        "health_check_interval": 30,
-        "retry_on_timeout": True,
-    }
+    # The broker's own connection dict is forwarded, so the resilience kwargs are
+    # present (alongside host/port/db); the FastStream-only one is not.
+    ck = transport._connection_kwargs
+    assert ck["socket_timeout"] == 10.0
+    assert ck["socket_keepalive"] is True
+    assert ck["health_check_interval"] == 30
+    assert ck["retry_on_timeout"] is True
+    assert "graceful_timeout" not in ck
 
     async def handler(message):
         return message
@@ -1700,18 +1702,15 @@ async def handler(message):
     )
 
     assert transport._reclaimer_manager is not None
-    assert transport._reclaimer_manager._connection_kwargs == {
-        "socket_timeout": 10.0,
-        "socket_keepalive": True,
-        "health_check_interval": 30,
-        "retry_on_timeout": True,
-    }
+    # The reclaimer gets the same connection dict the transport forwards.
+    assert transport._reclaimer_manager._connection_kwargs is ck
 
 
 @pytest.mark.asyncio
 async def test_reclaimer_start_applies_connection_kwargs(monkeypatch):
-    """start() forwards connection_kwargs to from_url while pinning
-    decode_responses=False (callers cannot override the binary-passthrough flag)."""
+    """start() builds its client from the forwarded connection dict — credentials, db
+    and resilience kwargs all reach it — while pinning decode_responses=False (callers
+    cannot override the binary-passthrough flag)."""
     from eggai.transport import pending_reclaimer
     from eggai.transport.pending_reclaimer import PendingReclaimerManager
 
@@ -1721,27 +1720,36 @@ class _FakeClient:
         async def aclose(self):
             pass
 
-    def fake_from_url(url, **kwargs):
-        captured["url"] = url
+    def fake_pool(**kwargs):
         captured["kwargs"] = kwargs
-        return _FakeClient()
+        return object()
 
-    monkeypatch.setattr(pending_reclaimer.aioredis, "from_url", fake_from_url)
+    monkeypatch.setattr(pending_reclaimer.aioredis, "ConnectionPool", fake_pool)
+    monkeypatch.setattr(
+        pending_reclaimer.aioredis.Redis,
+        "from_pool",
+        staticmethod(lambda pool: _FakeClient()),
+    )
 
+    provider = object()
     manager = PendingReclaimerManager(
-        "redis://localhost:6379",
         connection_kwargs={
+            "host": "db.example",
+            "port": 6380,
+            "db": 3,
+            "password": "s3cret",
+            "credential_provider": provider,
             "socket_timeout": 10.0,
-            "socket_keepalive": True,
             # Even if a caller sneaks decode_responses in, it must stay False.
             "decode_responses": True,
         },
     )
     await manager.start()
 
-    assert captured["url"] == "redis://localhost:6379"
+    assert captured["kwargs"]["db"] == 3
+    assert captured["kwargs"]["password"] == "s3cret"
+    assert captured["kwargs"]["credential_provider"] is provider
     assert captured["kwargs"]["socket_timeout"] == 10.0
-    assert captured["kwargs"]["socket_keepalive"] is True
     assert captured["kwargs"]["decode_responses"] is False
 
     await manager.stop()
@@ -1749,15 +1757,17 @@ def fake_from_url(url, **kwargs):
 
 @pytest.mark.asyncio
 async def test_monitor_applies_connection_kwargs(monkeypatch):
-    """The group monitor's background client gets the transport's resilience
-    kwargs while pinning decode_responses=True (its string commands need decoding,
-    and that flag must not be overridable)."""
+    """The group monitor's background client dials the same server as the broker — db,
+    password and credential_provider all reach it — while pinning decode_responses=True
+    (its string commands need decoding, and that flag must not be overridable)."""
     from eggai.transport import redis as redis_module
 
+    provider = object()
     transport = RedisTransport(
+        url="redis://localhost:6379/3",
+        password="s3cret",
+        credential_provider=provider,
         socket_timeout=10.0,
-        socket_keepalive=True,
-        health_check_interval=30,
     )
 
     captured: dict = {}
@@ -1766,24 +1776,49 @@ class _FakeClient:
         async def aclose(self):
             pass
 
-    def fake_from_url(url, **kwargs):
-        captured["url"] = url
+    def fake_pool(**kwargs):
         captured["kwargs"] = kwargs
-        return _FakeClient()
+        return object()
 
-    monkeypatch.setattr(redis_module.aioredis, "from_url", fake_from_url)
+    monkeypatch.setattr(redis_module.aioredis, "ConnectionPool", fake_pool)
+    monkeypatch.setattr(
+        redis_module.aioredis.Redis,
+        "from_pool",
+        staticmethod(lambda pool: _FakeClient()),
+    )
 
     # _running stays False, so the monitor creates its client, skips the loop body,
-    # and closes the client via the finally block — enough to capture from_url.
+    # and closes the client via the finally block — enough to capture the pool kwargs.
     await transport._monitor_stream_groups()
 
-    assert captured["url"] == transport._redis_url
+    assert captured["kwargs"]["db"] == 3
+    assert captured["kwargs"]["password"] == "s3cret"
+    assert captured["kwargs"]["credential_provider"] is provider
     assert captured["kwargs"]["socket_timeout"] == 10.0
-    assert captured["kwargs"]["socket_keepalive"] is True
-    assert captured["kwargs"]["health_check_interval"] == 30
     assert captured["kwargs"]["decode_responses"] is True
 
 
+def test_prebuilt_broker_forwards_its_connection_settings():
+    """With an explicit broker=, the background clients take their settings from the
+    broker's own connection dict, so they dial its server/db/credentials rather than a
+    default localhost. This is the case the old kwargs whitelist could not cover."""
+    from faststream.redis import RedisBroker
+
+    provider = object()
+    broker = RedisBroker(
+        "rediss://redis.example:10000/4",
+        password="s3cret",
+        credential_provider=provider,
+    )
+    transport = RedisTransport(broker=broker)
+
+    ck = transport._connection_kwargs
+    assert ck["host"] == "redis.example"
+    assert ck["db"] == 4
+    assert ck["password"] == "s3cret"
+    assert ck["credential_provider"] is provider
+
+
 @pytest.mark.asyncio
 async def test_publish_maxlen_bounds_stream_length():
     """Integration: publishing well past max_len keeps the stream approximately
@@ -2007,7 +2042,7 @@ async def test_effective_idle_ms_escalates_and_caps():
     """_effective_idle_ms grows geometrically with retry count and respects the cap."""
     from eggai.transport.pending_reclaimer import PendingReclaimerManager
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     config = _make_backoff_config(
         "s", "s.retry", backoff_multiplier=2.0, backoff_max_ms=5000
     )
@@ -2031,7 +2066,7 @@ async def test_effective_idle_ms_huge_retry_count_does_not_overflow():
     """
     from eggai.transport.pending_reclaimer import PendingReclaimerManager
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
 
     # 2.0 ** 5000 overflows the max float (~1.8e308) several times over.
     capped = _make_backoff_config(
@@ -2054,7 +2089,7 @@ async def test_effective_idle_ms_constant_when_multiplier_one():
     """multiplier=1.0 reproduces the original constant cadence at every count."""
     from eggai.transport.pending_reclaimer import PendingReclaimerManager
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     config = _make_backoff_config("s", "s.retry")  # multiplier defaults to 1.0
     for count in range(5):
         assert manager._effective_idle_ms(config, count) == 1000
@@ -2072,7 +2107,7 @@ async def test_effective_idle_ms_jitter_spreads_upward_above_base():
     """
     from eggai.transport.pending_reclaimer import PendingReclaimerManager
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     config = _make_backoff_config("s", "s.retry", backoff_jitter=0.5)  # multiplier 1.0
     samples = [manager._effective_idle_ms(config, 0) for _ in range(200)]
     # Upward only: never below the base floor (1000), never above base*(1+jitter).
@@ -2090,7 +2125,7 @@ async def test_effective_idle_ms_jitter_stays_under_cap():
     """Jitter is applied before the cap, so backoff_max_ms remains a hard ceiling."""
     from eggai.transport.pending_reclaimer import PendingReclaimerManager
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     # base*mult^count = 1000*2**3 = 8000; jitter would push to 12000, but the cap
     # pins it at 5000 regardless.
     config = _make_backoff_config(
@@ -2194,7 +2229,7 @@ async def test_backoff_skips_high_retry_count_message_not_yet_due():
     envelope = await _make_envelope({"type": "t", "_retry_count": "3", "id": test_id})
     await _deliver_to_pel(client, stream, group, f"{group}-live", envelope)
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     manager._redis_client = redis.Redis(
         host="localhost", port=6379, decode_responses=False
     )
@@ -2235,7 +2270,7 @@ async def test_backoff_reclaims_message_once_threshold_elapsed():
     envelope = await _make_envelope({"type": "t", "_retry_count": "0", "id": test_id})
     await _deliver_to_pel(client, stream, group, f"{group}-live", envelope)
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     manager._redis_client = redis.Redis(
         host="localhost", port=6379, decode_responses=False
     )
@@ -2277,7 +2312,7 @@ async def test_reclaimer_stop_completes_when_cancellation_is_swallowed(monkeypat
         ReclaimerConfig,
     )
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     manager.add(
         ReclaimerConfig(
             stream="eggai.swallow",
diff --git a/sdk/tests/test_redis_delete_on_ack.py b/sdk/tests/test_redis_delete_on_ack.py
index 9f411806..7729a544 100644
--- a/sdk/tests/test_redis_delete_on_ack.py
+++ b/sdk/tests/test_redis_delete_on_ack.py
@@ -433,9 +433,7 @@ async def test_reclaimer_ack_respects_runtime_disabled_streams():
     )
 
     disabled: set[str] = set()
-    manager = PendingReclaimerManager(
-        "redis://localhost:6379", delete_on_ack_disabled=disabled
-    )
+    manager = PendingReclaimerManager(delete_on_ack_disabled=disabled)
     client = MagicMock()
     client.xack = AsyncMock()
     manager._redis_client = client
diff --git a/sdk/tests/test_redis_shared_dlq.py b/sdk/tests/test_redis_shared_dlq.py
index d8724f0f..17d50242 100644
--- a/sdk/tests/test_redis_shared_dlq.py
+++ b/sdk/tests/test_redis_shared_dlq.py
@@ -563,7 +563,7 @@ async def test_reclaimer_wraps_poison_entry_into_dlq():
     await redis_client.xreadgroup(group, "worker", {stream: ">"}, count=2)
     await asyncio.sleep(0.05)
 
-    manager = PendingReclaimerManager("redis://localhost:6379")
+    manager = PendingReclaimerManager()
     config = ReclaimerConfig(
         stream=stream,
         group=group,