From 6b7d1928b84a5ff618b958b199b52836ebe45499 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 13 Oct 2026 09:04:36 -0400 Subject: [PATCH] fix(a2a): add per-agent a2a_protocol_version override for langgraph-api 0.15 cards (#44463) * fix(a2a): add per-agent a2a_protocol_version override for langgraph-api 0.15 cards Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(a2a): cover streamed 0.3 replies through the real client Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(a2a): accept new kwargs in create_a2a_client fake; add protocol-version integration tests The create_a2a_client fake in tests/agent_tests/test_a2a_agent.py listed a fixed signature, so asend_message's new protocol_version= kwarg raised TypeError in test_a2a_non_streaming and test_a2a_streaming. Accept and ignore extra kwargs with **_, matching the new _capture fake in test_main.py. Add tests/integration/compatibility cells for litellm_params.a2a_protocol_version against a real 2-worker proxy with Postgres and Redis and scripted A2A peers modeled on langgraph-api 0.15.4: pinned send/stream on all three routes, aliases, the 1.0 opt-out, unchanged genuine v1 agents, PATCH taking effect on both workers, persistence via /v1/agents, the localhost retry, invalid values, upstream errors, the dialect hint, and the completion bridge stripping the key. Merge base 5f969982fc: 33 red, 7 green. Head ba709d3de5: 40/40 green twice. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: shivam Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Mrinal Chanshetty Co-authored-by: Claude Opus 5.5 --- litellm/a2a_protocol/card_resolver.py | 42 ++- .../a2a_protocol/exception_mapping_utils.py | 8 +- .../litellm_completion_bridge/handler.py | 3 +- litellm/a2a_protocol/main.py | 13 +- tests/integration/compatibility/_a2a_peers.py | 283 ++++++++++++++++ .../test_a2a_protocol_version_bridge.py | 86 +++++ .../test_a2a_protocol_version_errors.py | 158 +++++++++ .../test_a2a_protocol_version_override.py | 304 ++++++++++++++++++ .../test_a2a_exception_mapping_utils.py | 16 +- tests/unit/a2a_protocol/test_card_resolver.py | 72 +++++ .../test_completion_bridge_streaming.py | 2 + tests/unit/a2a_protocol/test_main.py | 219 ++++++++++++- 12 files changed, 1190 insertions(+), 16 deletions(-) create mode 100644 tests/integration/compatibility/_a2a_peers.py create mode 100644 tests/integration/compatibility/test_a2a_protocol_version_bridge.py create mode 100644 tests/integration/compatibility/test_a2a_protocol_version_errors.py create mode 100644 tests/integration/compatibility/test_a2a_protocol_version_override.py diff --git a/litellm/a2a_protocol/card_resolver.py b/litellm/a2a_protocol/card_resolver.py index 8614c794ac4..08e95f99c87 100644 --- a/litellm/a2a_protocol/card_resolver.py +++ b/litellm/a2a_protocol/card_resolver.py @@ -21,6 +21,7 @@ AGENT_CARD_WELL_KNOWN_PATH: str = "/.well-known/agent-card.json" PREV_AGENT_CARD_WELL_KNOWN_PATH: str = "/.well-known/agent.json" FOUNDRY_AGENT_CARD_PATH: Final = "/agentCard/v1.0" AGENT_CARD_PATH_PARAM: Final = "agent_card_path" +A2A_PROTOCOL_VERSION_PARAM: Final = "a2a_protocol_version" try: from a2a.client import A2ACardResolver as _A2ACardResolver @@ -76,22 +77,43 @@ _CANONICAL_PROTOCOL_BINDINGS: Final = MappingProxyType( ) _LEGACY_PROTOCOL_VERSION: Final = "0.3" +_A2A_PROTOCOL_VERSION_ALIASES: Final = MappingProxyType( + { + "0.3": "0.3", + "0.3.0": "0.3", + "1.0": "1.0", + "1.0.0": "1.0", + } +) -def normalize_agent_card_interfaces(agent_card: "AgentCard") -> "AgentCard": +def resolve_a2a_protocol_version(value: object) -> str | None: + if value is None: + return None + if isinstance(value, bool): + verbose_logger.warning("Ignoring invalid %s value: %r", A2A_PROTOCOL_VERSION_PARAM, value) + return None + if not isinstance(value, (str, int, float)): + verbose_logger.warning("Ignoring invalid %s value: %r", A2A_PROTOCOL_VERSION_PARAM, value) + return None + + protocol_version: Final = _A2A_PROTOCOL_VERSION_ALIASES.get(str(value).strip()) + if protocol_version is None: + verbose_logger.warning("Ignoring invalid %s value: %r", A2A_PROTOCOL_VERSION_PARAM, value) + return protocol_version + + +def normalize_agent_card_interfaces(agent_card: "AgentCard", protocol_version: str | None = None) -> "AgentCard": """ Canonicalize the supported interfaces of spec-adjacent agent cards. Some A2A servers (e.g. LangGraph Platform) serve agent cards with lowercase bindings like "jsonrpc", but a2a-sdk's ClientFactory matches bindings case-sensitively against its uppercase TransportProtocol constants and fails - with "no compatible transports found." for spec-adjacent casings. - - The same servers also speak the A2A 0.3 JSON dialect ("kind"-discriminated - payloads) while declaring protocolVersion "1.0", which a2a-sdk's strict v1 - proto parsing rejects. A mis-cased binding fingerprints such a server, so its - declared version is downgraded to 0.3 to route the SDK's ClientFactory onto - its v0.3 compat transport, which speaks that dialect. + with "no compatible transports found." for spec-adjacent casings. The casing + fingerprint stopped working with langgraph-api 0.15, so operators can pin + the version per agent via ``litellm_params.a2a_protocol_version``. An explicit + "1.0" opts out of the mis-cased downgrade. """ normalized: Final = type(agent_card)() normalized.CopyFrom(agent_card) @@ -101,6 +123,10 @@ def normalize_agent_card_interfaces(agent_card: "AgentCard") -> "AgentCard": continue interface.protocol_binding = canonical interface.protocol_version = _LEGACY_PROTOCOL_VERSION + if protocol_version is not None: + for interface in normalized.supported_interfaces: + if interface.protocol_binding in _CANONICAL_PROTOCOL_BINDINGS.values(): + interface.protocol_version = protocol_version return normalized diff --git a/litellm/a2a_protocol/exception_mapping_utils.py b/litellm/a2a_protocol/exception_mapping_utils.py index 62bef6e02ae..1abf32915f5 100644 --- a/litellm/a2a_protocol/exception_mapping_utils.py +++ b/litellm/a2a_protocol/exception_mapping_utils.py @@ -23,6 +23,12 @@ if TYPE_CHECKING: from a2a.client import Client as A2AClientType +_A2A_KIND_DIALECT_HINT: Final = ( + " (the agent replied in the A2A 0.3 dialect while its card declares 1.0; " + 'set litellm_params.a2a_protocol_version: "0.3" on this agent)' +) + + try: from a2a.client import Client, ClientConfig, create_client @@ -152,7 +158,7 @@ def map_a2a_exception( # Default: wrap in generic A2AError raise A2AError( - message=error_str, + message=error_str + _A2A_KIND_DIALECT_HINT if 'has no field named "kind"' in error_str else error_str, model=model, ) diff --git a/litellm/a2a_protocol/litellm_completion_bridge/handler.py b/litellm/a2a_protocol/litellm_completion_bridge/handler.py index bad17f05923..d664acb5d58 100644 --- a/litellm/a2a_protocol/litellm_completion_bridge/handler.py +++ b/litellm/a2a_protocol/litellm_completion_bridge/handler.py @@ -15,7 +15,7 @@ from typing import Any, Final import litellm from litellm._logging import verbose_logger -from litellm.a2a_protocol.card_resolver import AGENT_CARD_PATH_PARAM +from litellm.a2a_protocol.card_resolver import A2A_PROTOCOL_VERSION_PARAM, AGENT_CARD_PATH_PARAM from litellm.a2a_protocol.litellm_completion_bridge.transformation import ( A2ACompletionBridgeTransformation, A2AStreamingContext, @@ -38,6 +38,7 @@ _AGENT_ONLY_PARAMS: Final = frozenset( "agent_id", "agent_card_params", AGENT_CARD_PATH_PARAM, + A2A_PROTOCOL_VERSION_PARAM, A2A_USER_API_KEY_HASH_PARAM, } ) diff --git a/litellm/a2a_protocol/main.py b/litellm/a2a_protocol/main.py index 0ee2906da9a..4de2ff49b56 100644 --- a/litellm/a2a_protocol/main.py +++ b/litellm/a2a_protocol/main.py @@ -72,10 +72,12 @@ except ImportError: # Import our custom card resolver that supports multiple well-known paths from litellm.a2a_protocol.card_resolver import ( + A2A_PROTOCOL_VERSION_PARAM, AGENT_CARD_PATH_PARAM, LiteLLMA2ACardResolver, get_agent_card_url, normalize_agent_card_interfaces, + resolve_a2a_protocol_version, ) from litellm.a2a_protocol.exception_mapping_utils import ( handle_a2a_localhost_retry, @@ -153,6 +155,10 @@ def _agent_card_path(litellm_params: Mapping[str, object]) -> str | None: return configured_path if isinstance(configured_path, str) and configured_path else None +def _agent_protocol_version(litellm_params: Mapping[str, object]) -> str | None: + return resolve_a2a_protocol_version(litellm_params.get(A2A_PROTOCOL_VERSION_PARAM)) + + def _set_litellm_params_on_logging_obj( kwargs: Mapping[str, object], litellm_params: Mapping[str, object], @@ -498,6 +504,7 @@ async def asend_message( base_url=api_base, extra_headers=extra_headers, relative_card_path=_agent_card_path(litellm_params), + protocol_version=_agent_protocol_version(litellm_params), ) # Type assertion: a2a_client is guaranteed to be non-None here @@ -723,6 +730,7 @@ async def asend_message_streaming( extra_headers=extra_headers, streaming=True, relative_card_path=_agent_card_path(litellm_params), + protocol_version=_agent_protocol_version(litellm_params), ) assert a2a_client is not None @@ -770,6 +778,7 @@ async def create_a2a_client( extra_headers: dict[str, str] | None = None, streaming: bool = False, relative_card_path: str | None = None, + protocol_version: str | None = None, ) -> "A2AClientType": """ Create an A2A client for the given agent URL. @@ -783,6 +792,7 @@ async def create_a2a_client( extra_headers: Optional additional headers to include in requests relative_card_path: Optional card path relative to ``base_url`` (e.g. ``agentCard/v1.0`` for a Microsoft Foundry agent); when None the well-known paths are probed in order + protocol_version: Optional per-agent protocol version override for supported interfaces Returns: An initialized a2a.client.A2AClient instance @@ -819,7 +829,8 @@ async def create_a2a_client( await resolver.get_agent_card( relative_card_path=relative_card_path, http_kwargs=_card_http_kwargs(extra_headers), - ) + ), + protocol_version=protocol_version, ) a2a_client: Final = await create_client( # pyright: ignore[reportOptionalCall] diff --git a/tests/integration/compatibility/_a2a_peers.py b/tests/integration/compatibility/_a2a_peers.py new file mode 100644 index 00000000000..6a34bd78f41 --- /dev/null +++ b/tests/integration/compatibility/_a2a_peers.py @@ -0,0 +1,283 @@ +"""Scripted A2A agents shaped like the servers litellm proxies to, for the a2a_protocol_version cells. + +The ``langgraph`` dialect reproduces langgraph-api 0.15.4 byte for byte where it matters: its card declares +``protocolBinding: "JSONRPC"`` with ``protocolVersion: "1.0"``, and every reply, whether the caller used the +0.3 ``message/send`` or the v1 ``SendMessage`` method, is in the 0.3 ``kind``-discriminated dialect (a v1 +``SendMessage`` reply wraps the 0.3 task under ``result.task``, as langgraph-api does). ``v1`` is a genuine +A2A 1.0 agent that answers ``SendMessage`` and ``SendStreamingMessage`` in the proto JSON dialect and rejects +the 0.3 method names. +""" + +from __future__ import annotations + +import json +import uuid +from collections.abc import Callable, Iterator, Mapping +from contextlib import contextmanager +from dataclasses import dataclass, field +from typing import Final, Literal + +from integration._support.client import Gateway +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import JsonValue + +Dialect = Literal["langgraph", "v1"] +LEGACY_METHODS: Final = frozenset({"message/send", "message/stream"}) +V1_METHODS: Final = frozenset({"SendMessage", "SendStreamingMessage"}) +HINT: Final = ( + "(the agent replied in the A2A 0.3 dialect while its card declares 1.0; " + 'set litellm_params.a2a_protocol_version: "0.3" on this agent)' +) +KIND_ERROR: Final = 'has no field named "kind"' + + +@dataclass(slots=True) +class Peer: + """One scripted agent: ``card`` is what the card endpoint serves, ``dialect`` how the POSTs answer.""" + + marker: str + dialect: Dialect + interfaces: list[dict[str, str]] | None = None + card_url: str | None = None + error: Mapping[str, JsonValue] | None = None + post_status: int = 200 + card_status: int = 200 + wire: Wire | None = None + extra_targets: frozenset[str] = field(default_factory=frozenset) + + @property + def url(self) -> str: + assert self.wire is not None + return self.wire.url + "/a2a/" + self.marker + + def card(self) -> dict[str, JsonValue]: + interfaces: Final = self.interfaces if self.interfaces is not None else [{"protocolBinding": "JSONRPC"}] + return { + "name": self.marker, + "description": "Synthetic langgraph-api 0.15 echo assistant", + "url": self.card_url or self.url, + "supportedInterfaces": [ + { + "url": interface.get("url", self.card_url or self.url), + "protocolBinding": interface["protocolBinding"], + "protocolVersion": interface.get("protocolVersion", "1.0"), + } + for interface in interfaces + ], + "capabilities": {"streaming": True, "pushNotifications": False}, + "defaultInputModes": ["application/json", "text/plain"], + "defaultOutputModes": ["application/json", "text/plain"], + "skills": [], + "version": "0.15.4", + } + + def respond(self, request: Request) -> Reply: + if request.method == "GET": + if self.card_status != 200: + return Reply(status=self.card_status, body=b'{"detail": "no card here"}') + return Reply(body=json.dumps(self.card()).encode()) + if self.post_status != 200: + return Reply(status=self.post_status, body=b'{"detail": "scripted failure"}') + body: Final = json.loads(request.body) + method: Final = body["method"] + if self.error is not None: + return _json({"jsonrpc": "2.0", "id": body["id"], "error": dict(self.error)}) + if self.dialect == "v1" and method not in V1_METHODS: + return _json({"jsonrpc": "2.0", "id": body["id"], "error": {"code": -32601, "message": "Method not found"}}) + text: Final = _input_text(body["params"]["message"]) + reply: Final = "echo: " + text + if method in {"message/stream", "SendStreamingMessage"}: + frames: Final = ( + self._status_frame(body["id"], reply, final=False), + self._artifact_frame(body["id"], reply), + self._status_frame(body["id"], None, final=True), + ) + return Reply( + content_type="text/event-stream", + chunks=tuple(f"event: message\ndata: {json.dumps(frame)}\n\n".encode() for frame in frames), + ) + task: Final = self._task(reply, text, v1=self.dialect == "v1") + result: Final = {"task": task} if method == "SendMessage" else task + return _json({"jsonrpc": "2.0", "id": body["id"], "result": result}) + + def _ids(self) -> tuple[str, str]: + return self.marker + "-task", self.marker + "-context" + + def _task(self, reply: str, text: str, *, v1: bool) -> dict[str, JsonValue]: + task_id, context_id = self._ids() + if v1: + return { + "id": task_id, + "contextId": context_id, + "status": {"state": "TASK_STATE_COMPLETED"}, + "artifacts": [{"artifactId": self.marker + "-artifact", "parts": [{"text": reply}]}], + "history": [{"role": "ROLE_USER", "messageId": self.marker + "-in", "parts": [{"text": text}]}], + } + return { + "kind": "task", + "id": task_id, + "contextId": context_id, + "status": {"state": "completed"}, + "artifacts": [{"artifactId": self.marker + "-artifact", "parts": [{"kind": "text", "text": reply}]}], + "history": [ + { + "kind": "message", + "role": "user", + "messageId": self.marker + "-in", + "parts": [{"kind": "text", "text": text}], + } + ], + } + + def _status_frame(self, request_id: JsonValue, reply: str | None, *, final: bool) -> dict[str, JsonValue]: + task_id, context_id = self._ids() + if self.dialect == "v1": + status: dict[str, JsonValue] = {"state": "TASK_STATE_COMPLETED" if final else "TASK_STATE_WORKING"} + if reply is not None: + status["message"] = { + "role": "ROLE_AGENT", + "messageId": self.marker + "-out", + "parts": [{"text": reply}], + } + return { + "jsonrpc": "2.0", + "id": request_id, + "result": {"statusUpdate": {"taskId": task_id, "contextId": context_id, "status": status}}, + } + status = {"state": "completed" if final else "working"} + if reply is not None: + status["message"] = { + "kind": "message", + "role": "agent", + "messageId": self.marker + "-out", + "parts": [{"kind": "text", "text": reply}], + } + return { + "jsonrpc": "2.0", + "id": request_id, + "result": { + "kind": "status-update", + "taskId": task_id, + "contextId": context_id, + "status": status, + "final": final, + }, + } + + def _artifact_frame(self, request_id: JsonValue, reply: str) -> dict[str, JsonValue]: + task_id, context_id = self._ids() + if self.dialect == "v1": + return { + "jsonrpc": "2.0", + "id": request_id, + "result": { + "artifactUpdate": { + "taskId": task_id, + "contextId": context_id, + "artifact": {"artifactId": self.marker + "-artifact", "parts": [{"text": reply}]}, + "lastChunk": True, + } + }, + } + return { + "jsonrpc": "2.0", + "id": request_id, + "result": { + "kind": "artifact-update", + "taskId": task_id, + "contextId": context_id, + "artifact": {"artifactId": self.marker + "-artifact", "parts": [{"kind": "text", "text": reply}]}, + "lastChunk": True, + }, + } + + +def _json(value: Mapping[str, JsonValue]) -> Reply: + return Reply(body=json.dumps(value).encode()) + + +def _input_text(message: Mapping[str, JsonValue]) -> str: + parts: Final = message["parts"] + assert isinstance(parts, list) and parts, message + first: Final = parts[0] + assert isinstance(first, dict), message + return str(first["text"]) + + +@contextmanager +def peer(marker_prefix: str, dialect: Dialect, **fields: object) -> Iterator[Peer]: + agent: Final = Peer(marker=marker_prefix + uuid.uuid4().hex, dialect=dialect, **fields) # type: ignore[arg-type] + with wire_server(agent.respond) as wire: + agent.wire = wire + yield agent + + +def register( + gateway: Gateway, + cleanups: Callable[..., object], + name: str, + url: str, + litellm_params: Mapping[str, JsonValue] | None, +) -> str: + body: dict[str, JsonValue] = { + "agent_name": name, + "agent_card_params": { + "protocolVersion": "1.0", + "name": name, + "description": "Synthetic langgraph-api 0.15 echo assistant", + "version": "0.15.4", + "url": url, + "capabilities": {"streaming": True}, + "defaultInputModes": ["text"], + "defaultOutputModes": ["text"], + "skills": [], + }, + } + if litellm_params is not None: + body["litellm_params"] = dict(litellm_params) + created: Final = gateway.request("POST", "/v1/agents", body) + assert created.status_code == 200, created.text + identity: Final = str(created.json()["agent_id"]) + + def delete() -> None: + deleted: Final = gateway.request("DELETE", f"/v1/agents/{identity}") + assert deleted.status_code == 200, deleted.text + assert read_rows('SELECT agent_id FROM "LiteLLM_AgentsTable" WHERE agent_id=%s', (identity,)) == [] + + cleanups(delete) + return identity + + +def legacy_send(marker: str, text: str = "synthetic ping", method: str = "message/send") -> dict[str, JsonValue]: + return { + "jsonrpc": "2.0", + "id": marker, + "method": method, + "params": { + "message": { + "kind": "message", + "role": "user", + "messageId": marker + "-in", + "parts": [{"kind": "text", "text": text}], + } + }, + } + + +def posts(wire: Wire) -> tuple[dict[str, JsonValue], ...]: + return tuple(json.loads(item.body) for item in wire.drain() if item.method == "POST") + + +def sse_frames(text: str) -> tuple[dict[str, JsonValue], ...]: + return tuple( + json.loads(line[len("data: ") :]) + for line in text.splitlines() + if line.startswith("data: ") and line[6:].strip() + ) + + +def spend_rows(request_id: str) -> list[dict[str, JsonValue]]: + return read_rows( + 'SELECT request_id, model, call_type, status FROM "LiteLLM_SpendLogs" WHERE request_id=%s', (request_id,) + ) diff --git a/tests/integration/compatibility/test_a2a_protocol_version_bridge.py b/tests/integration/compatibility/test_a2a_protocol_version_bridge.py new file mode 100644 index 00000000000..6b65f1280c5 --- /dev/null +++ b/tests/integration/compatibility/test_a2a_protocol_version_bridge.py @@ -0,0 +1,86 @@ +"""``a2a_protocol_version`` is agent-only metadata: the completion bridge must strip it before +``litellm.acompletion`` so it never reaches a provider, and paths that do not build an a2a-sdk client (card +discovery, the ``a2a/`` chat-completions provider) must not change for a pinned agent.""" + +from __future__ import annotations + +import json +import uuid +from typing import Final + +import httpx +import pytest +from integration._support.client import Gateway, eventually +from integration.compatibility._a2a_peers import legacy_send, peer, posts, register, spend_rows, sse_frames +from pydantic import JsonValue + + +def _observations(gateway: Gateway) -> list[dict[str, JsonValue]]: + response: Final = httpx.get(gateway.upstream_url + "/__observations", trust_env=False, timeout=15) + response.raise_for_status() + return list(response.json()["requests"]) + + +@pytest.mark.parametrize("method", ("message/send", "message/stream")) +def test_completion_bridge_drops_protocol_version_before_the_provider(gateway: Gateway, method: str) -> None: + marker: Final = "bridge" + uuid.uuid4().hex + key: Final = "sk-scripted-" + marker + with gateway.scenario() as scenario: + identity: Final = register( + gateway, + scenario.cleanups.callback, + marker, + gateway.upstream_url + "/v1", + {"custom_llm_provider": "openai", "model": "gpt-4o-mini", "api_key": key, "a2a_protocol_version": "0.3"}, + ) + text: Final = "bridge ping " + marker + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(marker, text=text, method=method)) + assert response.status_code == 200, response.text + if method == "message/send": + body: Final = response.json() + assert body["id"] == marker and "error" not in body, response.text + parts = body["result"]["message"]["parts"] + assert parts == [{"text": "Hello! This is a mock response from the fake OpenAI endpoint."}], response.text + else: + frames: Final = sse_frames(response.text) + assert frames and all("error" not in frame and frame["id"] == marker for frame in frames), response.text + assert "TASK_STATE_COMPLETED" in response.text, response.text + mine: Final = [item for item in _observations(gateway) if item["authorization"] == "Bearer " + key] + assert len(mine) == 1, mine + sent: Final = mine[0]["body"] + assert isinstance(sent, dict), mine + assert "a2a_protocol_version" not in sent and "a2a_protocol_version" not in json.dumps(sent), sent + assert sent["model"] == "gpt-4o-mini" and sent["messages"] == [{"role": "user", "content": text}], sent + assert sent.get("stream", False) is (method == "message/stream"), sent + if method == "message/send": + rows: Final = eventually(lambda: spend_rows(marker), lambda values: len(values) == 1, seconds=70) + assert rows[0]["call_type"] == "asend_message" and rows[0]["status"] == "success", rows + + +def test_card_discovery_of_pinned_agent_is_unchanged(gateway: Gateway) -> None: + with peer("lgcard", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": "0.3"} + ) + discovered: Final = gateway.request("GET", f"/a2a/{identity}/.well-known/agent-card.json") + assert discovered.status_code == 200, discovered.text + card: Final = discovered.json() + assert card["protocolVersion"] == "1.0" and card["name"] == agent.marker, discovered.text + assert "a2a_protocol_version" not in discovered.text, discovered.text + assert agent.wire.drain() == () # type: ignore[union-attr] + + +def test_chat_completions_a2a_provider_unaffected_by_pin(gateway: Gateway) -> None: + with peer("lgchat", "langgraph") as agent, gateway.scenario() as scenario: + register(gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": "0.3"}) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": "a2a/" + agent.marker, "messages": [{"role": "user", "content": "chat ping " + agent.marker}]}, + ) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["choices"][0]["message"]["content"] == "echo: user: chat ping " + agent.marker, response.text + sent: Final = posts(agent.wire) # type: ignore[arg-type] + assert len(sent) == 1 and sent[0]["method"] == "message/send", sent + assert "a2a_protocol_version" not in json.dumps(sent), sent diff --git a/tests/integration/compatibility/test_a2a_protocol_version_errors.py b/tests/integration/compatibility/test_a2a_protocol_version_errors.py new file mode 100644 index 00000000000..1303195bc34 --- /dev/null +++ b/tests/integration/compatibility/test_a2a_protocol_version_errors.py @@ -0,0 +1,158 @@ +"""Failure shapes around ``litellm_params.a2a_protocol_version``: the hint naming the setting, values the proxy +must ignore, and agent failures that must reach the caller unchanged while the proxy keeps serving.""" + +from __future__ import annotations + +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration.compatibility._a2a_peers import ( + HINT, + KIND_ERROR, + legacy_send, + peer, + posts, + register, + sse_frames, +) +from pydantic import JsonValue + +INVALID: Final[tuple[JsonValue, ...]] = ("2.0", "abc", True, False, 1, [], {"v": "0.3"}, "", "0" * 5120) +INVALID_IDS: Final = ("future", "word", "true", "false", "int", "list", "object", "empty", "five-kb") + + +def _neighbour_answers(gateway: Gateway, identity: str, marker: str) -> None: + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(marker)) + assert response.status_code == 200, response.text + assert "error" not in response.json(), response.text + + +@pytest.mark.parametrize("method", ("message/send", "message/stream")) +def test_unpinned_langgraph_agent_error_names_the_setting(gateway: Gateway, method: str) -> None: + with peer("lghint", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, None) + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(agent.marker, method=method)) + if method == "message/send": + assert response.status_code == 500, response.text + error = response.json()["error"] + else: + assert response.status_code == 200, response.text + frames: Final = sse_frames(response.text) + assert len(frames) == 1, response.text + error = frames[0]["error"] + assert error["code"] == -32603, response.text # type: ignore[index] + assert KIND_ERROR in error["message"] and error["message"].endswith(HINT), response.text # type: ignore[index] + sent: Final = posts(agent.wire) # type: ignore[arg-type] + expected_method: Final = "SendMessage" if method == "message/send" else "SendStreamingMessage" + assert [item["method"] for item in sent] == [expected_method], sent + + +@pytest.mark.parametrize("value", INVALID, ids=INVALID_IDS) +def test_invalid_protocol_version_is_ignored(gateway: Gateway, value: JsonValue) -> None: + with ( + peer("lginvalid", "langgraph") as agent, + peer("lgneighbour", "langgraph") as neighbour, + gateway.scenario() as scenario, + ): + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": value} + ) + pinned: Final = register( + gateway, scenario.cleanups.callback, neighbour.marker, neighbour.url, {"a2a_protocol_version": "0.3"} + ) + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(agent.marker)) + assert response.status_code == 500, response.text + message: Final = response.json()["error"]["message"] + assert KIND_ERROR in message and message.endswith(HINT), response.text + assert [item["method"] for item in posts(agent.wire)] == ["SendMessage"] # type: ignore[arg-type] + _neighbour_answers(gateway, pinned, neighbour.marker) + assert [item["method"] for item in posts(neighbour.wire)] == ["message/send"] # type: ignore[arg-type] + assert gateway.request("GET", "/health/liveliness").status_code == 200 + + +def test_missing_and_null_protocol_version_keep_card_version(gateway: Gateway) -> None: + for params in (None, {"a2a_protocol_version": None}): + with peer("lgabsent", "langgraph") as agent, gateway.scenario() as scenario: + identity = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, params) + response = gateway.request("POST", f"/a2a/{identity}", legacy_send(agent.marker)) + assert response.status_code == 500, response.text + assert response.json()["error"]["message"].endswith(HINT), response.text + assert [item["method"] for item in posts(agent.wire)] == ["SendMessage"] # type: ignore[arg-type] + + +@pytest.mark.parametrize( + ("fields", "status"), + ( + ({"error": {"code": -32603, "message": "scripted agent failure"}}, 500), + ({"post_status": 500}, 500), + ({"post_status": 404}, 500), + ), + ids=("jsonrpc-error", "http-500", "http-404"), +) +def test_pinned_agent_upstream_failures_reach_caller_without_hint( + gateway: Gateway, fields: dict[str, object], status: int +) -> None: + with ( + peer("lgfail", "langgraph", **fields) as agent, + peer("lgok", "langgraph") as healthy, + gateway.scenario() as scenario, + ): + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": "0.3"} + ) + other: Final = register( + gateway, scenario.cleanups.callback, healthy.marker, healthy.url, {"a2a_protocol_version": "0.3"} + ) + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(agent.marker)) + assert response.status_code == status, response.text + body: Final = response.json() + assert body["id"] == agent.marker and "result" not in body, response.text + message: Final = str(body["error"]["message"]) + assert HINT not in message and KIND_ERROR not in message, response.text + if "error" in fields: + assert "scripted agent failure" in message, response.text + else: + assert str(fields["post_status"]) in message, response.text + assert [item["method"] for item in posts(agent.wire)] == ["message/send"] # type: ignore[arg-type] + failure: Final = eventually( + lambda: read_rows('SELECT status FROM "LiteLLM_SpendLogs" WHERE model=%s', ("a2a_agent/" + agent.marker,)), + lambda rows: len(rows) == 1, + seconds=70, + ) + assert failure[0]["status"] == "failure", failure + _neighbour_answers(gateway, other, healthy.marker) + + +def test_pinned_agent_card_unreachable_reaches_caller(gateway: Gateway) -> None: + with peer("lgnocard", "langgraph", card_status=404) as agent, gateway.scenario() as scenario: + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": "0.3"} + ) + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(agent.marker)) + assert response.status_code == 500, response.text + body: Final = response.json() + assert body["id"] == agent.marker and "result" not in body, response.text + assert "404" in body["error"]["message"] and HINT not in response.text, response.text + traffic: Final = agent.wire.drain() # type: ignore[union-attr] + assert traffic and all(item.method == "GET" for item in traffic), traffic + assert gateway.request("GET", "/health/liveliness").status_code == 200 + + +def test_upstream_authored_kind_message_gets_hint(gateway: Gateway) -> None: + """User ruling 2026-10-10: an agent-authored error that quotes the SDK's ``kind`` parse failure gets the hint + too, since the substring check cannot tell the two apart and the hint is still the right next step.""" + authored: Final = 'upstream says: Message type "x" has no field named "kind"' + with ( + peer("lgauthored", "langgraph", error={"code": -32000, "message": authored}) as agent, + gateway.scenario() as scenario, + ): + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": "0.3"} + ) + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(agent.marker)) + assert response.status_code == 500, response.text + message: Final = str(response.json()["error"]["message"]) + assert authored in message and message.endswith(HINT), response.text + assert [item["method"] for item in posts(agent.wire)] == ["message/send"] # type: ignore[arg-type] diff --git a/tests/integration/compatibility/test_a2a_protocol_version_override.py b/tests/integration/compatibility/test_a2a_protocol_version_override.py new file mode 100644 index 00000000000..299259714c4 --- /dev/null +++ b/tests/integration/compatibility/test_a2a_protocol_version_override.py @@ -0,0 +1,304 @@ +"""Per-agent ``litellm_params.a2a_protocol_version`` pins the A2A transport the proxy speaks to the agent. + +langgraph-api 0.15 serves cards declaring ``JSONRPC``/``1.0`` while answering in the 0.3 ``kind`` dialect, so +a2a-sdk picks its strict v1 transport and rejects every reply. Pinning "0.3" routes the SDK onto its v0.3 +compat transport; pinning "1.0" keeps v1 even where the mis-cased-binding fingerprint would downgrade. +Every cell asserts the caller's response, the exact request the scripted agent received, and Postgres. +""" + +from __future__ import annotations + +import json +import os +import socket +import time +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration.compatibility._a2a_peers import ( + HINT, + KIND_ERROR, + legacy_send, + peer, + posts, + register, + spend_rows, + sse_frames, +) +from pydantic import JsonValue + +from litellm.constants import PROXY_CONFIG_RELOAD_INTERVAL_SECONDS + +PROXY_WORKERS: Final = int(os.environ.get("INTEGRATION_PROXY_WORKERS", "1")) +WORKER_SYNC_SECONDS: Final = 0.0 if PROXY_WORKERS == 1 else PROXY_CONFIG_RELOAD_INTERVAL_SECONDS + 5.0 +FRESH_CONNECTION: Final = {"Connection": "close"} +PINNED: Final = {"a2a_protocol_version": "0.3"} +ROUTES: Final = ("/a2a/{id}", "/a2a/{id}/message/send", "/v1/a2a/{id}/message/send") + + +def _artifact_text(result: dict[str, JsonValue]) -> str: + task: Final = result.get("task", result) + assert isinstance(task, dict), result + artifacts: Final = task["artifacts"] + assert isinstance(artifacts, list) and len(artifacts) == 1, result + parts: Final = artifacts[0]["parts"] # type: ignore[index] + return str(parts[0]["text"]) # type: ignore[index] + + +def _assert_legacy_wire(sent: tuple[dict[str, JsonValue], ...], marker: str, method: str = "message/send") -> None: + assert len(sent) == 1, sent + assert sent[0]["method"] == method, sent + message: Final = sent[0]["params"]["message"] # type: ignore[index] + assert message["kind"] == "message" and message["role"] == "user", sent + assert message["parts"] == [{"kind": "text", "text": "synthetic ping"}], sent + assert message["messageId"] == marker + "-in", sent + + +def _assert_v1_wire(sent: tuple[dict[str, JsonValue], ...], marker: str, method: str = "SendMessage") -> None: + assert len(sent) == 1, sent + assert sent[0]["method"] == method, sent + message: Final = sent[0]["params"]["message"] # type: ignore[index] + assert "kind" not in message and message["role"] == "ROLE_USER", sent + assert message["parts"] == [{"text": "synthetic ping"}], sent + assert message["messageId"] == marker + "-in", sent + + +def _one_spend_row(request_id: str, call_type: str, status: str) -> None: + rows: Final = eventually(lambda: spend_rows(request_id), lambda values: len(values) == 1, seconds=70) + assert rows[0]["call_type"] == call_type and rows[0]["status"] == status, rows + + +def _send_ok(gateway: Gateway, identity: str, marker: str, path: str = "/a2a/{id}") -> dict[str, JsonValue]: + response: Final = gateway.request("POST", path.format(id=identity), legacy_send(marker)) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["id"] == marker and "error" not in body, response.text + assert _artifact_text(body["result"]) == "echo: synthetic ping", response.text + return body + + +def _send_kind_error(gateway: Gateway, identity: str, marker: str) -> str: + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(marker)) + assert response.status_code == 500, response.text + message: Final = str(response.json()["error"]["message"]) + assert KIND_ERROR in message, response.text + return message + + +@pytest.mark.parametrize("route", ROUTES) +def test_pinned_langgraph_agent_answers_message_send_in_the_legacy_dialect(gateway: Gateway, route: str) -> None: + with peer("lgsend", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, PINNED) + _send_ok(gateway, identity, agent.marker, route) + _assert_legacy_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message", "success") + + +def test_pinned_langgraph_agent_streams_message_stream_to_the_final_frame(gateway: Gateway) -> None: + with peer("lgstream", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, PINNED) + response: Final = gateway.request( + "POST", f"/a2a/{identity}", legacy_send(agent.marker, method="message/stream") + ) + assert response.status_code == 200, response.text + assert response.headers["content-type"].startswith("text/event-stream"), response.headers + frames: Final = sse_frames(response.text) + assert all(frame["id"] == agent.marker and "error" not in frame for frame in frames), response.text + results: Final = [frame["result"] for frame in frames] + assert len(results) == 3, response.text + assert results[0]["statusUpdate"]["status"]["state"] == "TASK_STATE_WORKING", response.text # type: ignore[index] + assert results[0]["statusUpdate"]["status"]["message"]["parts"] == [{"text": "echo: synthetic ping"}] # type: ignore[index] + assert results[1]["artifactUpdate"]["artifact"]["parts"] == [{"text": "echo: synthetic ping"}] # type: ignore[index] + assert results[2]["statusUpdate"]["status"]["state"] == "TASK_STATE_COMPLETED", response.text # type: ignore[index] + _assert_legacy_wire(posts(agent.wire), agent.marker, "message/stream") # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message_streaming", "success") + + +def test_v1_caller_reaches_pinned_legacy_agent_in_its_own_envelope(gateway: Gateway) -> None: + with peer("lgv1caller", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, PINNED) + response: Final = gateway.client.post( + f"/a2a/{identity}", + headers={"Authorization": f"Bearer {gateway.key}", "a2a-version": "1.0"}, + json={ + "jsonrpc": "2.0", + "id": agent.marker, + "method": "SendMessage", + "params": { + "message": { + "role": "ROLE_USER", + "messageId": agent.marker + "-in", + "parts": [{"text": "synthetic ping"}], + } + }, + }, + ) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["id"] == agent.marker and "error" not in body, response.text + task: Final = body["result"]["task"] + assert "kind" not in task and task["status"]["state"] == "TASK_STATE_COMPLETED", response.text + assert task["artifacts"][0]["parts"] == [{"text": "echo: synthetic ping"}], response.text + _assert_legacy_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message", "success") + + +@pytest.mark.parametrize("value", ("0.3.0", 0.3, " 0.3 "), ids=("dotted", "number", "spaced")) +def test_protocol_version_aliases_pin_the_legacy_transport(gateway: Gateway, value: JsonValue) -> None: + with peer("lgalias", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": value} + ) + _send_ok(gateway, identity, agent.marker) + _assert_legacy_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message", "success") + + +def test_genuine_v1_agent_pinned_one_point_zero_is_unchanged(gateway: Gateway) -> None: + with peer("v1pinned", "v1") as agent, gateway.scenario() as scenario: + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": "1.0"} + ) + _send_ok(gateway, identity, agent.marker) + _assert_v1_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message", "success") + + +def test_genuine_v1_agent_unpinned_is_unchanged(gateway: Gateway) -> None: + with peer("v1plain", "v1") as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, None) + _send_ok(gateway, identity, agent.marker) + _assert_v1_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message", "success") + + +@pytest.mark.parametrize("pin", (None, "0.3"), ids=("unpinned", "pinned-legacy")) +def test_lowercase_binding_still_downgrades_to_the_legacy_transport(gateway: Gateway, pin: str | None) -> None: + with ( + peer("lglower", "langgraph", interfaces=[{"protocolBinding": "jsonrpc"}]) as agent, + gateway.scenario() as scenario, + ): + params: Final = None if pin is None else {"a2a_protocol_version": pin} + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, params) + _send_ok(gateway, identity, agent.marker) + _assert_legacy_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message", "success") + + +def test_explicit_one_point_zero_opts_out_of_the_lowercase_downgrade(gateway: Gateway) -> None: + with peer("v1lower", "v1", interfaces=[{"protocolBinding": "jsonrpc"}]) as agent, gateway.scenario() as scenario: + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": "1.0"} + ) + _send_ok(gateway, identity, agent.marker) + _assert_v1_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message", "success") + + +def test_one_point_zero_pin_on_a_legacy_card_sends_v1(gateway: Gateway) -> None: + legacy_card: Final = [{"protocolBinding": "JSONRPC", "protocolVersion": "0.3"}] + with peer("v1oncard03", "v1", interfaces=legacy_card) as agent, gateway.scenario() as scenario: + identity: Final = register( + gateway, scenario.cleanups.callback, agent.marker, agent.url, {"a2a_protocol_version": "1.0"} + ) + _send_ok(gateway, identity, agent.marker) + _assert_v1_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + _one_spend_row(agent.marker, "asend_message", "success") + + +def test_override_rewrites_only_canonical_interfaces(gateway: Gateway) -> None: + with peer("lgmulti", "langgraph") as agent, peer("lgdecoy", "langgraph") as decoy, gateway.scenario() as scenario: + agent.interfaces = [ + {"protocolBinding": "JSONRPC", "url": agent.url}, + {"protocolBinding": "custom-ws", "url": decoy.url}, + {"protocolBinding": "HTTP+JSON", "url": decoy.url}, + ] + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, PINNED) + _send_ok(gateway, identity, agent.marker) + _assert_legacy_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + assert decoy.wire.drain() == (), "the HTTP+JSON and custom bindings must not be selected" # type: ignore[union-attr] + _one_spend_row(agent.marker, "asend_message", "success") + + +def test_override_added_and_removed_by_patch_takes_effect_on_both_workers(gateway: Gateway) -> None: + with peer("lgpatch", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, None) + assert HINT in _send_kind_error(gateway, identity, agent.marker + "-before") + before: Final = posts(agent.wire) # type: ignore[arg-type] + assert [item["method"] for item in before] == ["SendMessage"], before + + def send(suffix: str) -> int: + response: Final = gateway.request( + "POST", f"/a2a/{identity}", legacy_send(agent.marker + suffix), headers=FRESH_CONNECTION + ) + return response.status_code + + def settle(params: dict[str, JsonValue], expected: int) -> None: + patched: Final = gateway.request("PATCH", f"/v1/agents/{identity}", {"litellm_params": params}) + assert patched.status_code == 200, patched.text + assert patched.json()["litellm_params"].get("a2a_protocol_version") == params.get("a2a_protocol_version") + written_at: Final = time.monotonic() + counter = iter(range(10_000)) + eventually( + lambda: (time.monotonic() - written_at, tuple(send(f"-{next(counter)}") for _ in range(4))), + lambda stamped: stamped[0] >= WORKER_SYNC_SECONDS and set(stamped[1]) == {expected}, + seconds=WORKER_SYNC_SECONDS + 30, + ) + agent.wire.drain() # type: ignore[union-attr] + + settle(dict(PINNED), 200) + response: Final = gateway.request("POST", f"/a2a/{identity}", legacy_send(agent.marker)) + assert response.status_code == 200, response.text + _assert_legacy_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + settle({}, 500) + assert HINT in _send_kind_error(gateway, identity, agent.marker + "-after") + + +def test_repeated_pinned_sends_log_one_row_each(gateway: Gateway) -> None: + with peer("lgrepeat", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, PINNED) + markers: Final = tuple(f"{agent.marker}-{index}" for index in range(10)) + for marker in markers: + response = gateway.request("POST", f"/a2a/{identity}", legacy_send(marker)) + assert response.status_code == 200 and response.json()["id"] == marker, response.text + sent: Final = posts(agent.wire) # type: ignore[arg-type] + assert [item["params"]["message"]["messageId"] for item in sent] == [m + "-in" for m in markers], sent # type: ignore[index] + assert len({item["id"] for item in sent}) == len(markers), sent + assert {item["method"] for item in sent} == {"message/send"}, sent + for marker in markers: + _one_spend_row(marker, "asend_message", "success") + + +def test_registered_agent_persists_protocol_version(gateway: Gateway) -> None: + with peer("lgstored", "langgraph") as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, PINNED) + assert gateway.get(f"/v1/agents/{identity}")["litellm_params"]["a2a_protocol_version"] == "0.3" # type: ignore[index] + rows: Final = read_rows('SELECT litellm_params FROM "LiteLLM_AgentsTable" WHERE agent_id=%s', (identity,)) + assert len(rows) == 1, rows + stored: Final = rows[0]["litellm_params"] + stored_params: Final = json.loads(stored) if isinstance(stored, str) else stored + assert stored_params["a2a_protocol_version"] == "0.3", rows # type: ignore[index] + _send_ok(gateway, identity, agent.marker) + _assert_legacy_wire(posts(agent.wire), agent.marker) # type: ignore[arg-type] + + +def _closed_local_port() -> int: + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + return int(probe.getsockname()[1]) + + +def test_pinned_agent_survives_localhost_retry(gateway: Gateway) -> None: + dead: Final = f"http://localhost:{_closed_local_port()}/a2a/unreachable" + with peer("lgretry", "langgraph", card_url=dead) as agent, gateway.scenario() as scenario: + identity: Final = register(gateway, scenario.cleanups.callback, agent.marker, agent.url, PINNED) + _send_ok(gateway, identity, agent.marker) + traffic: Final = agent.wire.drain() # type: ignore[union-attr] + sent: Final = tuple(json.loads(item.body) for item in traffic if item.method == "POST") + assert tuple(item.target for item in traffic if item.method == "POST") == ("/a2a/" + agent.marker + "/",), ( + traffic + ) + _assert_legacy_wire(sent, agent.marker) + _one_spend_row(agent.marker, "asend_message", "success") diff --git a/tests/unit/a2a_protocol/test_a2a_exception_mapping_utils.py b/tests/unit/a2a_protocol/test_a2a_exception_mapping_utils.py index 5f097570bc2..4043a95c794 100644 --- a/tests/unit/a2a_protocol/test_a2a_exception_mapping_utils.py +++ b/tests/unit/a2a_protocol/test_a2a_exception_mapping_utils.py @@ -6,7 +6,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest from litellm.a2a_protocol import exception_mapping_utils as emu -from litellm.a2a_protocol.exceptions import A2ALocalhostURLError +from litellm.a2a_protocol.exceptions import A2AError, A2ALocalhostURLError def _localhost_error() -> A2ALocalhostURLError: @@ -17,6 +17,20 @@ def _localhost_error() -> A2ALocalhostURLError: ) +def test_map_a2a_exception_adds_protocol_version_hint_for_kind_field_error(): + with pytest.raises(A2AError) as error: + emu.map_a2a_exception(RuntimeError('Message type "lf.a2a.v1.Task" has no field named "kind"')) + + assert "a2a_protocol_version: " in str(error.value) + + +def test_map_a2a_exception_does_not_add_protocol_version_hint_for_other_errors(): + with pytest.raises(A2AError) as error: + emu.map_a2a_exception(RuntimeError("unrelated agent error")) + + assert "a2a_protocol_version" not in str(error.value) + + @pytest.mark.asyncio async def test_localhost_retry_reuses_stashed_httpx_client(): """The retry must reuse the httpx client LiteLLM attached at creation (it carries diff --git a/tests/unit/a2a_protocol/test_card_resolver.py b/tests/unit/a2a_protocol/test_card_resolver.py index fdfb51987a3..4cfa8dab45b 100644 --- a/tests/unit/a2a_protocol/test_card_resolver.py +++ b/tests/unit/a2a_protocol/test_card_resolver.py @@ -16,6 +16,7 @@ from litellm.a2a_protocol.card_resolver import ( fix_agent_card_url, is_localhost_or_internal_url, normalize_agent_card_interfaces, + resolve_a2a_protocol_version, set_agent_card_url, ) from litellm.a2a_protocol.exceptions import A2AAgentCardDiscoveryError @@ -139,6 +140,77 @@ def test_normalize_agent_card_interfaces_downgrades_miscased_interfaces_to_the_0 assert card.supported_interfaces[0].protocol_version == "1.0" +def test_normalize_agent_card_interfaces_applies_override_to_canonical_bindings_without_mutating_input(): + pb2 = pytest.importorskip("a2a.types.a2a_pb2") + card = pb2.AgentCard( + name="langgraph", + supported_interfaces=[ + pb2.AgentInterface(url="http://a/", protocol_binding="jsonrpc", protocol_version="1.0"), + pb2.AgentInterface(url="http://b/", protocol_binding="JSONRPC", protocol_version="1.0"), + pb2.AgentInterface(url="http://c/", protocol_binding="HTTP+JSON", protocol_version="1.0"), + pb2.AgentInterface(url="http://d/", protocol_binding="GRPC", protocol_version="1.0"), + pb2.AgentInterface(url="http://e/", protocol_binding="websocket", protocol_version="1.0"), + ], + ) + + normalized = normalize_agent_card_interfaces(card, protocol_version="0.3") + + assert [(item.protocol_binding, item.protocol_version) for item in normalized.supported_interfaces] == [ + ("JSONRPC", "0.3"), + ("JSONRPC", "0.3"), + ("HTTP+JSON", "0.3"), + ("GRPC", "0.3"), + ("websocket", "1.0"), + ] + assert card.supported_interfaces[0].protocol_binding == "jsonrpc" + assert [(item.protocol_binding, item.protocol_version) for item in card.supported_interfaces] == [ + ("jsonrpc", "1.0"), + ("JSONRPC", "1.0"), + ("HTTP+JSON", "1.0"), + ("GRPC", "1.0"), + ("websocket", "1.0"), + ] + + +def test_normalize_agent_card_interfaces_override_opts_out_of_mis_cased_downgrade(): + pb2 = pytest.importorskip("a2a.types.a2a_pb2") + card = pb2.AgentCard( + name="langgraph", + supported_interfaces=[ + pb2.AgentInterface(url="http://a/", protocol_binding="jsonrpc", protocol_version="1.0"), + ], + ) + + normalized = normalize_agent_card_interfaces(card, protocol_version="1.0") + + assert normalized.supported_interfaces[0].protocol_binding == "JSONRPC" + assert normalized.supported_interfaces[0].protocol_version == "1.0" + assert card.supported_interfaces[0].protocol_binding == "jsonrpc" + assert card.supported_interfaces[0].protocol_version == "1.0" + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("0.3", "0.3"), + (" 0.3 ", "0.3"), + ("0.3.0", "0.3"), + (0.3, "0.3"), + (1.0, "1.0"), + ("1.0.0", "1.0"), + (None, None), + (True, None), + (1, None), + ("2.0", None), + ("abc", None), + ("", None), + (["0.3"], None), + ], +) +def test_resolve_a2a_protocol_version(value, expected): + assert resolve_a2a_protocol_version(value) == expected + + _FOUNDRY_BASE_URL: Final = "https://foundry.example.com/a2a" _FOUNDRY_CARD_JSON: Final = { diff --git a/tests/unit/a2a_protocol/test_completion_bridge_streaming.py b/tests/unit/a2a_protocol/test_completion_bridge_streaming.py index 5ed2cf0aeb6..cf92aaf90d6 100644 --- a/tests/unit/a2a_protocol/test_completion_bridge_streaming.py +++ b/tests/unit/a2a_protocol/test_completion_bridge_streaming.py @@ -356,6 +356,7 @@ async def test_handle_streaming_keeps_agent_card_path_out_of_the_completion_call "custom_llm_provider": "langgraph", "model": "agent", "agent_card_path": "agentCard/v1.0", + "a2a_protocol_version": "0.3", }, api_base="http://localhost:2024", ) @@ -363,6 +364,7 @@ async def test_handle_streaming_keeps_agent_card_path_out_of_the_completion_call assert len(events) == 4 assert "agent_card_path" not in mock_acompletion.call_args.kwargs + assert "a2a_protocol_version" not in mock_acompletion.call_args.kwargs AGENTCORE_ARN = "arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/unit-agent-abc123" diff --git a/tests/unit/a2a_protocol/test_main.py b/tests/unit/a2a_protocol/test_main.py index 68368ddb77a..bab14dd26d6 100644 --- a/tests/unit/a2a_protocol/test_main.py +++ b/tests/unit/a2a_protocol/test_main.py @@ -1,6 +1,7 @@ """Tests for litellm/a2a_protocol/main.py non-streaming send behavior.""" import asyncio +import json import httpx import pytest @@ -23,6 +24,7 @@ from litellm.a2a_protocol.main import ( asend_message, create_a2a_client, ) +from litellm.a2a_protocol.exceptions import A2AError from litellm.caching.llm_caching_handler import LLMClientCache from litellm.constants import DEFAULT_A2A_AGENT_TIMEOUT from litellm.llms.custom_httpx.http_handler import ( @@ -222,6 +224,77 @@ _LANGGRAPH_TASK_REPLY = { } +_LANGGRAPH_STREAM_EVENTS = [ + { + "jsonrpc": "2.0", + "id": "reply", + "result": { + "kind": "task", + "id": "run-1:task-1", + "contextId": "thread-1", + "history": [ + { + "kind": "message", + "role": "user", + "parts": [{"kind": "text", "text": "hi"}], + "messageId": "m-user", + "taskId": "run-1:task-1", + "contextId": "thread-1", + } + ], + "status": {"state": "submitted"}, + }, + }, + { + "jsonrpc": "2.0", + "id": "reply", + "result": { + "kind": "status-update", + "taskId": "run-1:task-1", + "contextId": "thread-1", + "status": { + "state": "working", + "message": { + "kind": "message", + "role": "agent", + "parts": [{"kind": "text", "text": "langgraph echo: hi"}], + "messageId": "m-agent", + "taskId": "run-1:task-1", + "contextId": "thread-1", + }, + }, + "final": False, + }, + }, + { + "jsonrpc": "2.0", + "id": "reply", + "result": { + "kind": "artifact-update", + "taskId": "run-1:task-1", + "contextId": "thread-1", + "artifact": { + "artifactId": "art-1", + "name": "Assistant Response", + "parts": [{"kind": "text", "text": "langgraph echo: hi"}], + }, + "lastChunk": True, + }, + }, + { + "jsonrpc": "2.0", + "id": "reply", + "result": { + "kind": "status-update", + "taskId": "run-1:task-1", + "contextId": "thread-1", + "status": {"state": "completed"}, + "final": True, + }, + }, +] + + _LOWERCASE_BINDING_CARD = { "name": "langgraph-agent", "version": "1.0.0", @@ -232,16 +305,28 @@ _LOWERCASE_BINDING_CARD = { "supportedInterfaces": [{"url": "http://127.0.0.1:9/", "protocolBinding": "jsonrpc", "protocolVersion": "1.0"}], } +_UPPERCASE_BINDING_CARD = { + "name": "langgraph-agent", + "version": "1.0.0", + "capabilities": {"streaming": True}, + "defaultInputModes": ["text/plain"], + "defaultOutputModes": ["text/plain"], + "skills": [], + "supportedInterfaces": [{"url": "http://127.0.0.1:9/", "protocolBinding": "JSONRPC", "protocolVersion": "1.0"}], +} + class _RequestRecorder: """Records the headers httpx put on the wire, per outbound request.""" - def __init__(self, card=_AGENT_CARD, rpc_reply=_RPC_REPLY): + def __init__(self, card=_AGENT_CARD, rpc_reply=_RPC_REPLY, rpc_stream_events=None): self.card = card self.rpc_reply = rpc_reply + self.rpc_stream_events = rpc_stream_events self.card_requests = [] self.card_urls = [] self.rpc_requests = [] + self.rpc_bodies = [] self.client = None def __call__(self, request: httpx.Request) -> httpx.Response: @@ -251,6 +336,10 @@ class _RequestRecorder: self.card_urls.append(str(request.url)) return httpx.Response(200, json=self.card) self.rpc_requests.append(headers) + self.rpc_bodies.append(json.loads(request.content)) + if self.rpc_stream_events is not None: + body = "".join(f"data: {json.dumps(event)}\n\n" for event in self.rpc_stream_events) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body) return httpx.Response(200, json=self.rpc_reply) @@ -259,7 +348,10 @@ def _a2a_client_cache_key(timeout: float, provider: str = httpxSpecialProvider.A async def _seed_shared_a2a_client( - card=_AGENT_CARD, rpc_reply=_RPC_REPLY, provider: str = httpxSpecialProvider.A2AProvider + card=_AGENT_CARD, + rpc_reply=_RPC_REPLY, + rpc_stream_events=None, + provider: str = httpxSpecialProvider.A2AProvider, ) -> _RequestRecorder: """Put the one A2A client the cache will hand out behind a mock transport. @@ -267,7 +359,7 @@ async def _seed_shared_a2a_client( it. The injected client is a real httpx.AsyncClient, so the merge of per-request headers over client defaults, which is what these tests are about, stays real. """ - recorder = _RequestRecorder(card=card, rpc_reply=rpc_reply) + recorder = _RequestRecorder(card=card, rpc_reply=rpc_reply, rpc_stream_events=rpc_stream_events) handler = AsyncHTTPHandler(timeout=DEFAULT_A2A_AGENT_TIMEOUT) owned_client = handler.client handler.client = httpx.AsyncClient(transport=httpx.MockTransport(recorder)) @@ -394,6 +486,125 @@ async def test_lowercase_protocol_binding_card_round_trips_the_langgraph_dialect assert interface.protocol_version == "0.3" +@pytest.mark.asyncio +async def test_uppercase_binding_card_without_override_fails_with_actionable_hint(isolated_client_cache): + await _seed_shared_a2a_client(card=_UPPERCASE_BINDING_CARD, rpc_reply=_LANGGRAPH_TASK_REPLY) + + with pytest.raises(A2AError) as error: + await asend_message(request=_send_request("uppercase-default"), api_base="http://127.0.0.1:9") + + assert 'has no field named "kind"' in str(error.value) + assert "a2a_protocol_version" in str(error.value) + + +@pytest.mark.parametrize("card", [_LOWERCASE_BINDING_CARD, _UPPERCASE_BINDING_CARD]) +@pytest.mark.asyncio +async def test_protocol_version_override_round_trips_the_langgraph_dialect(card, isolated_client_cache): + recorder = await _seed_shared_a2a_client(card=card, rpc_reply=_LANGGRAPH_TASK_REPLY) + + a2a_client = await create_a2a_client(base_url="http://127.0.0.1:9", protocol_version="0.3") + response = await _send_message(a2a_client, _send_request("override-round-trip")) + + assert type(response.root.result).__name__ == "Task" + assert response.root.result.artifacts[0].parts[0].root.text == "langgraph echo: hi" + assert a2a_client._litellm_agent_card.supported_interfaces[0].protocol_version == "0.3" + assert recorder.rpc_bodies[-1]["method"] == "message/send" + + +@pytest.mark.parametrize("card", [_LOWERCASE_BINDING_CARD, _UPPERCASE_BINDING_CARD]) +@pytest.mark.asyncio +async def test_protocol_version_override_streams_the_langgraph_dialect(card, isolated_client_cache): + recorder = await _seed_shared_a2a_client(card=card, rpc_stream_events=_LANGGRAPH_STREAM_EVENTS) + + a2a_client = await create_a2a_client( + base_url="http://127.0.0.1:9", + streaming=True, + protocol_version="0.3", + ) + request = SendStreamingMessageRequest( + id="reply", + params=MessageSendParams( + message={"messageId": "m-user", "role": "user", "parts": [{"kind": "text", "text": "hi"}]} + ), + ) + streamed = [chunk async for chunk in _stream_messages(a2a_client, request)] + + artifact_event = next(chunk.root.result for chunk in streamed if chunk.root.result.kind == "artifact-update") + completed_status = next( + chunk.root.result for chunk in streamed if chunk.root.result.kind == "status-update" and chunk.root.result.final + ) + assert artifact_event.artifact.parts[0].root.text == "langgraph echo: hi" + assert completed_status.status.state.value == "completed" + assert recorder.rpc_bodies[-1]["method"] == "message/stream" + + +@pytest.mark.asyncio +async def test_uppercase_binding_card_stream_without_override_fails(isolated_client_cache): + await _seed_shared_a2a_client(card=_UPPERCASE_BINDING_CARD, rpc_stream_events=_LANGGRAPH_STREAM_EVENTS) + + a2a_client = await create_a2a_client(base_url="http://127.0.0.1:9", streaming=True) + request = SendStreamingMessageRequest( + id="reply", + params=MessageSendParams( + message={"messageId": "m-user", "role": "user", "parts": [{"kind": "text", "text": "hi"}]} + ), + ) + + with pytest.raises(Exception, match='has no field named "kind"'): + async for _ in _stream_messages(a2a_client, request): + pass + + +@pytest.mark.asyncio +async def test_asend_message_resolves_float_protocol_version_from_agent_params(isolated_client_cache): + recorder = await _seed_shared_a2a_client(card=_UPPERCASE_BINDING_CARD, rpc_reply=_LANGGRAPH_TASK_REPLY) + + response = await asend_message( + request=_send_request("float-protocol-override"), + api_base="http://127.0.0.1:9", + litellm_params={"a2a_protocol_version": 0.3}, + ) + + assert response.result["artifacts"][0]["parts"][0]["text"] == "langgraph echo: hi" + assert recorder.rpc_bodies[-1]["method"] == "message/send" + + +@pytest.mark.parametrize( + ("litellm_params", "expected_protocol_version"), + [({"a2a_protocol_version": "0.3"}, "0.3"), (None, None)], +) +@pytest.mark.asyncio +async def test_streaming_passes_agent_protocol_version_to_client(litellm_params, expected_protocol_version): + from unittest.mock import AsyncMock, patch + + from litellm.a2a_protocol import main as a2a_main + + request = SendStreamingMessageRequest( + id="stream-protocol-version", + params=MessageSendParams( + message={"messageId": "m1", "role": "user", "parts": [{"kind": "text", "text": "hi"}]} + ), + ) + captured = {} + + async def _capture(*, base_url, extra_headers=None, streaming=False, protocol_version=None, **_): + captured["protocol_version"] = protocol_version + raise RuntimeError("stop") + + with ( + patch.object(a2a_main, "create_a2a_client", new=AsyncMock(side_effect=_capture)), + pytest.raises(RuntimeError, match="stop"), + ): + async for _ in a2a_main.asend_message_streaming( + request=request, + api_base="http://upstream.local", + litellm_params=litellm_params, + ): + pass + + assert captured["protocol_version"] == expected_protocol_version + + @pytest.mark.asyncio async def test_agent_card_fetch_carries_the_callers_headers(isolated_client_cache): """Agent cards can sit behind the same auth as the agent, so the card fetch must stay @@ -585,7 +796,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, relative_card_path=None + base_url, timeout=60.0, extra_headers=None, streaming=False, relative_card_path=None, **_ ): return MockA2AClient()