Skip to content

Commit 7e55413

Browse files
Sonica-BIwaniukooo11
Sonica-B
and
Iwaniukooo11
authored
fix(client): raise A2A errors for malformed JSON-RPC responses (#1375)
# Description `JsonRpcTransport` turned every response body into a `json-rpc` `JSONRPC20Response` (`JSONRPC20Response(**response_data)` in the nine unary methods, `JSONRPC20Response.from_json` for SSE events). That package validates the envelope with its own exceptions, so a body that is not a JSON-RPC 2.0 response object reached the caller as a raw `TypeError`, `ValueError` or `json.JSONDecodeError` instead of an `A2AClientError`: | Response body | Before | After | |---|---|---| | a list, or a string (e.g. from a proxy) | `TypeError: ... argument after ** must be a mapping` | `A2AClientError` | | an object with neither `result` nor `error` | `ValueError: Either result or error should be used` | `A2AClientError` | | `"error": "boom"` | `TypeError: ... argument after ** must be a mapping` | `A2AClientError` | | error object with an extra member, e.g. `{"code": -32001, "message": "...", "reason": "x"}` | `TypeError: JSONRPCError.__init__() got an unexpected keyword argument 'reason'` | `TaskNotFoundError` | | error object without `message`, e.g. `{"code": -32001}` | `ValueError: Error message should be string` | `TaskNotFoundError` | | SSE `data:` that is not JSON | `json.JSONDecodeError` (raised in the consumer loop, outside `handle_http_exceptions`) | `A2AClientError` | | `event: error` whose error object has an extra member | `TypeError` | `TaskNotFoundError` | The last two error-object rows matter most: the server sent a valid A2A error code, and `_create_jsonrpc_error` already handles a missing `message` (`error_dict.get('message', str(error_dict))`), but the `json-rpc` validation failed before that code ran. It is also the error-object counterpart of the spec 5.7 leniency added for results in #1354. Change, all in `src/a2a/client/transports/jsonrpc.py`: - New `_check_jsonrpc_response`: an `error` object raises the mapped A2A error through the existing `_create_jsonrpc_error`; a non-object body, a non-object `error`, or a body with neither `result` nor `error` raises `A2AClientError`. A falsy `error` next to a `result` is still treated as success, as before. - `_send_request` and the SSE loop call it, so the unary methods read `response_data['result']` and no longer need their own error checks. `_handle_sse_error` parses with `json.loads` and keeps its `SSE stream error: ...` fallback. - `JSONRPC20Response` is no longer used on the client (requests still use `JSONRPC20Request`). Well-formed success and error responses behave as before; the existing tests pass unchanged. Tests: `TestMalformedJsonRpcResponses` in `tests/client/transports/test_jsonrpc_client.py`, 11 cases (unary invalid envelopes, non-standard error objects, streaming `data:` and `event: error`, plus a guard that `delete_task_push_notification_config` still raises on an error response). 10 of them fail on `main` with the exceptions in the table and pass with the change. The v0.3 `CompatJsonRpcTransport` has the same `JSONRPC20Response(**response_data)` pattern (an error object with an extra member raises `TypeError` there too). I kept this PR to the 1.0 client and can send a `fix(compat)` follow-up. Checks, run on Windows 11 with Python 3.10.19: - `bash ./scripts/lint.sh`: ruff check, ruff format and `ty` pass (three existing unused `type: ignore` warnings in files this PR does not touch). - `uv run pytest tests/client/transports/test_jsonrpc_client.py`: 51 passed. - `uv run pytest --ignore=tests/integration/cross_version --ignore=tests/integration/test_scenarios.py --cov=src`: 2213 passed, 199 skipped, 3 xfailed; total coverage 93%. The 3 failures (`test_samples_smoke.py`, console encoding) and 4 setup errors (`push_notifications` e2e, `Can't pickle local object` under Windows spawn) fail the same way on `main` here. `tests/integration/cross_version` (uses `os.killpg`) and `tests/integration/test_scenarios.py` (hangs on Windows on `main` too; it only builds gRPC clients) were not run locally, so the Linux CI run is the check for those. - [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) Co-authored-by: Mateusz Iwaniuk <45974414+Iwaniukooo11@users.noreply.github.com>
1 parent b21a144 commit 7e55413

2 files changed

Lines changed: 193 additions & 54 deletions

File tree

‎src/a2a/client/transports/jsonrpc.py‎

Lines changed: 61 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import json
12
import logging
23

34
from collections.abc import AsyncGenerator
@@ -7,7 +8,7 @@
78
import httpx
89

910
from google.protobuf import json_format
10-
from jsonrpc.jsonrpc2 import JSONRPC20Request, JSONRPC20Response
11+
from jsonrpc.jsonrpc2 import JSONRPC20Request
1112

1213
from a2a.client.client import ClientCallContext
1314
from a2a.client.errors import A2AClientError
@@ -80,11 +81,8 @@ async def send_message(
8081
response_data = await self._send_request(
8182
dict(rpc_request.data), context
8283
)
83-
json_rpc_response = JSONRPC20Response(**response_data)
84-
if json_rpc_response.error:
85-
raise self._create_jsonrpc_error(json_rpc_response.error)
8684
response: SendMessageResponse = parse_lenient(
87-
json_rpc_response.result, SendMessageResponse()
85+
response_data['result'], SendMessageResponse()
8886
)
8987
return response
9088

@@ -121,10 +119,7 @@ async def get_task(
121119
response_data = await self._send_request(
122120
dict(rpc_request.data), context
123121
)
124-
json_rpc_response = JSONRPC20Response(**response_data)
125-
if json_rpc_response.error:
126-
raise self._create_jsonrpc_error(json_rpc_response.error)
127-
response: Task = parse_lenient(json_rpc_response.result, Task())
122+
response: Task = parse_lenient(response_data['result'], Task())
128123
return response
129124

130125
async def list_tasks(
@@ -142,11 +137,8 @@ async def list_tasks(
142137
response_data = await self._send_request(
143138
dict(rpc_request.data), context
144139
)
145-
json_rpc_response = JSONRPC20Response(**response_data)
146-
if json_rpc_response.error:
147-
raise self._create_jsonrpc_error(json_rpc_response.error)
148140
response: ListTasksResponse = parse_lenient(
149-
json_rpc_response.result, ListTasksResponse()
141+
response_data['result'], ListTasksResponse()
150142
)
151143
return response
152144

@@ -165,10 +157,7 @@ async def cancel_task(
165157
response_data = await self._send_request(
166158
dict(rpc_request.data), context
167159
)
168-
json_rpc_response = JSONRPC20Response(**response_data)
169-
if json_rpc_response.error:
170-
raise self._create_jsonrpc_error(json_rpc_response.error)
171-
response: Task = parse_lenient(json_rpc_response.result, Task())
160+
response: Task = parse_lenient(response_data['result'], Task())
172161
return response
173162

174163
async def create_task_push_notification_config(
@@ -186,11 +175,8 @@ async def create_task_push_notification_config(
186175
response_data = await self._send_request(
187176
dict(rpc_request.data), context
188177
)
189-
json_rpc_response = JSONRPC20Response(**response_data)
190-
if json_rpc_response.error:
191-
raise self._create_jsonrpc_error(json_rpc_response.error)
192178
response: TaskPushNotificationConfig = parse_lenient(
193-
json_rpc_response.result, TaskPushNotificationConfig()
179+
response_data['result'], TaskPushNotificationConfig()
194180
)
195181
return response
196182

@@ -209,11 +195,8 @@ async def get_task_push_notification_config(
209195
response_data = await self._send_request(
210196
dict(rpc_request.data), context
211197
)
212-
json_rpc_response = JSONRPC20Response(**response_data)
213-
if json_rpc_response.error:
214-
raise self._create_jsonrpc_error(json_rpc_response.error)
215198
response: TaskPushNotificationConfig = parse_lenient(
216-
json_rpc_response.result, TaskPushNotificationConfig()
199+
response_data['result'], TaskPushNotificationConfig()
217200
)
218201
return response
219202

@@ -232,11 +215,8 @@ async def list_task_push_notification_configs(
232215
response_data = await self._send_request(
233216
dict(rpc_request.data), context
234217
)
235-
json_rpc_response = JSONRPC20Response(**response_data)
236-
if json_rpc_response.error:
237-
raise self._create_jsonrpc_error(json_rpc_response.error)
238218
response: ListTaskPushNotificationConfigsResponse = parse_lenient(
239-
json_rpc_response.result, ListTaskPushNotificationConfigsResponse()
219+
response_data['result'], ListTaskPushNotificationConfigsResponse()
240220
)
241221
return response
242222

@@ -252,12 +232,7 @@ async def delete_task_push_notification_config(
252232
params=json_format.MessageToDict(request),
253233
_id=str(uuid4()),
254234
)
255-
response_data = await self._send_request(
256-
dict(rpc_request.data), context
257-
)
258-
json_rpc_response = JSONRPC20Response(**response_data)
259-
if json_rpc_response.error:
260-
raise self._create_jsonrpc_error(json_rpc_response.error)
235+
await self._send_request(dict(rpc_request.data), context)
261236

262237
async def subscribe(
263238
self,
@@ -297,17 +272,11 @@ async def get_extended_agent_card(
297272
dict(rpc_request.data),
298273
context,
299274
)
300-
json_rpc_response = JSONRPC20Response(**response_data)
301-
if json_rpc_response.error:
302-
raise self._create_jsonrpc_error(json_rpc_response.error)
275+
result = response_data['result']
303276
# Validate type of the response
304-
if not isinstance(json_rpc_response.result, dict):
305-
raise A2AClientError(
306-
f'Invalid response type: {type(json_rpc_response.result)}'
307-
)
308-
response: AgentCard = parse_lenient(
309-
json_rpc_response.result, AgentCard()
310-
)
277+
if not isinstance(result, dict):
278+
raise A2AClientError(f'Invalid response type: {type(result)}')
279+
response: AgentCard = parse_lenient(result, AgentCard())
311280

312281
return response
313282

@@ -357,7 +326,37 @@ async def _send_request(
357326
request = self.httpx_client.build_request(
358327
'POST', self.url, json=payload, **(http_kwargs or {})
359328
)
360-
return await send_http_request(self.httpx_client, request)
329+
response_data = await send_http_request(self.httpx_client, request)
330+
return self._check_jsonrpc_response(response_data)
331+
332+
def _check_jsonrpc_response(self, response_data: Any) -> dict[str, Any]:
333+
"""Returns a JSON-RPC success response, raising for anything else.
334+
335+
An error response raises the A2A error mapped from its code, even when
336+
the error object lacks a ``message`` or carries extra members. A body
337+
that is not a JSON-RPC 2.0 response object (a list, a string, an object
338+
with neither ``result`` nor ``error``, or an ``error`` that is not an
339+
object) raises ``A2AClientError`` instead of leaking a ``TypeError`` or
340+
``ValueError`` from the ``json-rpc`` package.
341+
"""
342+
if not isinstance(response_data, dict):
343+
raise A2AClientError(
344+
'Invalid JSON-RPC response: expected an object, got '
345+
f'{type(response_data).__name__}'
346+
)
347+
error = response_data.get('error')
348+
if error:
349+
if not isinstance(error, dict):
350+
raise A2AClientError(
351+
f'Invalid JSON-RPC response: error is not an object: '
352+
f'{error!r}'
353+
)
354+
raise self._create_jsonrpc_error(error)
355+
if 'result' not in response_data:
356+
raise A2AClientError(
357+
"Invalid JSON-RPC response: it has neither 'result' nor 'error'"
358+
)
359+
return response_data
361360

362361
async def _send_stream_request(
363362
self,
@@ -375,17 +374,26 @@ async def _send_stream_request(
375374
json=rpc_request_payload,
376375
**http_kwargs,
377376
):
378-
json_rpc_response = JSONRPC20Response.from_json(sse_data)
379-
if json_rpc_response.error:
380-
raise self._create_jsonrpc_error(json_rpc_response.error)
377+
try:
378+
response_data = json.loads(sse_data)
379+
except json.JSONDecodeError as e:
380+
raise A2AClientError(
381+
f'Invalid JSON-RPC response in SSE stream: {e}'
382+
) from e
383+
response_data = self._check_jsonrpc_response(response_data)
381384
response: StreamResponse = parse_lenient(
382-
json_rpc_response.result, StreamResponse()
385+
response_data['result'], StreamResponse()
383386
)
384387
yield response
385388

386389
def _handle_sse_error(self, sse_data: str) -> NoReturn:
387390
"""Handles SSE error events by parsing JSON-RPC error payload and raising the appropriate domain error."""
388-
json_rpc_response = JSONRPC20Response.from_json(sse_data)
389-
if json_rpc_response.error:
390-
raise self._create_jsonrpc_error(json_rpc_response.error)
391+
try:
392+
response_data = json.loads(sse_data)
393+
except json.JSONDecodeError:
394+
response_data = None
395+
if isinstance(response_data, dict) and isinstance(
396+
response_data.get('error'), dict
397+
):
398+
raise self._create_jsonrpc_error(response_data['error'])
391399
raise A2AClientError(f'SSE stream error: {sse_data}')

‎tests/client/transports/test_jsonrpc_client.py‎

Lines changed: 132 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
TaskPushNotificationConfig,
3131
TaskState,
3232
)
33-
from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP
33+
from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP, TaskNotFoundError
3434
from google.protobuf import json_format
3535

3636

@@ -828,3 +828,134 @@ async def test_send_message_streaming_ignores_unrecognized_fields(
828828
assert len(events) == 1
829829
assert events[0].message.message_id == 'r1'
830830
assert events[0].message.parts[0].text == 'hello'
831+
832+
833+
# A JSON-RPC error object with a member beyond code, message and data.
834+
_ERROR_WITH_EXTRA_MEMBER = {
835+
'jsonrpc': '2.0',
836+
'id': '1',
837+
'error': {'code': -32001, 'message': 'Task not found', 'reason': 'x'},
838+
}
839+
840+
841+
class TestMalformedJsonRpcResponses:
842+
"""Responses that are not valid JSON-RPC 2.0 raise A2A errors.
843+
844+
The transport used to build ``JSONRPC20Response`` objects from the raw
845+
body, so these responses leaked ``TypeError``/``ValueError`` from the
846+
``json-rpc`` package instead of an ``A2AClientError``.
847+
"""
848+
849+
@pytest.mark.parametrize(
850+
'body',
851+
[
852+
pytest.param(
853+
[{'jsonrpc': '2.0', 'id': '1', 'result': {}}], id='list'
854+
),
855+
pytest.param('Bad Gateway', id='string'),
856+
pytest.param(
857+
{'jsonrpc': '2.0', 'id': '1'}, id='no-result-or-error'
858+
),
859+
pytest.param(
860+
{'jsonrpc': '2.0', 'id': '1', 'error': 'boom'},
861+
id='error-not-an-object',
862+
),
863+
],
864+
)
865+
@pytest.mark.asyncio
866+
async def test_invalid_envelope_raises_client_error(
867+
self, transport, mock_httpx_client, body
868+
):
869+
mock_response = MagicMock()
870+
mock_response.json.return_value = body
871+
mock_response.raise_for_status = MagicMock()
872+
mock_httpx_client.send.return_value = mock_response
873+
874+
with pytest.raises(A2AClientError, match='Invalid JSON-RPC response'):
875+
await transport.get_task(GetTaskRequest(id='task-1'))
876+
877+
@pytest.mark.parametrize(
878+
'error',
879+
[
880+
pytest.param(_ERROR_WITH_EXTRA_MEMBER['error'], id='extra-member'),
881+
pytest.param({'code': -32001}, id='no-message'),
882+
],
883+
)
884+
@pytest.mark.asyncio
885+
async def test_nonstandard_error_object_maps_to_a2a_error(
886+
self, transport, mock_httpx_client, error
887+
):
888+
mock_response = MagicMock()
889+
mock_response.json.return_value = {
890+
'jsonrpc': '2.0',
891+
'id': '1',
892+
'error': error,
893+
}
894+
mock_response.raise_for_status = MagicMock()
895+
mock_httpx_client.send.return_value = mock_response
896+
897+
with pytest.raises(TaskNotFoundError):
898+
await transport.get_task(GetTaskRequest(id='task-1'))
899+
900+
@pytest.mark.asyncio
901+
async def test_delete_push_config_error_still_raises(
902+
self, transport, mock_httpx_client
903+
):
904+
"""The delete call ignores the result but must still raise errors."""
905+
mock_response = MagicMock()
906+
mock_response.json.return_value = {
907+
'jsonrpc': '2.0',
908+
'id': '1',
909+
'error': {'code': -32001, 'message': 'Task not found'},
910+
}
911+
mock_response.raise_for_status = MagicMock()
912+
mock_httpx_client.send.return_value = mock_response
913+
914+
with pytest.raises(TaskNotFoundError):
915+
await transport.delete_task_push_notification_config(
916+
DeleteTaskPushNotificationConfigRequest(
917+
task_id='task-1', id='config-1'
918+
)
919+
)
920+
921+
@pytest.mark.parametrize(
922+
('lines', 'expected_error'),
923+
[
924+
pytest.param(['data: not json', ''], A2AClientError, id='not-json'),
925+
pytest.param(
926+
[f'data: {json.dumps({"jsonrpc": "2.0", "id": "1"})}', ''],
927+
A2AClientError,
928+
id='no-result-or-error',
929+
),
930+
pytest.param(
931+
[f'data: {json.dumps(_ERROR_WITH_EXTRA_MEMBER)}', ''],
932+
TaskNotFoundError,
933+
id='error-extra-member',
934+
),
935+
pytest.param(
936+
[
937+
'event: error',
938+
f'data: {json.dumps(_ERROR_WITH_EXTRA_MEMBER)}',
939+
'',
940+
],
941+
TaskNotFoundError,
942+
id='error-event-extra-member',
943+
),
944+
],
945+
)
946+
@pytest.mark.asyncio
947+
@patch('a2a.client.transports.http_helpers._SSEEventSource')
948+
async def test_streaming_malformed_event_raises_a2a_error(
949+
self, mock_sse_source: AsyncMock, transport, lines, expected_error
950+
):
951+
mock_response = AsyncMock(spec=httpx.Response)
952+
mock_response.raise_for_status = MagicMock()
953+
mock_response.headers = {'content-type': 'text/event-stream'}
954+
mock_response.aiter_lines.return_value = async_iterable_from_list(lines)
955+
mock_sse_source.return_value.__aenter__.return_value = mock_response
956+
957+
with pytest.raises(expected_error):
958+
async for _ in transport.send_message_streaming(
959+
create_send_message_request()
960+
):
961+
pass

0 commit comments

Comments
 (0)