Skip to content
Open
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
15 changes: 12 additions & 3 deletions src/a2a/compat/v0_3/grpc_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
from a2a.compat.v0_3.extension_headers import add_legacy_extension_header
from a2a.types import a2a_pb2
from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER
from a2a.utils.task import apply_history_length
from a2a.utils.telemetry import SpanKind, trace_class


Expand Down Expand Up @@ -130,8 +131,11 @@ async def send_message(
which = resp_proto.WhichOneof('payload')
if which == 'task':
return a2a_pb2.SendMessageResponse(
task=conversions.to_core_task(
proto_utils.FromProto.task(resp_proto.task)
task=apply_history_length(
conversions.to_core_task(
proto_utils.FromProto.task(resp_proto.task)
),
request.configuration,
)
)
if which == 'msg':
Expand Down Expand Up @@ -217,7 +221,12 @@ async def get_task(
req_proto,
metadata=self._get_grpc_metadata(context),
)
return conversions.to_core_task(proto_utils.FromProto.task(resp_proto))
# v0.3 has no way to ask for an empty history (0 means unlimited), so
# the v1.0 `history_length` semantics are applied to the result here.
return apply_history_length(
conversions.to_core_task(proto_utils.FromProto.task(resp_proto)),
request,
)

@_handle_grpc_exception
async def list_tasks(
Expand Down
17 changes: 13 additions & 4 deletions src/a2a/compat/v0_3/jsonrpc_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
)
from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER
from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP
from a2a.utils.task import apply_history_length
from a2a.utils.telemetry import SpanKind, trace_class


Expand Down Expand Up @@ -105,8 +106,11 @@ async def send_message(

if kind == 'task':
return SendMessageResponse(
task=conversions.to_core_task(
types_v03.Task.model_validate(result_dict)
task=apply_history_length(
conversions.to_core_task(
types_v03.Task.model_validate(result_dict)
),
request.configuration,
)
)
if kind == 'message':
Expand Down Expand Up @@ -164,8 +168,13 @@ async def get_task(
json_rpc_response = JSONRPC20Response(**response_data)
if json_rpc_response.error:
raise self._create_jsonrpc_error(json_rpc_response.error)
return conversions.to_core_task(
types_v03.Task.model_validate(json_rpc_response.result)
# v0.3 has no way to ask for an empty history (0 means unlimited), so
# the v1.0 `history_length` semantics are applied to the result here.
return apply_history_length(
conversions.to_core_task(
types_v03.Task.model_validate(json_rpc_response.result)
),
request,
)

async def list_tasks(
Expand Down
15 changes: 12 additions & 3 deletions src/a2a/compat/v0_3/rest_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
)
from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER
from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP, MethodNotFoundError
from a2a.utils.task import apply_history_length
from a2a.utils.telemetry import SpanKind, trace_class


Expand Down Expand Up @@ -107,8 +108,11 @@ async def send_message(
which = resp_proto.WhichOneof('payload')
if which == 'task':
return SendMessageResponse(
task=conversions.to_core_task(
proto_utils.FromProto.task(resp_proto.task)
task=apply_history_length(
conversions.to_core_task(
proto_utils.FromProto.task(resp_proto.task)
),
request.configuration,
)
)
if which == 'msg':
Expand Down Expand Up @@ -165,7 +169,12 @@ async def get_task(
resp_proto = ParseDict(
response_data, a2a_v0_3_pb2.Task(), ignore_unknown_fields=True
)
return conversions.to_core_task(proto_utils.FromProto.task(resp_proto))
# v0.3 has no way to ask for an empty history (0 means unlimited), so
# the v1.0 `history_length` semantics are applied to the result here.
return apply_history_length(
conversions.to_core_task(proto_utils.FromProto.task(resp_proto)),
request,
)

async def list_tasks(
self,
Expand Down
30 changes: 30 additions & 0 deletions tests/compat/v0_3/test_grpc_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from a2a.compat.v0_3 import a2a_v0_3_pb2
from a2a.compat.v0_3.grpc_transport import CompatGrpcTransport
from a2a.types.a2a_pb2 import (
GetTaskRequest,
Message,
Role,
SendMessageRequest,
Expand Down Expand Up @@ -66,3 +67,32 @@ def test_compat_grpc_transport_no_extension_metadata():

assert 'a2a-extensions' not in metadata
assert 'x-a2a-extensions' not in metadata


@pytest.mark.asyncio
@pytest.mark.parametrize(
'history_length, expected_ids',
[
(None, ['m0', 'm1', 'm2']),
(0, []),
(1, ['m2']),
(3, ['m0', 'm1', 'm2']),
(5, ['m0', 'm1', 'm2']),
],
)
async def test_get_task_history_length(
history_length: int | None, expected_ids: list[str]
) -> None:
transport = CompatGrpcTransport(
channel=AsyncMock(spec=Channel), agent_card=None
)
task = a2a_v0_3_pb2.Task(id='task-1', context_id='ctx-1')
task.status.state = a2a_v0_3_pb2.TaskState.TASK_STATE_WORKING
for i in range(3):
task.history.add(message_id=f'm{i}', role=a2a_v0_3_pb2.Role.ROLE_USER)
transport.stub = MagicMock(GetTask=AsyncMock(return_value=task))
req = GetTaskRequest(id='task-1')
if history_length is not None:
req.history_length = history_length
result = await transport.get_task(req)
assert [message.message_id for message in result.history] == expected_ids
77 changes: 77 additions & 0 deletions tests/compat/v0_3/test_jsonrpc_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -565,3 +565,80 @@ async def test_compat_jsonrpc_transport_mirrors_extension_header(
headers = kwargs['headers']
assert headers['A2A-Extensions'] == 'foo,bar'
assert headers['X-A2A-Extensions'] == 'foo,bar'


@pytest.mark.asyncio
@pytest.mark.parametrize(
'history_length, expected_ids',
[
(None, ['m0', 'm1', 'm2']),
(0, []),
(1, ['m2']),
(3, ['m0', 'm1', 'm2']),
(5, ['m0', 'm1', 'm2']),
],
)
async def test_get_task_history_length(
transport: CompatJsonRpcTransport,
history_length: int | None,
expected_ids: list[str],
) -> None:
# v0.3 treats zero as unlimited; v1.0 requires an empty local history.
payload = {
'id': 'task-1',
'contextId': 'ctx-1',
'status': {'state': 'working'},
'history': [
{'messageId': f'm{i}', 'role': 'user', 'parts': []}
for i in range(3)
],
}
with patch.object(
transport, '_send_request', new_callable=AsyncMock
) as request:
request.return_value = {'jsonrpc': '2.0', 'id': '0', 'result': payload}
req = GetTaskRequest(id='task-1')
if history_length is not None:
req.history_length = history_length
task = await transport.get_task(req)
assert [message.message_id for message in task.history] == expected_ids


@pytest.mark.asyncio
@pytest.mark.parametrize(
'history_length, expected_ids',
[
(None, ['m0', 'm1', 'm2']),
(0, []),
(1, ['m2']),
(3, ['m0', 'm1', 'm2']),
(5, ['m0', 'm1', 'm2']),
],
)
async def test_send_message_history_length(
transport: CompatJsonRpcTransport,
history_length: int | None,
expected_ids: list[str],
) -> None:
# v0.3 treats zero as unlimited; v1.0 requires an empty local history.
payload = {
'id': 'task-1',
'contextId': 'ctx-1',
'status': {'state': 'working'},
'history': [
{'messageId': f'm{i}', 'role': 'user', 'parts': []}
for i in range(3)
],
}
payload['kind'] = 'task'
with patch.object(
transport, '_send_request', new_callable=AsyncMock
) as request:
request.return_value = {'jsonrpc': '2.0', 'id': '0', 'result': payload}
req = SendMessageRequest()
if history_length is not None:
req.configuration.history_length = history_length
response = await transport.send_message(req)
assert response.HasField('task')
task = response.task
assert [message.message_id for message in task.history] == expected_ids
76 changes: 76 additions & 0 deletions tests/compat/v0_3/test_rest_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -690,3 +690,79 @@ async def test_compat_rest_transport_execute_request(
mock_send_http_request.assert_called_once_with(
transport.httpx_client, mock_request, transport._handle_http_error
)


@pytest.mark.asyncio
@pytest.mark.parametrize(
'history_length, expected_ids',
[
(None, ['m0', 'm1', 'm2']),
(0, []),
(1, ['m2']),
(3, ['m0', 'm1', 'm2']),
(5, ['m0', 'm1', 'm2']),
],
)
async def test_get_task_history_length(
transport: CompatRestTransport,
history_length: int | None,
expected_ids: list[str],
) -> None:
# v0.3 treats zero as unlimited; v1.0 requires an empty local history.
payload = {
'id': 'task-1',
'contextId': 'ctx-1',
'status': {'state': 'TASK_STATE_WORKING'},
'history': [
{'messageId': f'm{i}', 'role': 'ROLE_USER', 'parts': []}
for i in range(3)
],
}
with patch.object(
transport, '_execute_request', new_callable=AsyncMock
) as request:
request.return_value = payload
req = GetTaskRequest(id='task-1')
if history_length is not None:
req.history_length = history_length
task = await transport.get_task(req)
assert [message.message_id for message in task.history] == expected_ids


@pytest.mark.asyncio
@pytest.mark.parametrize(
'history_length, expected_ids',
[
(None, ['m0', 'm1', 'm2']),
(0, []),
(1, ['m2']),
(3, ['m0', 'm1', 'm2']),
(5, ['m0', 'm1', 'm2']),
],
)
async def test_send_message_history_length(
transport: CompatRestTransport,
history_length: int | None,
expected_ids: list[str],
) -> None:
# v0.3 treats zero as unlimited; v1.0 requires an empty local history.
payload = {
'id': 'task-1',
'contextId': 'ctx-1',
'status': {'state': 'TASK_STATE_WORKING'},
'history': [
{'messageId': f'm{i}', 'role': 'ROLE_USER', 'parts': []}
for i in range(3)
],
}
with patch.object(
transport, '_execute_request', new_callable=AsyncMock
) as request:
request.return_value = {'task': payload}
req = SendMessageRequest()
if history_length is not None:
req.configuration.history_length = history_length
response = await transport.send_message(req)
assert response.HasField('task')
task = response.task
assert [message.message_id for message in task.history] == expected_ids
Loading