Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 43 additions & 1 deletion src/a2a/compat/v0_3/jsonrpc_adapter.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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


Expand Down Expand Up @@ -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(
{
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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'
Expand All @@ -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(
Comment thread
mykytanetipa marked this conversation as resolved.
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
)
95 changes: 95 additions & 0 deletions tests/compat/v0_3/test_jsonrpc_app_compat.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import json
import logging

from collections.abc import AsyncIterator
Expand All @@ -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

Expand Down Expand Up @@ -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'
Loading