feat(compat): AgentCard backward compatibility helpers and tests

This commit implements the backwards compatibility helpers for exchanging
legacy v0.3 Agent Cards across the v1.0 protocol bounds.

Also resolves linting and validation constraints.

Author: bartek-w

src/a2a/client/card_resolver.py

@@ -6,12 +6,13 @@
 
 import httpx
 
-from google.protobuf.json_format import ParseDict, ParseError
+from google.protobuf.json_format import ParseError
 
 from a2a.client.errors import (
     A2AClientHTTPError,
     A2AClientJSONError,
 )
+from a2a.client.helpers import parse_agent_card
 from a2a.types.a2a_pb2 import (
     AgentCard,
 )
@@ -88,7 +89,7 @@ async def get_agent_card(
                 target_url,
                 agent_card_data,
             )
-            agent_card = ParseDict(agent_card_data, AgentCard())
+            agent_card = parse_agent_card(agent_card_data)
             if signature_verifier:
                 signature_verifier(agent_card)
         except httpx.HTTPStatusError as e:

src/a2a/client/helpers.py

@@ -1,8 +1,135 @@
 """Helper functions for the A2A client."""
 
+from typing import Any
 from uuid import uuid4
 
-from a2a.types.a2a_pb2 import Message, Part, Role
+from google.protobuf.json_format import ParseDict
+
+from a2a.types.a2a_pb2 import AgentCard, Message, Part, Role
+
+
+def parse_agent_card(agent_card_data: dict[str, Any]) -> AgentCard:
+    """Parse AgentCard JSON dictionary and handle backward compatibility."""
+    _handle_extended_card_compatibility(agent_card_data)
+    _handle_connection_fields_compatibility(agent_card_data)
+    _handle_security_compatibility(agent_card_data)
+
+    return ParseDict(agent_card_data, AgentCard(), ignore_unknown_fields=True)
+
+
+def _handle_extended_card_compatibility(
+    agent_card_data: dict[str, Any],
+) -> None:
+    """Map legacy supportsAuthenticatedExtendedCard to capabilities."""
+    supports_extended_card = agent_card_data.pop(
+        'supportsAuthenticatedExtendedCard', None
+    )
+    if supports_extended_card is None:
+        supports_extended_card = agent_card_data.pop(
+            'supports_authenticated_extended_card', None
+        )
+
+    if supports_extended_card:
+        capabilities = agent_card_data.setdefault('capabilities', {})
+        if (
+            'extendedAgentCard' not in capabilities
+            and 'extended_agent_card' not in capabilities
+        ):
+            capabilities['extendedAgentCard'] = True
+
+
+def _handle_connection_fields_compatibility(
+    agent_card_data: dict[str, Any],
+) -> None:
+    """Map legacy connection and transport fields to supportedInterfaces."""
+    main_url = agent_card_data.pop('url', None)
+
+    main_transport = agent_card_data.pop('preferredTransport', None)
+    if main_transport is None:
+        main_transport = agent_card_data.pop('preferred_transport', 'JSONRPC')
+
+    version = agent_card_data.pop('protocolVersion', None)
+    if version is None:
+        version = agent_card_data.pop('protocol_version', '0.3.0')
+
+    additional_interfaces = agent_card_data.pop('additionalInterfaces', None)
+    if additional_interfaces is None:
+        additional_interfaces = agent_card_data.pop('additional_interfaces', [])
+
+    if (
+        'supportedInterfaces' not in agent_card_data
+        and 'supported_interfaces' not in agent_card_data
+        and main_url
+    ):
+        supported_interfaces = []
+        supported_interfaces.append(
+            {
+                'url': main_url,
+                'protocolBinding': main_transport,
+                'protocolVersion': version,
+            }
+        )
+        supported_interfaces.extend(
+            {
+                'url': iface.get('url'),
+                'protocolBinding': iface.get('transport'),
+                'protocolVersion': version,
+            }
+            for iface in additional_interfaces
+        )
+        agent_card_data['supportedInterfaces'] = supported_interfaces
+
+
+def _map_legacy_security(sec_list: list[dict[str, list[str]]]) -> list[dict[str, Any]]:
+    """Convert a legacy security requirement list into the 1.0.0 Protobuf format."""
+    return [
+        {
+            'schemes': {
+                scheme_name: {'list': scopes}
+                for scheme_name, scopes in sec_dict.items()
+            }
+        }
+        for sec_dict in sec_list
+    ]
+
+
+def _handle_security_compatibility(agent_card_data: dict[str, Any]) -> None:
+    """Map legacy security requirements and schemas to their 1.0.0 Protobuf equivalents."""
+    legacy_security = agent_card_data.pop('security', None)
+    if (
+        'securityRequirements' not in agent_card_data
+        and legacy_security is not None
+    ):
+        agent_card_data['securityRequirements'] = _map_legacy_security(
+            legacy_security
+        )
+
+    for skill in agent_card_data.get('skills', []):
+        legacy_skill_sec = skill.pop('security', None)
+        if 'securityRequirements' not in skill and legacy_skill_sec is not None:
+            skill['securityRequirements'] = _map_legacy_security(
+                legacy_skill_sec
+            )
+
+    if 'securitySchemes' in agent_card_data:
+        type_mapping = {
+            'apiKey': 'apiKeySecurityScheme',
+            'http': 'httpAuthSecurityScheme',
+            'oauth2': 'oauth2SecurityScheme',
+            'openIdConnect': 'openIdConnectSecurityScheme',
+            'mutualTLS': 'mtlsSecurityScheme',
+        }
+        for scheme in agent_card_data['securitySchemes'].values():
+            scheme_type = scheme.pop('type', None)
+            if scheme_type in type_mapping:
+                # Map legacy 'in' to modern 'location'
+                if scheme_type == 'apiKey' and 'in' in scheme:
+                    scheme['location'] = scheme.pop('in')
+
+                mapped_name = type_mapping[scheme_type]
+                new_scheme_wrapper = {mapped_name: scheme.copy()}
+                scheme.clear()
+                scheme.update(new_scheme_wrapper)
 
 
 def create_text_message_object(

src/a2a/client/transports/jsonrpc.py

@@ -17,6 +17,7 @@
     A2AClientJSONRPCError,
     A2AClientTimeoutError,
 )
