feat(agent): add stream_final_turn_only parameter to stream_async

strands-py/src/strands/agent/agent.py

@@ -65,7 +65,15 @@
 from ..tools.registry import ToolRegistry
 from ..tools.structured_output._structured_output_context import StructuredOutputContext
 from ..tools.watcher import ToolWatcher
-from ..types._events import AgentResultEvent, EventLoopStopEvent, InitEventLoopEvent, ModelStreamChunkEvent, TypedEvent
+from ..types._events import (
+    AgentResultEvent,
+    EventLoopStopEvent,
+    InitEventLoopEvent,
+    ModelStreamChunkEvent,
+    StartEventLoopEvent,
+    TextStreamEvent,
+    TypedEvent,
+)
 from ..types.agent import AgentInput, ConcurrentInvocationMode, Limits
 from ..types.content import ContentBlock, Message, Messages, SystemContentBlock
 from ..types.exceptions import ConcurrencyException, ContextWindowOverflowException
@@ -792,6 +800,7 @@ async def stream_async(
         invocation_state: dict[str, Any] | None = None,
         structured_output_model: type[BaseModel] | None = None,
         structured_output_prompt: str | None = None,
+        stream_final_turn_only: bool = False,
         limits: Limits | None = None,
         **kwargs: Any,
     ) -> AsyncIterator[Any]:
@@ -812,6 +821,19 @@ async def stream_async(
             invocation_state: Additional parameters to pass through the event loop.
             structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
             structured_output_prompt: Custom prompt for forcing structured output (overrides agent default).
+            stream_final_turn_only: When True, buffers text events from intermediate turns and only yields
+                text events from the final turn. A turn is considered intermediate when it ends with a
+                ``tool_use`` stop reason; any other stop reason (``end_turn``, ``max_tokens``,
+                ``content_filtered``, ``cancelled``, etc.) flushes the buffered text so partial output
+                is not silently lost when the model terminates abnormally on the final turn.
+
+                Note: This setting only filters ``TextStreamEvent`` instances (events with a ``"data"``
+                key). Reasoning events from intermediate turns are still yielded because they are a
+                distinct event type (``ReasoningTextStreamEvent``). Non-text events such as lifecycle,
+                tool use, reasoning, and citation events are yielded normally regardless of this setting.
+
+                When False (default), all events are yielded as they are produced with no change in
+                behavior.
             limits: Per-invocation budget caps (turns / output_tokens / total_tokens).
                 See :class:`~strands.types.agent.Limits`. When a cap is reached, the loop
                 terminates gracefully at the next turn boundary with a corresponding
@@ -835,11 +857,21 @@ async def stream_async(
             Exception: Any exceptions from the agent invocation will be propagated to the caller.
 
         Example:
+            Stream all events (default behavior):
+
             ```python
             async for event in agent.stream_async("Analyze this data"):
                 if "data" in event:
                     yield event["data"]
             ```
+
+            Stream only the final answer (skip intermediate tool-use turns):
+
+            ```python
+            async for event in agent.stream_async("Analyze this data", stream_final_turn_only=True):
+                if "data" in event:
+                    yield event["data"]  # Only receives final turn text
+            ```
         """
         self._validate_limits(limits)
         # Conditionally acquire lock based on concurrent_invocation_mode
@@ -882,9 +914,31 @@ async def stream_async(
                         messages, merged_state, structured_output_model, structured_output_prompt, limits
                     )
 
+                    text_event_buffer: list[dict[str, Any]] = []
+
                     async for event in events:
                         event.prepare(invocation_state=merged_state)
 
+                        if stream_final_turn_only:
+                            if isinstance(event, StartEventLoopEvent):
+                                text_event_buffer.clear()
+                            elif isinstance(event, TextStreamEvent):
+                                text_event_buffer.append(event.as_dict())
+                                continue
+                            elif isinstance(event, EventLoopStopEvent):
+                                stop_reason = event["stop"][0]
+                                # Flush buffered text for any stop reason except tool_use.
+                                # tool_use is the only stop reason that means "this is an
+                                # intermediate turn — more model turns will follow". For all
+                                # other stop reasons (end_turn, max_tokens, content_filtered,
+                                # cancelled, etc.) the buffered text represents the model's
+                                # final output and should be delivered to the caller.
+                                if stop_reason != "tool_use":
+                                    for buffered in text_event_buffer:
+                                        callback_handler(**buffered)
+                                        yield buffered
+                                text_event_buffer.clear()
+
                         if event.is_callback_event:
                             as_dict = event.as_dict()
                             callback_handler(**as_dict)