fix: lint and formatting fixes for A2A lifecycle states

- Fix B904: add 'from None' to re-raised ServerError in cancel()
- Fix F401: remove unused imports (TaskStatus, Part, TextPart, Role) in tests
- Fix I001: sort import blocks in test files
- Fix ruff formatting in executor.py and test_executor.py

Author: agent-of-mkmeral

src/strands/multiagent/a2a/executor.py

@@ -107,9 +107,7 @@ async def execute(
             logger.exception("Agent execution failed, transitioning task to failed state")
             try:
                 await updater.failed(
-                    message=updater.new_agent_message(
-                        parts=[Part(root=TextPart(text=f"Agent execution failed: {e}"))]
-                    )
+                    message=updater.new_agent_message(parts=[Part(root=TextPart(text=f"Agent execution failed: {e}"))])
                 )
             except RuntimeError:
                 # Task already in terminal state (e.g., completed before error in cleanup)
@@ -196,11 +194,7 @@ async def _handle_interrupt_result(self, result: SAAgentResult, updater: TaskUpd
 
         input_message = "Agent requires input:\n" + "\n".join(interrupt_descriptions)
 
-        await updater.requires_input(
-            message=updater.new_agent_message(
-                parts=[Part(root=TextPart(text=input_message))]
-            )
-        )
+        await updater.requires_input(message=updater.new_agent_message(parts=[Part(root=TextPart(text=input_message))]))
 
     async def _handle_streaming_event(self, event: dict[str, Any], updater: TaskUpdater) -> None:
         """Handle a single streaming event from the Strands Agent.
@@ -286,20 +280,18 @@ async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None
         task = context.current_task
         if not task:
             logger.warning("Cancellation requested but no current task found")
-            raise ServerError(error=UnsupportedOperationError())
+            raise ServerError(error=UnsupportedOperationError()) from None
 
         updater = TaskUpdater(event_queue, task.id, task.context_id)
 
         try:
             await updater.cancel(
-                message=updater.new_agent_message(
-                    parts=[Part(root=TextPart(text="Task cancelled by client request"))]
-                )
+                message=updater.new_agent_message(parts=[Part(root=TextPart(text="Task cancelled by client request"))])
             )
         except RuntimeError:
             # Task already in terminal state
             logger.warning("Cannot cancel task %s: already in terminal state", task.id)
-            raise ServerError(error=UnsupportedOperationError())
+            raise ServerError(error=UnsupportedOperationError()) from None
 
     def _get_file_type_from_mime_type(self, mime_type: str | None) -> Literal["document", "image", "video", "unknown"]:
         """Classify file type based on MIME type.

tests/strands/agent/test_a2a_agent.py

@@ -724,7 +724,8 @@ async def mock_send_message(*args, **kwargs):
 def test_is_complete_event_failed_state(a2a_agent):
     """Test that failed state is recognized as complete."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatusUpdateEvent
 
     task = MagicMock()
     status = MagicMock()
@@ -738,7 +739,8 @@ def test_is_complete_event_failed_state(a2a_agent):
 def test_is_complete_event_canceled_state(a2a_agent):
     """Test that canceled state is recognized as complete."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatusUpdateEvent
 
     task = MagicMock()
     status = MagicMock()
@@ -752,7 +754,8 @@ def test_is_complete_event_canceled_state(a2a_agent):
 def test_is_complete_event_rejected_state(a2a_agent):
     """Test that rejected state is recognized as complete."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatusUpdateEvent
 
     task = MagicMock()
     status = MagicMock()
@@ -766,7 +769,8 @@ def test_is_complete_event_rejected_state(a2a_agent):
 def test_is_complete_event_input_required_state(a2a_agent):
     """Test that input_required state is recognized as complete (pausing)."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatusUpdateEvent
 
     task = MagicMock()
     status = MagicMock()
@@ -780,7 +784,8 @@ def test_is_complete_event_input_required_state(a2a_agent):
 def test_is_complete_event_auth_required_state(a2a_agent):
     """Test that auth_required state is recognized as complete (pausing)."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatusUpdateEvent
 
     task = MagicMock()
     status = MagicMock()
@@ -794,7 +799,8 @@ def test_is_complete_event_auth_required_state(a2a_agent):
 def test_is_complete_event_working_state_not_complete(a2a_agent):
     """Test that working state is NOT recognized as complete."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatusUpdateEvent
 
     task = MagicMock()
     status = MagicMock()
@@ -808,7 +814,8 @@ def test_is_complete_event_working_state_not_complete(a2a_agent):
 def test_is_complete_event_submitted_state_not_complete(a2a_agent):
     """Test that submitted state is NOT recognized as complete."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatusUpdateEvent
 
     task = MagicMock()
     status = MagicMock()

tests/strands/multiagent/a2a/test_converters.py

@@ -253,7 +253,8 @@ def test_convert_response_handles_missing_data():
 def test_convert_response_completed_state_maps_to_end_turn():
     """Test that completed state maps to end_turn stop_reason."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus, Part, TextPart
+
+    from a2a.types import TaskState, TaskStatus, TaskStatusUpdateEvent
 
     task = MagicMock()
     task.artifacts = None
@@ -269,7 +270,8 @@ def test_convert_response_completed_state_maps_to_end_turn():
 def test_convert_response_failed_state_maps_to_end_turn():
     """Test that failed state maps to end_turn stop_reason with error content."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus, Message, Part, TextPart, Role
+
+    from a2a.types import Message, TaskState, TaskStatus, TaskStatusUpdateEvent
 
     task = MagicMock()
     task.artifacts = None
@@ -295,7 +297,8 @@ def test_convert_response_failed_state_maps_to_end_turn():
 def test_convert_response_input_required_maps_to_interrupt():
     """Test that input_required state maps to interrupt stop_reason."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus, Message, Part, TextPart, Role
+
+    from a2a.types import Message, TaskState, TaskStatus, TaskStatusUpdateEvent
 
     task = MagicMock()
     task.artifacts = None
@@ -320,7 +323,8 @@ def test_convert_response_input_required_maps_to_interrupt():
 def test_convert_response_canceled_state_maps_to_end_turn():
     """Test that canceled state maps to end_turn stop_reason."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatus, TaskStatusUpdateEvent
 
     task = MagicMock()
     task.artifacts = None
@@ -337,7 +341,8 @@ def test_convert_response_canceled_state_maps_to_end_turn():
 def test_convert_response_rejected_state_maps_to_end_turn():
     """Test that rejected state maps to end_turn stop_reason."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatus, TaskStatusUpdateEvent
 
     task = MagicMock()
     task.artifacts = None
@@ -354,7 +359,8 @@ def test_convert_response_rejected_state_maps_to_end_turn():
 def test_convert_response_auth_required_maps_to_interrupt():
     """Test that auth_required state maps to interrupt stop_reason."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatus, TaskStatusUpdateEvent
 
     task = MagicMock()
     task.artifacts = None
@@ -371,7 +377,9 @@ def test_convert_response_auth_required_maps_to_interrupt():
 def test_extract_task_state_from_status_update():
     """Test _extract_task_state helper."""
     from unittest.mock import MagicMock
-    from a2a.types import TaskState, TaskStatusUpdateEvent, TaskStatus
+
+    from a2a.types import TaskState, TaskStatus, TaskStatusUpdateEvent
+
     from strands.multiagent.a2a._converters import _extract_task_state
 
     task = MagicMock()
@@ -386,7 +394,9 @@ def test_extract_task_state_from_status_update():
 def test_extract_task_state_from_message_returns_none():
     """Test _extract_task_state returns None for Message responses."""
     from unittest.mock import MagicMock
+
     from a2a.types import Message
+
     from strands.multiagent.a2a._converters import _extract_task_state
 
     message = MagicMock(spec=Message)

tests/strands/multiagent/a2a/test_executor.py

@@ -616,10 +616,10 @@ async def test_execute_streaming_mode_handles_agent_exception(
 
     # Verify a failed status event was enqueued
     enqueued_events = [call[0][0] for call in mock_event_queue.enqueue_event.call_args_list]
-    from a2a.types import TaskStatusUpdateEvent, TaskState
+    from a2a.types import TaskState, TaskStatusUpdateEvent
+
     failed_events = [
-        e for e in enqueued_events
-        if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.failed
+        e for e in enqueued_events if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.failed
     ]
     assert len(failed_events) == 1
     assert "Agent execution failed" in failed_events[0].status.message.parts[0].root.text
@@ -633,6 +633,7 @@ async def test_execute_streaming_mode_handles_agent_exception(
     # Verify the error is a ServerError containing an UnsupportedOperationError
     assert isinstance(excinfo.value.error, UnsupportedOperationError)
 
+
 @pytest.mark.asyncio
 async def test_handle_agent_result_with_none_result(mock_strands_agent, mock_request_context, mock_event_queue):
     """Test that _handle_agent_result handles None result correctly."""
@@ -1379,8 +1380,7 @@ async def mock_stream(content_blocks, **kwargs):
     # Verify failed state was enqueued
     enqueued_events = [call[0][0] for call in mock_event_queue.enqueue_event.call_args_list]
     failed_events = [
-        e for e in enqueued_events
-        if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.failed
+        e for e in enqueued_events if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.failed
     ]
     assert len(failed_events) == 1
     assert "Connection lost" in failed_events[0].status.message.parts[0].root.text
@@ -1403,8 +1403,7 @@ async def test_cancel_with_valid_task(mock_strands_agent, mock_request_context,
     # Verify canceled state was enqueued
     enqueued_events = [call[0][0] for call in mock_event_queue.enqueue_event.call_args_list]
     canceled_events = [
-        e for e in enqueued_events
-        if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.canceled
+        e for e in enqueued_events if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.canceled
     ]
     assert len(canceled_events) == 1
     assert "cancelled" in canceled_events[0].status.message.parts[0].root.text.lower()
@@ -1428,6 +1427,7 @@ async def test_execute_with_interrupt_transitions_to_input_required(
 ):
     """Test that agent interrupts map to input_required state."""
     from a2a.types import TaskState, TaskStatusUpdateEvent, TextPart
+
     from strands.interrupt import Interrupt
 
     # Create a mock result with interrupts
@@ -1462,7 +1462,8 @@ async def mock_stream(content_blocks, **kwargs):
     # Verify input_required state was enqueued
     enqueued_events = [call[0][0] for call in mock_event_queue.enqueue_event.call_args_list]
     input_required_events = [
-        e for e in enqueued_events
+        e
+        for e in enqueued_events
         if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.input_required
     ]
     assert len(input_required_events) == 1
@@ -1475,6 +1476,7 @@ async def mock_stream(content_blocks, **kwargs):
 async def test_execute_with_multiple_interrupts(mock_strands_agent, mock_request_context, mock_event_queue):
     """Test handling of multiple interrupts in a single result."""
     from a2a.types import TaskState, TaskStatusUpdateEvent, TextPart
+
     from strands.interrupt import Interrupt
 
     mock_result = MagicMock(spec=SAAgentResult)
@@ -1508,7 +1510,8 @@ async def mock_stream(content_blocks, **kwargs):
 
     enqueued_events = [call[0][0] for call in mock_event_queue.enqueue_event.call_args_list]
     input_required_events = [
-        e for e in enqueued_events
+        e
+        for e in enqueued_events
         if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.input_required
     ]
     assert len(input_required_events) == 1
@@ -1555,14 +1558,14 @@ async def mock_stream(content_blocks, **kwargs):
     # Verify completed state was enqueued (not input_required)
     enqueued_events = [call[0][0] for call in mock_event_queue.enqueue_event.call_args_list]
     completed_events = [
-        e for e in enqueued_events
-        if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.completed
+        e for e in enqueued_events if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.completed
     ]
     assert len(completed_events) == 1
 
     # Verify no input_required events
     input_required_events = [
-        e for e in enqueued_events
+        e
+        for e in enqueued_events
         if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.input_required
     ]
     assert len(input_required_events) == 0