fix: Add validation for empty artifact lists in completed_task (#308)

This update addresses an issue where the `completed_task` function in
`a2a/utils/task.py` did not perform any validation on the `artifacts`
list. This could lead to unexpected behavior if an empty or invalid list
was provided.

This change introduces a validation check to ensure that the `artifacts`
list is a non-empty list of `Artifact` objects, raising a `ValueError`
if the validation fails.

**Changes:**

- Modified `a2a/utils/task.py` to add a validation check for the
`artifacts` parameter in the `completed_task` function.
- Updated `tests/utils/test_task.py` to include tests for the new
validation logic, covering cases with empty lists and lists containing
invalid items.

Author: pankaj-bind

src/a2a/utils/task.py

@@ -55,6 +55,11 @@ def completed_task(
     Returns:
         A `Task` object with status set to 'completed'.
     """
+    if not artifacts or not all(isinstance(a, Artifact) for a in artifacts):
+        raise ValueError(
+            'artifacts must be a non-empty list of Artifact objects'
+        )
+
     if history is None:
         history = []
     return Task(

tests/utils/test_task.py

@@ -3,7 +3,9 @@
 
 from unittest.mock import patch
 
-from a2a.types import Message, Part, Role, TextPart
+import pytest
+
+from a2a.types import Artifact, Message, Part, Role, TextPart
 from a2a.utils.task import completed_task, new_task
 
 
@@ -57,7 +59,12 @@ def test_new_task_initial_message_in_history(self):
     def test_completed_task_status(self):
         task_id = str(uuid.uuid4())
         context_id = str(uuid.uuid4())
-        artifacts = []  # Artifacts should be of type Artifact
+        artifacts = [
+            Artifact(
+                artifactId='artifact_1',
+                parts=[Part(root=TextPart(text='some content'))],
+            )
+        ]
         task = completed_task(
             task_id=task_id,
             context_id=context_id,
@@ -69,7 +76,12 @@ def test_completed_task_status(self):
     def test_completed_task_assigns_ids_and_artifacts(self):
         task_id = str(uuid.uuid4())
         context_id = str(uuid.uuid4())
-        artifacts = []  # Artifacts should be of type Artifact
+        artifacts = [
+            Artifact(
+                artifactId='artifact_1',
+                parts=[Part(root=TextPart(text='some content'))],
+            )
+        ]
         task = completed_task(
             task_id=task_id,
             context_id=context_id,
@@ -83,7 +95,12 @@ def test_completed_task_assigns_ids_and_artifacts(self):
     def test_completed_task_empty_history_if_not_provided(self):
         task_id = str(uuid.uuid4())
         context_id = str(uuid.uuid4())
-        artifacts = []  # Artifacts should be of type Artifact
+        artifacts = [
+            Artifact(
+                artifactId='artifact_1',
+                parts=[Part(root=TextPart(text='some content'))],
+            )
+        ]
         task = completed_task(
             task_id=task_id, context_id=context_id, artifacts=artifacts
         )
@@ -92,7 +109,12 @@ def test_completed_task_empty_history_if_not_provided(self):
     def test_completed_task_uses_provided_history(self):
         task_id = str(uuid.uuid4())
         context_id = str(uuid.uuid4())
-        artifacts = []  # Artifacts should be of type Artifact
+        artifacts = [
+            Artifact(
+                artifactId='artifact_1',
+                parts=[Part(root=TextPart(text='some content'))],
+            )
+        ]
         history = [
             Message(
                 role=Role.user,
@@ -132,6 +154,30 @@ def test_new_task_invalid_message_none_role(self):
             )
             new_task(msg)
 
+    def test_completed_task_empty_artifacts(self):
+        with pytest.raises(
+            ValueError,
+            match='artifacts must be a non-empty list of Artifact objects',
+        ):
+            completed_task(
+                task_id='task-123',
+                context_id='ctx-456',
+                artifacts=[],
+                history=[],
+            )
+
+    def test_completed_task_invalid_artifact_type(self):
+        with pytest.raises(
+            ValueError,
+            match='artifacts must be a non-empty list of Artifact objects',
+        ):
+            completed_task(
+                task_id='task-123',
+                context_id='ctx-456',
+                artifacts=['not an artifact'],
+                history=[],
+            )
+
 
 if __name__ == '__main__':
     unittest.main()