Add server support

Author: lkawka

src/a2a/server/request_handlers/default_request_handler.py

@@ -45,6 +45,7 @@
     TaskState,
     UnsupportedOperationError,
 )
+from a2a.utils.constants import DEFAULT_LIST_TASKS_PAGE_SIZE
 from a2a.utils.errors import ServerError
 from a2a.utils.task import apply_history_length
 from a2a.utils.telemetry import SpanKind, trace_class
@@ -129,8 +130,13 @@ async def on_list_tasks(
         context: ServerCallContext | None = None,
     ) -> ListTasksResult:
         """Default handler for 'tasks/list'."""
-        # TODO: #515 - Implement method
-        raise NotImplementedError('tasks/list not implemented')
+        page = await self.task_store.list(params, context)
+        return ListTasksResult(
+            next_page_token=page.next_page_token,
+            page_size=params.page_size or DEFAULT_LIST_TASKS_PAGE_SIZE,
+            tasks=page.tasks,
+            total_size=page.total_size,
+        )
 
     async def on_cancel_task(
         self, params: TaskIdParams, context: ServerCallContext | None = None
@@ -590,3 +596,9 @@ async def on_delete_task_push_notification_config(
         await self._push_config_store.delete_info(
             params.id, params.push_notification_config_id
         )
+
+
+def _next_page_token(current_page_token: str) -> str:
+    if not current_page_token:
+        return '1'
+    return str(int(current_page_token) + 1)

src/a2a/server/request_handlers/grpc_handler.py

@@ -345,8 +345,15 @@ async def ListTasks(
         Returns:
             A `ListTasksResponse` object.
         """
-        # TODO: #515 - Implement method
-        raise NotImplementedError('tasks/list not implemented')
+        try:
+            server_context = self.context_builder.build(context)
+            result = await self.request_handler.on_list_tasks(
+                proto_utils.FromProto.list_tasks_params(request), server_context
+            )
+            return proto_utils.ToProto.list_tasks_response(result)
+        except ServerError as e:
+            await self.abort_context(e, context)
+        return a2a_pb2.ListTasksResponse()
 
     async def GetAgentCard(
         self,

src/a2a/server/request_handlers/jsonrpc_handler.py

@@ -28,8 +28,11 @@
     ListTaskPushNotificationConfigRequest,
     ListTaskPushNotificationConfigResponse,
     ListTaskPushNotificationConfigSuccessResponse,
+    ListTasksParams,
     ListTasksRequest,
     ListTasksResponse,
+    ListTasksResult,
+    ListTasksSuccessResponse,
     Message,
     SendMessageRequest,
     SendMessageResponse,
@@ -375,8 +378,23 @@ async def list_tasks(
         Returns:
             A `ListTasksResponse` object containing the Task or a JSON-RPC error.
         """
-        # TODO: #515 - Implement method
-        raise NotImplementedError('tasks/list not implemented')
+        try:
+            result = await self.request_handler.on_list_tasks(
+                request.params or ListTasksParams(), context
+            )
+        except ServerError as e:
+            return ListTasksResponse(
+                root=JSONRPCErrorResponse(
+                    id=request.id, error=e.error if e.error else InternalError()
+                )
+            )
+        return prepare_response_object(
+            request.id,
+            result,
+            (ListTasksResult,),
+            ListTasksSuccessResponse,
+            ListTasksResponse,
+        )
 
     async def list_push_notification_config(
         self,

src/a2a/server/request_handlers/response_helpers.py

@@ -18,6 +18,9 @@
     JSONRPCErrorResponse,
     ListTaskPushNotificationConfigResponse,
     ListTaskPushNotificationConfigSuccessResponse,
+    ListTasksResponse,
+    ListTasksResult,
+    ListTasksSuccessResponse,
     Message,
     SendMessageResponse,
     SendMessageSuccessResponse,
@@ -42,6 +45,7 @@
     SendStreamingMessageResponse,
     ListTaskPushNotificationConfigResponse,
     DeleteTaskPushNotificationConfigResponse,
+    ListTasksResponse,
 )
 """Type variable for RootModel response types."""
 
@@ -56,6 +60,7 @@
     SendStreamingMessageSuccessResponse,
     ListTaskPushNotificationConfigSuccessResponse,
     DeleteTaskPushNotificationConfigSuccessResponse,
+    ListTasksSuccessResponse,
 )
 """Type variable for SuccessResponse types."""
 
@@ -69,6 +74,7 @@
     | A2AError
     | JSONRPCError
     | list[TaskPushNotificationConfig]
+    | ListTasksResult
 )
 """Type alias for possible event types produced by handlers."""
 

src/a2a/server/request_handlers/rest_handler.py

@@ -21,6 +21,7 @@
 from a2a.types import (
     AgentCard,
     GetTaskPushNotificationConfigParams,
+    ListTasksParams,
     TaskIdParams,
     TaskNotFoundError,
     TaskQueryParams,
@@ -264,12 +265,12 @@ async def on_get_task(
             return MessageToDict(proto_utils.ToProto.task(task))
         raise ServerError(error=TaskNotFoundError())
 
-    async def list_push_notifications(
+    async def list_tasks(
         self,
         request: Request,
         context: ServerCallContext,
     ) -> dict[str, Any]:
-        """Handles the 'tasks/pushNotificationConfig/list' REST method.
+        """Handles the 'tasks/list' REST method.
 
         This method is currently not implemented.
 
@@ -278,19 +279,21 @@ async def list_push_notifications(
             context: Context provided by the server.
 
         Returns:
-            A list of `dict` representing the `TaskPushNotificationConfig` objects.
+            A list of dict representing the`Task` objects.
 
         Raises:
             NotImplementedError: This method is not yet implemented.
         """
-        raise NotImplementedError('list notifications not implemented')
+        params = ListTasksParams.model_validate(request.query_params)
+        result = await self.request_handler.on_list_tasks(params, context)
+        return MessageToDict(proto_utils.ToProto.list_tasks_response(result))
 
-    async def list_tasks(
+    async def list_push_notifications(
         self,
         request: Request,
         context: ServerCallContext,
     ) -> dict[str, Any]:
-        """Handles the 'tasks/list' REST method.
+        """Handles the 'tasks/pushNotificationConfig/list' REST method.
 
         This method is currently not implemented.
 
@@ -299,10 +302,9 @@ async def list_tasks(
             context: Context provided by the server.
 
         Returns:
-            A list of dict representing the`Task` objects.
+            A list of `dict` representing the `TaskPushNotificationConfig` objects.
 
         Raises:
             NotImplementedError: This method is not yet implemented.
         """
-        # TODO: #515 - Implement method
-        raise NotImplementedError('list tasks not implemented')
+        raise NotImplementedError('list notifications not implemented')

src/a2a/server/tasks/database_task_store.py

@@ -2,7 +2,7 @@
 
 
 try:
-    from sqlalchemy import Table, delete, select
+    from sqlalchemy import Table, delete, func, select
     from sqlalchemy.ext.asyncio import (
         AsyncEngine,
         AsyncSession,
@@ -21,8 +21,9 @@
 
 from a2a.server.context import ServerCallContext
 from a2a.server.models import Base, TaskModel, create_task_model
-from a2a.server.tasks.task_store import TaskStore
-from a2a.types import Task  # Task is the Pydantic model
+from a2a.server.tasks.task_store import TaskStore, TasksPage
+from a2a.types import ListTasksParams, Task
+from a2a.utils.constants import DEFAULT_LIST_TASKS_PAGE_SIZE
 
 
 logger = logging.getLogger(__name__)
@@ -147,6 +148,54 @@ async def get(
             logger.debug('Task %s not found in store.', task_id)
             return None
 
+    async def list(
+        self, params: ListTasksParams, context: ServerCallContext | None = None
+    ) -> TasksPage:
+        """Retrieves all tasks from the database."""
+        await self._ensure_initialized()
+        async with self.async_session_maker() as session:
+            page_number = int(params.page_token) if params.page_token else 0
+            page_size = params.page_size or DEFAULT_LIST_TASKS_PAGE_SIZE
+            offset = page_number * page_size
+
+            # Base query for filtering
+            base_stmt = select(self.task_model)
+            if params.context_id:
+                base_stmt = base_stmt.where(
+                    self.task_model.context_id == params.context_id
+                )
+            if params.status is not None:
+                base_stmt = base_stmt.where(
+                    self.task_model.status['state'].as_string()
+                    == params.status.value
+                )
+
+            # Get total count
+            count_stmt = select(func.count()).select_from(base_stmt.alias())
+            total_count = (await session.execute(count_stmt)).scalar_one()
+
+            # Get paginated results
+            stmt = (
+                base_stmt.order_by(self.task_model.id.desc())
+                .limit(page_size)
+                .offset(offset)
+            )
+            result = await session.execute(stmt)
+            tasks_models = result.scalars().all()
+            tasks = [self._from_orm(task_model) for task_model in tasks_models]
+
+            next_page_token = (
+                str(page_number + 1)
+                if total_count > (page_number + 1) * page_size
+                else ''
+            )
+
+            return TasksPage(
+                tasks=tasks,
+                total_size=total_count,
+                next_page_token=next_page_token,
+            )
+
     async def delete(
         self, task_id: str, context: ServerCallContext | None = None
     ) -> None:

src/a2a/server/tasks/inmemory_task_store.py

@@ -2,8 +2,9 @@
 import logging
 
 from a2a.server.context import ServerCallContext
-from a2a.server.tasks.task_store import TaskStore
-from a2a.types import Task
+from a2a.server.tasks.task_store import TaskStore, TasksPage
+from a2a.types import ListTasksParams, Task
+from a2a.utils.constants import DEFAULT_LIST_TASKS_PAGE_SIZE
 
 
 logger = logging.getLogger(__name__)
@@ -43,6 +44,59 @@ async def get(
                 logger.debug('Task %s not found in store.', task_id)
             return task
 
+    async def list(
+        self,
+        params: ListTasksParams,
+        context: ServerCallContext | None = None,
+    ) -> TasksPage:
+        """Retrieves a list of tasks from the store."""
+        async with self.lock:
+            tasks = list(self.tasks.values())
+
+        # Apply filtering
+        if params.context_id:
+            tasks = [
+                task for task in tasks if task.context_id == params.context_id
+            ]
+        if params.status is not None:
+            tasks = [
+                task for task in tasks if task.status.state == params.status
+            ]
+
+        # Reduce payload
+        base_updates = {}
+        if not params.include_artifacts:
+            base_updates = {'artifacts': []}
+        for i in range(len(tasks)):
+            updates = dict(base_updates)
+            history = tasks[i].history
+            if params.history_length is not None and history:
+                limited_history = (
+                    history[-params.history_length :]
+                    if params.history_length > 0
+                    else []
+                )
+                updates['history'] = limited_history
+            tasks[i] = tasks[i].model_copy(update=updates)
+
+        # Apply pagination
+        total_size = len(tasks)
+        page_token = int(params.page_token) if params.page_token else 0
+        page_size = params.page_size or DEFAULT_LIST_TASKS_PAGE_SIZE
+        tasks = tasks[page_token * page_size : (page_token + 1) * page_size]
+
+        next_page_token = (
+            str(page_token + 1)
+            if (page_token + 1) * page_size < total_size
+            else ''
+        )
+
+        return TasksPage(
+            next_page_token=next_page_token,
+            tasks=tasks,
+            total_size=total_size,
+        )
+
     async def delete(
         self, task_id: str, context: ServerCallContext | None = None
     ) -> None:

src/a2a/server/tasks/task_store.py

@@ -1,7 +1,17 @@
 from abc import ABC, abstractmethod
 
+from pydantic import BaseModel
+
 from a2a.server.context import ServerCallContext
-from a2a.types import Task
+from a2a.types import ListTasksParams, Task
+
+
+class TasksPage(BaseModel):
+    """Page with tasks."""
+
+    next_page_token: str = ''
+    tasks: list[Task]
+    total_size: int
 
 
 class TaskStore(ABC):
@@ -22,6 +32,14 @@ async def get(
     ) -> Task | None:
         """Retrieves a task from the store by ID."""
 
+    @abstractmethod
+    async def list(
+        self,
+        params: ListTasksParams,
+        context: ServerCallContext | None = None,
+    ) -> TasksPage:
+        """Retrieves a list of tasks from the store."""
+
     @abstractmethod
     async def delete(
         self, task_id: str, context: ServerCallContext | None = None