Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/a2a/agent_card.py: 93%
55 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Pure logic for merging an upstream A2A agent card with LiteLLM-specific overrides.
4The merge produces the card that LiteLLM exposes to A2A clients at
5``/a2a/{agent_id}/.well-known/agent-card.json``. The upstream card is taken as
6the base; specific fields are replaced so all traffic flows through the proxy
7and uses LiteLLM auth.
8"""
10import re
11from collections.abc import Mapping
12from copy import deepcopy
13from typing import Final, Literal
15SupportedA2AVersion = Literal["0.3", "1.0"]
17# Protocol versions LiteLLM can serve to A2A clients. The admin pins one per agent;
18# responses are normalized to it regardless of the upstream agent's own version.
19SUPPORTED_A2A_PROTOCOL_VERSIONS: Final[tuple[SupportedA2AVersion, ...]] = ("0.3", "1.0")
21# Default served version when the agent card does not pin one.
22LITELLM_A2A_PROTOCOL_VERSION: Final = "1.0"
25_PROTOCOL_VERSION_PATTERN: Final = re.compile(
26 r"^(\d+\.\d+)(?:\.\d+(?:-[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?)?$"
27)
30def normalize_protocol_version(version: object) -> SupportedA2AVersion | None:
31 """Map a raw ``protocolVersion`` value to the supported canonical major.minor version.
33 Accepts the bare major.minor convention of the 1.0 spec (``"0.3"``, ``"1.0"``) and the
34 full semver forms older SDKs emit (``"0.3.0"``, ``"1.0.1"``, including prerelease and
35 build suffixes like ``"0.3.0-rc1"``). Malformed strings, versions outside the
36 supported set, and non-strings yield ``None``.
37 """
38 if not isinstance(version, str):
39 return None
40 match: Final = _PROTOCOL_VERSION_PATTERN.match(version)
41 if match is None: 41 ↛ 43line 41 didn't jump to line 43 because the condition on line 41 was always true
42 return None
43 major_minor: Final = match.group(1)
44 return next((supported for supported in SUPPORTED_A2A_PROTOCOL_VERSIONS if supported == major_minor), None)
47def resolve_served_protocol_version(card: Mapping[str, object] | None) -> str:
48 """Return the validated protocol version an agent card pins, else the default."""
49 normalized: Final = normalize_protocol_version(card.get("protocolVersion") if card else None)
50 return normalized if normalized is not None else LITELLM_A2A_PROTOCOL_VERSION
53# Security scheme exposed by the LiteLLM-fronted agent card. Always replaces
54# whatever upstream advertised — the client must authenticate to the proxy,
55# not the upstream agent.
56LITELLM_SECURITY_SCHEMES: Final[dict[str, dict[str, str]]] = {
57 "LiteLLMKey": {
58 "type": "http",
59 "scheme": "bearer",
60 "description": "LiteLLM virtual key",
61 },
62}
64LITELLM_SECURITY_REQUIREMENTS: Final[list[dict[str, list[str]]]] = [{"LiteLLMKey": []}]
66# Capabilities LiteLLM can faithfully proxy today. Anything not in this set is
67# dropped during merge so we don't advertise behavior the proxy can't deliver.
68#
69# TODO: re-enable ``streaming`` once the A2A streaming endpoint at
70# ``POST /a2a/{agent_id}/message/stream`` is exercised end-to-end with
71# cost tracking + guardrails. It's wired in ``a2a_endpoints.py`` but not
72# yet covered by tests, so we keep it gated on the upstream advertising it.
73# TODO: ``pushNotifications`` — proxy has no webhook plumbing yet.
74# TODO: ``extendedAgentCard`` — no separate authenticated-extended-card
75# endpoint exposed by the proxy.
76# TODO: ``extensions`` — protocol extensions aren't validated/forwarded yet.
77_ALLOWED_CAPABILITY_KEYS: Final = {"streaming"}
79# v1.0 AgentCard top-level fields. Anything else is stripped from the merged
80# card as a defense against upstream drift. ``supportedInterfaces`` is kept
81# verbatim per product spec even though it is not in the v1.0 schema — clients
82# that expect it will find it; clients that don't will ignore it.
83#
84# ``additionalInterfaces`` is deliberately excluded: it advertises alternate
85# upstream URLs (HTTP/JSONRPC/gRPC backends) that, if persisted and served,
86# would let authenticated agent callers reach the backend directly and bypass
87# the proxy's auth/budget/logging. The proxy publishes its own entrypoint via
88# ``supportedInterfaces`` instead.
89_ALLOWED_TOP_LEVEL_KEYS: Final = {
90 "protocolVersion",
91 "name",
92 "description",
93 "version",
94 "capabilities",
95 "defaultInputModes",
96 "defaultOutputModes",
97 "skills",
98 "preferredTransport",
99 "supportedInterfaces",
100 "iconUrl",
101 "provider",
102 "documentationUrl",
103 "securitySchemes",
104 "security",
105 "supportsAuthenticatedExtendedCard",
106 "signatures",
107 # ``url`` is retained on the stored card because the runtime A2A invocation
108 # path (``a2a_endpoints.py``) reads ``agent.agent_card_params['url']`` to
109 # locate the upstream backend. The public ``/.well-known/agent-card.json``
110 # endpoint rewrites this field to the proxy URL before serving it to
111 # clients, so retaining it here does not leak the upstream to A2A callers.
112 "url",
113}
115_DEFAULT_SKILLS: Final[list[dict[str, str | list[str]]]] = [
116 {
117 "id": "chat",
118 "name": "Chat",
119 "description": "Conversational interaction with the agent.",
120 "tags": ["chat"],
121 }
122]
124_DEFAULT_MODES: Final[list[str]] = ["text"]
126# Fallback ``version`` when the upstream card omits the field. The A2A v1.0
127# schema requires ``version`` on every card, so without this default the
128# merged card would fail validation on clients that ``model_validate`` it.
129_DEFAULT_AGENT_VERSION: Final = "1.0.0"
132def _filter_capabilities(upstream_capabilities: object) -> dict[str, object]:
133 """Return a capabilities dict containing only allowlisted, truthy keys."""
134 if not isinstance(upstream_capabilities, dict):
135 return {}
136 return {
137 key: value for key, value in upstream_capabilities.items() if key in _ALLOWED_CAPABILITY_KEYS and bool(value)
138 }
141def _default_litellm_provider(proxy_base_url: str) -> dict[str, str]:
142 return {"organization": "LiteLLM Proxy", "url": proxy_base_url}
145def merge_agent_card(
146 upstream_card: Mapping[str, object] | None,
147 *,
148 proxy_url: str,
149 proxy_base_url: str,
150 name: str | None = None,
151 description: str | None = None,
152) -> dict[str, object]:
153 """
154 Build the LiteLLM-fronted agent card.
156 Args:
157 upstream_card: Card returned by the upstream agent's well-known endpoint.
158 May be ``None``/empty when the upstream did not expose one.
159 proxy_url: Full URL clients should hit to invoke this agent through
160 the proxy, e.g. ``https://proxy.example.com/a2a/<agent_id>``.
161 proxy_base_url: Root URL of the LiteLLM proxy, used as a fallback when
162 we synthesize a provider record.
163 name: User-supplied agent name from the LiteLLM UI. Takes precedence
164 over the upstream card's ``name``.
165 description: User-supplied description from the LiteLLM UI. Takes
166 precedence over the upstream card's ``description``.
168 Returns:
169 A dict suitable for serving as the proxy's agent card. Only keys in
170 the v1.0 AgentCard schema (plus ``supportedInterfaces``) are emitted.
171 """
172 base: Final[dict[str, object]] = deepcopy(dict(upstream_card)) if upstream_card else {}
174 # Keep the upstream ``url`` on the stored card: the runtime A2A
175 # invocation path reads it from ``agent_card_params`` to know where to
176 # proxy requests. The public well-known endpoint rewrites this field
177 # to the proxy URL before exposing the card to clients.
179 served_version: Final = resolve_served_protocol_version(upstream_card)
180 base["protocolVersion"] = served_version
182 if name:
183 base["name"] = name
184 if description: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true
185 base["description"] = description
187 if not base.get("version"):
188 base["version"] = _DEFAULT_AGENT_VERSION
190 base["capabilities"] = _filter_capabilities(base.get("capabilities"))
192 if not base.get("skills"):
193 base["skills"] = deepcopy(_DEFAULT_SKILLS)
194 if not base.get("defaultInputModes"):
195 base["defaultInputModes"] = list(_DEFAULT_MODES)
196 if not base.get("defaultOutputModes"):
197 base["defaultOutputModes"] = list(_DEFAULT_MODES)
199 if not base.get("provider"):
200 base["provider"] = _default_litellm_provider(proxy_base_url)
202 base["supportedInterfaces"] = [
203 {
204 "url": proxy_url,
205 "protocolBinding": "JSONRPC",
206 "protocolVersion": served_version,
207 }
208 ]
210 base["securitySchemes"] = deepcopy(LITELLM_SECURITY_SCHEMES)
211 # Use the standard A2A/OpenAPI ``security`` field for requirements, not
212 # the non-standard ``securityRequirements`` alias. The upstream's own
213 # ``security`` selector is overwritten here because the proxy enforces its
214 # own scheme regardless of what upstream required.
215 base["security"] = deepcopy(LITELLM_SECURITY_REQUIREMENTS)
217 return {key: value for key, value in base.items() if key in _ALLOWED_TOP_LEVEL_KEYS}