+from a2a.client.helpers import parse_agent_card
 from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
 from a2a.client.transports.base import ClientTransport
 from a2a.extensions.common import update_extension_header
@@ -521,9 +522,7 @@ async def get_extended_agent_card(
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise A2AClientJSONRPCError(json_rpc_response.error)
-        response: AgentCard = json_format.ParseDict(
-            json_rpc_response.result, AgentCard()
-        )
+        response: AgentCard = parse_agent_card(json_rpc_response.result)
         if signature_verifier:
             signature_verifier(response)
 

src/a2a/client/transports/rest.py

@@ -15,6 +15,7 @@
     A2AClientJSONError,
     A2AClientTimeoutError,
 )
+from a2a.client.helpers import parse_agent_card
 from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
 from a2a.client.transports.base import ClientTransport
 from a2a.extensions.common import update_extension_header
@@ -478,7 +479,7 @@ async def get_extended_agent_card(
         response_data = await self._send_get_request(
             '/v1/card', {}, modified_kwargs
         )
-        response: AgentCard = ParseDict(response_data, AgentCard())
+        response: AgentCard = parse_agent_card(response_data)
 
         if signature_verifier:
             signature_verifier(response)

src/a2a/server/apps/jsonrpc/jsonrpc_app.py

@@ -9,7 +9,7 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from typing import TYPE_CHECKING, Any
 
-from google.protobuf.json_format import MessageToDict, ParseDict
+from google.protobuf.json_format import ParseDict
 from jsonrpc.jsonrpc2 import JSONRPC20Request
 
 from a2a.auth.user import UnauthenticatedUser
@@ -29,7 +29,10 @@
 )
 from a2a.server.request_handlers.jsonrpc_handler import JSONRPCHandler
 from a2a.server.request_handlers.request_handler import RequestHandler
