Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/a2a/discovery.py: 71%

1""" 

2Fetch an A2A agent's well-known card from the upstream agent. 

3 

4Different agent runtimes publish the card at different URL shapes, so the 

5fetcher dispatches by ``discovery_mode``: 

6 

7- ``well_known_fallback`` (pure A2A): the card lives at one of the standard 

8 well-known paths on the agent's own base URL. We try the canonical path, 

9 then the previous-spec path, then a non-standard root fallback. 

10 

11- ``langgraph_platform``: LangGraph Platform mounts a single card endpoint at 

12 ``{base}/.well-known/agent-card.json`` and disambiguates assistants via the 

13 ``assistant_id`` query parameter. There is no per-assistant subpath, so the 

14 pure-A2A fallback strategy returns 404 for these deployments. 

15""" 

16 

17from collections.abc import Mapping 

18from enum import Enum 

19from typing import Any, Final 

20from urllib.parse import urlencode 

21 

22from litellm._logging import verbose_proxy_logger 

23from litellm.litellm_core_utils.url_utils import SSRFError, async_safe_get 

24from litellm.llms.custom_httpx.http_handler import get_async_httpx_client 

25from litellm.types.llms.custom_http import httpxSpecialProvider 

26 

27 

28class DiscoveryMode(str, Enum): 

29 """How to locate the upstream agent card. 

30 

31 String-valued so it serializes cleanly over JSON / Pydantic. 

32 """ 

33 

34 WELL_KNOWN_FALLBACK = "well_known_fallback" 

35 LANGGRAPH_PLATFORM = "langgraph_platform" 

36 

37 

38# Paths the pure-A2A fetcher tries in order. The first two are the current and 

39# previous A2A spec locations; ``/agent.json`` is a non-standard root fallback 

40# some agents still serve. 

41AGENT_CARD_WELL_KNOWN_PATHS: Final[tuple[str, ...]] = ( 

42 "/.well-known/agent-card.json", 

43 "/.well-known/agent.json", 

44 "/agent.json", 

45) 

46 

47DEFAULT_DISCOVERY_TIMEOUT_SECONDS: Final = 10.0 

48 

49 

50class AgentCardDiscoveryError(Exception): 

51 """Raised when none of the well-known paths returned a usable agent card.""" 

52 

53 

54def _normalize_base_url(base_url: str) -> str: 

55 return base_url.rstrip("/") 

56 

57 

58def _build_langgraph_platform_paths( 

59 params: Mapping[str, object] | None, 

60) -> tuple[str, ...]: 

61 """Build the paths to try for LangGraph Platform discovery. 

62 

63 LangGraph serves the card at ``/.well-known/agent-card.json`` with the 

64 ``assistant_id`` carried as a query parameter. We still try the other 

65 A2A path variants (with the same query string appended) so we degrade 

66 gracefully if a deployment uses an older spec name. 

67 """ 

68 assistant_id: Final = (params or {}).get("assistant_id") 

69 if not assistant_id: 69 ↛ 71line 69 didn't jump to line 71 because the condition on line 69 was always true

70 raise AgentCardDiscoveryError("langgraph_platform discovery requires params.assistant_id") 

71 query: Final = urlencode({"assistant_id": str(assistant_id)}) 

72 return tuple(f"{path}?{query}" for path in AGENT_CARD_WELL_KNOWN_PATHS) 

73 

74 

75def _paths_for_mode(mode: DiscoveryMode, params: Mapping[str, object] | None) -> tuple[str, ...]: 

76 if mode == DiscoveryMode.WELL_KNOWN_FALLBACK: 

77 return AGENT_CARD_WELL_KNOWN_PATHS 

78 if mode == DiscoveryMode.LANGGRAPH_PLATFORM: 78 ↛ 80line 78 didn't jump to line 80 because the condition on line 78 was always true

79 return _build_langgraph_platform_paths(params) 

80 raise AgentCardDiscoveryError(f"unsupported discovery_mode: {mode}") 

81 

82 

83async def fetch_well_known_card( 

84 base_url: str, 

85 *, 

86 discovery_mode: DiscoveryMode = DiscoveryMode.WELL_KNOWN_FALLBACK, 

87 params: Mapping[str, object] | None = None, 

88 timeout: float = DEFAULT_DISCOVERY_TIMEOUT_SECONDS, 

89 headers: dict[str, str] | None = None, 

90) -> dict[str, Any]: 

91 """ 

92 Fetch an agent card from ``base_url`` using the strategy chosen by 

93 ``discovery_mode``. Returns the parsed JSON from the first path that 

94 responds with a JSON body. 

95 

96 Raises: 

97 AgentCardDiscoveryError: if every path fails (network error, non-2xx, 

98 or non-JSON body), or if the chosen mode is missing required params. 

99 """ 

100 if not base_url: 

101 raise AgentCardDiscoveryError("base_url is required") 

102 

103 normalized: Final = _normalize_base_url(base_url) 

104 paths: Final = _paths_for_mode(discovery_mode, params) 

105 client: Final = get_async_httpx_client( 

106 llm_provider=httpxSpecialProvider.A2A, 

107 params={"timeout": timeout}, 

108 ) 

109 

110 last_error: str | None = None 

111 for path in paths: 

112 url = f"{normalized}{path}" 

113 try: 

114 # ``async_safe_get`` validates the URL against the SSRF blocklist 

115 # (private/loopback IPs, cloud metadata endpoints, etc.) on every 

116 # redirect hop. Even though the discovery endpoint is admin-only, 

117 # we don't want a compromised admin key to be able to probe 

118 # internal infrastructure through this fetcher. 

119 # Pass ``headers or {}`` because ``async_safe_get`` (in the 

120 # URL-validation path) uses ``kwargs.pop("headers", {})`` which 

121 # returns ``None`` when the key is present-but-None, then crashes 

122 # on ``{**None, "Host": ...}``. Default the kwarg to an empty 

123 # dict so production (``user_url_validation=True``) doesn't 500. 

124 response = await async_safe_get(client, url, headers=headers or {}) 

125 except SSRFError as exc: 

126 last_error = f"{url}: {exc}" 

127 verbose_proxy_logger.debug("A2A discovery blocked by SSRF guard for %s: %s", url, exc) 

128 continue 

129 except Exception as exc: 

130 last_error = f"{url}: {exc}" 

131 verbose_proxy_logger.debug("A2A discovery failed for %s: %s", url, exc) 

132 continue 

133 

134 if response.status_code >= 400: 134 ↛ 139line 134 didn't jump to line 139 because the condition on line 134 was always true

135 last_error = f"{url}: HTTP {response.status_code}" 

136 verbose_proxy_logger.debug("A2A discovery HTTP %s for %s", response.status_code, url) 

137 continue 

138 

139 try: 

140 card = response.json() 

141 except Exception as exc: 

142 last_error = f"{url}: invalid JSON ({exc})" 

143 continue 

144 

145 if not isinstance(card, dict): 

146 last_error = f"{url}: expected JSON object, got {type(card).__name__}" 

147 continue 

148 

149 verbose_proxy_logger.debug("A2A discovery succeeded at %s", url) 

150 return card 

151 

152 raise AgentCardDiscoveryError( 

153 f"Could not fetch agent card from {base_url} (mode={discovery_mode.value}). Last error: {last_error}" 

154 )