Formatting

Author: holtskinner

examples/helloworld/__main__.py

@@ -5,7 +5,13 @@
 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 (
+    AgentAuthentication,
+    AgentCapabilities,
+    AgentCard,
+    AgentSkill,
+)
+
 
 if __name__ == '__main__':
     skill = AgentSkill(
@@ -32,7 +38,7 @@
         defaultInputModes=['text'],
         defaultOutputModes=['text'],
         capabilities=AgentCapabilities(streaming=True),
-        skills=[skill, extended_skill], # Include both skills
+        skills=[skill, extended_skill],  # Include both skills
         authentication=AgentAuthentication(schemes=['public']),
         # Adding this line to enable extended card support:
         supportsAuthenticatedExtendedCard=True,

src/a2a/client/client.py

@@ -1,10 +1,11 @@
 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 a2a.client.errors import A2AClientHTTPError, A2AClientJSONError
@@ -53,7 +54,10 @@ async def get_agent_card(
             )
             response.raise_for_status()
             public_agent_card_data = response.json()
-            logger.info("Successfully fetched public agent card data: %s", public_agent_card_data) # Added for verbosity
+            logger.info(
+                'Successfully fetched public agent card data: %s',
+                public_agent_card_data,
+            )  # Added for verbosity
             # print(f"DEBUG: Fetched public agent card data:\n{json.dumps(public_agent_card_data, indent=2)}") # Added for direct output
             agent_card = AgentCard.model_validate(public_agent_card_data)
         except httpx.HTTPStatusError as e:
@@ -77,9 +81,9 @@ async def get_agent_card(
             # The extended card URL is relative to the agent's base URL specified *in* the agent card.
             if not agent_card.url:
                 logger.warning(
-                    "Agent card (from %s) indicates support for an extended card "
+                    'Agent card (from %s) indicates support for an extended card '
                     "but does not specify its own base 'url' field. "
-                    "Cannot fetch extended card. Proceeding with public card.",
+                    'Cannot fetch extended card. Proceeding with public card.',
                     public_card_url,
                 )
                 return agent_card
@@ -102,19 +106,30 @@ async def get_agent_card(
                 )
                 extended_response.raise_for_status()
                 extended_agent_card_data = extended_response.json()
-                logger.info("Successfully fetched extended agent card data: %s", extended_agent_card_data) # Added for verbosity
-                print(f"DEBUG: Fetched extended agent card data:\n{json.dumps(extended_agent_card_data, indent=2)}") # Added for direct output
+                logger.info(
+                    'Successfully fetched extended agent card data: %s',
+                    extended_agent_card_data,
+                )  # Added for verbosity
+                print(
+                    f'DEBUG: Fetched extended agent card data:\n{json.dumps(extended_agent_card_data, indent=2)}'
+                )  # Added for direct output
                 # This new card data replaces the old one entirely
                 agent_card = AgentCard.model_validate(extended_agent_card_data)
                 logger.info(
                     'Successfully fetched and using extended agent card from %s',
                     full_extended_card_url,
                 )
-            except (httpx.HTTPStatusError, httpx.RequestError, json.JSONDecodeError, ValidationError) as e:
+            except (
+                httpx.HTTPStatusError,
+                httpx.RequestError,
+                json.JSONDecodeError,
+                ValidationError,
+            ) as e:
                 logger.warning(
                     'Failed to fetch or parse extended agent card from %s. Error: %s. '
                     'Proceeding with the initially fetched public agent card.',
-                    full_extended_card_url, e
+                    full_extended_card_url,
+                    e,
                 )
                 # Fallback to the already parsed public_agent_card (which is 'agent_card' at this point)
 

src/a2a/server/apps/starlette_app.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import traceback
+
 from collections.abc import AsyncGenerator
 from typing import Any
 
@@ -37,6 +38,7 @@
 )
 from a2a.utils.errors import MethodNotImplementedError
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -202,7 +204,7 @@ async def _process_non_streaming_request(
     def _create_response(
         self,
         handler_result: (
-            AsyncGenerator[SendStreamingMessageResponse, None]
+            AsyncGenerator[SendStreamingMessageResponse]
             | JSONRPCErrorResponse
             | JSONRPCResponse
         ),
@@ -225,8 +227,8 @@ def _create_response(
         if isinstance(handler_result, AsyncGenerator):
             # Result is a stream of SendStreamingMessageResponse objects
             async def event_generator(
-                stream: AsyncGenerator[SendStreamingMessageResponse, None],
-            ) -> AsyncGenerator[dict[str, str], None]:
+                stream: AsyncGenerator[SendStreamingMessageResponse],
+            ) -> AsyncGenerator[dict[str, str]]:
                 async for item in stream:
                     yield {'data': item.root.model_dump_json(exclude_none=True)}
 
@@ -247,29 +249,35 @@ async def _handle_get_agent_card(self, request: Request) -> JSONResponse:
         """Handles GET requests for the agent card."""
         # Construct the public view of the agent card.
         public_card_data = {
-            "version": self.agent_card.version,
-            "name": self.agent_card.name,
-            "providerName": self.agent_card.provider.organization if self.agent_card.provider else None,
-            "url": self.agent_card.url,
-            "authentication": self.agent_card.authentication.model_dump(mode='json', exclude_none=True)
-            if self.agent_card.authentication else None, # authentication is a single object, can be None if made Optional
-            "skills": [
+            'version': self.agent_card.version,
+            'name': self.agent_card.name,
+            'providerName': self.agent_card.provider.organization
+            if self.agent_card.provider
+            else None,
+            'url': self.agent_card.url,
+            'authentication': self.agent_card.authentication.model_dump(
+                mode='json', exclude_none=True
+            )
+            if self.agent_card.authentication
+            else None,  # authentication is a single object, can be None if made Optional
+            'skills': [
                 f.model_dump(mode='json', exclude_none=True)
-                for f in self.agent_card.skills if f.id == 'hello_world' # Explicitly filter for public skills
+                for f in self.agent_card.skills
+                if f.id == 'hello_world'  # Explicitly filter for public skills
             ]
             if self.agent_card.skills
-            else [], # Default to empty list if no skills
-            "capabilities": self.agent_card.capabilities.model_dump(
+            else [],  # Default to empty list if no skills
+            'capabilities': self.agent_card.capabilities.model_dump(
                 mode='json', exclude_none=True
             ),
-            "supportsAuthenticatedExtendedCard": (
+            'supportsAuthenticatedExtendedCard': (
                 self.agent_card.supportsAuthenticatedExtendedCard
             ),
             # Include other fields from types.py AgentCard designated as public
-            "description": self.agent_card.description,
-            "documentationUrl": self.agent_card.documentationUrl,
-            "defaultInputModes": self.agent_card.defaultInputModes,
-            "defaultOutputModes": self.agent_card.defaultOutputModes,
+            'description': self.agent_card.description,
+            'documentationUrl': self.agent_card.documentationUrl,
+            'defaultInputModes': self.agent_card.defaultInputModes,
+            'defaultOutputModes': self.agent_card.defaultOutputModes,
         }
         # Filter out None values from the public card data.
         public_card_data_cleaned = {
@@ -283,7 +291,7 @@ async def _handle_get_authenticated_extended_agent_card(
         """Handles GET requests for the authenticated extended agent card."""
         if not self.agent_card.supportsAuthenticatedExtendedCard:
             return JSONResponse(
-                {"error": "Extended agent card not supported or not enabled."},
+                {'error': 'Extended agent card not supported or not enabled.'},
                 status_code=404,
             )
 
@@ -357,7 +365,9 @@ def build(
         Returns:
             A configured Starlette application instance.
         """
-        app_routes = self.routes(agent_card_url, extended_agent_card_url, rpc_url)
+        app_routes = self.routes(
+            agent_card_url, extended_agent_card_url, rpc_url
+        )
         if 'routes' in kwargs:
             kwargs['routes'].extend(app_routes)
         else:

src/a2a/types.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 from enum import Enum
-from typing import Any, Literal, Optional
+from typing import Any, Literal
 
 from pydantic import BaseModel, Field, RootModel
 
@@ -1408,7 +1408,7 @@ class AgentCard(BaseModel):
     """
     The version of the agent - format is up to the provider.
     """
-    supportsAuthenticatedExtendedCard: Optional[bool] = Field(default=None)
+    supportsAuthenticatedExtendedCard: bool | None = Field(default=None)
     """
     Optional field indicating there is an extended card available post authentication at the /agent/authenticatedExtendedCard endpoint.
     """