add GetExtendedCard method to compat request handler

Author: sokoliva

src/a2a/compat/v0_3/grpc_handler.py

@@ -29,7 +29,7 @@
 from a2a.server.request_handlers.request_handler import RequestHandler
 from a2a.types.a2a_pb2 import AgentCard
 from a2a.utils.errors import A2AError, InvalidParamsError
-from a2a.utils.helpers import maybe_await, validate
+from a2a.utils.helpers import maybe_await
 
 
 logger = logging.getLogger(__name__)
@@ -177,10 +177,6 @@ async def SendStreamingMessage(
     ) -> AsyncIterable[a2a_v0_3_pb2.StreamResponse]:
         """Handles the 'SendStreamingMessage' gRPC method (v0.3)."""
 
-        @validate(
-            lambda _: self.agent_card.capabilities.streaming,
-            'Streaming is not supported by the agent',
-        )
         async def _handler(
             server_context: ServerCallContext,
         ) -> AsyncIterable[a2a_v0_3_pb2.StreamResponse]:
@@ -240,10 +236,6 @@ async def TaskSubscription(
     ) -> AsyncIterable[a2a_v0_3_pb2.StreamResponse]:
         """Handles the 'TaskSubscription' gRPC method (v0.3)."""
 
-        @validate(
-            lambda _: self.agent_card.capabilities.streaming,
-            'Streaming is not supported by the agent',
-        )
         async def _handler(
             server_context: ServerCallContext,
         ) -> AsyncIterable[a2a_v0_3_pb2.StreamResponse]:
@@ -267,10 +259,6 @@ async def CreateTaskPushNotificationConfig(
     ) -> a2a_v0_3_pb2.TaskPushNotificationConfig:
         """Handles the 'CreateTaskPushNotificationConfig' gRPC method (v0.3)."""
 
-        @validate(
-            lambda _: self.agent_card.capabilities.push_notifications,
-            'Push notifications are not supported by the agent',
-        )
         async def _handler(
             server_context: ServerCallContext,
         ) -> a2a_v0_3_pb2.TaskPushNotificationConfig:
@@ -389,3 +377,23 @@ async def _handler(
             return empty_pb2.Empty()
 
         return await self._handle_unary(context, _handler, empty_pb2.Empty())
+
+    async def GetExtendedCard(
+        self,
+        request: a2a_v0_3_pb2.GetAgentCardRequest,
+        context: grpc.aio.ServicerContext,
+    ) -> a2a_v0_3_pb2.AgentCard:
+        """Get the authenticated extended agent card (v0.3)."""
+
+        async def _handler(
+            server_context: ServerCallContext,
+        ) -> a2a_v0_3_pb2.AgentCard:
+            req_v03 = types_v03.GetAuthenticatedExtendedCardRequest(id=0)
+            res_v03 = await self.handler03.on_get_extended_agent_card(
+                req_v03, server_context
+            )
+            return proto_utils.ToProto.agent_card(res_v03)
+
+        return await self._handle_unary(
+            context, _handler, a2a_v0_3_pb2.AgentCard()
+        )

src/a2a/compat/v0_3/jsonrpc_adapter.py

@@ -1,6 +1,6 @@
 import logging
 
-from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterable, AsyncIterator
 from typing import TYPE_CHECKING, Any
 
 from sse_starlette.sse import EventSourceResponse
@@ -12,7 +12,6 @@
 
     from a2a.server.request_handlers.request_handler import RequestHandler
     from a2a.server.routes import CallContextBuilder
-    from a2a.types.a2a_pb2 import AgentCard
 
     _package_starlette_installed = True
 else:
@@ -25,7 +24,6 @@
 
         _package_starlette_installed = False
 
-from a2a.compat.v0_3 import conversions
 from a2a.compat.v0_3 import types as types_v03
 from a2a.compat.v0_3.request_handler import RequestHandler03
 from a2a.server.context import ServerCallContext
@@ -39,8 +37,7 @@
     JSONRPCError as CoreJSONRPCError,
 )
 from a2a.utils import constants
