refactor: switch cli and agent_registry to the a2a-sdk compatibility shim

make cli/fast_api.py and agent_registry/agent_registry.py compatible with both
a2a 0.3 and 1.0 by calling _compat helpers instead of a2a.* directly

PiperOrigin-RevId: 944988024

Author: google-genai-bot

src/google/adk/cli/fast_api.py

@@ -679,12 +679,9 @@ async def _a2a_lifespan(app_instance: FastAPI):
   _register_builder_endpoints(app, web, agents_dir)
 
   if a2a and a2a_task_store is not None:
-    from a2a.server.apps import A2AStarletteApplication
-    from a2a.server.request_handlers import DefaultRequestHandler
     from a2a.server.tasks import InMemoryPushNotificationConfigStore
-    from a2a.types import AgentCard
-    from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH
 
+    from ..a2a import _compat
     from ..a2a.executor.a2a_agent_executor import A2aAgentExecutor
 
     # locate all a2a agent apps in the agents directory
@@ -720,29 +717,19 @@ async def _get_a2a_runner_async() -> Runner:
 
           push_config_store = InMemoryPushNotificationConfigStore()
 
-          request_handler = DefaultRequestHandler(
-              agent_executor=agent_executor,
-              task_store=a2a_task_store,
-              push_config_store=push_config_store,
-          )
-
           with (p / "agent.json").open("r", encoding="utf-8") as f:
             data = json.load(f)
-            agent_card = AgentCard(**data)
+            agent_card = _compat.parse_agent_card(data)
 
-          a2a_app = A2AStarletteApplication(
+          _compat.attach_a2a_routes_to_app(
+              app,
               agent_card=agent_card,
-              http_handler=request_handler,
-          )
-
-          routes = a2a_app.routes(
-              rpc_url=f"/a2a/{app_name}",
-              agent_card_url=f"/a2a/{app_name}{AGENT_CARD_WELL_KNOWN_PATH}",
+              agent_executor=agent_executor,
+              task_store=a2a_task_store,
+              push_config_store=push_config_store,
+              prefix=f"/a2a/{app_name}",
           )
 
-          for new_route in routes:
-            app.router.routes.append(new_route)
-
           logger.info("Successfully configured A2A agent: %s", app_name)
 
         except Exception as e:
@@ -756,7 +743,6 @@ async def _get_a2a_runner_async() -> Runner:
       )
 
     import inspect
-    import json
 
     from google.adk.agents import Agent
     import google.auth

src/google/adk/integrations/agent_registry/agent_registry.py

@@ -48,10 +48,8 @@
 
 # pylint: disable=g-import-not-at-top
 try:
-  from a2a.types import AgentCapabilities
-  from a2a.types import AgentCard
   from a2a.types import AgentSkill
-  from a2a.types import TransportProtocol as A2ATransport
+  from google.adk.a2a import _compat
   from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
 except ImportError as e:
   raise ImportError(
@@ -66,9 +64,9 @@
 AGENT_REGISTRY_MTLS_BASE_URL = "https://agentregistry.mtls.googleapis.com/v1"
 
 _TRANSPORT_MAPPING = {
-    "HTTP_JSON": A2ATransport.http_json,
-    "JSONRPC": A2ATransport.jsonrpc,
-    "GRPC": A2ATransport.grpc,
+    "HTTP_JSON": _compat.TP_HTTP_JSON,
+    "JSONRPC": _compat.TP_JSONRPC,
+    "GRPC": _compat.TP_GRPC,
 }
 
 
@@ -273,8 +271,8 @@ def _get_connection_uri(
       self,
       resource_details: Mapping[str, Any],
       protocol_type: _ProtocolType | None = None,
-      protocol_binding: A2ATransport | None = None,
-  ) -> str | None:
+      protocol_binding: _compat.TransportProtocol | None = None,
+  ) -> tuple[Any, Any, Any]:
     """Extracts the first matching URI based on type and binding filters."""
     protocols = list(resource_details.get("protocols", []))
     if "interfaces" in resource_details:
