Add forward_subagent_text option

e2e-tests/test_forward_subagent_text.py

@@ -0,0 +1,98 @@
+"""End-to-end tests for the forward_subagent_text option with real Claude API calls."""
+
+import pytest
+
+from claude_agent_sdk import (
+    AgentDefinition,
+    AssistantMessage,
+    ClaudeAgentOptions,
+    ResultMessage,
+    TextBlock,
+    ToolUseBlock,
+    query,
+)
+
+_AGENT_TOOL_NAMES = {"Agent", "Task"}
+
+
+def _options(forward_subagent_text: bool) -> ClaudeAgentOptions:
+    return ClaudeAgentOptions(
+        forward_subagent_text=forward_subagent_text,
+        agents={
+            "greeter": AgentDefinition(
+                description="Replies with a short greeting. Use for greeting tasks.",
+                prompt="Reply with one short friendly sentence. Do not use any tools.",
+                model="haiku",
+                # Foreground so the run exercises the synchronous subagent
+                # path, where text forwarding is opt-in.
+                background=False,
+            )
+        },
+        allowed_tools=list(_AGENT_TOOL_NAMES),
+        max_turns=4,
+    )
+
+
+_PROMPT = (
+    "Use the Agent tool exactly once with subagent_type 'greeter', prompt "
+    "'say hi', and run_in_background set to false. Then reply with the single "
+    "word DONE."
+)
+
+
+async def _collect(forward_subagent_text: bool):
+    """Run the prompt and return (agent_tool_use_ids, attributed assistant msgs)."""
+    agent_tool_use_ids: set[str] = set()
+    attributed: list[AssistantMessage] = []
+    result: ResultMessage | None = None
+    async for message in query(prompt=_PROMPT, options=_options(forward_subagent_text)):
+        if isinstance(message, AssistantMessage):
+            if message.parent_tool_use_id is None:
+                for block in message.content:
+                    if (
+                        isinstance(block, ToolUseBlock)
+                        and block.name in _AGENT_TOOL_NAMES
+                    ):
+                        agent_tool_use_ids.add(block.id)
+            else:
+                attributed.append(message)
+        elif isinstance(message, ResultMessage):
+            result = message
+    assert result is not None and not result.is_error, f"run failed: {result}"
+    assert agent_tool_use_ids, "the model should have called the Agent tool"
+    return agent_tool_use_ids, attributed
+
+
+@pytest.mark.e2e
+@pytest.mark.anyio
+async def test_forward_subagent_text_delivers_attributed_text():
+    """With forward_subagent_text=True the subagent's text blocks arrive as
+    AssistantMessages whose parent_tool_use_id is the Agent tool_use id."""
+    agent_tool_use_ids, attributed = await _collect(forward_subagent_text=True)
+
+    text_messages = [
+        m for m in attributed if any(isinstance(b, TextBlock) for b in m.content)
+    ]
+    assert text_messages, (
+        "expected at least one subagent text message with parent_tool_use_id set, "
+        f"got attributed={[[type(b).__name__ for b in m.content] for m in attributed]}"
+    )
+    for m in text_messages:
+        assert m.parent_tool_use_id in agent_tool_use_ids, (
+            m.parent_tool_use_id,
+            agent_tool_use_ids,
+        )
+
+
+@pytest.mark.e2e
+@pytest.mark.anyio
+async def test_subagent_text_not_forwarded_by_default():
+    """Without the option, a foreground subagent only surfaces tool_use /
+    tool_result blocks — its text stays out of the parent stream."""
+    _, attributed = await _collect(forward_subagent_text=False)
+
+    leaked = [m for m in attributed if any(isinstance(b, TextBlock) for b in m.content)]
+    assert not leaked, (
+        "subagent text should not be forwarded by default, got: "
+        f"{[[type(b).__name__ for b in m.content] for m in leaked]}"
+    )

src/claude_agent_sdk/_internal/client.py

@@ -183,6 +183,7 @@ async def _process_query_inner(
             agents=agents_dict,
             exclude_dynamic_sections=exclude_dynamic_sections,
             skills=configured_options.skills,
+            forward_subagent_text=configured_options.forward_subagent_text,
         )
 
         if configured_options.session_store is not None:

src/claude_agent_sdk/_internal/query.py

@@ -101,6 +101,7 @@ def __init__(
         agents: dict[str, dict[str, Any]] | None = None,
         exclude_dynamic_sections: bool | None = None,
         skills: list[str] | Literal["all"] | None = None,
+        forward_subagent_text: bool = False,
     ):
         """Initialize Query with transport and callbacks.
 
@@ -116,6 +117,8 @@ def __init__(
                 initialize (see ``SystemPromptPreset``)
             skills: Optional skill allowlist to send via initialize so the CLI
                 can filter which skills are loaded into the system prompt
+            forward_subagent_text: Ask the CLI (via initialize) to forward
+                subagent text/thinking blocks, not just tool_use/tool_result
         """
         self._initialize_timeout = initialize_timeout
         self.transport = transport
