mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-15 04:19:08 +00:00
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 base5f969982fc: 33 red, 7 green. Headba709d3de5: 40/40 green twice. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: shivam <shivam@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Mrinal Chanshetty <mrinal@berri.ai> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
7dd53d6adc
commit
6b7d1928b8
12 changed files with 1190 additions and 16 deletions
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
283
tests/integration/compatibility/_a2a_peers.py
Normal file
283
tests/integration/compatibility/_a2a_peers.py
Normal file
|
|
@ -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,)
|
||||
)
|
||||
|
|
@ -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/<agent>`` 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
|
||||
|
|
@ -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]
|
||||
|
|
@ -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")
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue