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

@@ -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

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)

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)

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}")

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)"}