Merge branch 'main' into kthota/deprecated-agent-card-path

Author: kthota-g

.github/actions/spelling/allow.txt

@@ -45,6 +45,7 @@ INR
 isready
 JPY
 JSONRPCt
+JWS
 kwarg
 langgraph
 lifecycles

.github/workflows/update-a2a-types.yml

@@ -47,10 +47,10 @@ jobs:
           token: ${{ secrets.A2A_BOT_PAT }}
           committer: a2a-bot <[email protected]>
           author: a2a-bot <[email protected]>
-          commit-message: 'feat(spec): Update A2A types from specification 🤖'
-          title: 'feat(spec): Update A2A types from specification 🤖'
+          commit-message: '${{ github.event.client_payload.message }}'
+          title: '${{ github.event.client_payload.message }}'
           body: |
-            This PR updates `src/a2a/types.py` based on the latest `specification/json/a2a.json` from [a2aproject/A2A](https://github.com/a2aproject/A2A/commit/${{ github.event.client_payload.sha }}).
+            Commit: https://github.com/a2aproject/A2A/commit/${{ github.event.client_payload.sha }}
           branch: auto-update-a2a-types-${{ github.event.client_payload.sha }}
           base: main
           labels: |

src/a2a/_base.py

@@ -64,7 +64,7 @@ def __setattr__(self, name: str, value: Any) -> None:
         # Get the map and find the corresponding snake_case field name.
         field_name = type(self)._get_alias_map().get(name)  # noqa: SLF001
 
-        if field_name:
+        if field_name and field_name != name:
             # An alias was used, issue a warning.
             warnings.warn(
                 (
@@ -83,7 +83,7 @@ def __getattr__(self, name: str) -> Any:
         # Get the map and find the corresponding snake_case field name.
         field_name = type(self)._get_alias_map().get(name)  # noqa: SLF001
 
-        if field_name:
+        if field_name and field_name != name:
             # An alias was used, issue a warning.
             warnings.warn(
                 (

src/a2a/grpc/a2a_pb2.py

@@ -202,7 +202,7 @@ class AgentInterface(_message.Message):
     def __init__(self, url: _Optional[str] = ..., transport: _Optional[str] = ...) -> None: ...
 
 class AgentCard(_message.Message):
-    __slots__ = ("protocol_version", "name", "description", "url", "preferred_transport", "additional_interfaces", "provider", "version", "documentation_url", "capabilities", "security_schemes", "security", "default_input_modes", "default_output_modes", "skills", "supports_authenticated_extended_card")
+    __slots__ = ("protocol_version", "name", "description", "url", "preferred_transport", "additional_interfaces", "provider", "version", "documentation_url", "capabilities", "security_schemes", "security", "default_input_modes", "default_output_modes", "skills", "supports_authenticated_extended_card", "signatures")
     class SecuritySchemesEntry(_message.Message):
         __slots__ = ("key", "value")
         KEY_FIELD_NUMBER: _ClassVar[int]
@@ -226,6 +226,7 @@ class AgentCard(_message.Message):
     DEFAULT_OUTPUT_MODES_FIELD_NUMBER: _ClassVar[int]
     SKILLS_FIELD_NUMBER: _ClassVar[int]
     SUPPORTS_AUTHENTICATED_EXTENDED_CARD_FIELD_NUMBER: _ClassVar[int]
+    SIGNATURES_FIELD_NUMBER: _ClassVar[int]
     protocol_version: str
     name: str
     description: str
@@ -242,7 +243,8 @@ class AgentCard(_message.Message):
     default_output_modes: _containers.RepeatedScalarFieldContainer[str]
     skills: _containers.RepeatedCompositeFieldContainer[AgentSkill]
     supports_authenticated_extended_card: bool
-    def __init__(self, protocol_version: _Optional[str] = ..., name: _Optional[str] = ..., description: _Optional[str] = ..., url: _Optional[str] = ..., preferred_transport: _Optional[str] = ..., additional_interfaces: _Optional[_Iterable[_Union[AgentInterface, _Mapping]]] = ..., provider: _Optional[_Union[AgentProvider, _Mapping]] = ..., version: _Optional[str] = ..., documentation_url: _Optional[str] = ..., capabilities: _Optional[_Union[AgentCapabilities, _Mapping]] = ..., security_schemes: _Optional[_Mapping[str, SecurityScheme]] = ..., security: _Optional[_Iterable[_Union[Security, _Mapping]]] = ..., default_input_modes: _Optional[_Iterable[str]] = ..., default_output_modes: _Optional[_Iterable[str]] = ..., skills: _Optional[_Iterable[_Union[AgentSkill, _Mapping]]] = ..., supports_authenticated_extended_card: bool = ...) -> None: ...
+    signatures: _containers.RepeatedCompositeFieldContainer[AgentCardSignature]
+    def __init__(self, protocol_version: _Optional[str] = ..., name: _Optional[str] = ..., description: _Optional[str] = ..., url: _Optional[str] = ..., preferred_transport: _Optional[str] = ..., additional_interfaces: _Optional[_Iterable[_Union[AgentInterface, _Mapping]]] = ..., provider: _Optional[_Union[AgentProvider, _Mapping]] = ..., version: _Optional[str] = ..., documentation_url: _Optional[str] = ..., capabilities: _Optional[_Union[AgentCapabilities, _Mapping]] = ..., security_schemes: _Optional[_Mapping[str, SecurityScheme]] = ..., security: _Optional[_Iterable[_Union[Security, _Mapping]]] = ..., default_input_modes: _Optional[_Iterable[str]] = ..., default_output_modes: _Optional[_Iterable[str]] = ..., skills: _Optional[_Iterable[_Union[AgentSkill, _Mapping]]] = ..., supports_authenticated_extended_card: bool = ..., signatures: _Optional[_Iterable[_Union[AgentCardSignature, _Mapping]]] = ...) -> None: ...
 
 class AgentProvider(_message.Message):
     __slots__ = ("url", "organization")
@@ -292,6 +294,16 @@ class AgentSkill(_message.Message):
     output_modes: _containers.RepeatedScalarFieldContainer[str]
     def __init__(self, id: _Optional[str] = ..., name: _Optional[str] = ..., description: _Optional[str] = ..., tags: _Optional[_Iterable[str]] = ..., examples: _Optional[_Iterable[str]] = ..., input_modes: _Optional[_Iterable[str]] = ..., output_modes: _Optional[_Iterable[str]] = ...) -> None: ...
 
+class AgentCardSignature(_message.Message):
+    __slots__ = ("protected", "signature", "header")
+    PROTECTED_FIELD_NUMBER: _ClassVar[int]
+    SIGNATURE_FIELD_NUMBER: _ClassVar[int]
+    HEADER_FIELD_NUMBER: _ClassVar[int]
+    protected: str
+    signature: str
+    header: _struct_pb2.Struct
+    def __init__(self, protected: _Optional[str] = ..., signature: _Optional[str] = ..., header: _Optional[_Union[_struct_pb2.Struct, _Mapping]] = ...) -> None: ...
+
 class TaskPushNotificationConfig(_message.Message):
     __slots__ = ("name", "push_notification_config")
     NAME_FIELD_NUMBER: _ClassVar[int]

src/a2a/grpc/a2a_pb2.pyi

@@ -280,12 +280,18 @@ async def on_message_send(
         consumer = EventConsumer(queue)
         producer_task.add_done_callback(consumer.agent_task_callback)
 
-        interrupted = False
+        blocking = True  # Default to blocking behavior
+        if params.configuration and params.configuration.blocking is False:
+            blocking = False
+
+        interrupted_or_non_blocking = False
         try:
             (
                 result,
-                interrupted,
-            ) = await result_aggregator.consume_and_break_on_interrupt(consumer)
+                interrupted_or_non_blocking,
+            ) = await result_aggregator.consume_and_break_on_interrupt(
+                consumer, blocking=blocking
+            )
             if not result:
                 raise ServerError(error=InternalError())
 
@@ -300,7 +306,7 @@ async def on_message_send(
             logger.error(f'Agent execution failed. Error: {e}')
             raise
         finally:
-            if interrupted:
+            if interrupted_or_non_blocking:
                 # TODO: Track this disconnected cleanup task.
                 asyncio.create_task(  # noqa: RUF006
                     self._cleanup_producer(producer_task, task_id)

src/a2a/server/request_handlers/default_request_handler.py

@@ -92,15 +92,19 @@ async def consume_all(
         return await self.task_manager.get_task()
 
     async def consume_and_break_on_interrupt(
-        self, consumer: EventConsumer
+        self, consumer: EventConsumer, blocking: bool = True
     ) -> tuple[Task | Message | None, bool]:
         """Processes the event stream until completion or an interruptable state is encountered.
 
-        Interruptable states currently include `TaskState.auth_required`.
+        If `blocking` is False, it returns after the first event that creates a Task or Message.
+        If `blocking` is True, it waits for completion unless an `auth_required`
+        state is encountered, which is always an interruption.
         If interrupted, consumption continues in a background task.
 
         Args:
             consumer: The `EventConsumer` to read events from.
+            blocking: If `False`, the method returns as soon as a task/message
+                      is available. If `True`, it waits for a terminal state.
 
         Returns:
             A tuple containing:
@@ -117,10 +121,15 @@ async def consume_and_break_on_interrupt(
                 self._message = event
                 return event, False
             await self.task_manager.process(event)
-            if (
+
+            should_interrupt = False
+            is_auth_required = (
                 isinstance(event, Task | TaskStatusUpdateEvent)
                 and event.status.state == TaskState.auth_required
-            ):
+            )
+
+            # Always interrupt on auth_required, as it needs external action.
+            if is_auth_required:
                 # auth-required is a special state: the message should be
                 # escalated back to the caller, but the agent is expected to
                 # continue producing events once the authorization is received
@@ -130,6 +139,16 @@ async def consume_and_break_on_interrupt(
                 logger.debug(
                     'Encountered an auth-required task: breaking synchronous message/send flow.'
                 )
+                should_interrupt = True
+            # For non-blocking calls, interrupt as soon as a task is available.
+            elif not blocking:
+                logger.debug(
+                    'Non-blocking call: returning task after first event.'
+                )
+                should_interrupt = True
+
+            if should_interrupt:
+                # Continue consuming the rest of the events in the background.
                 # TODO: We should track all outstanding tasks to ensure they eventually complete.
                 asyncio.create_task(self._continue_consuming(event_stream))  # noqa: RUF006
                 interrupted = True

src/a2a/server/tasks/result_aggregator.py

@@ -48,6 +48,27 @@ class APIKeySecurityScheme(A2ABaseModel):
     """
 
 
+class AgentCardSignature(A2ABaseModel):
+    """
+    AgentCardSignature represents a JWS signature of an AgentCard.
+    This follows the JSON format of an RFC 7515 JSON Web Signature (JWS).
+    """
+
+    header: dict[str, Any] | None = None
+    """
+    The unprotected JWS header values.
+    """
+    protected: str
+    """
+    The protected JWS header for the signature. This is a Base64url-encoded
+    JSON object, as per RFC 7515.
+    """
+    signature: str
+    """
+    The computed signature, Base64url-encoded.
+    """
+
+
 class AgentExtension(A2ABaseModel):
     """
     A declaration of a protocol extension supported by an Agent.
@@ -75,16 +96,23 @@ class AgentExtension(A2ABaseModel):
 class AgentInterface(A2ABaseModel):
     """
     Declares a combination of a target URL and a transport protocol for interacting with the agent.
+    This allows agents to expose the same functionality over multiple transport mechanisms.
     """
 
-    transport: str
+    transport: str = Field(..., examples=['JSONRPC', 'GRPC', 'HTTP+JSON'])
     """
-    The transport protocol supported at this URL. This is a string to allow for future
-    extension. Core supported transports include 'JSONRPC', 'GRPC', and 'HTTP+JSON'.
+    The transport protocol supported at this URL.
     """
-    url: str
+    url: str = Field(
+        ...,
+        examples=[
+            'https://api.example.com/a2a/v1',
+            'https://grpc.example.com/a2a',
+            'https://rest.example.com/v1',
+        ],
+    )
     """
-    The URL where this interface is available.
+    The URL where this interface is available. Must be a valid absolute HTTPS URL in production.
     """
 
 
@@ -928,6 +956,16 @@ class TextPart(A2ABaseModel):
     """
 
 
+class TransportProtocol(str, Enum):
+    """
+    Supported A2A transport protocols.
+    """
+
+    jsonrpc = 'JSONRPC'
+    grpc = 'GRPC'
+    http_json = 'HTTP+JSON'
+
+
 class UnsupportedOperationError(A2ABaseModel):
     """
     An A2A-specific error indicating that the requested operation is not supported by the agent.
@@ -1615,7 +1653,16 @@ class AgentCard(A2ABaseModel):
     additional_interfaces: list[AgentInterface] | None = None
     """
     A list of additional supported interfaces (transport and URL combinations).
-    A client can use any of these to communicate with the agent.
+    This allows agents to expose multiple transports, potentially at different URLs.
+
+    Best practices:
+    - SHOULD include all supported transports for completeness
+    - SHOULD include an entry matching the main 'url' and 'preferredTransport'
+    - MAY reuse URLs if multiple transports are available at the same endpoint
+    - MUST accurately declare the transport available at each URL
+
+    Clients can select any interface from this list based on their transport capabilities
+    and preferences. This enables transport negotiation and fallback scenarios.
     """
     capabilities: AgentCapabilities
     """
@@ -1650,9 +1697,16 @@ class AgentCard(A2ABaseModel):
     """
     A human-readable name for the agent.
     """
-    preferred_transport: str | None = None
+    preferred_transport: str | None = Field(
+        default='JSONRPC', examples=['JSONRPC', 'GRPC', 'HTTP+JSON']
+    )
     """
-    The transport protocol for the preferred endpoint. Defaults to 'JSONRPC' if not specified.
+    The transport protocol for the preferred endpoint (the main 'url' field).
+    If not specified, defaults to 'JSONRPC'.
+
+    IMPORTANT: The transport specified here MUST be available at the main 'url'.
+    This creates a binding between the main URL and its supported transport protocol.
+    Clients should prefer this transport and URL combination when both are supported.
     """
     protocol_version: str | None = '0.2.6'
     """
@@ -1672,6 +1726,10 @@ class AgentCard(A2ABaseModel):
     A declaration of the security schemes available to authorize requests. The key is the
     scheme name. Follows the OpenAPI 3.0 Security Scheme Object.
     """
+    signatures: list[AgentCardSignature] | None = None
+    """
+    JSON Web Signatures computed for this AgentCard.
+    """
     skills: list[AgentSkill]
     """
     The set of skills, or distinct capabilities, that the agent can perform.
@@ -1681,9 +1739,10 @@ class AgentCard(A2ABaseModel):
     If true, the agent can provide an extended agent card with additional details
     to authenticated users. Defaults to false.
     """
-    url: str
+    url: str = Field(..., examples=['https://api.example.com/a2a/v1'])
     """
     The preferred endpoint URL for interacting with the agent.
+    This URL MUST support the transport specified by 'preferredTransport'.
     """
     version: str = Field(..., examples=['1.0.0'])
     """

src/a2a/types.py

@@ -384,6 +384,44 @@ async def raiser_gen_interrupt():
         )
         self.mock_task_manager.get_task.assert_not_called()
 
+    @patch('asyncio.create_task')
+    async def test_consume_and_break_non_blocking(
+        self, mock_create_task: MagicMock
+    ):
+        """Test that with blocking=False, the method returns after the first event."""
+        first_event = create_sample_task('non_blocking_task')
+        event_after = create_sample_message('should be consumed later')
+
+        async def mock_consume_generator():
+            yield first_event
+            yield event_after
+
+        self.mock_event_consumer.consume_all.return_value = (
+            mock_consume_generator()
+        )
+        # After processing `first_event`, the current result will be that task.
+        self.aggregator.task_manager.get_task.return_value = first_event
+
+        self.aggregator._continue_consuming = AsyncMock()
+        mock_create_task.side_effect = lambda coro: asyncio.ensure_future(coro)
+
+        (
+            result,
+            interrupted,
+        ) = await self.aggregator.consume_and_break_on_interrupt(
+            self.mock_event_consumer, blocking=False
+        )
+
+        self.assertEqual(result, first_event)
+        self.assertTrue(interrupted)
+        self.mock_task_manager.process.assert_called_once_with(first_event)
+        mock_create_task.assert_called_once()
+        # The background task should be created with the remaining stream
+        self.aggregator._continue_consuming.assert_called_once()
+        self.assertIsInstance(
+            self.aggregator._continue_consuming.call_args[0][0], AsyncIterator
+        )
+
     @patch('asyncio.create_task')  # To verify _continue_consuming is called
     async def test_continue_consuming_processes_remaining_events(
         self, mock_create_task: MagicMock

tests/server/tasks/test_result_aggregator.py

@@ -1552,15 +1552,13 @@ def test_camelCase() -> None:
     )
 
     # Test setting an attribute via camelCase alias
-    # We expect a DeprecationWarning with a specific message
     with pytest.warns(
         DeprecationWarning,
         match="Setting field 'supportsAuthenticatedExtendedCard'",
     ):
         agent_card.supportsAuthenticatedExtendedCard = False
 
     # Test getting an attribute via camelCase alias
-    # We expect another DeprecationWarning with a specific message
     with pytest.warns(
         DeprecationWarning, match="Accessing field 'defaultInputModes'"
     ):