From 5e68a003476d0a2ecce31036ea25597cf0a549d7 Mon Sep 17 00:00:00 2001 From: yassin Date: Tue, 21 Jul 2026 00:10:49 +0000 Subject: [PATCH] test(e2e): add live A2A agent e2e suite Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- tests/e2e/CLAUDE.md | 1 + tests/e2e/a2a/a2a_client.py | 217 +++++++++++++++++++++++++ tests/e2e/a2a/conftest.py | 17 ++ tests/e2e/a2a/test_a2a_agent_e2e.py | 139 ++++++++++++++++ tests/e2e/coverage_registry/other.yaml | 6 + 5 files changed, 380 insertions(+) create mode 100644 tests/e2e/a2a/a2a_client.py create mode 100644 tests/e2e/a2a/conftest.py create mode 100644 tests/e2e/a2a/test_a2a_agent_e2e.py diff --git a/tests/e2e/CLAUDE.md b/tests/e2e/CLAUDE.md index 47f3c74d7f1..c35fb2fa435 100644 --- a/tests/e2e/CLAUDE.md +++ b/tests/e2e/CLAUDE.md @@ -13,6 +13,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family - `realtime/` - realtime websocket sessions, including the pipecat audio path - `quota_management/` - quota enforcement and accounting, one subfolder per behavior: `ratelimit/` (rpm/tpm blocks, window reset, pacing headers on live traffic), `budgets/` (budget definition, enforcement, and reset windows: key, team, tag, soft, multi-window), and `spend_tracking/` (spend logging and cost attribution on `/spend/*`) - `management/` - key/team/user/organization management routes: create/update/delete persistence via the info routes, team membership, and llm-only-key route denials (API surface; not Playwright) +- `a2a/` - the A2A (agent-to-agent) surface: admin registration via `/v1/agents`, proxy-fronted card discovery at `/.well-known/agent-card.json`, and JSON-RPC `message/send` invocation, driving agents backed by the litellm completion bridge (a real provider) and asserting protocol-version normalization (0.3 vs 1.0) - `mcp/` - the MCP server surface over api_key auth against the real Datadog remote MCP server only (see "MCP suite: real Datadog only" below) - `logging/` - logging-integration delivery (datadog and friends) - `security/` - secret handling and log-leak protection diff --git a/tests/e2e/a2a/a2a_client.py b/tests/e2e/a2a/a2a_client.py new file mode 100644 index 00000000000..5274ec15383 --- /dev/null +++ b/tests/e2e/a2a/a2a_client.py @@ -0,0 +1,217 @@ +"""Client for the proxy's A2A (agent-to-agent) surface. + +An A2A agent is registered admin-side via POST /v1/agents with an agent card and +litellm_params; the proxy fronts it at /a2a/{id}, serving a proxy-owned agent card +at /.well-known/agent-card.json and accepting A2A JSON-RPC calls at /a2a/{id}. This +suite registers agents backed by the litellm_completion_bridge (custom_llm_provider ++ model), so message/send runs a real provider completion and comes back in the +agent's pinned A2A protocol version. The A2A request/response models are co-located +here because only this suite uses them. +""" + +from __future__ import annotations + +import warnings +from dataclasses import dataclass + +from pydantic import BaseModel, ConfigDict, Field + +from e2e_http import NoBody, Result, is_ok +from proxy_client import ProxyClient + + +class A2ACapabilities(BaseModel): + streaming: bool | None = None + push_notifications: bool | None = Field(default=None, serialization_alias="pushNotifications") + + +class A2ASkill(BaseModel): + id: str + name: str + description: str + tags: list[str] + + +class AgentCardParams(BaseModel): + """The upstream agent card an admin registers. `protocolVersion` is the field the + proxy validates against SUPPORTED_A2A_PROTOCOL_VERSIONS on registration.""" + + protocol_version: str = Field(serialization_alias="protocolVersion") + name: str + description: str + version: str + capabilities: A2ACapabilities = A2ACapabilities() + skills: list[A2ASkill] + default_input_modes: list[str] = Field(default=["text"], serialization_alias="defaultInputModes") + default_output_modes: list[str] = Field(default=["text"], serialization_alias="defaultOutputModes") + + +class A2ABridgeParams(BaseModel): + """litellm_params that route the agent through the completion bridge: an A2A + message/send is transformed into a litellm.acompletion against this provider.""" + + model_config = ConfigDict(protected_namespaces=()) + + custom_llm_provider: str + model: str + + +class AgentRegisterBody(BaseModel): + agent_name: str + agent_card_params: AgentCardParams + litellm_params: A2ABridgeParams + + +class A2ASecurityScheme(BaseModel): + type: str + scheme: str + + +class A2AInterface(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + url: str + protocol_version: str | None = Field(default=None, alias="protocolVersion") + + +class ServedAgentCard(BaseModel): + """The proxy-owned card, either nested under a registration response's + `agent_card_params` or served raw at /.well-known/agent-card.json. The proxy + rewrites `url`/`supportedInterfaces` to itself and replaces the security scheme + with its own virtual-key bearer scheme.""" + + model_config = ConfigDict(populate_by_name=True) + + protocol_version: str = Field(alias="protocolVersion") + name: str + url: str | None = None + security_schemes: dict[str, A2ASecurityScheme] | None = Field(default=None, alias="securitySchemes") + security: list[dict[str, list[str]]] | None = None + supported_interfaces: list[A2AInterface] | None = Field(default=None, alias="supportedInterfaces") + + +class AgentResponse(BaseModel): + agent_id: str + agent_name: str + agent_card_params: ServedAgentCard + + +class A2ATextPart(BaseModel): + kind: str = "text" + text: str + + +class A2AOutboundMessage(BaseModel): + role: str = "user" + parts: list[A2ATextPart] + message_id: str = Field(serialization_alias="messageId") + + +class A2AMessageSendParams(BaseModel): + message: A2AOutboundMessage + + +class A2AJsonRpcRequest(BaseModel): + jsonrpc: str = "2.0" + id: str + method: str = "message/send" + params: A2AMessageSendParams + + +class A2AResponsePart(BaseModel): + kind: str | None = None + text: str | None = None + + +class A2AResponseMessage(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + message_id: str | None = Field(default=None, alias="messageId") + role: str | None = None + parts: list[A2AResponsePart] = [] + + +class A2AResult(BaseModel): + """A message/send result. In 0.3 the message fields sit directly on the result + (`kind`/`role`/`parts`); in 1.0 they are nested under `message`. `text` reads the + agent's reply from whichever shape the served version produced.""" + + model_config = ConfigDict(populate_by_name=True) + + kind: str | None = None + role: str | None = None + message_id: str | None = Field(default=None, alias="messageId") + parts: list[A2AResponsePart] = [] + message: A2AResponseMessage | None = None + + @property + def text(self) -> str: + parts = self.message.parts if self.message is not None else self.parts + return "".join(part.text or "" for part in parts) + + @property + def is_nested_v1_shape(self) -> bool: + return self.message is not None + + +class A2AError(BaseModel): + code: int + message: str + + +class A2AResponse(BaseModel): + jsonrpc: str + id: str | None = None + result: A2AResult | None = None + error: A2AError | None = None + + +@dataclass(frozen=True, slots=True) +class A2AClient: + proxy: ProxyClient + + def register_agent(self, body: AgentRegisterBody) -> Result[AgentResponse]: + return self.proxy.transport.post( + "/v1/agents", + headers=self.proxy.transport.master, + json=body, + response_type=AgentResponse, + ) + + def get_agent(self, agent_id: str) -> Result[AgentResponse]: + return self.proxy.transport.get( + f"/v1/agents/{agent_id}", + headers=self.proxy.transport.master, + params=NoBody(), + response_type=AgentResponse, + ) + + def delete_agent(self, agent_id: str) -> None: + result = self.proxy.transport.delete( + f"/v1/agents/{agent_id}", + headers=self.proxy.transport.master, + json=NoBody(), + response_type=NoBody, + ) + if not is_ok(result): + warnings.warn(f"delete_agent({agent_id!r}) failed: {result}", stacklevel=2) + + def agent_card(self, agent_id: str, key: str) -> Result[ServedAgentCard]: + return self.proxy.transport.get( + f"/a2a/{agent_id}/.well-known/agent-card.json", + headers=self.proxy.transport.bearer(key), + params=NoBody(), + response_type=ServedAgentCard, + ) + + def send_message(self, agent_id: str, key: str, body: A2AJsonRpcRequest) -> Result[A2AResponse]: + return self.proxy.transport.post( + f"/a2a/{agent_id}", + headers=self.proxy.transport.bearer(key), + json=body, + response_type=A2AResponse, + ) + + +def build_a2a_client(proxy: ProxyClient) -> A2AClient: + return A2AClient(proxy=proxy) diff --git a/tests/e2e/a2a/conftest.py b/tests/e2e/a2a/conftest.py new file mode 100644 index 00000000000..93f3b56c8f7 --- /dev/null +++ b/tests/e2e/a2a/conftest.py @@ -0,0 +1,17 @@ +"""A2A suite's `client` fixture. + +The shared lifecycle (resources/scoped_key), proxy liveness gate, and e2e marker +live in the parent tests/e2e/conftest.py. A2AClient holds the shared ProxyClient, +so the `resources` fixture cleans up keys this suite creates; agents are torn down +via `resources.defer(...)` in each test. +""" + +import pytest + +from a2a_client import A2AClient, build_a2a_client +from proxy_client import ProxyClient + + +@pytest.fixture(scope="session") +def client(proxy: ProxyClient) -> A2AClient: + return build_a2a_client(proxy) diff --git a/tests/e2e/a2a/test_a2a_agent_e2e.py b/tests/e2e/a2a/test_a2a_agent_e2e.py new file mode 100644 index 00000000000..eb61ace238c --- /dev/null +++ b/tests/e2e/a2a/test_a2a_agent_e2e.py @@ -0,0 +1,139 @@ +"""A2A agents end to end, against a live proxy. + +An admin registers an agent whose card pins an A2A protocol version and whose +litellm_params route it through the completion bridge; a caller then discovers the +proxy-owned card and drives it over A2A JSON-RPC. These tests assert the recorded +state (the agent persists, a spend row lands) and the enforced behavior (the served +card points back at the proxy, message/send returns a real completion in the pinned +protocol version, and an unsupported version is refused at registration). +""" + +from __future__ import annotations + +import pytest + +from a2a_client import ( + A2ABridgeParams, + A2AClient, + A2AJsonRpcRequest, + A2AMessageSendParams, + A2AOutboundMessage, + A2ASkill, + A2ATextPart, + AgentCardParams, + AgentRegisterBody, + AgentResponse, +) +from e2e_config import unique_marker +from e2e_http import UnknownApiError, unwrap +from lifecycle import ResourceManager + +BRIDGE = A2ABridgeParams(custom_llm_provider="anthropic", model="claude-haiku-4-5") + +pytestmark = pytest.mark.e2e + + +def _register(client: A2AClient, resources: ResourceManager, protocol_version: str) -> AgentResponse: + marker = unique_marker() + body = AgentRegisterBody( + agent_name=f"e2e-a2a-{marker}", + agent_card_params=AgentCardParams( + protocol_version=protocol_version, + name=f"E2E A2A {marker}", + description="e2e agent backed by the litellm completion bridge", + version="1.0.0", + skills=[A2ASkill(id="chat", name="Chat", description="general chat", tags=["chat"])], + ), + litellm_params=BRIDGE, + ) + agent = unwrap(client.register_agent(body)) + resources.defer(lambda: client.delete_agent(agent.agent_id)) + return agent + + +def _ask(text: str) -> A2AJsonRpcRequest: + return A2AJsonRpcRequest( + id=f"e2e-{unique_marker()}", + params=A2AMessageSendParams( + message=A2AOutboundMessage(parts=[A2ATextPart(text=text)], message_id=unique_marker()) + ), + ) + + +class TestA2AAgentLifecycle: + @pytest.mark.covers("other.a2a.register.persists") + def test_register_persists(self, client: A2AClient, resources: ResourceManager) -> None: + agent = _register(client, resources, "0.3") + fetched = unwrap(client.get_agent(agent.agent_id)) + assert fetched.agent_id == agent.agent_id + assert fetched.agent_name == agent.agent_name + assert fetched.agent_card_params.protocol_version == "0.3" + + @pytest.mark.covers("other.a2a.discovery.proxy_fronted_card") + def test_discovery_card_is_proxy_fronted(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None: + agent = _register(client, resources, "0.3") + card = unwrap(client.agent_card(agent.agent_id, scoped_key)) + assert card.url is not None and card.url.endswith(f"/a2a/{agent.agent_id}") + assert card.security_schemes is not None + scheme = next(iter(card.security_schemes.values())) + assert scheme.scheme == "bearer" + assert card.supported_interfaces is not None + assert card.supported_interfaces[0].url == card.url + + @pytest.mark.covers("other.a2a.message_send.bridge_invokes") + def test_message_send_runs_completion_bridge(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None: + agent = _register(client, resources, "0.3") + request = _ask("Reply with exactly the word PONG and nothing else") + response = unwrap(client.send_message(agent.agent_id, scoped_key, request)) + assert response.error is None + assert response.result is not None + assert "PONG" in response.result.text.upper() + + rows = client.proxy.poll_logs_for_request_id(request.id) + assert rows, f"no spend log row landed for a2a request {request.id}" + assert rows[0].call_type == "asend_message" + assert rows[0].model == f"a2a_agent/{agent.agent_card_params.name}" + + @pytest.mark.covers("other.a2a.version.serves_pinned_0_3") + def test_pinned_v0_3_serves_flat_message_shape(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None: + agent = _register(client, resources, "0.3") + request = _ask("Say hi in one word") + result = unwrap(client.send_message(agent.agent_id, scoped_key, request)).result + assert result is not None + assert not result.is_nested_v1_shape + assert result.kind == "message" + assert result.role == "agent" + assert result.text != "" + + @pytest.mark.covers("other.a2a.version.serves_pinned_1_0") + def test_pinned_v1_0_serves_nested_message_shape(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None: + agent = _register(client, resources, "1.0") + request = _ask("Say hi in one word") + result = unwrap(client.send_message(agent.agent_id, scoped_key, request)).result + assert result is not None + assert result.is_nested_v1_shape + assert result.message is not None + assert result.message.role == "ROLE_AGENT" + assert result.text != "" + + @pytest.mark.covers("other.a2a.register.unsupported_version_rejected") + def test_unsupported_protocol_version_rejected(self, client: A2AClient, resources: ResourceManager) -> None: + marker = unique_marker() + body = AgentRegisterBody( + agent_name=f"e2e-a2a-bad-{marker}", + agent_card_params=AgentCardParams( + protocol_version="9.9", + name=f"E2E A2A bad {marker}", + description="unsupported version", + version="1.0.0", + skills=[A2ASkill(id="chat", name="Chat", description="c", tags=["chat"])], + ), + litellm_params=BRIDGE, + ) + result = client.register_agent(body) + match result: + case UnknownApiError(status_code=status, body=detail): + assert status == 400 + assert "protocolVersion" in detail + case _: + pytest.fail(f"expected 400 for unsupported protocolVersion, got {result}") diff --git a/tests/e2e/coverage_registry/other.yaml b/tests/e2e/coverage_registry/other.yaml index 6b183cbf9f3..f4d0120e085 100644 --- a/tests/e2e/coverage_registry/other.yaml +++ b/tests/e2e/coverage_registry/other.yaml @@ -28,3 +28,9 @@ - {id: other.config.overrides.audit_logged, module: other, tier: P1, area: config, assertions: [audit_logged], source: "config_override_endpoints.py:67-100", rationale: "Config override mutations audit-logged, values redacted"} - {id: other.key_mgmt.regenerate.grace_period_honored, module: other, tier: P1, area: auth, assertions: [grace_period_honored], source: "key_management_endpoints.py:4503-4560", rationale: "Old key valid during grace_period then revoked"} - {id: other.key_mgmt.spend_reset.resets_to_value, module: other, tier: P1, area: auth, assertions: [resets_to_value], source: "key_management_endpoints.py:4841", rationale: "reset_spend resets accumulated spend"} +- {id: other.a2a.register.persists, module: other, tier: P1, area: a2a, assertions: [persists], source: "agent_endpoints/endpoints.py:325-443", rationale: "POST /v1/agents registers an agent card; GET /v1/agents/{id} reads it back"} +- {id: other.a2a.register.unsupported_version_rejected, module: other, tier: P1, area: a2a, assertions: [unsupported_version_rejected], source: "agent_endpoints/endpoints.py _validate_protocol_version", rationale: "A card pinning a protocolVersion outside SUPPORTED_A2A_PROTOCOL_VERSIONS is refused with 400"} +- {id: other.a2a.discovery.proxy_fronted_card, module: other, tier: P1, area: a2a, assertions: [proxy_fronted_card], source: "agent_endpoints/a2a_endpoints.py get_agent_card", rationale: "/.well-known/agent-card.json serves the proxy url + supportedInterfaces and the LiteLLM virtual-key bearer scheme, not the upstream"} +- {id: other.a2a.message_send.bridge_invokes, module: other, tier: P1, area: a2a, assertions: [bridge_invokes], source: "a2a_protocol/litellm_completion_bridge/handler.py", rationale: "A2A message/send routes through the completion bridge to a real provider and logs an asend_message spend row"} +- {id: other.a2a.version.serves_pinned_0_3, module: other, tier: P1, area: a2a, assertions: [serves_pinned_0_3], source: "agent_endpoints/a2a_endpoints.py _served_version", rationale: "An agent pinning 0.3 returns the flat 0.3 message shape (parts on the result)"} +- {id: other.a2a.version.serves_pinned_1_0, module: other, tier: P1, area: a2a, assertions: [serves_pinned_1_0], source: "agent_endpoints/a2a_endpoints.py _served_version", rationale: "An agent pinning 1.0 returns the nested 1.0 message shape (result.message with ROLE_AGENT)"}