Dedent task docstrings used as doc_md

task-sdk/src/airflow/sdk/bases/decorator.py

@@ -528,8 +528,10 @@ def __call__(self, *args: FParams.args, **kwargs: FParams.kwargs) -> XComArg:
         op.returns_dag_result = self.returns_dag_result
         op_doc_attrs = [op.doc, op.doc_json, op.doc_md, op.doc_rst, op.doc_yaml]
         # Set the task's doc_md to the function's docstring if it exists and no other doc* args are set.
+        # cleandoc strips the leading indentation the docstring inherits from the nested function so it
+        # renders as Markdown rather than an indented code block.
         if self.function.__doc__ and not any(op_doc_attrs):
-            op.doc_md = self.function.__doc__
+            op.doc_md = inspect.cleandoc(self.function.__doc__)
         return XComArg(op)
 
     def _validate_arg_names(self, func: ValidationSource, kwargs: dict[str, Any]):

task-sdk/tests/task_sdk/bases/test_decorator.py

@@ -68,6 +68,18 @@ def test_get_python_source_strips_decorator_and_comment(self, tmp_path: Path):
         ast.parse(cleaned)
         assert cleaned.lstrip().splitlines()[0].startswith("def a_task")
 
+    def test_docstring_used_as_doc_md_is_dedented(self):
+        @task
+        def my_task():
+            """Summary line.
+
+            Body line indented under the function.
+            """
+
+        op = my_task().operator
+
+        assert op.doc_md == "Summary line.\n\nBody line indented under the function."
+
 
 class DummyDecoratedOperator(DecoratedOperator):
     custom_operator_name = "@task.dummy"