Add client support for task/list method

Author: lkawka

src/a2a/client/transports/grpc.py

@@ -2,6 +2,8 @@
 
 from collections.abc import AsyncGenerator
 
+from a2a.utils.constants import DEFAULT_LIST_TASKS_PAGE_SIZE
+
 
 try:
     import grpc
@@ -154,8 +156,11 @@ async def list_tasks(
         context: ClientCallContext | None = None,
     ) -> ListTasksResult:
         """Retrieves tasks for an agent."""
-        # TODO: #515 - Implement method
-        raise NotImplementedError('tasks/list not implemented')
+        response = await self.stub.ListTasks(
+            proto_utils.ToProto.list_tasks_request(request)
+        )
+        page_size = request.page_size or DEFAULT_LIST_TASKS_PAGE_SIZE
+        return proto_utils.FromProto.list_tasks_result(response, page_size)
 
     async def cancel_task(
         self,

src/a2a/client/transports/jsonrpc.py

@@ -31,6 +31,8 @@
     GetTaskResponse,
     JSONRPCErrorResponse,
     ListTasksParams,
+    ListTasksRequest,
+    ListTasksResponse,
     ListTasksResult,
     Message,
     MessageSendParams,
@@ -231,8 +233,18 @@ async def list_tasks(
         context: ClientCallContext | None = None,
     ) -> ListTasksResult:
         """Retrieves tasks for an agent."""
-        # TODO: #515 - Implement method
-        raise NotImplementedError('tasks/list not implemented')
+        rpc_request = ListTasksRequest(params=request, id=str(uuid4()))
+        payload, modified_kwargs = await self._apply_interceptors(
+            'tasks/list',
+            rpc_request.model_dump(mode='json', exclude_none=True),
+            self._get_http_args(context),
+            context,
+        )
+        response_data = await self._send_request(payload, modified_kwargs)
+        response = ListTasksResponse.model_validate(response_data)
+        if isinstance(response.root, JSONRPCErrorResponse):
+            raise A2AClientJSONRPCError(response.root)
+        return response.root.result
 
     async def cancel_task(
         self,

src/a2a/client/transports/rest.py

@@ -8,6 +8,7 @@
 
 from google.protobuf.json_format import MessageToDict, Parse, ParseDict
 from httpx_sse import SSEError, aconnect_sse
+from pydantic import BaseModel
 
 from a2a.client.card_resolver import A2ACardResolver
 from a2a.client.errors import A2AClientHTTPError, A2AClientJSONError
@@ -29,6 +30,7 @@
     TaskStatusUpdateEvent,
 )
 from a2a.utils import proto_utils
+from a2a.utils.constants import DEFAULT_LIST_TASKS_PAGE_SIZE
 from a2a.utils.telemetry import SpanKind, trace_class
 
 
@@ -231,8 +233,20 @@ async def list_tasks(
         context: ClientCallContext | None = None,
     ) -> ListTasksResult:
         """Retrieves tasks for an agent."""
-        # TODO: #515 - Implement method
-        raise NotImplementedError('tasks/list not implemented')
+        _, modified_kwargs = await self._apply_interceptors(
+            request.model_dump(mode='json', exclude_none=True),
+            self._get_http_args(context),
+            context,
+        )
+        response_data = await self._send_get_request(
+            '/v1/tasks',
+            _model_to_query_params(request),
+            modified_kwargs,
+        )
+        response = a2a_pb2.ListTasksResponse()
+        ParseDict(response_data, response)
+        page_size = request.page_size or DEFAULT_LIST_TASKS_PAGE_SIZE
+        return proto_utils.FromProto.list_tasks_result(response, page_size)
 
     async def cancel_task(
         self,
@@ -375,3 +389,21 @@ async def get_card(
     async def close(self) -> None:
         """Closes the httpx client."""
         await self.httpx_client.aclose()
+
+
+def _model_to_query_params(instance: BaseModel) -> dict[str, str]:
+    data = instance.model_dump(mode='json', exclude_none=True)
+    return _json_to_query_params(data)
+
+
+def _json_to_query_params(data: dict[str, Any]) -> dict[str, str]:
+    query_dict = {}
+    for key, value in data.items():
+        if isinstance(value, list):
+            query_dict[key] = ','.join(map(str, value))
+        elif isinstance(value, bool):
+            query_dict[key] = str(value).lower()
+        else:
+            query_dict[key] = str(value)
+
+    return query_dict

src/a2a/utils/constants.py

@@ -4,3 +4,5 @@
 PREV_AGENT_CARD_WELL_KNOWN_PATH = '/.well-known/agent.json'
 EXTENDED_AGENT_CARD_PATH = '/agent/authenticatedExtendedCard'
 DEFAULT_RPC_URL = '/'
+DEFAULT_LIST_TASKS_PAGE_SIZE = 50
+"""Default page size for the `tasks/list` method."""

src/a2a/utils/proto_utils.py

@@ -8,6 +8,7 @@
 from typing import Any
 
 from google.protobuf import json_format, struct_pb2
+from google.protobuf.timestamp_pb2 import Timestamp
 
 from a2a import types
 from a2a.grpc import a2a_pb2
@@ -566,6 +567,24 @@ def role(cls, role: types.Role) -> a2a_pb2.Role:
             case _:
                 return a2a_pb2.Role.ROLE_UNSPECIFIED
 
+    @classmethod
+    def list_tasks_request(
+        cls, params: types.ListTasksParams
+    ) -> a2a_pb2.ListTasksRequest:
+        last_updated_time = None
+        if params.last_updated_after is not None:
+            last_updated_time = Timestamp()
+            last_updated_time.FromMilliseconds(params.last_updated_after)
+        return a2a_pb2.ListTasksRequest(
+            context_id=params.context_id,
+            status=cls.task_state(params.status) if params.status else None,
+            page_size=params.page_size,
+            page_token=params.page_token,
+            history_length=params.history_length,
+            last_updated_time=last_updated_time,
+            include_artifacts=params.include_artifacts,
+        )
+
 
 class FromProto:
     """Converts proto types to Python types."""
@@ -796,6 +815,28 @@ def task_id_params(
             )
         return types.TaskIdParams(id=m.group(1))
 
+    @classmethod
+    def list_tasks_result(
+        cls,
+        response: a2a_pb2.ListTasksResponse,
+        page_size: int,
+    ) -> types.ListTasksResult:
+        """Converts a ListTasksResponse to a ListTasksResult.
+
+        Args:
+            response: The ListTasksResponse to convert.
+            page_size: The maximum number of tasks returned in this response.
+
+        Returns:
+            A `ListTasksResult` object.
+        """
+        return types.ListTasksResult(
+            next_page_token=response.next_page_token,
+            page_size=page_size,
+            tasks=[cls.task(t) for t in response.tasks],
+            total_size=response.total_size,
+        )
+
     @classmethod
     def task_push_notification_config_request(
         cls,

tests/client/test_grpc_client.py

@@ -25,6 +25,7 @@
     TaskStatus,
     TaskStatusUpdateEvent,
     TextPart,
+    ListTasksParams,
 )
 from a2a.utils import get_text_parts, proto_utils
 from a2a.utils.errors import ServerError
@@ -37,6 +38,7 @@ def mock_grpc_stub() -> AsyncMock:
     stub.SendMessage = AsyncMock()
     stub.SendStreamingMessage = MagicMock()
     stub.GetTask = AsyncMock()
+    stub.ListTasks = AsyncMock()
     stub.CancelTask = AsyncMock()
     stub.CreateTaskPushNotificationConfig = AsyncMock()
     stub.GetTaskPushNotificationConfig = AsyncMock()
@@ -91,6 +93,16 @@ def sample_task() -> Task:
     )
 
 
+@pytest.fixture
+def sample_task_2() -> Task:
+    """Provides a sample Task object."""
+    return Task(
+        id='task-2',
+        context_id='ctx-2',
+        status=TaskStatus(state=TaskState.failed),
+    )
+
+
 @pytest.fixture
 def sample_message() -> Message:
     """Provides a sample Message object."""
@@ -283,6 +295,32 @@ async def test_get_task(
     assert response.id == sample_task.id
 
 
+@pytest.mark.asyncio
+async def test_list_tasks(
+    grpc_transport: GrpcTransport,
+    mock_grpc_stub: AsyncMock,
+    sample_task: Task,
+    sample_task_2: Task,
+):
+    """Test listing tasks."""
+    mock_grpc_stub.ListTasks.return_value = a2a_pb2.ListTasksResponse(
+        tasks=[
+            proto_utils.ToProto.task(t) for t in [sample_task, sample_task_2]
+        ],
+        total_size=2,
+    )
+    params = ListTasksParams()
+
+    result = await grpc_transport.list_tasks(params)
+
+    mock_grpc_stub.ListTasks.assert_awaited_once_with(
+        proto_utils.ToProto.list_tasks_request(params)
+    )
+    assert result.total_size == 2
+    assert not result.next_page_token
+    assert [t.id for t in result.tasks] == [sample_task.id, sample_task_2.id]
+
+
 @pytest.mark.asyncio
 async def test_get_task_with_history(
     grpc_transport: GrpcTransport, mock_grpc_stub: AsyncMock, sample_task: Task

tests/client/test_jsonrpc_client.py

@@ -31,6 +31,8 @@
     TaskIdParams,
     TaskPushNotificationConfig,
     TaskQueryParams,
+    ListTasksParams,
+    ListTasksResult,
 )
 from a2a.utils import AGENT_CARD_WELL_KNOWN_PATH
 
@@ -560,6 +562,42 @@ async def test_get_task_success(
         sent_payload = mock_send_request.call_args.args[0]
         assert sent_payload['method'] == 'tasks/get'
 
+    @pytest.mark.asyncio
+    async def test_list_tasks_success(
+        self, mock_httpx_client: AsyncMock, mock_agent_card: MagicMock
+    ):
+        client = JsonRpcTransport(
+            httpx_client=mock_httpx_client, agent_card=mock_agent_card
+        )
+        params = ListTasksParams()
+        mock_rpc_response = {
+            'id': '123',
+            'jsonrpc': '2.0',
+            'result': {
+                'nextPageToken': '',
+                'tasks': [MINIMAL_TASK],
+                'pageSize': 10,
+                'totalSize': 1,
+            },
+        }
+
+        with patch.object(
+            client, '_send_request', new_callable=AsyncMock
+        ) as mock_send_request:
+            mock_send_request.return_value = mock_rpc_response
+            response = await client.list_tasks(request=params)
+
+        assert isinstance(response, ListTasksResult)
+        assert (
+            response.model_dump()
+            == ListTasksResult(
+                next_page_token='',
+                page_size=10,
+                tasks=[Task.model_validate(MINIMAL_TASK)],
+                total_size=1,
+            ).model_dump()
+        )
+
     @pytest.mark.asyncio
     async def test_cancel_task_success(
         self, mock_httpx_client: AsyncMock, mock_agent_card: MagicMock