fix(compat): return A2A error codes from the v0.3 JSON-RPC adapter (#1369)

# Description

A2A errors raised while handling v0.3 JSON-RPC requests fell through to
the generic `except Exception` in `JSONRPC03Adapter` and were returned
as `-32603 InternalError`. v0.3 clients could not distinguish e.g.
`TaskNotFoundError` (`-32001`) from a server failure.

This maps `A2AError`s to their JSON-RPC codes via
`JSON_RPC_ERROR_CODE_MAP` in both paths, matching what
`JsonRpcDispatcher` does for v1.0:

- non-streaming: `handle_request` catches `A2AError` before the generic
handler, and `_generate_error_response` builds a v0.3 `JSONRPCError`
from it
- streaming: `event_generator` yields the mapped error event instead of
`InternalError`

Errors that only exist in v1.0 (`-32008` ExtensionSupportRequired,
`-32009` VersionNotSupported) are not part of the v0.3 spec, so they
still map to `-32603`. This keeps the existing expectation in
`tests/integration/test_version_header.py`.

**Tests** (`tests/compat/v0_3/test_jsonrpc_app_compat.py`): `tasks/get`
not found, `tasks/cancel` with `TaskNotFoundError` /
`TaskNotCancelableError` / `UnsupportedOperationError`, and an error
raised mid-stream on `message/stream`. All fail on `main` and pass with
this change. Full suite passes locally; `ruff check`, `ruff format` and
`ty check` are clean.

- [x] Follow the [`CONTRIBUTING`
Guide](https://github.com/a2aproject/a2a-python/blob/main/CONTRIBUTING.md).
- [x] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- [x] Ensure the tests and linter pass
- [x] Appropriate docs were updated (if necessary)

Fixes #1368 🦕

---------

Co-authored-by: mykytanetipa <mykytanetipa@google.com>

Author: shwetangsinha0509

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
+    )

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'