-from a2a.server.request_handlers.response_helpers import build_error_response
+from a2a.server.request_handlers.response_helpers import (
+    agent_card_to_dict,
+    build_error_response,
+)
 from a2a.types import A2ARequest
 from a2a.types.a2a_pb2 import (
     AgentCard,
@@ -575,7 +578,7 @@ async def _handle_get_agent_card(self, request: Request) -> JSONResponse:
             card_to_serve = await maybe_await(self.card_modifier(card_to_serve))
 
         return JSONResponse(
-            MessageToDict(
+            agent_card_to_dict(
                 card_to_serve,
                 preserving_proto_field_name=False,
             )

src/a2a/server/apps/rest/rest_adapter.py

@@ -4,8 +4,7 @@
 from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable
 from typing import TYPE_CHECKING, Any
 
-from google.protobuf.json_format import MessageToDict
-
+from a2a.server.request_handlers.response_helpers import agent_card_to_dict
 from a2a.utils.helpers import maybe_await
 
 
@@ -158,7 +157,7 @@ async def handle_get_agent_card(
         if self.card_modifier:
             card_to_serve = await maybe_await(self.card_modifier(card_to_serve))
 
-        return MessageToDict(card_to_serve)
+        return agent_card_to_dict(card_to_serve)
 
     async def handle_authenticated_agent_card(
         self, request: Request, call_context: ServerCallContext | None = None
@@ -192,7 +191,9 @@ async def handle_authenticated_agent_card(
         elif self.card_modifier:
             card_to_serve = await maybe_await(self.card_modifier(card_to_serve))
 
-        return MessageToDict(card_to_serve, preserving_proto_field_name=True)
+        return agent_card_to_dict(
+            card_to_serve, preserving_proto_field_name=True
+        )
 
     def routes(self) -> dict[tuple[str, str], Callable[[Request], Any]]:
         """Constructs a dictionary of API routes and their corresponding handlers.

src/a2a/server/request_handlers/jsonrpc_handler.py

@@ -16,6 +16,7 @@
     JSONRPCError,
 )
 from a2a.server.request_handlers.request_handler import RequestHandler
+from a2a.server.request_handlers.response_helpers import agent_card_to_dict
 from a2a.types.a2a_pb2 import (
     AgentCard,
     CancelTaskRequest,
@@ -470,5 +471,7 @@ async def get_authenticated_extended_card(
         elif self.card_modifier:
             card_to_serve = await maybe_await(self.card_modifier(base_card))
 
-        result = MessageToDict(card_to_serve, preserving_proto_field_name=False)
+        result = agent_card_to_dict(
+            card_to_serve, preserving_proto_field_name=False
+        )
         return _build_success_response(request_id, result)

src/a2a/server/request_handlers/response_helpers.py

@@ -6,13 +6,15 @@
 from google.protobuf.message import Message as ProtoMessage
 from jsonrpc.jsonrpc2 import JSONRPC20Response
 
+from a2a.compat.v0_3.conversions import to_compat_agent_card
 from a2a.server.jsonrpc_models import (
     InternalError as JSONRPCInternalError,
 )
 from a2a.server.jsonrpc_models import (
     JSONRPCError,
 )
 from a2a.types.a2a_pb2 import (
+    AgentCard,
     ListTasksResponse,
     Message,
     StreamResponse,
@@ -89,6 +91,43 @@
 """Type alias for possible event types produced by handlers."""
 
 
+def agent_card_to_dict(
+    card: AgentCard, preserving_proto_field_name: bool = False
+) -> dict[str, Any]:
+    """Convert AgentCard to dict and inject backward compatibility fields."""
+    result = MessageToDict(
+        card, preserving_proto_field_name=preserving_proto_field_name
+    )
+
+    compat_card = to_compat_agent_card(card)
+    compat_dict = compat_card.model_dump(
+        by_alias=not preserving_proto_field_name, exclude_none=True
+    )
+
+    # Do not include supportsAuthenticatedExtendedCard if false
+    key = (
+        'supports_authenticated_extended_card'
+        if preserving_proto_field_name
+        else 'supportsAuthenticatedExtendedCard'
+    )
+    if not compat_dict.get(key):
+        compat_dict.pop(key, None)
+
+    def merge(dict1: dict[str, Any], dict2: dict[str, Any]) -> dict[str, Any]:
+        for k, v in dict2.items():
+            if k not in dict1:
+                dict1[k] = v
+            elif isinstance(v, dict) and isinstance(dict1[k], dict):
+                merge(dict1[k], v)
+            elif isinstance(v, list) and isinstance(dict1[k], list):
+                for i in range(min(len(dict1[k]), len(v))):
+                    if isinstance(dict1[k][i], dict) and isinstance(v[i], dict):
+                        merge(dict1[k][i], v[i])
+        return dict1
+
+    return merge(result, compat_dict)
+
+
 def build_error_response(
     request_id: str | int | None,
     error: A2AError | JSONRPCError,

tests/client/test_card_resolver.py

@@ -260,7 +260,7 @@ async def test_get_agent_card_validation_error(
         valid_agent_card_data,
     ):
         """Test A2AClientJSONError is raised on agent card validation error."""
-        return_json = {'invalid': 'data'}
+        return_json = {'name': {'invalid': 'type'}}
         mock_response.json.return_value = return_json
         mock_httpx_client.get.return_value = mock_response
         with pytest.raises(A2AClientJSONError) as exc_info: