diff --git a/src/a2a/compat/v0_3/jsonrpc_adapter.py b/src/a2a/compat/v0_3/jsonrpc_adapter.py index 303f5f9f8..04df1d0d0 100644 --- a/src/a2a/compat/v0_3/jsonrpc_adapter.py +++ b/src/a2a/compat/v0_3/jsonrpc_adapter.py @@ -1,7 +1,7 @@ import logging from collections.abc import AsyncIterable, AsyncIterator -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, get_args from sse_starlette.sse import EventSourceResponse from starlette.responses import JSONResponse @@ -42,6 +42,7 @@ create_event_source_response, ) from a2a.utils import constants +from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP, A2AError from a2a.utils.version_validator import validate_version @@ -93,6 +94,12 @@ def _generate_error_response( {'jsonrpc': '2.0', 'id': request_id, 'error': err_dict} ) + if isinstance(error, A2AError): + err_dict = _to_v03_jsonrpc_error(error).model_dump(by_alias=True) + return JSONResponse( + {'jsonrpc': '2.0', 'id': request_id, 'error': err_dict} + ) + internal_error = CoreInternalError(message=str(error)) return JSONResponse( { @@ -141,6 +148,8 @@ async def handle_request( return await self._process_non_streaming_request( request_id, specific_request, call_context ) + except A2AError as e: + return self._generate_error_response(request_id, e) except Exception as e: logger.exception('Unhandled exception in v0.3 JSONRPCAdapter') return self._generate_error_response( @@ -264,6 +273,17 @@ async def event_generator( by_alias=True, exclude_none=True ) } + except A2AError as e: + err_resp = types_v03.SendStreamingMessageResponse( + root=types_v03.JSONRPCErrorResponse( + id=request_id, error=_to_v03_jsonrpc_error(e) + ) + ) + yield { + 'data': err_resp.model_dump_json( + by_alias=True, exclude_none=True + ) + } except Exception as e: logger.exception( 'Error during stream generation in v0.3 JSONRPCAdapter' @@ -284,3 +304,25 @@ async def event_generator( event_generator(stream_gen), shutdown_grace_period=self._shutdown_grace_period, ) + + +# Error codes defined by the A2A v0.3 specification, derived from the v0.3 +# error models. Codes added in v1.0 (e.g. -32009 VersionNotSupported) are +# unknown to v0.3 clients. +_V03_ERROR_CODES = frozenset( + cls.model_fields['code'].default + for cls in get_args(types_v03.A2AError.model_fields['root'].annotation) +) + + +def _to_v03_jsonrpc_error(error: A2AError) -> types_v03.JSONRPCError: + """Maps an A2AError to a v0.3 JSON-RPC error carrying its A2A error code. + + Errors without a v0.3 equivalent are reported as InternalError (-32603). + """ + code = JSON_RPC_ERROR_CODE_MAP.get(type(error), -32603) + if code not in _V03_ERROR_CODES: + code = -32603 + return types_v03.JSONRPCError( + code=code, message=error.message, data=error.data + ) diff --git a/tests/compat/v0_3/test_jsonrpc_app_compat.py b/tests/compat/v0_3/test_jsonrpc_app_compat.py index 8a480b013..aec4a56ee 100644 --- a/tests/compat/v0_3/test_jsonrpc_app_compat.py +++ b/tests/compat/v0_3/test_jsonrpc_app_compat.py @@ -1,3 +1,4 @@ +import json import logging from collections.abc import AsyncIterator @@ -20,6 +21,12 @@ TaskState as TaskState10, TaskStatus as TaskStatus10, ) +from a2a.utils.errors import ( + A2AError, + TaskNotCancelableError, + TaskNotFoundError, + UnsupportedOperationError, +) from starlette.applications import Starlette from starlette.testclient import TestClient @@ -176,3 +183,91 @@ def test_get_extended_agent_card_v03_compat( assert 'result' in data # The result should be a v0.3 AgentCard assert 'supportsAuthenticatedExtendedCard' in data['result'] + + +def test_get_task_not_found_v03_compat( + client: TestClient, mock_handler: AsyncMock +) -> None: + mock_handler.on_get_task.return_value = None + request_payload = { + 'jsonrpc': '2.0', + 'id': '4', + 'method': 'tasks/get', + 'params': {'id': 'missing_task_id'}, + } + + response = client.post('/', json=request_payload) + assert response.status_code == 200 + data = response.json() + + assert data['id'] == '4' + assert data['error']['code'] == -32001 + assert data['error']['message'] == 'Task not found' + + +@pytest.mark.parametrize( + ('error', 'expected_code'), + [ + (TaskNotFoundError(), -32001), + (TaskNotCancelableError(), -32002), + (UnsupportedOperationError(), -32004), + ], +) +def test_cancel_task_v03_compat_maps_a2a_error_codes( + client: TestClient, + mock_handler: AsyncMock, + error: A2AError, + expected_code: int, +) -> None: + mock_handler.on_cancel_task.side_effect = error + request_payload = { + 'jsonrpc': '2.0', + 'id': '5', + 'method': 'tasks/cancel', + 'params': {'id': 'test_task_id'}, + } + + response = client.post('/', json=request_payload) + assert response.status_code == 200 + data = response.json() + + assert data['id'] == '5' + assert data['error']['code'] == expected_code + assert data['error']['message'] == error.message + + +def test_message_stream_v03_compat_maps_a2a_error_code( + client: TestClient, mock_handler: AsyncMock +) -> None: + async def failing_stream( + *args: object, **kwargs: object + ) -> AsyncIterator[MagicMock]: + raise TaskNotFoundError + yield MagicMock() + + mock_handler.on_message_send_stream = failing_stream + request_payload = { + 'jsonrpc': '2.0', + 'id': '6', + 'method': 'message/stream', + 'params': { + 'message': { + 'messageId': 'req', + 'role': 'user', + 'parts': [{'text': 'hello'}], + } + }, + } + + response = client.post('/', json=request_payload) + assert response.status_code == 200 + events = [ + json.loads(line.removeprefix('data:')) + for line in response.text.splitlines() + if line.startswith('data:') + ] + + assert len(events) == 1 + assert events[0]['id'] == '6' + assert events[0]['error']['code'] == -32001 + assert events[0]['error']['message'] == 'Task not found'