scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-30 01:52:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 70

fix(a2a): treat CP4D token expiration as absolute Unix time

Browse source 
CP4D /authorize returns expiration as epoch seconds, not TTL. Compute
remaining lifetime against wall clock so cached tokens refresh before expiry.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Sameer Kankute 2026-06-01 12:57:14 +05:30
parent 9a4ab99fb4
commit 0fea4f90ef
No known key found for this signature in database
2 changed files with 23 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
litellm/a2a_protocol/providers/watsonx_orchestrate/handler.py
View file

@ -43,6 +43,13 @@ class WatsonxOrchestrateHandler:
material = f"{auth_mode}:{cp4d_host}:{username or ''}:{api_key}"
return hashlib.sha256(material.encode()).hexdigest()
@staticmethod
def _cp4d_token_ttl_seconds(expiration: Any, now_wall: Optional[float] = None) -> int:
# CP4D returns expiration as absolute Unix epoch seconds, not a duration.
expires_at = int(expiration)
wall = now_wall if now_wall is not None else time.time()
return max(expires_at - int(wall), 0)
@staticmethod
async def _get_bearer_token(
cp4d_host: str,
@ -89,7 +96,11 @@ class WatsonxOrchestrateHandler:
response.raise_for_status()
payload = response.json()
token = str(payload["token"])
ttl_s = int(payload.get("expiration", 3600))
expiration = payload.get("expiration")
if expiration is None:
ttl_s = 3600
else:
ttl_s = WatsonxOrchestrateHandler._cp4d_token_ttl_seconds(expiration)
expires_at = now + max(ttl_s - _TOKEN_CACHE_TTL_BUFFER_S, 60)
_token_cache[cache_key] = (token, expires_at)

11
tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py
View file

@ -6,6 +6,9 @@ import pytest
sys.path.insert(0, os.path.abspath("../../../../.."))
from litellm.a2a_protocol.providers.config_manager import A2AProviderConfigManager
from litellm.a2a_protocol.providers.watsonx_orchestrate.handler import (
WatsonxOrchestrateHandler,
)
from litellm.a2a_protocol.providers.watsonx_orchestrate.transformation import (
WatsonxOrchestrateTransformation,
)
@ -104,6 +107,14 @@ class TestWatsonxOrchestrateTransformation:
)
def test_cp4d_token_ttl_from_absolute_expiration():
wall = 1_750_000_000.0
assert (
WatsonxOrchestrateHandler._cp4d_token_ttl_seconds(1_750_003_600, wall) == 3600
)
assert WatsonxOrchestrateHandler._cp4d_token_ttl_seconds(1_749_999_000, wall) == 0
def test_config_manager_returns_wxo_provider():
config = A2AProviderConfigManager.get_provider_config(
custom_llm_provider="watsonx_orchestrate"