fix(a2a): keep the upstream status on card discovery failures and inject the card client in tests

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-09-16 23:55:21 +00:00
parent bc8e28cfcf
commit 75eec8712c
4 changed files with 49 additions and 32 deletions

View file

@ -24,6 +24,7 @@ AGENT_CARD_PATH_PARAM: Final = "agent_card_path"
try:
from a2a.client import A2ACardResolver as _A2ACardResolver
from a2a.client.errors import AgentCardResolutionError
from a2a.utils.constants import (
AGENT_CARD_WELL_KNOWN_PATH,
PREV_AGENT_CARD_WELL_KNOWN_PATH,
@ -32,6 +33,15 @@ except ImportError:
pass
def _discovery_status_code(failures: tuple[tuple[str, Exception], ...]) -> int:
statuses: Final = tuple(
error.status_code
for _, error in failures
if isinstance(error, AgentCardResolutionError) and error.status_code is not None and error.status_code != 404
)
return statuses[0] if statuses else 404
def is_localhost_or_internal_url(url: str | None) -> bool:
"""
Check if a URL is a localhost or internal URL.
@ -151,7 +161,7 @@ class LiteLLMA2ACardResolver(_A2ACardResolver):
Extends the base A2ACardResolver to try, in order:
- /.well-known/agent-card.json (standard)
- /.well-known/agent.json (previous/alternative)
- /agentCard/v1.0 (Microsoft Foundry agents, which serve no well-known card)
- /agentCard/v1.0
"""
async def get_agent_card(
@ -159,23 +169,7 @@ class LiteLLMA2ACardResolver(_A2ACardResolver):
relative_card_path: str | None = None,
http_kwargs: Mapping[str, object] | None = None,
) -> "AgentCard":
"""
Fetch the agent card, trying multiple well-known paths.
First tries the standard path, then the previous path, then Foundry's documented path.
Args:
relative_card_path: Optional path to the agent card endpoint.
If None, tries every known path in order.
http_kwargs: Optional dictionary of keyword arguments to pass to httpx.get
Returns:
AgentCard from the A2A agent
Raises:
A2AAgentCardDiscoveryError naming every probed path and its error when no path answers
"""
# If a specific path is provided, use the parent implementation
"""Fetch the agent card, probing every known path when none is given."""
if relative_card_path is not None:
return await super().get_agent_card(
relative_card_path=relative_card_path,
@ -191,11 +185,15 @@ class LiteLLMA2ACardResolver(_A2ACardResolver):
async def _get_agent_card_from_first_reachable_path(
self,
paths: tuple[str, ...],
http_kwargs: dict[str, Any] | None,
http_kwargs: Mapping[str, object] | None,
failures: tuple[tuple[str, Exception], ...],
) -> "AgentCard":
if not paths:
raise A2AAgentCardDiscoveryError(base_url=self.base_url, failures=failures)
raise A2AAgentCardDiscoveryError(
base_url=self.base_url,
failures=failures,
status_code=_discovery_status_code(failures),
)
path: Final = paths[0]
try:
verbose_logger.debug("Attempting to fetch agent card from %s%s", self.base_url, path)

View file

@ -102,11 +102,12 @@ class A2AAgentCardError(A2AError):
model: str | None = None,
response: httpx.Response | None = None,
litellm_debug_info: str | None = None,
status_code: int = 404,
):
self.url = url
super().__init__(
message=message,
status_code=404,
status_code=status_code,
llm_provider="a2a_agent",
model=model,
response=response,
@ -115,12 +116,14 @@ class A2AAgentCardError(A2AError):
class A2AAgentCardDiscoveryError(A2AAgentCardError):
"""Raised when no known agent card path answered; names every path probed and why each failed."""
def __init__(self, base_url: str, failures: tuple[tuple[str, Exception], ...]) -> None:
def __init__(self, base_url: str, failures: tuple[tuple[str, Exception], ...], status_code: int) -> None:
self.failures = failures
attempts: Final = ", ".join(f"{path} ({error})" for path, error in failures)
super().__init__(message=f"Failed to fetch agent card from {base_url}. Tried {attempts}", url=base_url)
super().__init__(
message=f"Failed to fetch agent card from {base_url}. Tried {attempts}",
url=base_url,
status_code=status_code,
)
class A2ALocalhostURLError(A2AConnectionError):

View file

@ -57,7 +57,7 @@ def mock_a2a_client(monkeypatch):
import litellm.a2a_protocol.main as a2a_main
async def _fake_create_a2a_client(
base_url, timeout=60.0, extra_headers=None, streaming=False
base_url, timeout=60.0, extra_headers=None, streaming=False, relative_card_path=None
):
return MockA2AClient()

View file

@ -18,6 +18,7 @@ from litellm.a2a_protocol.card_resolver import (
normalize_agent_card_interfaces,
set_agent_card_url,
)
from litellm.a2a_protocol.exceptions import A2AAgentCardDiscoveryError
@pytest.mark.asyncio
@ -174,8 +175,6 @@ class _FakeHttpxClient:
@pytest.mark.asyncio
async def test_card_resolver_falls_through_to_the_foundry_card_path():
"""Microsoft Foundry agents serve their card only at /agentCard/v1.0 and 404 both well-known
paths, so discovery must reach that path after the two well-known probes fail."""
httpx_client = _FakeHttpxClient(
base_url=_FOUNDRY_BASE_URL,
responses={
@ -209,10 +208,6 @@ async def test_card_resolver_explicit_path_skips_the_probes():
@pytest.mark.asyncio
async def test_card_resolver_names_every_probed_path_when_discovery_fails():
"""A Foundry agent 401s its well-known paths and 404s the rest; surfacing only the last probe's
error would hide the auth failure that actually explains the outage."""
from litellm.a2a_protocol.exceptions import A2AAgentCardDiscoveryError
httpx_client = _FakeHttpxClient(
base_url=_FOUNDRY_BASE_URL,
responses={
@ -226,8 +221,29 @@ async def test_card_resolver_names_every_probed_path_when_discovery_fails():
with pytest.raises(A2AAgentCardDiscoveryError) as raised:
await resolver.get_agent_card()
assert raised.value.status_code == 401
message = str(raised.value)
assert _FOUNDRY_BASE_URL in message
assert "/.well-known/agent-card.json (" in message and "HTTP 404" in message
assert "/.well-known/agent.json (" in message and "HTTP 401" in message
assert "/agentCard/v1.0 (" in message
@pytest.mark.asyncio
async def test_card_resolver_discovery_error_is_404_when_every_probe_is_404():
resolver = LiteLLMA2ACardResolver(
httpx_client=_FakeHttpxClient(
base_url=_FOUNDRY_BASE_URL,
responses={
"/.well-known/agent-card.json": (404, {"error": "not found"}),
"/.well-known/agent.json": (404, {"error": "not found"}),
"/agentCard/v1.0": (404, {"error": "not found"}),
},
),
base_url=_FOUNDRY_BASE_URL,
)
with pytest.raises(A2AAgentCardDiscoveryError) as raised:
await resolver.get_agent_card()
assert raised.value.status_code == 404