Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/_experimental/mcp_server/outbound_credentials/runtime_refresh_coordinator.py: 73%

1"""The runtime ``RefreshCoordinator``: cross-replica single-flight when Redis is wired. 

2 

3Builds ``RedisRefreshCoordinator`` over the proxy's shared Redis so one refresh runs per key 

4across the fleet, or returns ``None`` when Redis is absent so the caller keeps the foundation's 

5in-process default (correct for a single replica). The proxy globals it reads are not ready at 

6import time, so this is called per composition rather than held as module state. 

7 

8Shared by every credential arm that renews a stored grant: a rotating refresh token must be 

9redeemed once across all workers, so each arm electing its own winner with its own lock shape 

10would be a bug waiting to differ. 

11""" 

12 

13from __future__ import annotations 

14 

15from typing import Final 

16 

17from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import ( 

18 RefreshCoordinator, 

19) 

20from litellm.proxy._experimental.mcp_server.outbound_credentials.redis_distributed_lock import ( 

21 RedisDistributedLock, 

22) 

23from litellm.proxy._experimental.mcp_server.outbound_credentials.redis_refresh_coordinator import ( 

24 RedisRefreshCoordinator, 

25) 

26 

27 

28def runtime_refresh_coordinator() -> RefreshCoordinator | None: 

29 from litellm.proxy.proxy_server import user_api_key_cache # noqa: PLC0415 # runtime global 

30 

31 redis_cache: Final = user_api_key_cache.redis_cache 

32 if redis_cache is None: 32 ↛ 36line 32 didn't jump to line 36 because the condition on line 32 was always true

33 return None 

34 # The Redis client from init_async_client() is only partially typed; the lock validates every 

35 # reply it depends on, so the untyped boundary is contained here. 

36 redis_client: Final = redis_cache.init_async_client() # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType] # litellm redis wrapper is untyped 

37 lock: Final = RedisDistributedLock( 

38 redis_client, # pyright: ignore[reportArgumentType,reportUnknownArgumentType] # litellm redis wrapper is untyped 

39 namespace_key=redis_cache.check_and_fix_namespace, 

40 ) 

41 return RedisRefreshCoordinator(lock)