-from a2a.utils.errors import ExtendedAgentCardNotConfiguredError
-from a2a.utils.helpers import maybe_await, validate_version
+from a2a.utils.helpers import validate_version
 
 
 logger = logging.getLogger(__name__)
@@ -62,19 +59,11 @@ class JSONRPC03Adapter:
         'agent/authenticatedExtendedCard': types_v03.GetAuthenticatedExtendedCardRequest,
     }
 
-    def __init__(  # noqa: PLR0913
+    def __init__(
         self,
-        agent_card: 'AgentCard',
         http_handler: 'RequestHandler',
-        extended_agent_card: 'AgentCard | None' = None,
         context_builder: 'CallContextBuilder | None' = None,
-        card_modifier: 'Callable[[AgentCard], Awaitable[AgentCard] | AgentCard] | None' = None,
-        extended_card_modifier: 'Callable[[AgentCard, ServerCallContext], Awaitable[AgentCard] | AgentCard] | None' = None,
     ):
-        self.agent_card = agent_card
-        self.extended_agent_card = extended_agent_card
-        self.card_modifier = card_modifier
-        self.extended_card_modifier = extended_card_modifier
         self.handler = RequestHandler03(
             request_handler=http_handler,
         )
@@ -226,7 +215,7 @@ async def _process_non_streaming_request(
                 )
             )
         elif method == 'agent/authenticatedExtendedCard':
