fix: add history length and page size validations

Author: ishymko

src/a2a/server/request_handlers/default_request_handler.py

@@ -52,7 +52,11 @@
     TaskNotFoundError,
     UnsupportedOperationError,
 )
-from a2a.utils.task import apply_history_length
+from a2a.utils.task import (
+    apply_history_length,
+    validate_history_length,
+    validate_page_size,
+)
 from a2a.utils.telemetry import SpanKind, trace_class
 
 
@@ -122,6 +126,8 @@ async def on_get_task(
         context: ServerCallContext | None = None,
     ) -> Task | None:
         """Default handler for 'tasks/get'."""
+        validate_history_length(params)
+
         task_id = params.id
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
@@ -135,6 +141,10 @@ async def on_list_tasks(
         context: ServerCallContext | None = None,
     ) -> ListTasksResponse:
         """Default handler for 'tasks/list'."""
+        validate_history_length(params)
+        if params.HasField('page_size'):
+            validate_page_size(params.page_size)
+
         page = await self.task_store.list(params, context)
         for task in page.tasks:
             if not params.include_artifacts:
@@ -327,6 +337,8 @@ async def on_message_send(
         Starts the agent execution for the message and waits for the final
         result (Task or Message).
         """
+        validate_history_length(params.configuration)
+
         (
             _task_manager,
             task_id,

src/a2a/utils/task.py

@@ -13,6 +13,7 @@
     TaskState,
     TaskStatus,
 )
+from a2a.utils.errors import InvalidParamsError, ServerError
 
 
 def new_task(request: Message) -> Task:
@@ -96,6 +97,16 @@ def HasField(self, field_name: Literal['history_length']) -> bool:  # noqa: N802
         ...
 
 
+def validate_history_length(config: HistoryLengthConfig | None) -> None:
+    """Validates that history_length is non-negative."""
+    if config.history_length < 0:
+        raise ServerError(
+            error=InvalidParamsError(
+                message='history length must be non-negative'
+            )
+        )
+
+
 def apply_history_length(
     task: Task, config: HistoryLengthConfig | None
 ) -> Task:
@@ -136,6 +147,22 @@ def apply_history_length(
     return task
 
 
+def validate_page_size(page_size: int) -> None:
+    """Validates that page_size is in range [1, 100].
+
+    See Also:
+        https://a2a-protocol.org/latest/specification/#314-list-tasks
+    """
+    if page_size < 1:
+        raise ServerError(
+            error=InvalidParamsError(message='minimum page size is 1')
+        )
+    if page_size > 100:
+        raise ServerError(
+            error=InvalidParamsError(message='maximum page size is 100')
+        )
+
+
 _ENCODING = 'utf-8'