@@ -354,11 +352,11 @@ def get_mcp_toolset(
       mcp_server_id = None
 
     endpoint_uri, _, _ = self._get_connection_uri(
-        server_details, protocol_binding=A2ATransport.jsonrpc
+        server_details, protocol_binding=_compat.TP_JSONRPC
     )
     if not endpoint_uri:
       endpoint_uri, _, _ = self._get_connection_uri(
-          server_details, protocol_binding=A2ATransport.http_json
+          server_details, protocol_binding=_compat.TP_HTTP_JSON
       )
     if not endpoint_uri:
       raise ValueError(
@@ -492,7 +490,7 @@ def get_remote_a2a_agent(
     card = agent_info.get("card", {})
     card_content = card.get("content")
     if card.get("type") == "A2A_AGENT_CARD" and card_content:
-      agent_card = AgentCard(**card_content)
+      agent_card = _compat.parse_agent_card(card_content)
       # Clean the name to be a valid identifier
       name = self._clean_name(agent_card.name)
 
@@ -525,17 +523,17 @@ def get_remote_a2a_agent(
           )
       )
 
-    agent_card = AgentCard(
+    binding = protocol_binding or _compat.TP_HTTP_JSON
+    agent_card = _compat.build_agent_card(
         name=name,
         description=description,
         version=version,
-        preferredTransport=protocol_binding or A2ATransport.http_json,
-        protocolVersion=protocol_version or "0.3.0",
         url=url,
+        protocol_binding=getattr(binding, "value", binding),
+        protocol_version=protocol_version,
         skills=skills,
-        capabilities=AgentCapabilities(streaming=False, polling=False),
-        defaultInputModes=["text"],
-        defaultOutputModes=["text"],
+        default_input_modes=["text"],
+        default_output_modes=["text"],
     )
 
     return RemoteA2aAgent(

tests/unittests/cli/test_fast_api.py

@@ -22,16 +22,15 @@
 from typing import Any
 from typing import Optional
 from unittest.mock import AsyncMock
-from unittest.mock import call
 from unittest.mock import MagicMock
 from unittest.mock import patch
 from urllib.parse import quote
 
 from fastapi.testclient import TestClient
+from google.adk.a2a import _compat
 from google.adk.agents.base_agent import BaseAgent
 from google.adk.agents.llm_agent import LlmAgent
 from google.adk.agents.run_config import RunConfig
-from google.adk.apps.app import App
 from google.adk.artifacts.base_artifact_service import ArtifactVersion
 from google.adk.cli import fast_api as fast_api_module
 from google.adk.cli.fast_api import get_fast_api_app
@@ -46,7 +45,6 @@
 from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin
 from google.adk.runners import Runner
 from google.adk.sessions.in_memory_session_service import InMemorySessionService
-from google.adk.sessions.session import Session
 from google.genai import types
 from pydantic import BaseModel
 import pytest
@@ -841,8 +839,11 @@ def temp_agents_dir_with_a2a():
         "name": "test_a2a_agent",
         "description": "Test A2A agent",
         "version": "1.0.0",
-        "author": "test",
-        "capabilities": ["text"],
+        "url": "http://localhost:8000/a2a/test_a2a_agent",
+        "capabilities": {},
+        "defaultInputModes": ["text/plain"],
+        "defaultOutputModes": ["text/plain"],
+        "skills": [],
     }
 
     with open(agent_dir / "agent.json", "w") as f:
@@ -2041,6 +2042,12 @@ def test_openapi_json_schema_accessible(test_app):
   logger.info("OpenAPI /openapi.json endpoint is accessible")
 
 
+@pytest.mark.skipif(
+    _compat.IS_A2A_V1,
+    reason=(
+        "0.3.x-only: mocks server.apps.A2AStarletteApplication (gone in 1.x)"
+    ),
+)
 def test_a2a_agent_discovery(test_app_with_a2a):
   """Test that A2A agents are properly discovered and configured."""
   # This test mainly verifies that the A2A setup doesn't break the app
@@ -2049,6 +2056,12 @@ def test_a2a_agent_discovery(test_app_with_a2a):
   logger.info("A2A agent discovery test passed")
 
 
+@pytest.mark.skipif(
+    _compat.IS_A2A_V1,
+    reason=(
+        "0.3.x-only: mocks server.apps.A2AStarletteApplication (gone in 1.x)"
+    ),
+)
 def test_a2a_request_handler_uses_push_config_store(
     mock_session_service,
     mock_artifact_service,
@@ -2131,6 +2144,12 @@ def test_a2a_request_handler_uses_push_config_store(
     )
 
 
+@pytest.mark.skipif(
+    _compat.IS_A2A_V1,
+    reason=(
+        "0.3.x-only: mocks server.apps.A2AStarletteApplication (gone in 1.x)"
+    ),
+)
 def test_a2a_request_handler_uses_task_store_uri(
     mock_session_service,
     mock_artifact_service,
@@ -2211,6 +2230,12 @@ def test_a2a_request_handler_uses_task_store_uri(
     assert call_kwargs["task_store"] is custom_task_store
 
 
+@pytest.mark.skipif(
+    _compat.IS_A2A_V1,
+    reason=(
+        "0.3.x-only: mocks server.apps.A2AStarletteApplication (gone in 1.x)"
+    ),
+)
 def test_a2a_task_store_engine_disposed_on_shutdown(
     mock_session_service,
     mock_artifact_service,
@@ -2292,6 +2317,12 @@ def test_a2a_task_store_engine_disposed_on_shutdown(
     mock_engine.dispose.assert_awaited_once()
 
 
+@pytest.mark.skipif(
+    _compat.IS_A2A_V1,
+    reason=(
+        "0.3.x-only: mocks server.apps.A2AStarletteApplication (gone in 1.x)"
+    ),
+)
 def test_a2a_in_memory_task_store_no_engine_dispose(
     mock_session_service,
     mock_artifact_service,