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

1""" 

2FastAPI routes for the A2A registration flow. 

3 

4Today this exposes a single endpoint, ``POST /v1/a2a/discover``, used by the 

5LiteLLM UI when an admin registers a new A2A agent: the UI hands us the 

6upstream agent's base URL, we fetch its well-known card, and we return the 

7raw card so the UI can render the agent's skills/capabilities and let the 

8admin pick which ones to expose through the proxy. The actual merge into a 

9LiteLLM-fronted card happens when the agent is saved via ``POST /v1/agents``. 

10""" 

11 

12from typing import Any, Final 

13 

14from fastapi import APIRouter, Depends, HTTPException 

15from fastapi.responses import JSONResponse 

16from pydantic import BaseModel, Field 

17 

18from litellm._logging import verbose_proxy_logger 

19from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth 

20from litellm.proxy.a2a.discovery import ( 

21 AGENT_CARD_WELL_KNOWN_PATHS, 

22 AgentCardDiscoveryError, 

23 DiscoveryMode, 

24 fetch_well_known_card, 

25) 

26from litellm.proxy.auth.user_api_key_auth import user_api_key_auth 

27 

28router: Final = APIRouter() 

29 

30 

31class DiscoverAgentRequest(BaseModel): 

32 url: str = Field( 

33 ..., 

34 description=( 

35 "Base URL of the upstream agent. Behavior depends on " 

36 "``discovery_mode``: ``well_known_fallback`` (default) tries " 

37 f"{', '.join(AGENT_CARD_WELL_KNOWN_PATHS)} under this URL in " 

38 "order; ``langgraph_platform`` hits " 

39 "``/.well-known/agent-card.json?assistant_id=<id>`` instead." 

40 ), 

41 ) 

42 discovery_mode: DiscoveryMode = Field( 

43 default=DiscoveryMode.WELL_KNOWN_FALLBACK, 

44 description=( 

45 "How to locate the upstream card. " 

46 "``well_known_fallback`` for pure A2A agents (try standard paths); " 

47 "``langgraph_platform`` for LangGraph Platform deployments where " 

48 "the card is shared across assistants and disambiguated by a " 

49 "query parameter." 

50 ), 

51 ) 

52 params: dict[str, Any] | None = Field( 

53 default=None, 

54 description=( 

55 "Mode-specific parameters. ``langgraph_platform`` requires " 

56 "``{'assistant_id': <id>}``. ``well_known_fallback`` ignores this." 

57 ), 

58 ) 

59 

60 

61class DiscoverAgentResponse(BaseModel): 

62 url: str 

63 agent_card: dict[str, Any] 

64 

65 

66@router.post( 

67 "/v1/a2a/discover", 

68 tags=["[beta] A2A Agents"], 

69 dependencies=[Depends(user_api_key_auth)], 

70 response_model=DiscoverAgentResponse, 

71) 

72async def discover_agent_card( 

73 request: DiscoverAgentRequest, 

74 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), 

75) -> JSONResponse: 

76 """ 

77 Fetch the upstream agent's well-known card so the UI can show the admin 

78 which skills/capabilities the agent exposes. 

79 

80 Only proxy admins can call this — the UI uses it during agent registration, 

81 and we don't want arbitrary keys probing internal URLs. 

82 

83 Example: 

84 ```bash 

85 curl -X POST "http://localhost:4000/v1/a2a/discover" \\ 

86 -H "Authorization: Bearer <admin_key>" \\ 

87 -H "Content-Type: application/json" \\ 

88 -d '{"url": "https://upstream-agent.example.com"}' 

89 ``` 

90 """ 

91 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true

92 raise HTTPException( 

93 status_code=403, 

94 detail=(f"Only proxy admins can discover agent cards. Your role={user_api_key_dict.user_role}"), 

95 ) 

96 

97 try: 

98 card: Final = await fetch_well_known_card( 

99 request.url, 

100 discovery_mode=request.discovery_mode, 

101 params=request.params, 

102 ) 

103 except AgentCardDiscoveryError as exc: 

104 raise HTTPException(status_code=400, detail=str(exc)) 

105 except Exception as exc: 

106 verbose_proxy_logger.exception("Unexpected error during A2A discovery: %s", exc) 

107 raise HTTPException(status_code=500, detail=f"Discovery failed: {exc}") 

108 

109 return JSONResponse( 

110 content={"url": request.url, "agent_card": card}, 

111 media_type="application/json", 

112 )