feat: Add functionality for extended agent card. (#31)

* Add functionality for extended agent card. Does not support authentication yet

* Formatting

* Pass a default and an  extended card to the starlette app. Also add tests

* Address review comments

* Add docstring back and remove extra attribute not used anymore

* Use model_copy for specific_extended_agent_card

* return a 404 instead of public card from extended card EP

* eat any exception thrown during auth card get.

---------

Co-authored-by: Holt Skinner <[email protected]>
Co-authored-by: Holt Skinner <[email protected]>

Author: dmandar

examples/helloworld/__main__.py

@@ -5,7 +5,11 @@
 from a2a.server.apps import A2AStarletteApplication
 from a2a.server.request_handlers import DefaultRequestHandler
 from a2a.server.tasks import InMemoryTaskStore
-from a2a.types import AgentCapabilities, AgentCard, AgentSkill
+from a2a.types import (
+    AgentCapabilities,
+    AgentCard,
+    AgentSkill,
+)
 
 
 if __name__ == '__main__':
@@ -17,25 +21,48 @@
         examples=['hi', 'hello world'],
     )
 
-    agent_card = AgentCard(
+    extended_skill = AgentSkill(
+        id='super_hello_world',
+        name='Returns a SUPER Hello World',
+        description='A more enthusiastic greeting, only for authenticated users.',
+        tags=['hello world', 'super', 'extended'],
+        examples=['super hi', 'give me a super hello'],
+    )
+
+    # This will be the public-facing agent card
+    public_agent_card = AgentCard(
         name='Hello World Agent',
         description='Just a hello world agent',
         url='http://localhost:9999/',
         version='1.0.0',
         defaultInputModes=['text'],
         defaultOutputModes=['text'],
         capabilities=AgentCapabilities(streaming=True),
-        skills=[skill],
+        skills=[skill],  # Only the basic skill for the public card
+        supportsAuthenticatedExtendedCard=True,
+    )
+
+    # This will be the authenticated extended agent card
+    # It includes the additional 'extended_skill'
+    specific_extended_agent_card = public_agent_card.model_copy(
+        update={
+            'name': 'Hello World Agent - Extended Edition', # Different name for clarity
+            'description': 'The full-featured hello world agent for authenticated users.',
+            'version': '1.0.1', # Could even be a different version
+            # Capabilities and other fields like url, defaultInputModes, defaultOutputModes,
+            # supportsAuthenticatedExtendedCard are inherited from public_agent_card unless specified here.
+            'skills': [skill, extended_skill],  # Both skills for the extended card
+        }
     )
 
     request_handler = DefaultRequestHandler(
         agent_executor=HelloWorldAgentExecutor(),
         task_store=InMemoryTaskStore(),
     )
 
-    server = A2AStarletteApplication(
-        agent_card=agent_card, http_handler=request_handler
-    )
+    server = A2AStarletteApplication(agent_card=public_agent_card,
+                                     http_handler=request_handler,
+                                     extended_agent_card=specific_extended_agent_card)
     import uvicorn
 
     uvicorn.run(server.build(), host='0.0.0.0', port=9999)

examples/helloworld/test_client.py

@@ -1,19 +1,70 @@
-from a2a.client import A2AClient
+import logging  # Import the logging module
 from typing import Any
-import httpx
 from uuid import uuid4
-from a2a.types import (
-    SendMessageRequest,
-    MessageSendParams,
-    SendStreamingMessageRequest,
-)
+
+import httpx
+
+from a2a.client import A2ACardResolver, A2AClient
+from a2a.types import (AgentCard, MessageSendParams, SendMessageRequest,
+                       SendStreamingMessageRequest)
 
 
 async def main() -> None:
+    PUBLIC_AGENT_CARD_PATH = "/.well-known/agent.json"
+    EXTENDED_AGENT_CARD_PATH = "/agent/authenticatedExtendedCard"
+
+    # Configure logging to show INFO level messages
+    logging.basicConfig(level=logging.INFO)
+    logger = logging.getLogger(__name__)  # Get a logger instance
+
+    base_url = 'http://localhost:9999'
+
     async with httpx.AsyncClient() as httpx_client:
-        client = await A2AClient.get_client_from_agent_card_url(
-            httpx_client, 'http://localhost:9999'
+        # Initialize A2ACardResolver
+        resolver = A2ACardResolver(
+            httpx_client=httpx_client,
+            base_url=base_url,
+            # agent_card_path uses default, extended_agent_card_path also uses default
         )
+
+        # Fetch Public Agent Card and Initialize Client
+        final_agent_card_to_use: AgentCard | None = None
+
+        try:
+            logger.info(f"Attempting to fetch public agent card from: {base_url}{PUBLIC_AGENT_CARD_PATH}")
+            _public_card = await resolver.get_agent_card()  # Fetches from default public path
+            logger.info("Successfully fetched public agent card:")
+            logger.info(_public_card.model_dump_json(indent=2, exclude_none=True))
+            final_agent_card_to_use = _public_card
+            logger.info("\nUsing PUBLIC agent card for client initialization (default).")
+
+            if _public_card.supportsAuthenticatedExtendedCard:
+                try:
+                    logger.info(f"\nPublic card supports authenticated extended card. Attempting to fetch from: {base_url}{EXTENDED_AGENT_CARD_PATH}")
+                    auth_headers_dict = {"Authorization": "Bearer dummy-token-for-extended-card"}
+                    _extended_card = await resolver.get_agent_card(
+                        relative_card_path=EXTENDED_AGENT_CARD_PATH,
+                        http_kwargs={"headers": auth_headers_dict}
+                    )
+                    logger.info("Successfully fetched authenticated extended agent card:")
+                    logger.info(_extended_card.model_dump_json(indent=2, exclude_none=True))
+                    final_agent_card_to_use = _extended_card # Update to use the extended card
+                    logger.info("\nUsing AUTHENTICATED EXTENDED agent card for client initialization.")
+                except Exception as e_extended:
+                    logger.warning(f"Failed to fetch extended agent card: {e_extended}. Will proceed with public card.", exc_info=True)
+            elif _public_card: # supportsAuthenticatedExtendedCard is False or None
+                logger.info("\nPublic card does not indicate support for an extended card. Using public card.")
+
+        except Exception as e:
+            logger.error(f"Critical error fetching public agent card: {e}", exc_info=True)
+            raise RuntimeError("Failed to fetch the public agent card. Cannot continue.") from e
+
+       
+        client = A2AClient(
+            httpx_client=httpx_client, agent_card=final_agent_card_to_use
+        )
+        logger.info("A2AClient initialized.")
+
         send_message_payload: dict[str, Any] = {
             'message': {
                 'role': 'user',

src/a2a/client/client.py

@@ -1,31 +1,25 @@
 import json
-
+import logging
 from collections.abc import AsyncGenerator
 from typing import Any
 from uuid import uuid4
 
 import httpx
-
 from httpx_sse import SSEError, aconnect_sse
+from pydantic import ValidationError
 
 from a2a.client.errors import A2AClientHTTPError, A2AClientJSONError
-from a2a.types import (
-    AgentCard,
-    CancelTaskRequest,
-    CancelTaskResponse,
-    GetTaskPushNotificationConfigRequest,
-    GetTaskPushNotificationConfigResponse,
-    GetTaskRequest,
-    GetTaskResponse,
-    SendMessageRequest,
-    SendMessageResponse,
-    SendStreamingMessageRequest,
-    SendStreamingMessageResponse,
-    SetTaskPushNotificationConfigRequest,
-    SetTaskPushNotificationConfigResponse,
-)
+from a2a.types import (AgentCard, CancelTaskRequest, CancelTaskResponse,
+                       GetTaskPushNotificationConfigRequest,
+                       GetTaskPushNotificationConfigResponse, GetTaskRequest,
+                       GetTaskResponse, SendMessageRequest,
+                       SendMessageResponse, SendStreamingMessageRequest,
+                       SendStreamingMessageResponse,
+                       SetTaskPushNotificationConfigRequest,
+                       SetTaskPushNotificationConfigResponse)
 from a2a.utils.telemetry import SpanKind, trace_class
 
+logger = logging.getLogger(__name__)
 
 class A2ACardResolver:
     """Agent Card resolver."""
@@ -48,11 +42,19 @@ def __init__(
         self.httpx_client = httpx_client
 
     async def get_agent_card(
-        self, http_kwargs: dict[str, Any] | None = None
+        self,
+        relative_card_path: str | None = None,
+        http_kwargs: dict[str, Any] | None = None,
     ) -> AgentCard:
-        """Fetches the agent card from the specified URL.
+        """Fetches an agent card from a specified path relative to the base_url.
+
+        If relative_card_path is None, it defaults to the resolver's configured
+        agent_card_path (for the public agent card).
 
         Args:
+            relative_card_path: Optional path to the agent card endpoint,
+                relative to the base URL. If None, uses the default public
+                agent card path.
             http_kwargs: Optional dictionary of keyword arguments to pass to the
                 underlying httpx.get request.
 
@@ -64,21 +66,47 @@ async def get_agent_card(
             A2AClientJSONError: If the response body cannot be decoded as JSON
                 or validated against the AgentCard schema.
         """
+        if relative_card_path is None:
+            # Use the default public agent card path configured during initialization
+            path_segment = self.agent_card_path
+        else:
+            path_segment = relative_card_path.lstrip('/')
+
+        target_url = f'{self.base_url}/{path_segment}'
+
         try:
             response = await self.httpx_client.get(
-                f'{self.base_url}/{self.agent_card_path}',
+                target_url,
                 **(http_kwargs or {}),
             )
             response.raise_for_status()
-            return AgentCard.model_validate(response.json())
+            agent_card_data = response.json()
+            logger.info(
+                'Successfully fetched agent card data from %s: %s',
+                target_url,
+                agent_card_data,
+            )
+            agent_card = AgentCard.model_validate(agent_card_data)
         except httpx.HTTPStatusError as e:
-            raise A2AClientHTTPError(e.response.status_code, str(e)) from e
+            raise A2AClientHTTPError(
+                e.response.status_code,
+                f'Failed to fetch agent card from {target_url}: {e}',
+            ) from e
         except json.JSONDecodeError as e:
-            raise A2AClientJSONError(str(e)) from e
+            raise A2AClientJSONError(
+                f'Failed to parse JSON for agent card from {target_url}: {e}'
+            ) from e
         except httpx.RequestError as e:
             raise A2AClientHTTPError(
-                503, f'Network communication error: {e}'
+                503,
+                f'Network communication error fetching agent card from {target_url}: {e}',
             ) from e
+        except ValidationError as e:  # Pydantic validation error
+            raise A2AClientJSONError(
+                f'Failed to validate agent card structure from {target_url}: {e.json()}'
+            ) from e
+
+        return agent_card
 
 
 @trace_class(kind=SpanKind.CLIENT)
@@ -119,15 +147,19 @@ async def get_client_from_agent_card_url(
         agent_card_path: str = '/.well-known/agent.json',
         http_kwargs: dict[str, Any] | None = None,
     ) -> 'A2AClient':
-        """Fetches the AgentCard and initializes an A2A client.
+        """Fetches the public AgentCard and initializes an A2A client.
+
+        This method will always fetch the public agent card. If an authenticated
+        or extended agent card is required, the A2ACardResolver should be used
+        directly to fetch the specific card, and then the A2AClient should be
+        instantiated with it.
 
         Args:
             httpx_client: An async HTTP client instance (e.g., httpx.AsyncClient).
             base_url: The base URL of the agent's host.
             agent_card_path: The path to the agent card endpoint, relative to the base URL.
             http_kwargs: Optional dictionary of keyword arguments to pass to the
                 underlying httpx.get request when fetching the agent card.
-
         Returns:
             An initialized `A2AClient` instance.
 
@@ -137,7 +169,7 @@ async def get_client_from_agent_card_url(
         """
         agent_card: AgentCard = await A2ACardResolver(
             httpx_client, base_url=base_url, agent_card_path=agent_card_path
-        ).get_agent_card(http_kwargs=http_kwargs)
+        ).get_agent_card(http_kwargs=http_kwargs) # Fetches public card by default
         return A2AClient(httpx_client=httpx_client, agent_card=agent_card)
 
     async def send_message(