From bbc6ebbbc4fb8ff613410a0afb351e1d524cea1e Mon Sep 17 00:00:00 2001 From: Al Duncanson Date: Sun, 16 Aug 2026 22:43:42 -0400 Subject: [PATCH 1/2] feat: implement ListTasks across service, CLI, and MCP ListTasks is one of the eleven A2A v1.0 methods and Handler implemented it on no surface; there was no way to ask an agent what tasks exist. A2AService gains list_tasks (one page, raw ListTasksResponse) and list_all_tasks (follows next_page_token to the end, with a repeated- token guard so a misbehaving server cannot loop it forever). An unset proto3 page size is indistinguishable from zero and servers reject zero, so requests always carry an explicit page size (default 50). task_state_from_label and TASK_STATE_LABELS translate compact labels like "completed" or "input-required" into TaskState values for filters. CLI: handler task list with --context-id, --status, --page-size, --history-length, and --include-artifacts. Text output is a compact one-line-per-task listing; --output json wraps the full wire-format tasks with a count; ndjson emits one task per line. All pages are always fetched rather than silently truncating at the first. MCP: a list_tasks tool with the same filters, returning count + tasks. Verified against the local streaming test agent: listing returns all tasks across pages (page-size 2 over 6 tasks), and filtering by context and by status both narrow correctly. The TUI task browser this unlocks is left as a follow-up. Closes #101 Co-Authored-By: Claude Fable 5 --- src/a2a_handler/cli/task.py | 114 ++++++++++++++++++++++++++++++++ src/a2a_handler/mcp/server.py | 77 ++++++++++++++++++++++ src/a2a_handler/service.py | 121 +++++++++++++++++++++++++++++++++- tests/cli/test_task.py | 85 ++++++++++++++++++++++++ tests/core/test_service.py | 118 ++++++++++++++++++++++++++++++++- tests/mcp/test_server.py | 58 ++++++++++++++++ 6 files changed, 571 insertions(+), 2 deletions(-) diff --git a/src/a2a_handler/cli/task.py b/src/a2a_handler/cli/task.py index e2cacb7..0e3c55a 100644 --- a/src/a2a_handler/cli/task.py +++ b/src/a2a_handler/cli/task.py @@ -19,8 +19,12 @@ from a2a.types import Task from a2a_handler.service import ( A2AService, + TASK_STATE_LABELS, protocol_dump, push_config_dump, + response_state, + state_label, + task_state_from_label, ) from ._helpers import ( @@ -128,6 +132,116 @@ async def do_get() -> None: asyncio.run(do_get()) +@task.command("list") +@click.option("--url", "agent_url", help="Agent URL") +@click.option("--server", "-s", "server_name", help="Named server from servers.toml") +@click.option("--context-id", help="Only list tasks in this context") +@click.option( + "--status", + type=click.Choice(TASK_STATE_LABELS), + help="Only list tasks in this state", +) +@click.option( + "--page-size", + type=int, + help="Tasks per request page (all pages are still fetched)", +) +@click.option( + "--history-length", "-n", type=int, help="History messages to include per task" +) +@click.option( + "--include-artifacts", is_flag=True, help="Include task artifacts in the output" +) +@click.option( + "--bearer-env", "-b", help="Env var containing bearer token (overrides saved)" +) +@click.option( + "--api-key-env", "-k", help="Env var containing API key (overrides saved)" +) +def task_list( + agent_url: Optional[str], + server_name: Optional[str], + context_id: Optional[str], + status: Optional[str], + page_size: Optional[int], + history_length: Optional[int], + include_artifacts: bool, + bearer_env: Optional[str], + api_key_env: Optional[str], +) -> None: + """List the agent's tasks, following pagination to the end. + + \b + Examples: + $ handler task list --server my_agent + $ handler task list --url http://localhost:8000 --status completed + $ handler task list --server my_agent --context-id ctx-123 + $ handler --output json task list --server my_agent --include-artifacts + """ + output = Output() + + selection = resolve_agent_selection(agent_url, server_name) + resolved_url = selection.agent_url + + status_value: Optional[int] = None + try: + validate_agent_url(resolved_url) + if context_id: + validate_resource_id(context_id, "context_id") + if status: + status_value = task_state_from_label(status) + except InputValidationError as error: + handle_validation_error(error, output) + raise click.Abort() from error + + log.info("Listing tasks at %s", resolved_url) + + credentials = resolve_selection_credentials(selection, bearer_env, api_key_env) + + async def do_list() -> None: + try: + async with build_http_client(credentials=credentials) as http_client: + service = A2AService(http_client, resolved_url, credentials=credentials) + tasks = await service.list_all_tasks( + context_id=context_id, + status=status_value, + page_size=page_size, + history_length=history_length, + include_artifacts=include_artifacts, + ) + _format_task_list(tasks, output) + except Exception as e: + handle_client_error(e, resolved_url, output) + raise click.Abort() + + asyncio.run(do_list()) + + +def _format_task_list(tasks: list[Task], output: Output) -> None: + """Format a task listing for structured or human-readable output.""" + if output.output_format == "ndjson": + for task_item in tasks: + output.json(protocol_dump(task_item)) + return + if output.output_format == "json": + output.json( + { + "count": len(tasks), + "tasks": [protocol_dump(task_item) for task_item in tasks], + } + ) + return + + if not tasks: + output.text("No tasks found.") + return + for task_item in tasks: + line = f"{task_item.id} {state_label(response_state(task_item))}" + if task_item.context_id: + line += f" context={task_item.context_id}" + output.text(line) + + @task.command("cancel") @click.argument("task_id_arg", metavar="[TASK_ID]", required=False) @click.option("--url", "agent_url", help="Agent URL") diff --git a/src/a2a_handler/mcp/server.py b/src/a2a_handler/mcp/server.py index 554e797..84daccd 100644 --- a/src/a2a_handler/mcp/server.py +++ b/src/a2a_handler/mcp/server.py @@ -29,6 +29,7 @@ push_config_dump, response_context_id, response_task_id, + task_state_from_label, to_json_dict, ) from a2a_handler.session import ( @@ -388,6 +389,82 @@ async def get_task( return protocol_dump(task) + @mcp.tool() + async def list_tasks( + agent_url: str, + context_id: str | None = None, + status: str | None = None, + page_size: int | None = None, + history_length: int | None = None, + include_artifacts: bool = False, + bearer_token: str | None = None, + api_key: str | None = None, + cert_path: str | None = None, + key_path: str | None = None, + ca_cert_path: str | None = None, + custom_headers: dict[str, str] | None = None, + ) -> dict: + """List an agent's tasks, following pagination to the end. + + Retrieves every task the agent will show this client, optionally + filtered by context or state. + + Args: + agent_url: Base URL of the A2A agent + context_id: Only list tasks in this context + status: Only list tasks in this state (e.g., "completed", + "working", "input_required") + page_size: Tasks per request page (all pages are still fetched) + history_length: History messages to include per task + include_artifacts: Whether to include task artifacts + bearer_token: Optional bearer token for authentication + api_key: Optional API key for authentication + + Returns: + A dictionary containing: + - count: Number of tasks returned + - tasks: The tasks in A2A wire format + """ + logger.info("Listing tasks at %s", agent_url) + status_value: int | None = None + try: + validate_agent_url(agent_url) + if context_id: + validate_resource_id(context_id, "context_id") + if status: + status_value = task_state_from_label(status) + if bearer_token: + reject_control_chars(bearer_token, "bearer_token") + if api_key: + reject_control_chars(api_key, "api_key") + except InputValidationError as error: + raise _validation_error(error) from error + + credentials = _resolve_credentials( + agent_url, + bearer_token, + api_key, + cert_path, + key_path, + ca_cert_path, + custom_headers, + ) + + async with _build_http_client(credentials=credentials) as http_client: + service = A2AService(http_client, agent_url, credentials=credentials) + tasks = await service.list_all_tasks( + context_id=context_id, + status=status_value, + page_size=page_size, + history_length=history_length, + include_artifacts=include_artifacts, + ) + + return { + "count": len(tasks), + "tasks": [protocol_dump(task) for task in tasks], + } + @mcp.tool() async def cancel_task( agent_url: str, diff --git a/src/a2a_handler/service.py b/src/a2a_handler/service.py index 0ba1201..dae69b7 100644 --- a/src/a2a_handler/service.py +++ b/src/a2a_handler/service.py @@ -13,7 +13,7 @@ import uuid from dataclasses import dataclass from pathlib import Path -from typing import Any, AsyncIterator, Iterable, Sequence, Union +from typing import Any, AsyncIterator, Iterable, Sequence, Union, cast from urllib.parse import urlparse import httpx @@ -31,6 +31,8 @@ CancelTaskRequest, GetTaskPushNotificationConfigRequest, GetTaskRequest, + ListTasksRequest, + ListTasksResponse, Message, Part, Role, @@ -62,6 +64,10 @@ # path locally so Handler can still fall back to it for older servers. LEGACY_AGENT_CARD_WELL_KNOWN_PATH = "/.well-known/agent.json" +# Tasks fetched per ListTasks request when the caller does not choose a page +# size. Servers reject a zero (i.e. unset) page size outright. +DEFAULT_LIST_TASKS_PAGE_SIZE = 50 + TERMINAL_TASK_STATES = { TaskState.TASK_STATE_COMPLETED, TaskState.TASK_STATE_CANCELED, @@ -123,6 +129,29 @@ def state_label(state: int | None) -> str: return TaskState.Name(state).removeprefix("TASK_STATE_").lower() +#: Compact labels for every real task state, e.g. ``completed``, +#: ``input_required``. Used for CLI choices and label parsing. +TASK_STATE_LABELS = tuple(state_label(value) for value in TaskState.values() if value) + + +def task_state_from_label(label: str) -> int: + """Return the ``TaskState`` value for a compact label like ``completed``. + + Accepts hyphens or underscores (``input-required`` and ``input_required`` + both work). + """ + normalized = label.strip().lower().replace("-", "_") + try: + return TaskState.Value(f"TASK_STATE_{normalized.upper()}") + except ValueError: + raise InputValidationError( + code="invalid_task_state", + message=f"Unknown task state: {label}", + suggestion=f"Use one of: {', '.join(TASK_STATE_LABELS)}", + details={"field": "status"}, + ) from None + + def role_label(role: int | None) -> str: """Return a compact, human-readable label for a ``Role`` value.""" if not role: @@ -853,6 +882,96 @@ async def cancel_task(self, task_id: str) -> Task: return await client.cancel_task(CancelTaskRequest(id=task_id)) + async def list_tasks( + self, + context_id: str | None = None, + status: int | None = None, + page_size: int | None = None, + page_token: str | None = None, + history_length: int | None = None, + include_artifacts: bool = False, + ) -> ListTasksResponse: + """List tasks on the agent, one page at a time. + + Args: + context_id: Only return tasks in this context + status: Only return tasks in this ``TaskState`` + page_size: Maximum tasks per page (server may return fewer); + defaults to ``DEFAULT_LIST_TASKS_PAGE_SIZE`` + page_token: Continuation token from a previous page's + ``next_page_token`` + history_length: Number of history messages to include per task + include_artifacts: Whether to include task artifacts + + Returns: + The raw ``ListTasksResponse`` with tasks and the next page token. + """ + if context_id: + validate_resource_id(context_id, "context_id") + if page_token: + reject_control_chars(page_token, "page_token") + + client = await self._get_or_create_client() + + # An unset proto3 int is indistinguishable from 0, and servers reject a + # zero page size, so always send an explicit one. + request = ListTasksRequest( + context_id=context_id or "", + page_size=page_size or DEFAULT_LIST_TASKS_PAGE_SIZE, + page_token=page_token or "", + include_artifacts=include_artifacts, + ) + if status is not None: + request.status = cast("TaskState", status) + if history_length is not None: + request.history_length = history_length + + logger.info( + "Listing tasks (context_id=%s, status=%s, page_token=%s)", + context_id, + state_label(status) if status else "any", + page_token or "", + ) + + return await client.list_tasks(request) + + async def list_all_tasks( + self, + context_id: str | None = None, + status: int | None = None, + page_size: int | None = None, + history_length: int | None = None, + include_artifacts: bool = False, + ) -> list[Task]: + """List tasks across every page, following continuation tokens. + + A repeated token stops the loop, so a server that keeps returning the + same page cannot spin this forever. + """ + tasks: list[Task] = [] + page_token: str | None = None + seen_tokens: set[str] = set() + + while True: + response = await self.list_tasks( + context_id=context_id, + status=status, + page_size=page_size, + page_token=page_token, + history_length=history_length, + include_artifacts=include_artifacts, + ) + tasks.extend(response.tasks) + page_token = response.next_page_token + if not page_token or page_token in seen_tokens: + break + seen_tokens.add(page_token) + + logger.info( + "Listed %d task(s) across %d page(s)", len(tasks), len(seen_tokens) + 1 + ) + return tasks + async def resubscribe(self, task_id: str) -> AsyncIterator[StreamEvent]: """Resubscribe to a task's event stream. diff --git a/tests/cli/test_task.py b/tests/cli/test_task.py index 4447ba1..fa0b356 100644 --- a/tests/cli/test_task.py +++ b/tests/cli/test_task.py @@ -397,6 +397,91 @@ def test_task_id_from_json_params_alone(self, runner): mock_service.get_task.assert_awaited_once_with("task-123", None) +class TestTaskList: + """Tests for task list command.""" + + def _invoke_list(self, runner, args, tasks): + with ( + patch("a2a_handler.cli.task.build_http_client") as mock_client, + patch("a2a_handler.cli.task.A2AService") as mock_service_cls, + ): + mock_http = AsyncMock() + mock_http.__aenter__.return_value = mock_http + mock_http.__aexit__.return_value = None + mock_client.return_value = mock_http + + mock_service = AsyncMock() + mock_service.list_all_tasks.return_value = tasks + mock_service_cls.return_value = mock_service + + result = runner.invoke(task, ["list", *args]) + return result, mock_service + + def test_task_list_success(self, runner): + """Listing shows every returned task with its state.""" + tasks = [ + _make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1"), + _make_task(TaskState.TASK_STATE_WORKING, task_id="task-2"), + ] + result, _ = self._invoke_list(runner, ["--url", "http://localhost:8000"], tasks) + + assert result.exit_code == 0 + assert "task-1" in result.output + assert "completed" in result.output + assert "task-2" in result.output + assert "working" in result.output + + def test_task_list_empty(self, runner): + """An empty listing says so instead of printing nothing.""" + result, _ = self._invoke_list(runner, ["--url", "http://localhost:8000"], []) + + assert result.exit_code == 0 + assert "No tasks found" in result.output + + def test_task_list_passes_filters(self, runner): + """Context, status, and paging options reach the service.""" + result, mock_service = self._invoke_list( + runner, + [ + "--url", + "http://localhost:8000", + "--context-id", + "ctx-9", + "--status", + "completed", + "--page-size", + "10", + "--include-artifacts", + ], + [], + ) + + assert result.exit_code == 0 + mock_service.list_all_tasks.assert_called_once_with( + context_id="ctx-9", + status=TaskState.TASK_STATE_COMPLETED, + page_size=10, + history_length=None, + include_artifacts=True, + ) + + def test_task_list_rejects_unknown_status(self, runner): + """An unknown status label fails before any network call.""" + result = runner.invoke( + task, + ["list", "--url", "http://localhost:8000", "--status", "sleeping"], + ) + assert result.exit_code != 0 + + def test_task_list_rejects_invalid_context_id(self, runner): + """A malformed context ID fails before any network call.""" + result = runner.invoke( + task, + ["list", "--url", "http://localhost:8000", "--context-id", "ctx?bad"], + ) + assert result.exit_code != 0 + + class TestTaskCancel: """Tests for task cancel command.""" diff --git a/tests/core/test_service.py b/tests/core/test_service.py index 726e85c..d51fa8b 100644 --- a/tests/core/test_service.py +++ b/tests/core/test_service.py @@ -5,6 +5,7 @@ import httpx import pytest from a2a.types import ( + ListTasksResponse, Message, Part, Role, @@ -14,7 +15,7 @@ TaskStatus, TaskStatusUpdateEvent, ) -from typing import cast +from typing import Any, cast from a2a_handler.auth import create_bearer_auth, create_oauth2_auth from a2a_handler.common.input_validation import InputValidationError @@ -22,6 +23,7 @@ A2AService, MAX_INLINE_FILE_BYTES, StreamEvent, + TASK_STATE_LABELS, TERMINAL_TASK_STATES, attachment_part_from_spec, build_data_part, @@ -36,6 +38,7 @@ response_needs_auth, response_state, response_task_id, + task_state_from_label, to_json_dict, ) from a2a_handler.service import ( @@ -584,6 +587,119 @@ def test_empty_message_is_refused(self): self._service()._build_user_message("") +class TestTaskStateLabels: + """Tests for the state-label parsing used by task listing filters.""" + + def test_labels_cover_every_real_state(self): + assert "completed" in TASK_STATE_LABELS + assert "input_required" in TASK_STATE_LABELS + assert "unspecified" not in TASK_STATE_LABELS + + def test_label_round_trips(self): + assert task_state_from_label("completed") == TaskState.TASK_STATE_COMPLETED + + def test_label_accepts_hyphens(self): + assert ( + task_state_from_label("input-required") + == TaskState.TASK_STATE_INPUT_REQUIRED + ) + + def test_unknown_label_is_rejected(self): + with pytest.raises(InputValidationError) as exc_info: + task_state_from_label("sleeping") + assert isinstance(exc_info.value, InputValidationError) + assert exc_info.value.code == "invalid_task_state" + + +class _FakeListTasksClient: + """Replays one ListTasksResponse per call, recording each request.""" + + def __init__(self, pages: list[ListTasksResponse]) -> None: + self.pages = pages + self.requests: list[Any] = [] + + async def list_tasks(self, request): + self.requests.append(request) + index = min(len(self.requests) - 1, len(self.pages) - 1) + return self.pages[index] + + +@pytest.mark.asyncio +class TestA2AServiceListTasks: + """Tests for A2AService.list_tasks and list_all_tasks.""" + + def _service_with(self, fake_client: _FakeListTasksClient) -> A2AService: + service = A2AService( + http_client=cast(httpx.AsyncClient, AsyncMock()), + agent_url="http://example.com", + ) + + async def _get_client(): + return fake_client + + service._get_or_create_client = _get_client # type: ignore[method-assign] + return service + + async def test_list_tasks_passes_filters_through(self): + page = ListTasksResponse(tasks=[_make_task(TaskState.TASK_STATE_COMPLETED)]) + fake_client = _FakeListTasksClient([page]) + service = self._service_with(fake_client) + + response = await service.list_tasks( + context_id="ctx-1", + status=TaskState.TASK_STATE_COMPLETED, + page_size=25, + history_length=5, + include_artifacts=True, + ) + + assert len(response.tasks) == 1 + request = fake_client.requests[0] + assert request.context_id == "ctx-1" + assert request.status == TaskState.TASK_STATE_COMPLETED + assert request.page_size == 25 + assert request.history_length == 5 + assert request.include_artifacts is True + + async def test_list_tasks_rejects_bad_context_id(self): + service = self._service_with(_FakeListTasksClient([ListTasksResponse()])) + with pytest.raises(InputValidationError): + await service.list_tasks(context_id="ctx?bad") + + async def test_list_all_tasks_follows_pages(self): + pages = [ + ListTasksResponse( + tasks=[_make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1")], + next_page_token="page-2", + ), + ListTasksResponse( + tasks=[_make_task(TaskState.TASK_STATE_WORKING, task_id="task-2")], + ), + ] + fake_client = _FakeListTasksClient(pages) + service = self._service_with(fake_client) + + tasks = await service.list_all_tasks(page_size=1) + + assert [item.id for item in tasks] == ["task-1", "task-2"] + assert len(fake_client.requests) == 2 + assert fake_client.requests[1].page_token == "page-2" + + async def test_list_all_tasks_stops_on_repeated_token(self): + page = ListTasksResponse( + tasks=[_make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1")], + next_page_token="same-token", + ) + fake_client = _FakeListTasksClient([page]) + service = self._service_with(fake_client) + + tasks = await service.list_all_tasks() + + # First call, then one follow-up with the token; the repeat stops it. + assert len(fake_client.requests) == 2 + assert len(tasks) == 2 + + class _FakeStreamingClient: def __init__(self, events): self._events = events diff --git a/tests/mcp/test_server.py b/tests/mcp/test_server.py index af003b1..689c0dd 100644 --- a/tests/mcp/test_server.py +++ b/tests/mcp/test_server.py @@ -67,6 +67,7 @@ def test_mcp_server_registers_core_tools() -> None: assert "send_message" in names assert "get_task" in names + assert "list_tasks" in names assert "set_task_notification" in names assert "list_sessions" in names @@ -328,6 +329,63 @@ async def test_get_task_rejects_invalid_task_id() -> None: await fn(agent_url="http://localhost:8000", task_id="bad\x00id") +# --------------------------------------------------------------------------- +# list_tasks +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_list_tasks_success() -> None: + server = create_mcp_server() + fn = _tool_fn(server, "list_tasks") + + tasks = [ + _make_task(task_id="task-1", state=TaskState.TASK_STATE_COMPLETED), + _make_task(task_id="task-2", state=TaskState.TASK_STATE_WORKING), + ] + + mock_service = AsyncMock() + mock_service.list_all_tasks.return_value = tasks + + with ( + patch("a2a_handler.mcp.server._build_http_client", return_value=_mock_http()), + patch("a2a_handler.mcp.server.A2AService", return_value=mock_service), + ): + resp = await fn( + agent_url="http://localhost:8000", + context_id="ctx-1", + status="completed", + ) + + assert resp["count"] == 2 + assert resp["tasks"][0]["id"] == "task-1" + mock_service.list_all_tasks.assert_called_once_with( + context_id="ctx-1", + status=TaskState.TASK_STATE_COMPLETED, + page_size=None, + history_length=None, + include_artifacts=False, + ) + + +@pytest.mark.asyncio +async def test_list_tasks_rejects_invalid_url() -> None: + server = create_mcp_server() + fn = _tool_fn(server, "list_tasks") + + with pytest.raises(ValueError, match="invalid_agent_url"): + await fn(agent_url="nope") + + +@pytest.mark.asyncio +async def test_list_tasks_rejects_unknown_status() -> None: + server = create_mcp_server() + fn = _tool_fn(server, "list_tasks") + + with pytest.raises(ValueError, match="invalid_task_state"): + await fn(agent_url="http://localhost:8000", status="sleeping") + + # --------------------------------------------------------------------------- # cancel_task # --------------------------------------------------------------------------- From 4696e7ef0c44e8eb53392b478b1cfe22d79779b2 Mon Sep 17 00:00:00 2001 From: Al Duncanson Date: Tue, 8 Sep 2026 11:41:06 -0400 Subject: [PATCH 2/2] fix: address review findings on task listing - list_all_tasks deduplicates tasks by ID, stops when the server returns the token that was just sent (a guaranteed loop), and caps pagination at MAX_LIST_TASKS_PAGES so a server minting fresh tokens forever cannot spin the client; the old repeated-token guard appended the replayed page's tasks before breaking, double-counting them. - page_size below 1 is rejected with a clear InputValidationError at the service, CLI, and MCP layers instead of silently becoming the default (0) or reaching the wire (negatives); the 50-task default is now stated in the CLI and MCP help. - task_state_from_label only accepts the real states in TASK_STATE_LABELS: "unspecified" maps to the proto default and would silently drop the status filter, so it errors like any unknown label. - The CLI --status option uses the tolerant parser instead of click.Choice, so the documented hyphenated spellings (input-required) and mixed case work, matching the MCP tool. - The list_tasks log no longer reports "any" for a zero status value, and the test fake raises past its scripted pages instead of replaying the last one, so extra requests fail loudly. Co-Authored-By: Claude Fable 5 --- src/a2a_handler/cli/task.py | 15 ++++++-- src/a2a_handler/mcp/server.py | 7 ++++ src/a2a_handler/service.py | 69 ++++++++++++++++++++++++++--------- tests/cli/test_task.py | 28 ++++++++++++++ tests/core/test_service.py | 54 +++++++++++++++++++++++---- tests/mcp/test_server.py | 20 ++++++++++ 6 files changed, 165 insertions(+), 28 deletions(-) diff --git a/src/a2a_handler/cli/task.py b/src/a2a_handler/cli/task.py index 0e3c55a..1bc5019 100644 --- a/src/a2a_handler/cli/task.py +++ b/src/a2a_handler/cli/task.py @@ -138,13 +138,15 @@ async def do_get() -> None: @click.option("--context-id", help="Only list tasks in this context") @click.option( "--status", - type=click.Choice(TASK_STATE_LABELS), - help="Only list tasks in this state", + help=( + "Only list tasks in this state " + f"(one of: {', '.join(TASK_STATE_LABELS)}; hyphens also accepted)" + ), ) @click.option( "--page-size", type=int, - help="Tasks per request page (all pages are still fetched)", + help="Tasks per request page, default 50 (all pages are still fetched)", ) @click.option( "--history-length", "-n", type=int, help="History messages to include per task" @@ -190,6 +192,13 @@ def task_list( validate_resource_id(context_id, "context_id") if status: status_value = task_state_from_label(status) + if page_size is not None and page_size < 1: + raise InputValidationError( + code="invalid_page_size", + message="--page-size must be at least 1", + suggestion="Omit --page-size to use the default of 50", + details={"field": "page_size"}, + ) except InputValidationError as error: handle_validation_error(error, output) raise click.Abort() from error diff --git a/src/a2a_handler/mcp/server.py b/src/a2a_handler/mcp/server.py index 84daccd..3889207 100644 --- a/src/a2a_handler/mcp/server.py +++ b/src/a2a_handler/mcp/server.py @@ -433,6 +433,13 @@ async def list_tasks( validate_resource_id(context_id, "context_id") if status: status_value = task_state_from_label(status) + if page_size is not None and page_size < 1: + raise InputValidationError( + code="invalid_page_size", + message="page_size must be at least 1", + suggestion="Omit page_size to use the default of 50", + details={"field": "page_size"}, + ) if bearer_token: reject_control_chars(bearer_token, "bearer_token") if api_key: diff --git a/src/a2a_handler/service.py b/src/a2a_handler/service.py index dae69b7..3eb95a4 100644 --- a/src/a2a_handler/service.py +++ b/src/a2a_handler/service.py @@ -68,6 +68,10 @@ # size. Servers reject a zero (i.e. unset) page size outright. DEFAULT_LIST_TASKS_PAGE_SIZE = 50 +# Hard bound on pagination so a server minting a fresh continuation token on +# every response cannot spin list_all_tasks forever. +MAX_LIST_TASKS_PAGES = 1000 + TERMINAL_TASK_STATES = { TaskState.TASK_STATE_COMPLETED, TaskState.TASK_STATE_CANCELED, @@ -138,18 +142,19 @@ def task_state_from_label(label: str) -> int: """Return the ``TaskState`` value for a compact label like ``completed``. Accepts hyphens or underscores (``input-required`` and ``input_required`` - both work). + both work). Only the real states in ``TASK_STATE_LABELS`` are accepted: + ``unspecified`` maps to the proto default and would silently drop a + filter, so it is rejected like any unknown label. """ normalized = label.strip().lower().replace("-", "_") - try: - return TaskState.Value(f"TASK_STATE_{normalized.upper()}") - except ValueError: + if normalized not in TASK_STATE_LABELS: raise InputValidationError( code="invalid_task_state", message=f"Unknown task state: {label}", suggestion=f"Use one of: {', '.join(TASK_STATE_LABELS)}", details={"field": "status"}, - ) from None + ) + return TaskState.Value(f"TASK_STATE_{normalized.upper()}") def role_label(role: int | None) -> str: @@ -910,11 +915,18 @@ async def list_tasks( validate_resource_id(context_id, "context_id") if page_token: reject_control_chars(page_token, "page_token") + if page_size is not None and page_size < 1: + raise InputValidationError( + code="invalid_page_size", + message="page_size must be at least 1", + suggestion="Omit page_size to use the default", + details={"field": "page_size"}, + ) client = await self._get_or_create_client() - # An unset proto3 int is indistinguishable from 0, and servers reject a - # zero page size, so always send an explicit one. + # Servers commonly reject a zero page size and some treat an unset + # field as zero, so always send an explicit one. request = ListTasksRequest( context_id=context_id or "", page_size=page_size or DEFAULT_LIST_TASKS_PAGE_SIZE, @@ -929,7 +941,7 @@ async def list_tasks( logger.info( "Listing tasks (context_id=%s, status=%s, page_token=%s)", context_id, - state_label(status) if status else "any", + state_label(status) if status is not None else "any", page_token or "", ) @@ -945,12 +957,16 @@ async def list_all_tasks( ) -> list[Task]: """List tasks across every page, following continuation tokens. - A repeated token stops the loop, so a server that keeps returning the - same page cannot spin this forever. + Defenses against misbehaving servers: tasks are deduplicated by ID (a + replayed page adds nothing), a token identical to the one just sent + stops the loop (requesting again could only repeat), and pagination + is capped at ``MAX_LIST_TASKS_PAGES`` so a server minting fresh + tokens forever cannot spin the client. """ tasks: list[Task] = [] + seen_task_ids: set[str] = set() page_token: str | None = None - seen_tokens: set[str] = set() + pages_fetched = 0 while True: response = await self.list_tasks( @@ -961,15 +977,32 @@ async def list_all_tasks( history_length=history_length, include_artifacts=include_artifacts, ) - tasks.extend(response.tasks) - page_token = response.next_page_token - if not page_token or page_token in seen_tokens: + pages_fetched += 1 + for task in response.tasks: + if task.id and task.id in seen_task_ids: + continue + if task.id: + seen_task_ids.add(task.id) + tasks.append(task) + + next_token = response.next_page_token + if not next_token: + break + if next_token == (page_token or ""): + logger.warning( + "Server repeated page token %r; stopping pagination", + next_token, + ) + break + if pages_fetched >= MAX_LIST_TASKS_PAGES: + logger.warning( + "Stopping after %d pages; the task listing may be incomplete", + pages_fetched, + ) break - seen_tokens.add(page_token) + page_token = next_token - logger.info( - "Listed %d task(s) across %d page(s)", len(tasks), len(seen_tokens) + 1 - ) + logger.info("Listed %d task(s) across %d page(s)", len(tasks), pages_fetched) return tasks async def resubscribe(self, task_id: str) -> AsyncIterator[StreamEvent]: diff --git a/tests/cli/test_task.py b/tests/cli/test_task.py index fa0b356..9e18f56 100644 --- a/tests/cli/test_task.py +++ b/tests/cli/test_task.py @@ -472,6 +472,34 @@ def test_task_list_rejects_unknown_status(self, runner): ["list", "--url", "http://localhost:8000", "--status", "sleeping"], ) assert result.exit_code != 0 + assert "Unknown task state" in result.output + + def test_task_list_accepts_hyphenated_status(self, runner): + """The natural kebab-case spelling works on the CLI.""" + result, mock_service = self._invoke_list( + runner, + [ + "--url", + "http://localhost:8000", + "--status", + "input-required", + ], + [], + ) + assert result.exit_code == 0 + assert ( + mock_service.list_all_tasks.call_args.kwargs["status"] + == TaskState.TASK_STATE_INPUT_REQUIRED + ) + + def test_task_list_rejects_non_positive_page_size(self, runner): + """A page size below 1 fails before any network call.""" + result = runner.invoke( + task, + ["list", "--url", "http://localhost:8000", "--page-size", "0"], + ) + assert result.exit_code != 0 + assert "at least 1" in result.output def test_task_list_rejects_invalid_context_id(self, runner): """A malformed context ID fails before any network call.""" diff --git a/tests/core/test_service.py b/tests/core/test_service.py index d51fa8b..5e1bb86 100644 --- a/tests/core/test_service.py +++ b/tests/core/test_service.py @@ -610,9 +610,19 @@ def test_unknown_label_is_rejected(self): assert isinstance(exc_info.value, InputValidationError) assert exc_info.value.code == "invalid_task_state" + def test_unspecified_is_rejected(self): + # TASK_STATE_UNSPECIFIED is the proto default; accepting it would + # silently drop the status filter and return every task. + with pytest.raises(InputValidationError): + task_state_from_label("unspecified") + class _FakeListTasksClient: - """Replays one ListTasksResponse per call, recording each request.""" + """Replays one ListTasksResponse per call, recording each request. + + Raises IndexError past the scripted pages so a regression that issues + extra requests fails loudly instead of replaying the last page. + """ def __init__(self, pages: list[ListTasksResponse]) -> None: self.pages = pages @@ -620,8 +630,7 @@ def __init__(self, pages: list[ListTasksResponse]) -> None: async def list_tasks(self, request): self.requests.append(request) - index = min(len(self.requests) - 1, len(self.pages) - 1) - return self.pages[index] + return self.pages[len(self.requests) - 1] @pytest.mark.asyncio @@ -685,19 +694,50 @@ async def test_list_all_tasks_follows_pages(self): assert len(fake_client.requests) == 2 assert fake_client.requests[1].page_token == "page-2" - async def test_list_all_tasks_stops_on_repeated_token(self): + async def test_list_all_tasks_stops_on_repeated_token_without_duplicates(self): page = ListTasksResponse( tasks=[_make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1")], next_page_token="same-token", ) - fake_client = _FakeListTasksClient([page]) + fake_client = _FakeListTasksClient([page, page]) service = self._service_with(fake_client) tasks = await service.list_all_tasks() - # First call, then one follow-up with the token; the repeat stops it. + # First call, then one follow-up with the token; the replayed page + # stops the loop and its tasks are deduplicated, not double-counted. assert len(fake_client.requests) == 2 - assert len(tasks) == 2 + assert [item.id for item in tasks] == ["task-1"] + + async def test_list_all_tasks_is_capped_against_fresh_token_loops( + self, monkeypatch + ): + # A server minting a new token every page never repeats one; the + # page cap is what stops it. + monkeypatch.setattr("a2a_handler.service.MAX_LIST_TASKS_PAGES", 3) + pages = [ + ListTasksResponse( + tasks=[_make_task(TaskState.TASK_STATE_COMPLETED, task_id=f"task-{n}")], + next_page_token=f"token-{n}", + ) + for n in range(10) + ] + fake_client = _FakeListTasksClient(pages) + service = self._service_with(fake_client) + + tasks = await service.list_all_tasks() + + assert len(fake_client.requests) == 3 + assert [item.id for item in tasks] == ["task-0", "task-1", "task-2"] + + async def test_list_tasks_rejects_non_positive_page_size(self): + service = self._service_with(_FakeListTasksClient([])) + with pytest.raises(InputValidationError) as exc_info: + await service.list_tasks(page_size=0) + assert isinstance(exc_info.value, InputValidationError) + assert exc_info.value.code == "invalid_page_size" + with pytest.raises(InputValidationError): + await service.list_tasks(page_size=-5) class _FakeStreamingClient: diff --git a/tests/mcp/test_server.py b/tests/mcp/test_server.py index 689c0dd..19bdeca 100644 --- a/tests/mcp/test_server.py +++ b/tests/mcp/test_server.py @@ -386,6 +386,26 @@ async def test_list_tasks_rejects_unknown_status() -> None: await fn(agent_url="http://localhost:8000", status="sleeping") +@pytest.mark.asyncio +async def test_list_tasks_rejects_unspecified_status() -> None: + # "unspecified" would map to the proto default and silently drop the + # filter; it must error like any unknown label. + server = create_mcp_server() + fn = _tool_fn(server, "list_tasks") + + with pytest.raises(ValueError, match="invalid_task_state"): + await fn(agent_url="http://localhost:8000", status="unspecified") + + +@pytest.mark.asyncio +async def test_list_tasks_rejects_non_positive_page_size() -> None: + server = create_mcp_server() + fn = _tool_fn(server, "list_tasks") + + with pytest.raises(ValueError, match="invalid_page_size"): + await fn(agent_url="http://localhost:8000", page_size=0) + + # --------------------------------------------------------------------------- # cancel_task # ---------------------------------------------------------------------------