-            res_card = await self.get_authenticated_extended_card(
+            res_card = await self.handler.on_get_extended_agent_card(
                 request_obj, context
             )
             result = types_v03.GetAuthenticatedExtendedCardResponse(
@@ -243,31 +232,6 @@ async def _process_non_streaming_request(
             )
         )
 
-    async def get_authenticated_extended_card(
-        self,
-        request: types_v03.GetAuthenticatedExtendedCardRequest,
-        context: ServerCallContext,
-    ) -> types_v03.AgentCard:
-        """Handles the 'agent/authenticatedExtendedCard' JSON-RPC method."""
-        if not self.agent_card.capabilities.extended_agent_card:
-            raise ExtendedAgentCardNotConfiguredError(
-                message='Authenticated card not supported'
-            )
-
-        base_card = self.extended_agent_card
-        if base_card is None:
-            base_card = self.agent_card
-
-        card_to_serve = base_card
-        if self.extended_card_modifier and context:
-            card_to_serve = await maybe_await(
-                self.extended_card_modifier(base_card, context)
-            )
-        elif self.card_modifier:
-            card_to_serve = await maybe_await(self.card_modifier(base_card))
-
-        return conversions.to_compat_agent_card(card_to_serve)
-
     @validate_version(constants.PROTOCOL_VERSION_0_3)
     async def _process_streaming_request(
         self,

src/a2a/compat/v0_3/request_handler.py

@@ -9,9 +9,7 @@
 from a2a.server.request_handlers.request_handler import RequestHandler
 from a2a.types.a2a_pb2 import Task
 from a2a.utils import proto_utils as core_proto_utils
-from a2a.utils.errors import (
-    TaskNotFoundError,
-)
+from a2a.utils.errors import TaskNotFoundError
 
 
 logger = logging.getLogger(__name__)
@@ -170,3 +168,15 @@ async def on_delete_task_push_notification_config(
         await self.request_handler.on_delete_task_push_notification_config(
             v10_req, context
         )
+
+    async def on_get_extended_agent_card(
+        self,
+        request: types_v03.GetAuthenticatedExtendedCardRequest,
+        context: ServerCallContext,
+    ) -> types_v03.AgentCard:
+        """Gets the authenticated extended agent card using v0.3 protocol types."""
+        v10_req = conversions.to_core_get_extended_agent_card_request(request)
+        v10_card = await self.request_handler.on_get_extended_agent_card(
+            v10_req, context
+        )
+        return conversions.to_compat_agent_card(v10_card)

src/a2a/compat/v0_3/rest_adapter.py

@@ -11,8 +11,8 @@
     from starlette.requests import Request
     from starlette.responses import JSONResponse, Response
 
+    from a2a.server.context import ServerCallContext
     from a2a.server.request_handlers.request_handler import RequestHandler
-    from a2a.types.a2a_pb2 import AgentCard
 
     _package_starlette_installed = True
 else:
@@ -31,19 +31,15 @@
         _package_starlette_installed = False
 
 
-from a2a.compat.v0_3 import conversions
 from a2a.compat.v0_3.rest_handler import REST03Handler
-from a2a.server.context import ServerCallContext
 from a2a.server.routes import CallContextBuilder, DefaultCallContextBuilder
 from a2a.utils.error_handlers import (
     rest_error_handler,
     rest_stream_error_handler,
 )
 from a2a.utils.errors import (
-    ExtendedAgentCardNotConfiguredError,
     InvalidRequestError,
 )
-from a2a.utils.helpers import maybe_await
 
 
 logger = logging.getLogger(__name__)
@@ -55,22 +51,12 @@ class REST03Adapter:
     Defines v0.3 REST request processors and their routes, as well as managing response generation including Server-Sent Events (SSE).
     """
 
-    def __init__(  # noqa: PLR0913
+    def __init__(
         self,
-        agent_card: 'AgentCard',
         http_handler: 'RequestHandler',
-        extended_agent_card: 'AgentCard | None' = None,
         context_builder: 'CallContextBuilder | None' = None,
-        card_modifier: 'Callable[[AgentCard], Awaitable[AgentCard] | AgentCard] | None' = None,
-        extended_card_modifier: 'Callable[[AgentCard, ServerCallContext], Awaitable[AgentCard] | AgentCard] | None' = None,
     ):
-        self.agent_card = agent_card
-        self.extended_agent_card = extended_agent_card
-        self.card_modifier = card_modifier
-        self.extended_card_modifier = extended_card_modifier
-        self.handler = REST03Handler(
-            agent_card=agent_card, request_handler=http_handler
-        )
+        self.handler = REST03Handler(request_handler=http_handler)
         self._context_builder = context_builder or DefaultCallContextBuilder()
 
     @rest_error_handler
@@ -108,39 +94,6 @@ async def event_generator(
             event_generator(method(request, call_context))
         )
 
-    async def handle_get_agent_card(
-        self, request: Request, call_context: ServerCallContext
-    ) -> dict[str, Any]:
-        """Handles GET requests for the agent card endpoint."""
-        card_to_serve = self.agent_card
-        if self.card_modifier:
-            card_to_serve = await maybe_await(self.card_modifier(card_to_serve))
-        v03_card = conversions.to_compat_agent_card(card_to_serve)
-        return v03_card.model_dump(mode='json', exclude_none=True)
-
-    async def handle_authenticated_agent_card(
-        self, request: Request, call_context: ServerCallContext
-    ) -> dict[str, Any]:
-        """Hook for per credential agent card response."""
-        if not self.agent_card.capabilities.extended_agent_card:
-            raise ExtendedAgentCardNotConfiguredError(
-                message='Authenticated card not supported'
-            )
-        card_to_serve = self.extended_agent_card
-
-        if not card_to_serve:
-            card_to_serve = self.agent_card
-
-        if self.extended_card_modifier:
-            card_to_serve = await maybe_await(
-                self.extended_card_modifier(card_to_serve, call_context)
-            )
-        elif self.card_modifier:
-            card_to_serve = await maybe_await(self.card_modifier(card_to_serve))
-
-        v03_card = conversions.to_compat_agent_card(card_to_serve)
-        return v03_card.model_dump(mode='json', exclude_none=True)
-
     def routes(self) -> dict[tuple[str, str], Callable[[Request], Any]]:
         """Constructs a dictionary of API routes and their corresponding handlers."""
         routes: dict[tuple[str, str], Callable[[Request], Any]] = {
@@ -186,10 +139,9 @@ def routes(self) -> dict[tuple[str, str], Callable[[Request], Any]]:
             ('/v1/tasks', 'GET'): functools.partial(
                 self._handle_request, self.handler.list_tasks
             ),
+            ('/v1/card', 'GET'): functools.partial(
+                self._handle_request, self.handler.on_get_extended_agent_card
+            ),
         }
-        if self.agent_card.capabilities.extended_agent_card:
-            routes[('/v1/card', 'GET')] = functools.partial(
-                self._handle_request, self.handle_authenticated_agent_card
-            )
 
         return routes

src/a2a/compat/v0_3/rest_handler.py

@@ -10,7 +10,6 @@
     from starlette.requests import Request
 
     from a2a.server.request_handlers.request_handler import RequestHandler
-    from a2a.types.a2a_pb2 import AgentCard
 
     _package_starlette_installed = True
 else:
@@ -30,7 +29,6 @@
 from a2a.server.context import ServerCallContext
 from a2a.utils import constants
 from a2a.utils.helpers import (
-    validate,
     validate_version,
 )
 from a2a.utils.telemetry import SpanKind, trace_class
@@ -45,16 +43,13 @@ class REST03Handler:
 
     def __init__(
         self,
-        agent_card: 'AgentCard',
         request_handler: 'RequestHandler',
     ):
         """Initializes the REST03Handler.
 
         Args:
-          agent_card: The AgentCard describing the agent's capabilities (v1.0).
           request_handler: The underlying `RequestHandler` instance to delegate requests to (v1.0).
         """
-        self.agent_card = agent_card
         self.handler03 = RequestHandler03(request_handler=request_handler)
 
     @validate_version(constants.PROTOCOL_VERSION_0_3)
@@ -84,10 +79,6 @@ async def on_message_send(
         return MessageToDict(pb2_v03_resp)
 
     @validate_version(constants.PROTOCOL_VERSION_0_3)
-    @validate(
-        lambda self: self.agent_card.capabilities.streaming,
-        'Streaming is not supported by the agent',
-    )
     async def on_message_send_stream(
         self,
         request: Request,
@@ -142,10 +133,6 @@ async def on_cancel_task(
         return MessageToDict(pb2_v03_task)
 
     @validate_version(constants.PROTOCOL_VERSION_0_3)
-    @validate(
-        lambda self: self.agent_card.capabilities.streaming,
-        'Streaming is not supported by the agent',
-    )
     async def on_subscribe_to_task(
         self,
         request: Request,
@@ -208,10 +195,6 @@ async def get_push_notification(
         return MessageToDict(pb2_v03_config)
 
     @validate_version(constants.PROTOCOL_VERSION_0_3)
-    @validate(
-        lambda self: self.agent_card.capabilities.push_notifications,
-        'Push notifications are not supported by the agent',
-    )
     async def set_push_notification(
         self,
         request: Request,
@@ -317,3 +300,16 @@ async def list_tasks(
     ) -> dict[str, Any]:
         """Handles the 'tasks/list' REST method."""
         raise NotImplementedError('list tasks not implemented')
+
+    @validate_version(constants.PROTOCOL_VERSION_0_3)
+    async def on_get_extended_agent_card(
+        self,
+        request: Request,
+        context: ServerCallContext,
+    ) -> dict[str, Any]:
+        """Handles the 'v1/agent/authenticatedExtendedAgentCard' REST method."""
+        rpc_req = types_v03.GetAuthenticatedExtendedCardRequest(id=0)
+        v03_resp = await self.handler03.on_get_extended_agent_card(
+            rpc_req, context
+        )
+        return v03_resp.model_dump(mode='json', exclude_none=True)