Formatting

Author: holtskinner

src/a2a/server/agent_execution/context.py

@@ -3,8 +3,8 @@
 from a2a.types import (
     InvalidParamsError,
     Message,
-    MessageSendParams,
     MessageSendConfiguration,
+    MessageSendParams,
     Task,
 )
 from a2a.utils import get_message_text

src/a2a/server/request_handlers/default_request_handler.py

@@ -128,9 +128,14 @@ async def on_message_send(
         if task:
             task = task_manager.update_with_message(params.message, task)
             if self.should_add_push_info(params):
-                assert isinstance(self._push_notifier, PushNotifier) # For typechecker
-                assert isinstance(params.configuration, MessageSendConfiguration) # For typechecker
-                assert isinstance(params.configuration.pushNotificationConfig, PushNotificationConfig) # For typechecker
+                assert isinstance(self._push_notifier, PushNotifier)
+                assert isinstance(
+                    params.configuration, MessageSendConfiguration
+                )
+                assert isinstance(
+                    params.configuration.pushNotificationConfig,
+                    PushNotificationConfig,
+                )
                 await self._push_notifier.set_info(
                     task.id, params.configuration.pushNotificationConfig
                 )
@@ -193,9 +198,14 @@ async def on_message_send_stream(
             task = task_manager.update_with_message(params.message, task)
 
             if self.should_add_push_info(params):
-                assert isinstance(self._push_notifier, PushNotifier) # For typechecker
-                assert isinstance(params.configuration, MessageSendConfiguration) # For typechecker
-                assert isinstance(params.configuration.pushNotificationConfig, PushNotificationConfig) # For typechecker
+                assert isinstance(self._push_notifier, PushNotifier)
+                assert isinstance(
+                    params.configuration, MessageSendConfiguration
+                )
+                assert isinstance(
+                    params.configuration.pushNotificationConfig,
+                    PushNotificationConfig,
+                )
                 await self._push_notifier.set_info(
                     task.id, params.configuration.pushNotificationConfig
                 )
@@ -324,11 +334,8 @@ async def on_resubscribe_to_task(
             yield event
 
     def should_add_push_info(self, params: MessageSendParams) -> bool:
-        if (
+        return bool(
             self._push_notifier
             and params.configuration
             and params.configuration.pushNotificationConfig
-        ):
-            return True
-        else:
-            return False
+        )

src/a2a/server/tasks/task_updater.py

@@ -68,11 +68,7 @@ def complete(self, message: Message | None = None):
 
     def failed(self, message: Message | None = None):
         """Mark the task as failed."""
-        self.update_status(
-            TaskState.failed,
-            message=message,
-            final=True
-        )
+        self.update_status(TaskState.failed, message=message, final=True)
 
     def submit(self, message: Message | None = None):
         """Mark the task as submitted."""

src/a2a/utils/__init__.py

@@ -5,34 +5,34 @@
 )
 from a2a.utils.helpers import (
     append_artifact_to_task,
+    are_modalities_compatible,
     build_text_artifact,
     create_task_obj,
-    are_modalities_compatible,
 )
 from a2a.utils.message import (
     get_message_text,
     get_text_parts,
-    new_agent_text_message,
     new_agent_parts_message,
+    new_agent_text_message,
 )
 from a2a.utils.task import (
-    new_task,
     completed_task,
+    new_task,
 )
 
 
 __all__ = [
     'append_artifact_to_task',
+    'are_modalities_compatible',
     'build_text_artifact',
+    'completed_task',
     'create_task_obj',
     'get_message_text',
     'get_text_parts',
-    'new_agent_text_message',
-    'new_task',
-    'new_text_artifact',
     'new_agent_parts_message',
-    'completed_task',
+    'new_agent_text_message',
     'new_artifact',
     'new_data_artifact',
-    'are_modalities_compatible',
+    'new_task',
+    'new_text_artifact',
 ]

src/a2a/utils/artifact.py

@@ -1,6 +1,6 @@
 import uuid
 
-from a2a.types import Artifact, Part, TextPart
+from a2a.types import Artifact, DataPart, Part, TextPart
 
 
 def new_artifact(
@@ -13,20 +13,26 @@ def new_artifact(
         description=description,
     )
 
+
 def new_text_artifact(
     name: str,
     text: str,
     description: str = '',
 ) -> Artifact:
     return new_artifact(
-        [Part(root=TextPart(text=text))], name, description,
+        [Part(root=TextPart(text=text))],
+        name,
+        description,
     )
 
+
 def new_data_artifact(
     name: str,
     data: dict[str, any],
     description: str = '',
 ):
     return new_artifact(
-        [Part(root=DataPart(data=data))], name, description,
+        [Part(root=DataPart(data=data))],
+        name,
+        description,
     )

src/a2a/utils/message.py

@@ -22,6 +22,7 @@ def new_agent_text_message(
         contextId=context_id,
     )
 
+
 def new_agent_parts_message(
     parts: list[Part],
     context_id: str | None,
@@ -35,6 +36,7 @@ def new_agent_parts_message(
         contextId=context_id,
     )
 
+
 def get_text_parts(parts: list[Part]) -> list[str]:
     """Return all text parts from a list of parts."""
     return [part.root.text for part in parts if isinstance(part.root, TextPart)]

src/a2a/utils/task.py

@@ -18,8 +18,10 @@ def completed_task(
     task_id: str,
     context_id: str,
     artifacts: list[Artifact],
-    history: list[Message] = [],
+    history: list[Message] = None,
 ) -> Task:
+    if history is None:
+        history = []
     return Task(
         status=TaskStatus(state=TaskState.completed),
         id=task_id,

tests/utils/test_helpers.py

@@ -1,7 +1,5 @@
 from typing import Any
 import pytest
-from unittest.mock import MagicMock
-from uuid import uuid4
 from a2a.utils.helpers import (
     create_task_obj,
     append_artifact_to_task,
@@ -15,11 +13,10 @@
     Task,
     TaskArtifactUpdateEvent,
     TaskState,
-    TaskStatus,
     TextPart,
     Part,
 )
-from a2a.utils.errors import ServerError, UnsupportedOperationError
+from a2a.utils.errors import ServerError
 
 # --- Helper Data ---
 TEXT_PART_DATA: dict[str, Any] = {'type': 'text', 'text': 'Hello'}
@@ -40,6 +37,7 @@
     'type': 'task',
 }
 
+
 # Test create_task_obj
 def test_create_task_obj():
     message = Message(**MINIMAL_MESSAGE_USER)
@@ -55,7 +53,7 @@ def test_create_task_obj():
 
 # Test append_artifact_to_task
 def test_append_artifact_to_task():
-     # Prepare base task   
+    # Prepare base task
     task = Task(**MINIMAL_TASK)
     assert task.id == 'task-abc'
     assert task.contextId == 'session-xyz'
@@ -66,66 +64,86 @@ def test_append_artifact_to_task():
 
     # Prepare appending artifact and event
     artifact_1 = Artifact(
-        artifactId="artifact-123", parts=[Part(root=TextPart(text="Hello"))]
+        artifactId='artifact-123', parts=[Part(root=TextPart(text='Hello'))]
+    )
+    append_event_1 = TaskArtifactUpdateEvent(
+        artifact=artifact_1, append=False, taskId='123', contextId='123'
     )
-    append_event_1 = TaskArtifactUpdateEvent(artifact=artifact_1, append=False, taskId="123", contextId="123")
 
     # Test adding a new artifact (not appending)
     append_artifact_to_task(task, append_event_1)
     assert len(task.artifacts) == 1
-    assert task.artifacts[0].artifactId == "artifact-123"
+    assert task.artifacts[0].artifactId == 'artifact-123'
     assert task.artifacts[0].name == None
     assert len(task.artifacts[0].parts) == 1
-    assert task.artifacts[0].parts[0].root.text == "Hello"
+    assert task.artifacts[0].parts[0].root.text == 'Hello'
 
     # Test replacing the artifact
     artifact_2 = Artifact(
-        artifactId="artifact-123", name="updated name", parts=[Part(root=TextPart(text="Updated"))]
+        artifactId='artifact-123',
+        name='updated name',
+        parts=[Part(root=TextPart(text='Updated'))],
+    )
+    append_event_2 = TaskArtifactUpdateEvent(
+        artifact=artifact_2, append=False, taskId='123', contextId='123'
     )
-    append_event_2 = TaskArtifactUpdateEvent(artifact=artifact_2, append=False, taskId="123", contextId="123")
     append_artifact_to_task(task, append_event_2)
     assert len(task.artifacts) == 1  # Should still have one artifact
-    assert task.artifacts[0].artifactId == "artifact-123"
-    assert task.artifacts[0].name == "updated name"
+    assert task.artifacts[0].artifactId == 'artifact-123'
+    assert task.artifacts[0].name == 'updated name'
     assert len(task.artifacts[0].parts) == 1
-    assert task.artifacts[0].parts[0].root.text == "Updated"
+    assert task.artifacts[0].parts[0].root.text == 'Updated'
 
     # Test appending parts to an existing artifact
     artifact_with_parts = Artifact(
-        artifactId="artifact-123", parts=[Part(root=TextPart(text="Part 2"))]
+        artifactId='artifact-123', parts=[Part(root=TextPart(text='Part 2'))]
+    )
+    append_event_3 = TaskArtifactUpdateEvent(
+        artifact=artifact_with_parts, append=True, taskId='123', contextId='123'
     )
-    append_event_3 = TaskArtifactUpdateEvent(artifact=artifact_with_parts, append=True, taskId="123", contextId="123")
     append_artifact_to_task(task, append_event_3)
     assert len(task.artifacts[0].parts) == 2
-    assert task.artifacts[0].parts[0].root.text == "Updated"
-    assert task.artifacts[0].parts[1].root.text == "Part 2"
+    assert task.artifacts[0].parts[0].root.text == 'Updated'
+    assert task.artifacts[0].parts[1].root.text == 'Part 2'
 
     # Test adding another new artifact
     another_artifact_with_parts = Artifact(
-        artifactId="new_artifact", parts=[Part(root=TextPart(text="new artifact Part 1"))]
+        artifactId='new_artifact',
+        parts=[Part(root=TextPart(text='new artifact Part 1'))],
+    )
+    append_event_4 = TaskArtifactUpdateEvent(
+        artifact=another_artifact_with_parts,
+        append=False,
+        taskId='123',
+        contextId='123',
     )
-    append_event_4 = TaskArtifactUpdateEvent(artifact=another_artifact_with_parts, append=False, taskId="123", contextId="123")
     append_artifact_to_task(task, append_event_4)
     assert len(task.artifacts) == 2
-    assert task.artifacts[0].artifactId == "artifact-123"
-    assert task.artifacts[1].artifactId == "new_artifact"
+    assert task.artifacts[0].artifactId == 'artifact-123'
+    assert task.artifacts[1].artifactId == 'new_artifact'
     assert len(task.artifacts[0].parts) == 2
     assert len(task.artifacts[1].parts) == 1
 
     # Test appending part to a task that does not have a matching artifact
     non_existing_artifact_with_parts = Artifact(
-        artifactId="artifact-456", parts=[Part(root=TextPart(text="Part 1"))]
+        artifactId='artifact-456', parts=[Part(root=TextPart(text='Part 1'))]
+    )
+    append_event_5 = TaskArtifactUpdateEvent(
+        artifact=non_existing_artifact_with_parts,
+        append=True,
+        taskId='123',
+        contextId='123',
     )
-    append_event_5 = TaskArtifactUpdateEvent(artifact=non_existing_artifact_with_parts, append=True, taskId="123", contextId="123")
     append_artifact_to_task(task, append_event_5)
     assert len(task.artifacts) == 2
     assert len(task.artifacts[0].parts) == 2
     assert len(task.artifacts[1].parts) == 1
 
+
 # Test build_text_artifact
 def test_build_text_artifact():
-    artifact_id = "text_artifact"
-    text = "This is a sample text"
+    artifact_id = 'text_artifact'
+    text = 'This is a sample text'
     artifact = build_text_artifact(text, artifact_id)
 
     assert artifact.artifactId == artifact_id
@@ -138,17 +156,17 @@ def test_validate_decorator():
     class TestClass:
         condition = True
 
-        @validate(lambda self: self.condition, "Condition not met")
+        @validate(lambda self: self.condition, 'Condition not met')
         def test_method(self):
-            return "Success"
+            return 'Success'
 
     obj = TestClass()
 
     # Test passing condition
-    assert obj.test_method() == "Success"
+    assert obj.test_method() == 'Success'
 
     # Test failing condition
     obj.condition = False
     with pytest.raises(ServerError) as exc_info:
         obj.test_method()
-    assert "Condition not met" in str(exc_info.value)
+    assert 'Condition not met' in str(exc_info.value)