diff --git a/src/instana/options.py b/src/instana/options.py
index 480f68fa..c3fd28df 100644
--- a/src/instana/options.py
+++ b/src/instana/options.py
@@ -115,6 +115,11 @@ def set_trace_configurations(self) -> None:
                 "trace_correlation", True
             )
 
+        if "INSTANA_ASYNCIO_TASK_CONTEXT_PROPAGATION" in os.environ:
+            config["asyncio_task_context_propagation"]["enabled"] = is_truthy(
+                os.environ["INSTANA_ASYNCIO_TASK_CONTEXT_PROPAGATION"]
+            )
+
         self.set_disable_trace_configurations()
         self.set_stack_trace_configurations()
         self.set_span_filter_configurations()
@@ -397,34 +402,10 @@ def set_tracing(self, tracing: Dict[str, Any]) -> None:
         @return: None
         """
         if "filter" in tracing and not self._has_high_priority_span_filter_source():
-            parsed = parse_filter_rules(tracing["filter"])
-            for policy in ("exclude", "include"):
-                rules = parsed.get(policy, [])
-                if rules:
-                    if policy not in self.span_filters:
-                        self.span_filters[policy] = []
-                    self.span_filters[policy].extend(rules)
+            self._apply_agent_filter_config(tracing["filter"])
 
         if "kafka" in tracing:
-            if (
-                "INSTANA_KAFKA_TRACE_CORRELATION" not in os.environ
-                and not (
-                    isinstance(config.get("tracing"), dict)
-                    and "kafka" in config["tracing"]
-                )
-                and "trace-correlation" in tracing["kafka"]
-            ):
-                self.kafka_trace_correlation = is_truthy(
-                    tracing["kafka"].get("trace-correlation", True)
-                )
-
-            if (
-                "header-format" in tracing["kafka"]
-                and tracing["kafka"]["header-format"] == "binary"
-            ):
-                logger.warning(
-                    "Binary header format for Kafka is deprecated. Please use string header format."
-                )
+            self._apply_agent_kafka_config(tracing["kafka"])
 
         if "extra-http-headers" in tracing:
             self.extra_http_headers = tracing["extra-http-headers"]
@@ -436,6 +417,32 @@ def set_tracing(self, tracing: Dict[str, Any]) -> None:
         # Handle stack trace configuration from agent config
         self.set_stack_trace_from_agent(tracing)
 
+    def _apply_agent_filter_config(self, filter_config: Any) -> None:
+        """Apply span filter rules from agent config."""
+        parsed = parse_filter_rules(filter_config)
+        for policy in ("exclude", "include"):
+            rules = parsed.get(policy, [])
+            if rules:
+                if policy not in self.span_filters:
+                    self.span_filters[policy] = []
+                self.span_filters[policy].extend(rules)
+
+    def _apply_agent_kafka_config(self, kafka_config: Dict[str, Any]) -> None:
+        """Apply Kafka tracing configuration from agent config."""
+        no_env_override = "INSTANA_KAFKA_TRACE_CORRELATION" not in os.environ
+        no_code_override = not (
+            isinstance(config.get("tracing"), dict) and "kafka" in config["tracing"]
+        )
+        if no_env_override and no_code_override and "trace-correlation" in kafka_config:
+            self.kafka_trace_correlation = is_truthy(
+                kafka_config.get("trace-correlation", True)
+            )
+
+        if kafka_config.get("header-format") == "binary":
+            logger.warning(
+                "Binary header format for Kafka is deprecated. Please use string header format."
+            )
+
     def _has_high_priority_span_filter_source(self) -> bool:
         """Return True if a higher-priority span filter source (env var, YAML, or in-code config)
         has already been configured, in which case the agent-provided filter should be ignored."""
diff --git a/tests/test_options.py b/tests/test_options.py
index 3beeecf6..02be426a 100644
--- a/tests/test_options.py
+++ b/tests/test_options.py
@@ -864,6 +864,23 @@ def test_tracing_filter_environment_variables(self) -> None:
             ],
         }
 
+    def test_asyncio_task_context_propagation_default(self) -> None:
+        """INSTANA_ASYNCIO_TASK_CONTEXT_PROPAGATION is False by default."""
+        self.base_options = BaseOptions()
+        assert config["asyncio_task_context_propagation"]["enabled"] is False
+
+    @patch.dict(os.environ, {"INSTANA_ASYNCIO_TASK_CONTEXT_PROPAGATION": "true"})
+    def test_asyncio_task_context_propagation_enabled_via_env(self) -> None:
+        """INSTANA_ASYNCIO_TASK_CONTEXT_PROPAGATION=true enables the flag."""
+        self.base_options = BaseOptions()
+        assert config["asyncio_task_context_propagation"]["enabled"] is True
+
+    @patch.dict(os.environ, {"INSTANA_ASYNCIO_TASK_CONTEXT_PROPAGATION": "false"})
+    def test_asyncio_task_context_propagation_disabled_via_env(self) -> None:
+        """INSTANA_ASYNCIO_TASK_CONTEXT_PROPAGATION=false keeps the flag disabled."""
+        self.base_options = BaseOptions()
+        assert config["asyncio_task_context_propagation"]["enabled"] is False
+
 
 class TestStandardOptions:
     @pytest.fixture(autouse=True)