feat: Allow agent cards (default and extended) to be dynamic

Author: dmandar

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

@@ -1,6 +1,6 @@
 import logging
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Callable
 from contextlib import asynccontextmanager
 from typing import Any
 
@@ -10,13 +10,13 @@
     CallContextBuilder,
     JSONRPCApplication,
 )
+from a2a.server.context import ServerCallContext
 from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
 from a2a.types import A2ARequest, AgentCard
 from a2a.utils.constants import (
     AGENT_CARD_WELL_KNOWN_PATH,
     DEFAULT_RPC_URL,
     EXTENDED_AGENT_CARD_PATH,
-    PREV_AGENT_CARD_WELL_KNOWN_PATH,
 )
 
 
@@ -37,6 +37,11 @@ def __init__(
         http_handler: RequestHandler,
         extended_agent_card: AgentCard | None = None,
         context_builder: CallContextBuilder | None = None,
+        card_modifier: Callable[[AgentCard], AgentCard] | None = None,
+        extended_card_modifier: Callable[
+            [AgentCard, ServerCallContext], AgentCard
+        ]
+        | None = None,
     ) -> None:
         """Initializes the A2AStarletteApplication.
 
@@ -49,12 +54,19 @@ def __init__(
             context_builder: The CallContextBuilder used to construct the
               ServerCallContext passed to the http_handler. If None, no
               ServerCallContext is passed.
+            card_modifier: An optional callback to dynamically modify the public
+              agent card before it is served.
+            extended_card_modifier: An optional callback to dynamically modify
+              the extended agent card before it is served. It receives the
+              call context.
         """
         super().__init__(
             agent_card=agent_card,
             http_handler=http_handler,
             extended_agent_card=extended_agent_card,
             context_builder=context_builder,
+            card_modifier=card_modifier,
+            extended_card_modifier=extended_card_modifier,
         )
 
     def add_routes_to_app(
@@ -90,12 +102,6 @@ def add_routes_to_app(
         )(self._handle_requests)
         app.get(agent_card_url)(self._handle_get_agent_card)
 
-        # add deprecated path only if the agent_card_url uses default well-known path
-        if agent_card_url == AGENT_CARD_WELL_KNOWN_PATH:
-            app.get(PREV_AGENT_CARD_WELL_KNOWN_PATH, include_in_schema=False)(
-                self.handle_deprecated_agent_card_path
-            )
-
         if self.agent_card.supports_authenticated_extended_card:
             app.get(extended_agent_card_url)(
                 self._handle_get_authenticated_extended_agent_card

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

@@ -4,7 +4,7 @@
 import traceback
 
 from abc import ABC, abstractmethod
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Callable
 from typing import Any
 
 from fastapi import FastAPI
@@ -127,6 +127,11 @@ def __init__(
         http_handler: RequestHandler,
         extended_agent_card: AgentCard | None = None,
         context_builder: CallContextBuilder | None = None,
+        card_modifier: Callable[[AgentCard], AgentCard] | None = None,
+        extended_card_modifier: Callable[
+            [AgentCard, ServerCallContext], AgentCard
+        ]
+        | None = None,
     ) -> None:
         """Initializes the A2AStarletteApplication.
 
@@ -139,15 +144,23 @@ def __init__(
             context_builder: The CallContextBuilder used to construct the
               ServerCallContext passed to the http_handler. If None, no
               ServerCallContext is passed.
+            card_modifier: An optional callback to dynamically modify the public
+              agent card before it is served.
+            extended_card_modifier: An optional callback to dynamically modify
+              the extended agent card before it is served. It receives the
+              call context.
         """
         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 = JSONRPCHandler(
             agent_card=agent_card, request_handler=http_handler
         )
         if (
             self.agent_card.supports_authenticated_extended_card
             and self.extended_agent_card is None
+            and self.extended_card_modifier is None
         ):
             logger.error(
                 'AgentCard.supports_authenticated_extended_card is True, but no extended_agent_card was provided. The /agent/authenticatedExtendedCard endpoint will return 404.'
@@ -428,24 +441,23 @@ async def _handle_get_agent_card(self, request: Request) -> JSONResponse:
         Returns:
             A JSONResponse containing the agent card data.
         """
-        # The public agent card is a direct serialization of the agent_card
-        # provided at initialization.
+        if request.url.path == PREV_AGENT_CARD_WELL_KNOWN_PATH:
+            logger.warning(
+                f"Deprecated agent card endpoint '{PREV_AGENT_CARD_WELL_KNOWN_PATH}' accessed. "
+                f"Please use '{AGENT_CARD_WELL_KNOWN_PATH}' instead. This endpoint will be removed in a future version."
+            )
+
+        card_to_serve = self.agent_card
+        if self.card_modifier:
+            card_to_serve = self.card_modifier(card_to_serve)
+
         return JSONResponse(
-            self.agent_card.model_dump(
+            card_to_serve.model_dump(
                 exclude_none=True,
                 by_alias=True,
             )
         )
 
-    async def handle_deprecated_agent_card_path(
-        self, request: Request
-    ) -> JSONResponse:
-        """Handles GET requests for the deprecated agent card endpoint."""
-        logger.warning(
-            f"Deprecated agent card endpoint '{PREV_AGENT_CARD_WELL_KNOWN_PATH}' accessed. Please use '{AGENT_CARD_WELL_KNOWN_PATH}' instead. This endpoint will be removed in a future version."
-        )
-        return await self._handle_get_agent_card(request)
-
     async def _handle_get_authenticated_extended_agent_card(
         self, request: Request
     ) -> JSONResponse:
@@ -456,17 +468,24 @@ async def _handle_get_authenticated_extended_agent_card(
                 status_code=404,
             )
 
-        # If an explicit extended_agent_card is provided, serve that.
-        if self.extended_agent_card:
+        card_to_serve = self.extended_agent_card
+
+        if self.extended_card_modifier:
+            context = self._context_builder.build(request)
+            # If no base extended card is provided, pass the public card to the modifier
+            base_card = card_to_serve if card_to_serve else self.agent_card
+            card_to_serve = self.extended_card_modifier(base_card, context)
+
+        if card_to_serve:
             return JSONResponse(
-                self.extended_agent_card.model_dump(
+                card_to_serve.model_dump(
                     exclude_none=True,
                     by_alias=True,
                 )
             )
-        # If supports_authenticated_extended_card is true, but no specific
-        # extended_agent_card was provided during server initialization,
-        # return a 404
+        # If supports_authenticated_extended_card is true, but no
+        # extended_agent_card was provided, and no modifier produced a card,
+        # return a 404.
         return JSONResponse(
             {
                 'error': 'Authenticated extended agent card is supported but not configured on the server.'

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

@@ -1,5 +1,6 @@
 import logging
 
+from collections.abc import Callable
 from typing import Any
 
 from starlette.applications import Starlette
@@ -9,6 +10,7 @@
     CallContextBuilder,
     JSONRPCApplication,
 )
+from a2a.server.context import ServerCallContext
 from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
 from a2a.types import AgentCard
 from a2a.utils.constants import (
@@ -36,6 +38,11 @@ def __init__(
         http_handler: RequestHandler,
         extended_agent_card: AgentCard | None = None,
         context_builder: CallContextBuilder | None = None,
+        card_modifier: Callable[[AgentCard], AgentCard] | None = None,
+        extended_card_modifier: Callable[
+            [AgentCard, ServerCallContext], AgentCard
+        ]
+        | None = None,
     ) -> None:
         """Initializes the A2AStarletteApplication.
 
@@ -48,12 +55,19 @@ def __init__(
             context_builder: The CallContextBuilder used to construct the
               ServerCallContext passed to the http_handler. If None, no
               ServerCallContext is passed.
+            card_modifier: An optional callback to dynamically modify the public
+              agent card before it is served.
+            extended_card_modifier: An optional callback to dynamically modify
+              the extended agent card before it is served. It receives the
+              call context.
         """
         super().__init__(
             agent_card=agent_card,
             http_handler=http_handler,
             extended_agent_card=extended_agent_card,
             context_builder=context_builder,
+            card_modifier=card_modifier,
+            extended_card_modifier=extended_card_modifier,
         )
 
     def routes(
@@ -87,14 +101,13 @@ def routes(
             ),
         ]
 
-        # add deprecated path only if the agent_card_url uses default well-known path
         if agent_card_url == AGENT_CARD_WELL_KNOWN_PATH:
             app_routes.append(
                 Route(
                     PREV_AGENT_CARD_WELL_KNOWN_PATH,
-                    self.handle_deprecated_agent_card_path,
+                    self._handle_get_agent_card,
                     methods=['GET'],
-                    name='agent_card_path_deprecated',
+                    name='deprecated_agent_card',
                 )
             )
 

src/a2a/server/request_handlers/grpc_handler.py

@@ -18,6 +18,8 @@
         "'pip install a2a-sdk[grpc]'"
     ) from e
 
+from collections.abc import Callable
+
 import a2a.grpc.a2a_pb2_grpc as a2a_grpc
 
 from a2a import types
@@ -87,6 +89,7 @@ def __init__(
         agent_card: AgentCard,
         request_handler: RequestHandler,
         context_builder: CallContextBuilder | None = None,
+        card_modifier: Callable[[AgentCard], AgentCard] | None = None,
     ):
         """Initializes the GrpcHandler.
 
@@ -96,10 +99,13 @@ def __init__(
                              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.request_handler = request_handler
         self.context_builder = context_builder or DefaultCallContextBuilder()
+        self.card_modifier = card_modifier
 
     async def SendMessage(
         self,
@@ -331,7 +337,10 @@ async def GetAgentCard(
         context: grpc.aio.ServicerContext,
     ) -> a2a_pb2.AgentCard:
         """Get the agent card for the agent served."""
-        return proto_utils.ToProto.agent_card(self.agent_card)
+        card_to_serve = self.agent_card
+        if self.card_modifier:
+            card_to_serve = self.card_modifier(card_to_serve)
+        return proto_utils.ToProto.agent_card(card_to_serve)
 
     async def abort_context(
         self, error: ServerError, context: grpc.aio.ServicerContext

tests/server/apps/jsonrpc/test_serialization.py

@@ -58,7 +58,8 @@ def test_starlette_agent_card_with_api_key_scheme_alias(
     app_instance = A2AStarletteApplication(agent_card_with_api_key, handler)
     client = TestClient(app_instance.build())
 
-    response = client.get('/.well-known/agent.json')
+    response = client.get('/.well-known/agent-card.json')
+    print(response.status_code, response.content)
     assert response.status_code == 200
     response_data = response.json()
 
@@ -90,7 +91,8 @@ def test_fastapi_agent_card_with_api_key_scheme_alias(
     app_instance = A2AFastAPIApplication(agent_card_with_api_key, handler)
     client = TestClient(app_instance.build())
 
-    response = client.get('/.well-known/agent.json')
+    response = client.get('/.well-known/agent-card.json')
+    print(response.status_code, response.content)
     assert response.status_code == 200
     response_data = response.json()
 

tests/server/request_handlers/test_grpc_handler.py

@@ -201,6 +201,34 @@ async def test_get_agent_card(
     assert response.version == sample_agent_card.version
 
 
+@pytest.mark.asyncio
+async def test_get_agent_card_with_modifier(
+    mock_request_handler: AsyncMock,
+    sample_agent_card: types.AgentCard,
+    mock_grpc_context: AsyncMock,
+):
+    """Test GetAgentCard call with a card_modifier."""
+
+    def modifier(card: types.AgentCard) -> types.AgentCard:
+        modified_card = card.model_copy()
+        modified_card.name = 'Modified gRPC Agent'
+        return modified_card
+
+    grpc_handler_modified = GrpcHandler(
+        agent_card=sample_agent_card,
+        request_handler=mock_request_handler,
+        card_modifier=modifier,
+    )
+
+    request_proto = a2a_pb2.GetAgentCardRequest()
+    response = await grpc_handler_modified.GetAgentCard(
+        request_proto, mock_grpc_context
+    )
+
+    assert response.name == 'Modified gRPC Agent'
+    assert response.version == sample_agent_card.version
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     'server_error, grpc_status_code, error_message_part',

tests/server/tasks/test_database_push_notification_config_store.py

@@ -3,6 +3,14 @@
 from collections.abc import AsyncGenerator
 
 import pytest
+
+# Skip entire test module if SQLAlchemy is not installed
+pytest.importorskip('sqlalchemy', reason='Database tests require SQLAlchemy')
+pytest.importorskip(
+    'cryptography',
+    reason='Database tests require Cryptography. Install extra encryption',
+)
+
 import pytest_asyncio
 
 from _pytest.mark.structures import ParameterSet
@@ -12,14 +20,6 @@
     create_async_engine,
 )
 
-
-# Skip entire test module if SQLAlchemy is not installed
-pytest.importorskip('sqlalchemy', reason='Database tests require SQLAlchemy')
-pytest.importorskip(
-    'cryptography',
-    reason='Database tests require Cryptography. Install extra encryption',
-)
-
 # Now safe to import SQLAlchemy-dependent modules
 from cryptography.fernet import Fernet
 from sqlalchemy.inspection import inspect