feat: implement ListTasks across service, CLI, and MCP

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,125 @@ 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",
+    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, default 50 (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)
+        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
+
+    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")

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,89 @@ 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 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:
+                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,