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.

Author: bartek-w

src/a2a/client/card_resolver.py

@@ -6,9 +6,10 @@
 
 import httpx
 
-from google.protobuf.json_format import ParseDict, ParseError
+from google.protobuf.json_format import ParseError
 
 from a2a.client.errors import AgentCardResolutionError
+from a2a.client.helpers import parse_agent_card
 from a2a.types.a2a_pb2 import (
     AgentCard,
 )
@@ -85,7 +86,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,137 @@
 """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

@@ -10,6 +10,7 @@
 from jsonrpc.jsonrpc2 import JSONRPC20Request, JSONRPC20Response
 
 from a2a.client.errors import A2AClientError
+from a2a.client.helpers import parse_agent_card
 from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
 from a2a.client.transports.base import ClientTransport
 from a2a.client.transports.http_helpers import (
@@ -93,7 +94,7 @@ async def send_message(
         response_data = await self._send_request(payload, modified_kwargs)
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
-            raise self._create_jsonrpc_error(json_rpc_response.error)
+                        raise self._create_jsonrpc_error(json_rpc_response.error)
         response: SendMessageResponse = json_format.ParseDict(
             json_rpc_response.result, SendMessageResponse()
         )
@@ -154,7 +155,7 @@ async def get_task(
         response_data = await self._send_request(payload, modified_kwargs)
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
-            raise self._create_jsonrpc_error(json_rpc_response.error)
+                        raise self._create_jsonrpc_error(json_rpc_response.error)
         response: Task = json_format.ParseDict(json_rpc_response.result, Task())
         return response
 
@@ -184,7 +185,7 @@ async def list_tasks(
         response_data = await self._send_request(payload, modified_kwargs)
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
-            raise self._create_jsonrpc_error(json_rpc_response.error)
+                        raise self._create_jsonrpc_error(json_rpc_response.error)
         response: ListTasksResponse = json_format.ParseDict(
             json_rpc_response.result, ListTasksResponse()
         )
@@ -216,7 +217,7 @@ async def cancel_task(
         response_data = await self._send_request(payload, modified_kwargs)
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
-            raise self._create_jsonrpc_error(json_rpc_response.error)
+                        raise self._create_jsonrpc_error(json_rpc_response.error)
         response: Task = json_format.ParseDict(json_rpc_response.result, Task())
         return response
 
@@ -246,7 +247,7 @@ async def create_task_push_notification_config(
         response_data = await self._send_request(payload, modified_kwargs)
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
-            raise self._create_jsonrpc_error(json_rpc_response.error)
+                        raise self._create_jsonrpc_error(json_rpc_response.error)
         response: TaskPushNotificationConfig = json_format.ParseDict(
             json_rpc_response.result, TaskPushNotificationConfig()
         )
@@ -278,7 +279,7 @@ async def get_task_push_notification_config(
         response_data = await self._send_request(payload, modified_kwargs)
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
-            raise self._create_jsonrpc_error(json_rpc_response.error)
+                        raise self._create_jsonrpc_error(json_rpc_response.error)
         response: TaskPushNotificationConfig = json_format.ParseDict(
             json_rpc_response.result, TaskPushNotificationConfig()
         )
@@ -310,7 +311,7 @@ async def list_task_push_notification_configs(
         response_data = await self._send_request(payload, modified_kwargs)
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
-            raise self._create_jsonrpc_error(json_rpc_response.error)
+                        raise self._create_jsonrpc_error(json_rpc_response.error)
         response: ListTaskPushNotificationConfigsResponse = (
             json_format.ParseDict(
                 json_rpc_response.result,
@@ -345,7 +346,7 @@ async def delete_task_push_notification_config(
         response_data = await self._send_request(payload, modified_kwargs)
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
-            raise self._create_jsonrpc_error(json_rpc_response.error)
+                        raise self._create_jsonrpc_error(json_rpc_response.error)
 
     async def subscribe(
         self,
@@ -413,8 +414,13 @@ async def get_extended_agent_card(
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
-        response: AgentCard = json_format.ParseDict(
-            json_rpc_response.result, AgentCard()
+        # Validate type of the response
+        if not isinstance(json_rpc_response.result, dict):
+            raise A2AClientError(
+                f'Invalid response type: {type(json_rpc_response.result)}'
+            )
+        response: AgentCard = parse_agent_card(
+            cast('dict[str, Any]', json_rpc_response.result)
         )
         if signature_verifier:
             signature_verifier(response)
@@ -498,7 +504,7 @@ async def _send_stream_request(
         ):
             json_rpc_response = JSONRPC20Response.from_json(sse_data)
             if json_rpc_response.error:
-                raise self._create_jsonrpc_error(json_rpc_response.error)
+                            raise self._create_jsonrpc_error(json_rpc_response.error)
             response: StreamResponse = json_format.ParseDict(
                 json_rpc_response.result, StreamResponse()
             )

src/a2a/client/transports/rest.py

@@ -10,6 +10,7 @@
 from google.protobuf.message import Message
 
 from a2a.client.errors import A2AClientError
+from a2a.client.helpers import parse_agent_card
 from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
 from a2a.client.transports.base import ClientTransport
 from a2a.client.transports.http_helpers import (
@@ -347,7 +348,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,
@@ -458,5 +459,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)