test: add integration tests for ADK's A2A client-server interaction

PiperOrigin-RevId: 888114589

Author: google-genai-bot

tests/unittests/a2a/integration/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A2A integration tests package."""

tests/unittests/a2a/integration/client.py

@@ -0,0 +1,88 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A2A Client for integration tests."""
+
+from a2a.client.client import ClientConfig as A2AClientConfig
+from a2a.client.client_factory import ClientFactory as A2AClientFactory
+from a2a.extensions.common import HTTP_EXTENSION_HEADER
+from a2a.types import TransportProtocol as A2ATransport
+from google.adk.a2a.agent.interceptors.new_integration_extension import _NEW_A2A_ADK_INTEGRATION_EXTENSION
+from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
+import httpx
+
+from .server import agent_card
+
+
+def create_client(app, streaming: bool = False) -> RemoteA2aAgent:
+  """Creates a RemoteA2aAgent connected to the provided FastAPI app.
+
+  Args:
+    app: The FastAPI application (server) to connect to.
+    streaming: Whether to enable streaming mode in the client.
+
+  Returns:
+    A RemoteA2aAgent instance.
+  """
+
+  client = httpx.AsyncClient(
+      transport=httpx.ASGITransport(app=app), base_url="http://test"
+  )
+
+  client_config = A2AClientConfig(
+      httpx_client=client,
+      streaming=streaming,
+      polling=False,
+      supported_transports=[A2ATransport.jsonrpc],
+  )
+  factory = A2AClientFactory(config=client_config)
+
+  # use_legacy=False forces the new implementation
+  agent = RemoteA2aAgent(
+      name="remote_agent",
+      agent_card=agent_card,
+      a2a_client_factory=factory,
+      use_legacy=False,
+  )
+
+  return agent
+
+
+def create_a2a_client(app, streaming: bool = False):
+  """Creates a bare A2A Client connected to the provided FastAPI app.
+
+  This is in contrast to create_client, which wraps the a2a_client into a
+  RemoteA2aAgent for the standard runner framework ecosystem execution.
+
+  Args:
+    app: The FastAPI application (server) to connect to.
+    streaming: Whether to enable streaming mode in the client.
+
+  Returns:
+    An A2A Client instance.
+  """
+  client = httpx.AsyncClient(
+      transport=httpx.ASGITransport(app=app),
+      base_url="http://test",
+      headers={HTTP_EXTENSION_HEADER: _NEW_A2A_ADK_INTEGRATION_EXTENSION},
+  )
+
+  client_config = A2AClientConfig(
+      httpx_client=client,
+      streaming=streaming,
+      polling=False,
+      supported_transports=[A2ATransport.jsonrpc],
+  )
+  factory = A2AClientFactory(config=client_config)
+  return factory.create(agent_card)

tests/unittests/a2a/integration/server.py

@@ -0,0 +1,84 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A2A Server for integration tests."""
+
+from unittest.mock import Mock
+
+from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPIApplication
+from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler
+from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore
+from a2a.types import AgentCapabilities
+from a2a.types import AgentCard
+from a2a.types import AgentSkill
+from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
+from google.adk.agents.base_agent import BaseAgent
+from google.adk.runners import Runner
+from google.adk.sessions.in_memory_session_service import InMemorySessionService
+
+
+class FakeRunner(Runner):
+  """A Fake Runner that delegates run_async to a provided function."""
+
+  def __init__(self, run_async_fn):
+    agent = Mock(spec=BaseAgent)
+    agent.name = "FakeAgent"
+
+    session_service = InMemorySessionService()
+    super().__init__(
+        app_name="FakeApp",
+        agent=agent,
+        session_service=session_service,
+    )
+    self.run_async_fn = run_async_fn
+
+  async def run_async(self, **kwargs):
+    async for event in self.run_async_fn(**kwargs):
+      yield event
+
+
+agent_card = AgentCard(
+    name="remote_agent",
+    url="http://test",
+    description="A fun fact generator agent",
+    capabilities=AgentCapabilities(
+        streaming=True,
+        extensions=[{"uri": "https://a2a-adk/a2a-extension/new-integration"}],
+    ),
+    version="0.0.1",
+    default_input_modes=["text/plain"],
+    default_output_modes=["text/plain"],
+    skills=[],
+)
+
+
+def create_server_app(run_async_fn):
+  """Creates an A2A FastAPI application with a mocked runner.
+
+  Args:
+    run_async_fn: A generator function that takes **kwargs and yields Event
+      objects.
+
+  Returns:
+    A FastAPI application instance.
+  """
+  runner = FakeRunner(run_async_fn)
+  executor = A2aAgentExecutor(runner=runner)
+  task_store = InMemoryTaskStore()
+  handler = DefaultRequestHandler(
+      agent_executor=executor, task_store=task_store
+  )
+
+  app = A2AFastAPIApplication(agent_card=agent_card, http_handler=handler)
+  return app.build()