feat: Add GetExtendedAgentCard Support to RequestHandlers (#919)

# Description
The `GetExtendedAgentCard` capability was defined in the spec but not
implemented in the `request_handler.py`.

# Changes
- Added `on_get_extended_agent_card` to the base `RequestHandler` and
its child class `DefaultRequestHandler`.
- Removed `GetExtendedAgentCard` method implementations from the
Transport layer and consequently moved `AgentCard` informations from the
Transport layer to the `RequestHandlers`.
- moved `validate` logic from the transport layer to the default request
handler

Fixes #866 🦕

Author: sokoliva

itk/main.py

@@ -292,7 +292,11 @@ async def main_async(http_port: int, grpc_port: int) -> None:
         name='ITK v10 Agent',
         description='Python agent using SDK 1.0.',
         version='1.0.0',
-        capabilities=AgentCapabilities(streaming=True),
+        capabilities=AgentCapabilities(
+            streaming=True,
+            push_notifications=True,
+            extended_agent_card=True,
+        ),
         default_input_modes=['text/plain'],
         default_output_modes=['text/plain'],
         supported_interfaces=interfaces,
@@ -302,18 +306,25 @@ async def main_async(http_port: int, grpc_port: int) -> None:
     handler = DefaultRequestHandler(
         agent_executor=V10AgentExecutor(),
         task_store=task_store,
+        agent_card=agent_card,
         queue_manager=InMemoryQueueManager(),
     )
 
+    handler_extended = DefaultRequestHandler(
+        agent_executor=V10AgentExecutor(),
+        task_store=task_store,
+        agent_card=agent_card,
+        queue_manager=InMemoryQueueManager(),
+        extended_agent_card=agent_card,
+    )
+
     app = FastAPI()
 
     agent_card_routes = create_agent_card_routes(
         agent_card=agent_card, card_url='/.well-known/agent-card.json'
     )
     jsonrpc_routes = create_jsonrpc_routes(
-        agent_card=agent_card,
-        request_handler=handler,
-        extended_agent_card=agent_card,
+        request_handler=handler_extended,
         rpc_url='/',
         enable_v0_3_compat=True,
     )
@@ -323,17 +334,16 @@ async def main_async(http_port: int, grpc_port: int) -> None:
     )
 
     rest_routes = create_rest_routes(
-        agent_card=agent_card,
         request_handler=handler,
         enable_v0_3_compat=True,
     )
     app.mount('/rest', FastAPI(routes=rest_routes + agent_card_routes))
 
     server = grpc.aio.server()
 
-    compat_servicer = CompatGrpcHandler(agent_card, handler)
+    compat_servicer = CompatGrpcHandler(handler)
     a2a_v0_3_pb2_grpc.add_A2AServiceServicer_to_server(compat_servicer, server)
-    servicer = GrpcHandler(agent_card, handler)
+    servicer = GrpcHandler(handler)
     a2a_pb2_grpc.add_A2AServiceServicer_to_server(servicer, server)
 
     server.add_insecure_port(f'127.0.0.1:{grpc_port}')

samples/hello_world_agent.py

