feat: add auth header to push notifications

a2aproject · Wondr-design · Jan 25, 2026 · Jan 25, 2026 · Jan 25, 2026 · Jan 25, 2026
commit 17624b06aad01f4c48ca93305753ce601ce9a2e0
@@ -52,9 +52,7 @@ async def _dispatch_notification(
     ) -> bool:
         url = push_info.url
         try:
-            headers = None
-            if push_info.token:
-                headers = {'X-A2A-Notification-Token': push_info.token}
+            headers = self._build_headers(push_info)
             response = await self._client.post(
                 url,
                 json=task.model_dump(mode='json', exclude_none=True),
@@ -72,3 +70,30 @@ async def _dispatch_notification(
             )
             return False
         return True
+
+    @staticmethod
+    def _authorization_header(
+        push_info: PushNotificationConfig,
+    ) -> str | None:
+        auth = push_info.authentication
+        if not auth or not auth.credentials:
+            return None
+        schemes = [scheme for scheme in auth.schemes if scheme]
+        if not schemes:
+            return None
+        scheme = next(
+            (scheme for scheme in schemes if scheme.lower() == 'bearer'),
+            schemes[0],
+        )
+        return f'{scheme} {auth.credentials}'
+
+    def _build_headers(
+        self, push_info: PushNotificationConfig
+    ) -> dict[str, str] | None:
+        headers: dict[str, str] = {}
+        if push_info.token:
+            headers['X-A2A-Notification-Token'] = push_info.token
+        authorization = self._authorization_header(push_info)
+        if authorization:
+            headers['Authorization'] = authorization
+        return headers or None
@@ -8,6 +8,7 @@
     BasePushNotificationSender,
 )
 from a2a.types import (
+    PushNotificationAuthenticationInfo,
     PushNotificationConfig,
     Task,
     TaskState,
@@ -29,8 +30,14 @@ def create_sample_push_config(
     url: str = 'http://example.com/callback',
     config_id: str = 'cfg1',
     token: str | None = None,
+    authentication: PushNotificationAuthenticationInfo | None = None,
 ) -> PushNotificationConfig:
-    return PushNotificationConfig(id=config_id, url=url, token=token)
+    return PushNotificationConfig(
+        id=config_id,
+        url=url,
+        token=token,
+        authentication=authentication,
+    )
 
 
 class TestBasePushNotificationSender(unittest.IsolatedAsyncioTestCase):
@@ -92,6 +99,37 @@ async def test_send_notification_with_token_success(self) -> None:
         )
         mock_response.raise_for_status.assert_called_once()
 
+    async def test_send_notification_with_auth_header(self) -> None:
+        task_id = 'task_send_auth'
+        task_data = create_sample_task(task_id=task_id)
+        auth = PushNotificationAuthenticationInfo(
+            schemes=['Basic', 'Bearer'], credentials='token_or_jwt'
+        )
+        config = create_sample_push_config(
+            url='http://notify.me/here',
+            token='unique_token',
+            authentication=auth,
+        )
+        self.mock_config_store.get_info.return_value = [config]
+
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.status_code = 200
+        self.mock_httpx_client.post.return_value = mock_response
+
+        await self.sender.send_notification(task_data)
+
+        self.mock_config_store.get_info.assert_awaited_once_with
+
+        self.mock_httpx_client.post.assert_awaited_once_with(
+            config.url,
+            json=task_data.model_dump(mode='json', exclude_none=True),
+            headers={
+                'X-A2A-Notification-Token': 'unique_token',
+                'Authorization': 'Bearer token_or_jwt',
+            },
+        )
+        mock_response.raise_for_status.assert_called_once()
+
     async def test_send_notification_no_config(self) -> None:
         task_id = 'task_send_no_config'
         task_data = create_sample_task(task_id=task_id)