feat: add stop_task client method

Author: qing-ant

src/claude_agent_sdk/_internal/query.py

@@ -595,6 +595,19 @@ async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None:
             }
         )
 
+    async def stop_task(self, task_id: str) -> None:
+        """Stop a running task.
+
+        Args:
+            task_id: The task ID from task_notification events
+        """
+        await self._send_control_request(
+            {
+                "subtype": "stop_task",
+                "task_id": task_id,
+            }
+        )
+
     async def stream_input(self, stream: AsyncIterable[dict[str, Any]]) -> None:
         """Stream input messages to transport.
 

src/claude_agent_sdk/client.py

@@ -352,6 +352,29 @@ async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None:
             raise CLIConnectionError("Not connected. Call connect() first.")
         await self._query.toggle_mcp_server(server_name, enabled)
 
+    async def stop_task(self, task_id: str) -> None:
+        """Stop a running task (only works with streaming mode).
+
+        After this resolves, a `task_notification` system message with
+        status `'stopped'` will be emitted by the CLI in the message stream.
+
+        Args:
+            task_id: The task ID from `task_notification` events.
+
+        Example:
+            ```python
+            async with ClaudeSDKClient() as client:
+                await client.query("Start a long-running task")
+
+                # Listen for task_notification to get task_id, then:
+                await client.stop_task("task-abc123")
+                # A task_notification with status 'stopped' will follow
+            ```
+        """
+        if not self._query:
+            raise CLIConnectionError("Not connected. Call connect() first.")
+        await self._query.stop_task(task_id)
+
     async def get_mcp_status(self) -> dict[str, Any]:
         """Get current MCP server connection status (only works with streaming mode).
 

src/claude_agent_sdk/types.py

@@ -845,6 +845,11 @@ class SDKControlMcpToggleRequest(TypedDict):
     enabled: bool
 
 
+class SDKControlStopTaskRequest(TypedDict):
+    subtype: Literal["stop_task"]
+    task_id: str
+
+
 class SDKControlRequest(TypedDict):
     type: Literal["control_request"]
     request_id: str
@@ -858,6 +863,7 @@ class SDKControlRequest(TypedDict):
         | SDKControlRewindFilesRequest
         | SDKControlMcpReconnectRequest
         | SDKControlMcpToggleRequest
+        | SDKControlStopTaskRequest
     )
 
 

tests/test_streaming_client.py

@@ -637,6 +637,51 @@ async def _test():
 
         anyio.run(_test)
 
+    def test_stop_task(self):
+        """Test stop_task sends correct control request with task_id."""
+
+        async def _test():
+            with patch(
+                "claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
+            ) as mock_transport_class:
+                mock_transport = _create_mock_transport_with_control_responses()
+                mock_transport_class.return_value = mock_transport
+
+                async with ClaudeSDKClient() as client:
+                    await client.stop_task("task-abc123")
+                    # Check that a control request was sent via write
+                    write_calls = mock_transport.write.call_args_list
+                    request_found = False
+                    for call in write_calls:
+                        data = call[0][0]
+                        try:
+                            msg = json.loads(data.strip())
+                            req = msg.get("request", {})
+                            if (
+                                msg.get("type") == "control_request"
+                                and req.get("subtype") == "stop_task"
+                            ):
+                                assert req.get("task_id") == "task-abc123"
+                                request_found = True
+                                break
+                        except (json.JSONDecodeError, KeyError, AttributeError):
+                            pass
+                    assert request_found, (
+                        "stop_task control request with task_id not found"
+                    )
+
+        anyio.run(_test)
+
+    def test_stop_task_not_connected(self):
+        """Test stop_task when not connected raises error."""
+
+        async def _test():
+            client = ClaudeSDKClient()
+            with pytest.raises(CLIConnectionError, match="Not connected"):
+                await client.stop_task("task-abc123")
+
+        anyio.run(_test)
+
     def test_client_with_options(self):
         """Test client initialization with options."""