@@ -191,17 +191,17 @@ async def serve(
 
     task_store = InMemoryTaskStore()
     request_handler = DefaultRequestHandler(
-        agent_executor=SampleAgentExecutor(), task_store=task_store
+        agent_executor=SampleAgentExecutor(),
+        task_store=task_store,
+        agent_card=agent_card,
     )
 
     rest_routes = create_rest_routes(
-        agent_card=agent_card,
         request_handler=request_handler,
         path_prefix='/a2a/rest',
         enable_v0_3_compat=True,
     )
     jsonrpc_routes = create_jsonrpc_routes(
-        agent_card=agent_card,
         request_handler=request_handler,
         rpc_url='/a2a/jsonrpc',
         enable_v0_3_compat=True,
@@ -216,12 +216,12 @@ async def serve(
 
     grpc_server = grpc.aio.server()
     grpc_server.add_insecure_port(f'{host}:{grpc_port}')
-    servicer = GrpcHandler(agent_card, request_handler)
+    servicer = GrpcHandler(request_handler)
     a2a_pb2_grpc.add_A2AServiceServicer_to_server(servicer, grpc_server)
 
     compat_grpc_server = grpc.aio.server()
     compat_grpc_server.add_insecure_port(f'{host}:{compat_grpc_port}')
-    compat_servicer = CompatGrpcHandler(agent_card, request_handler)
+    compat_servicer = CompatGrpcHandler(request_handler)
     a2a_v0_3_pb2_grpc.add_A2AServiceServicer_to_server(
         compat_servicer, compat_grpc_server
     )

src/a2a/compat/v0_3/grpc_handler.py

@@ -12,7 +12,6 @@
 from a2a.compat.v0_3 import (
     a2a_v0_3_pb2,
     a2a_v0_3_pb2_grpc,
-    conversions,
     proto_utils,
 )
 from a2a.compat.v0_3 import (
@@ -27,9 +26,7 @@
     GrpcServerCallContextBuilder,
 )
 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
 
 
 logger = logging.getLogger(__name__)
@@ -42,29 +39,21 @@ class CompatGrpcHandler(a2a_v0_3_pb2_grpc.A2AServiceServicer):
 
     def __init__(
         self,
-        agent_card: AgentCard,
         request_handler: RequestHandler,
         context_builder: GrpcServerCallContextBuilder | None = None,
-        card_modifier: Callable[[AgentCard], Awaitable[AgentCard] | AgentCard]
-        | None = None,
     ):
         """Initializes the CompatGrpcHandler.
 
         Args:
-            agent_card: The AgentCard describing the agent's capabilities (v1.0).
             request_handler: The underlying `RequestHandler` instance to
                              delegate requests to.
             context_builder: The CallContextBuilder object. If none the
                              DefaultCallContextBuilder is used.
-            card_modifier: An optional callback to dynamically modify the public
-              agent card before it is served.
         """
-        self.agent_card = agent_card
         self.handler03 = RequestHandler03(request_handler=request_handler)
         self._context_builder = (
             context_builder or DefaultGrpcServerCallContextBuilder()
         )
-        self.card_modifier = card_modifier
 
     async def _handle_unary(
         self,
@@ -179,10 +168,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]:
@@ -242,10 +227,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]:
@@ -269,10 +250,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:
@@ -360,12 +337,19 @@ async def GetAgentCard(
         request: a2a_v0_3_pb2.GetAgentCardRequest,
         context: grpc.aio.ServicerContext,
     ) -> a2a_v0_3_pb2.AgentCard:
-        """Get the agent card for the agent served (v0.3)."""
-        card_to_serve = self.agent_card
-        if self.card_modifier:
-            card_to_serve = await maybe_await(self.card_modifier(card_to_serve))
-        return proto_utils.ToProto.agent_card(
-            conversions.to_compat_agent_card(card_to_serve)
+        """Get the extended agent card for the agent served (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()
         )
 
     async def DeleteTaskPushNotificationConfig(

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
@@ -11,7 +11,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:
@@ -24,7 +23,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
@@ -42,8 +40,7 @@
     ServerCallContextBuilder,
 )
 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__)
@@ -65,19 +62,11 @@ class JSONRPC03Adapter:
         'agent/getAuthenticatedExtendedCard': types_v03.GetAuthenticatedExtendedCardRequest,
     }
 
-    def __init__(  # noqa: PLR0913
+    def __init__(
         self,
-        agent_card: 'AgentCard',
         http_handler: 'RequestHandler',
-        extended_agent_card: 'AgentCard | None' = None,
         context_builder: 'ServerCallContextBuilder | 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,
         )
@@ -227,7 +216,7 @@ async def _process_non_streaming_request(
                 )
             )
         elif method == 'agent/getAuthenticatedExtendedCard':
-            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(
@@ -244,31 +233,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/getAuthenticatedExtendedCard' 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)