@@ -126,6 +129,7 @@ def __init__(
         self._agents = agents
         self._exclude_dynamic_sections = exclude_dynamic_sections
         self._skills = skills
+        self._forward_subagent_text = forward_subagent_text
 
         # Control protocol state
         self.pending_control_responses: dict[str, anyio.Event] = {}
@@ -238,6 +242,8 @@ async def initialize(self) -> dict[str, Any] | None:
         # only send the field when it's an explicit list.
         if isinstance(self._skills, list):
             request["skills"] = self._skills
+        if self._forward_subagent_text:
+            request["forwardSubagentText"] = True
 
         # Use longer timeout for initialize since MCP servers may take time to start
         response = await self._send_control_request(

src/claude_agent_sdk/client.py

@@ -238,6 +238,7 @@ async def _connect_inner(
             agents=agents_dict,
             exclude_dynamic_sections=exclude_dynamic_sections,
             skills=self.options.skills,
+            forward_subagent_text=self.options.forward_subagent_text,
         )
 
         if self.options.session_store is not None:

src/claude_agent_sdk/types.py

@@ -2091,6 +2091,18 @@ class ClaudeAgentOptions:
     TypeScript SDK's ``includeHookEvents``.
     """
 
+    forward_subagent_text: bool = False
+    """Forward subagent text and thinking blocks as messages in the stream.
+
+    By default only ``tool_use`` / ``tool_result`` blocks from subagents
+    (spawned via the Agent tool) are emitted, as ``AssistantMessage`` /
+    ``UserMessage`` objects whose ``parent_tool_use_id`` is the spawning
+    Agent ``tool_use`` id — enough for a progress heartbeat. When true, the
+    subagent's text and thinking blocks are forwarded the same way, so
+    consumers can render the full nested transcript. Matches the TypeScript
+    SDK's ``forwardSubagentText``.
+    """
+
     fork_session: bool = False
     """When true, resumed sessions fork to a new session ID rather than
     continuing the previous session. Use with ``resume``."""
@@ -2306,6 +2318,9 @@ class SDKControlInitializeRequest(TypedDict):
     subtype: Literal["initialize"]
     hooks: dict[HookEvent, Any] | None
     agents: NotRequired[dict[str, dict[str, Any]]]
+    excludeDynamicSections: NotRequired[bool]
+    skills: NotRequired[list[str]]
+    forwardSubagentText: NotRequired[bool]
 
 
 class SDKControlSetPermissionModeRequest(TypedDict):

tests/test_query.py

@@ -77,6 +77,51 @@ def test_initialize_omits_skills_for_none_and_all():
     assert "skills" not in _capture_initialize_request(skills="all")
 
 
+def test_initialize_sends_forward_subagent_text_when_enabled():
+    """forwardSubagentText is sent as an initialize capability, not a CLI flag."""
+    sent = _capture_initialize_request(forward_subagent_text=True)
+    assert sent["forwardSubagentText"] is True
+
+
+def test_initialize_omits_forward_subagent_text_by_default():
+    assert "forwardSubagentText" not in _capture_initialize_request()
+    assert "forwardSubagentText" not in _capture_initialize_request(
+        forward_subagent_text=False
+    )
+
+
+@pytest.mark.parametrize("enabled", [True, False])
+def test_forward_subagent_text_option_reaches_initialize(enabled):
+    """ClaudeAgentOptions.forward_subagent_text is plumbed through query()."""
+
+    async def _test():
+        mock_transport = _make_mock_transport(messages=_ASSISTANT_AND_RESULT)
+        captured: dict = {}
+
+        async def fake_send(self, request, timeout=60.0):
+            if request.get("subtype") == "initialize":
+                captured.update(request)
+            return {}
+
+        with (
+            patch(
+                "claude_agent_sdk._internal.client.SubprocessCLITransport"
+            ) as mock_cls,
+            patch.object(Query, "_send_control_request", fake_send),
+        ):
+            mock_cls.return_value = mock_transport
+            async for _ in query(
+                prompt="Hello",
+                options=ClaudeAgentOptions(forward_subagent_text=enabled),
+            ):
+                pass
+
+        assert captured["subtype"] == "initialize"
+        assert captured.get("forwardSubagentText") == (True if enabled else None)
+
+    anyio.run(_test)
+
+
 def _make_mock_transport(messages, control_requests=None):
     """Create a mock transport that yields messages and optionally sends control requests.
 

tests/test_streaming_client.py

@@ -223,6 +223,35 @@ async def test_connect_with_string_prompt(self):
             assert user_messages[0]["message"]["content"] == "Hello Claude"
             assert user_messages[0]["session_id"] == "default"
 
+    @pytest.mark.anyio
+    async def test_forward_subagent_text_sent_in_initialize(self):
+        """ClaudeAgentOptions.forward_subagent_text is sent as the
+        forwardSubagentText initialize capability; omitted when False."""
+
+        async def initialize_request_for(options: ClaudeAgentOptions) -> dict:
+            with patch(
+                "claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
+            ) as mock_transport_class:
+                mock_transport = create_mock_transport()
+                mock_transport_class.return_value = mock_transport
+                async with ClaudeSDKClient(options=options):
+                    pass
+            requests = [
+                json.loads(call.args[0].strip())
+                for call in mock_transport.write.call_args_list
+                if '"subtype": "initialize"' in call.args[0]
+            ]
+            assert len(requests) == 1
+            return requests[0]["request"]
+
+        enabled = await initialize_request_for(
+            ClaudeAgentOptions(forward_subagent_text=True)
+        )
+        assert enabled["forwardSubagentText"] is True
+
+        default = await initialize_request_for(ClaudeAgentOptions())
+        assert "forwardSubagentText" not in default
+
     @pytest.mark.anyio
     async def test_connect_with_async_iterable(self):
         """Test connecting with an async iterable."""