diff --git a/src/a2a/compat/v0_3/request_handler.py b/src/a2a/compat/v0_3/request_handler.py index a64b7a863..9a530815d 100644 --- a/src/a2a/compat/v0_3/request_handler.py +++ b/src/a2a/compat/v0_3/request_handler.py @@ -6,7 +6,7 @@ from a2a.compat.v0_3 import types as types_v03 from a2a.server.context import ServerCallContext from a2a.server.request_handlers.request_handler import RequestHandler -from a2a.types.a2a_pb2 import Task +from a2a.types.a2a_pb2 import ListTaskPushNotificationConfigsRequest, Task from a2a.utils import proto_utils as core_proto_utils from a2a.utils.errors import TaskNotFoundError @@ -85,6 +85,24 @@ async def on_get_task_push_notification_config( v10_req = conversions.to_core_get_task_push_notification_config_request( request ) + # v0.3 allows a get by task id alone. Prefer the config stored under + # the task id, which is where a config created without an id goes. + # Otherwise take the first, like v0.3 did. + if not v10_req.id: + v10_resp = await self.request_handler.on_list_task_push_notification_configs( + ListTaskPushNotificationConfigsRequest(task_id=v10_req.task_id), + context, + ) + configs = list(v10_resp.configs) + v10_config = next( + (c for c in configs if c.id == v10_req.task_id), + configs[0] if configs else None, + ) + if v10_config is None: + raise TaskNotFoundError + return conversions.to_compat_task_push_notification_config( + v10_config + ) v10_config = ( await self.request_handler.on_get_task_push_notification_config( v10_req, context diff --git a/tests/compat/v0_3/test_request_handler.py b/tests/compat/v0_3/test_request_handler.py index 26ad74264..002797338 100644 --- a/tests/compat/v0_3/test_request_handler.py +++ b/tests/compat/v0_3/test_request_handler.py @@ -2,10 +2,14 @@ import pytest +from a2a.auth.user import UnauthenticatedUser from a2a.compat.v0_3 import types as types_v03 from a2a.compat.v0_3.request_handler import RequestHandler03 +from a2a.server.agent_execution import AgentExecutor from a2a.server.context import ServerCallContext +from a2a.server.request_handlers import DefaultRequestHandlerV2 from a2a.server.request_handlers.request_handler import RequestHandler +from a2a.server.tasks import InMemoryPushNotificationConfigStore, TaskStore from a2a.types.a2a_pb2 import ( AgentCapabilities, AgentCard, @@ -241,6 +245,79 @@ async def test_on_get_task_push_notification_config( assert result.push_notification_config.url == 'http://example.com' +@pytest.fixture +def v03_handler_with_push_store(): + task_store = AsyncMock(spec=TaskStore) + task_store.get.return_value = V10Task(id='task-1', context_id='ctx-1') + core_handler = DefaultRequestHandlerV2( + agent_executor=MagicMock(spec=AgentExecutor), + task_store=task_store, + push_config_store=InMemoryPushNotificationConfigStore(), + agent_card=AgentCard( + capabilities=AgentCapabilities(push_notifications=True) + ), + ) + return RequestHandler03(request_handler=core_handler) + + +@pytest.mark.anyio +@pytest.mark.parametrize( + ('config_ids', 'expected_id'), + [ + ([None], 'task-1'), + (['cfg-1'], 'cfg-1'), + (['cfg-1', None], 'task-1'), + ], +) +async def test_on_get_task_push_notification_config_by_task_id( + v03_handler_with_push_store, config_ids, expected_id +): + """Without a config id, prefer the config stored under the task id.""" + context = ServerCallContext(user=UnauthenticatedUser()) + for config_id in config_ids: + await v03_handler_with_push_store.on_create_task_push_notification_config( + types_v03.SetTaskPushNotificationConfigRequest( + id='req-1', + method='tasks/pushNotificationConfig/set', + params=types_v03.TaskPushNotificationConfig( + task_id='task-1', + push_notification_config=types_v03.PushNotificationConfig( + id=config_id, url='http://example.com' + ), + ), + ), + context, + ) + + result = ( + await v03_handler_with_push_store.on_get_task_push_notification_config( + types_v03.GetTaskPushNotificationConfigRequest( + id='req-2', + method='tasks/pushNotificationConfig/get', + params=types_v03.TaskIdParams(id='task-1'), + ), + context, + ) + ) + + assert result.push_notification_config.id == expected_id + + +@pytest.mark.anyio +async def test_on_get_task_push_notification_config_by_task_id_none_set( + v03_handler_with_push_store, +): + with pytest.raises(TaskNotFoundError): + await v03_handler_with_push_store.on_get_task_push_notification_config( + types_v03.GetTaskPushNotificationConfigRequest( + id='req-1', + method='tasks/pushNotificationConfig/get', + params=types_v03.TaskIdParams(id='task-1'), + ), + ServerCallContext(user=UnauthenticatedUser()), + ) + + @pytest.mark.anyio async def test_on_create_task_push_notification_config( v03_handler, mock_core_handler, mock_context