diff --git a/src/crewai/task.py b/src/crewai/task.py
index 51e1cf009..aa38cfff9 100644
--- a/src/crewai/task.py
+++ b/src/crewai/task.py
@@ -67,6 +67,7 @@ class Task(BaseModel):
         description: Descriptive text detailing task's purpose and execution.
         expected_output: Clear definition of expected task outcome.
         output_file: File path for storing task output.
+        create_directory: Whether to create the directory for output_file if it doesn't exist.
         output_json: Pydantic model for structuring JSON output.
         output_pydantic: Pydantic model for task output.
         security_config: Security configuration including fingerprinting.
@@ -115,6 +116,10 @@ class Task(BaseModel):
         description="A file path to be used to create a file output.",
         default=None,
     )
+    create_directory: Optional[bool] = Field(
+        description="Whether to create the directory for output_file if it doesn't exist.",
+        default=True,
+    )
     output: Optional[TaskOutput] = Field(
         description="Task output, it's final result after being executed", default=None
     )
@@ -753,8 +758,10 @@ Follow these guidelines:
             resolved_path = Path(self.output_file).expanduser().resolve()
             directory = resolved_path.parent
 
-            if not directory.exists():
+            if self.create_directory and not directory.exists():
                 directory.mkdir(parents=True, exist_ok=True)
+            elif not self.create_directory and not directory.exists():
+                raise RuntimeError(f"Directory {directory} does not exist and create_directory is False")
 
             with resolved_path.open("w", encoding="utf-8") as file:
                 if isinstance(result, dict):
diff --git a/tests/task_test.py b/tests/task_test.py
index 89fa3c53c..40c094cab 100644
--- a/tests/task_test.py
+++ b/tests/task_test.py
@@ -1133,6 +1133,119 @@ def test_output_file_validation():
         )
 
 
+def test_create_directory_true():
+    """Test that directories are created when create_directory=True."""
+    from pathlib import Path
+    
+    output_path = "test_create_dir/output.txt"
+    
+    task = Task(
+        description="Test task",
+        expected_output="Test output",
+        output_file=output_path,
+        create_directory=True,
+    )
+    
+    resolved_path = Path(output_path).expanduser().resolve()
+    resolved_dir = resolved_path.parent
+    
+    if resolved_path.exists():
+        resolved_path.unlink()
+    if resolved_dir.exists():
+        import shutil
+        shutil.rmtree(resolved_dir)
+    
+    assert not resolved_dir.exists()
+    
+    task._save_file("test content")
+    
+    assert resolved_dir.exists()
+    assert resolved_path.exists()
+    
+    if resolved_path.exists():
+        resolved_path.unlink()
+    if resolved_dir.exists():
+        import shutil
+        shutil.rmtree(resolved_dir)
+
+
+def test_create_directory_false():
+    """Test that directories are not created when create_directory=False."""
+    from pathlib import Path
+    
+    output_path = "nonexistent_test_dir/output.txt"
+    
+    task = Task(
+        description="Test task",
+        expected_output="Test output",
+        output_file=output_path,
+        create_directory=False,
+    )
+    
+    resolved_path = Path(output_path).expanduser().resolve()
+    resolved_dir = resolved_path.parent
+    
+    if resolved_dir.exists():
+        import shutil
+        shutil.rmtree(resolved_dir)
+    
+    assert not resolved_dir.exists()
+    
+    with pytest.raises(RuntimeError, match="Directory .* does not exist and create_directory is False"):
+        task._save_file("test content")
+
+
+def test_create_directory_default():
+    """Test that create_directory defaults to True for backward compatibility."""
+    task = Task(
+        description="Test task",
+        expected_output="Test output",
+        output_file="output.txt",
+    )
+    
+    assert task.create_directory is True
+
+
+def test_create_directory_with_existing_directory():
+    """Test that create_directory=False works when directory already exists."""
+    from pathlib import Path
+    
+    output_path = "existing_test_dir/output.txt"
+    
+    resolved_path = Path(output_path).expanduser().resolve()
+    resolved_dir = resolved_path.parent
+    resolved_dir.mkdir(parents=True, exist_ok=True)
+    
+    task = Task(
+        description="Test task",
+        expected_output="Test output",
+        output_file=output_path,
+        create_directory=False,
+    )
+    
+    task._save_file("test content")
+    assert resolved_path.exists()
+    
+    if resolved_path.exists():
+        resolved_path.unlink()
+    if resolved_dir.exists():
+        import shutil
+        shutil.rmtree(resolved_dir)
+
+
+def test_github_issue_3149_reproduction():
+    """Test that reproduces the exact issue from GitHub issue #3149."""
+    task = Task(
+        description="Test task for issue reproduction",
+        expected_output="Test output",
+        output_file="test_output.txt",
+        create_directory=True,
+    )
+    
+    assert task.create_directory is True
+    assert task.output_file == "test_output.txt"
+
+
 @pytest.mark.vcr(filter_headers=["authorization"])
 def test_task_execution_times():
     researcher = Agent(