diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/streaming/streaming_response.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/streaming/streaming_response.py
index 34db4eb24..4466b98b1 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/streaming/streaming_response.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/streaming/streaming_response.py
@@ -47,6 +47,18 @@ def __init__(self, context: "TurnContext"):
             context: Context for the current turn of conversation with the user.
         """
         self._context = context
+        self._initialize_state()
+
+        # Set defaults based on channel
+        self._set_defaults(context)
+
+    def _initialize_state(self) -> None:
+        """
+        Initializes (or resets) all mutable streaming state to its default values.
+        Called from both __init__() and reset().
+        """
+        self._is_streaming_channel = False
+        self._interval = 0.1
         self._sequence_number = 1
         self._stream_id: Optional[str] = None
         self._message = ""
@@ -55,8 +67,6 @@ def __init__(self, context: "TurnContext"):
         self._chunk_queued = False
         self._ended = False
         self._cancelled = False
-        self._is_streaming_channel = False
-        self._interval = 0.1
         self._attachments: Optional[list[Attachment]] = None
         self._citations: list[ClientCitation] = []
         self._sensitivity_label: Optional[SensitivityUsageInfo] = None
@@ -64,9 +74,6 @@ def __init__(self, context: "TurnContext"):
         self._feedback_loop_type: Optional[Literal["default", "custom"]] = None
         self._enable_generated_by_ai_label = False
 
-        # Set defaults based on channel
-        self._set_defaults(context)
-
     def queue_informative_update(self, text: str) -> None:
         """
         Queues an informative update to be sent to the client.
@@ -150,6 +157,37 @@ def set_attachments(self, attachments: list[Attachment]) -> None:
         """
         self._attachments = attachments
 
+    def add_attachment(self, attachment: Attachment) -> None:
+        """
+        Adds an attachment to the collection of attachments for the final message.
+
+        Attachments are only included in the final message sent by `end_stream()`.
+        They are not sent in intermediate typing activities.
+
+        Args:
+            attachment: The attachment to add. Must not be None.
+
+        Raises:
+            ValueError: If attachment is None.
+        """
+        if attachment is None:
+            raise ValueError("attachment cannot be None")
+
+        if self._attachments is None:
+            self._attachments = []
+        self._attachments.append(attachment)
+
+    async def reset(self) -> None:
+        """
+        Resets the streaming response to its initial state.
+        If the stream is still running, this will wait for completion.
+        """
+        await self.wait_for_queue()
+        self._initialize_state()
+
+        # Set defaults based on channel
+        self._set_defaults(self._context)
+
     def set_sensitivity_label(self, sensitivity_label: SensitivityUsageInfo) -> None:
         """
         Sets the sensitivity label to attach to the final chunk.
diff --git a/tests/hosting_core/app/streaming/test_streaming_response.py b/tests/hosting_core/app/streaming/test_streaming_response.py
index 21e3d5b59..ee45d895f 100644
--- a/tests/hosting_core/app/streaming/test_streaming_response.py
+++ b/tests/hosting_core/app/streaming/test_streaming_response.py
@@ -9,6 +9,7 @@
 
 from microsoft_agents.activity import (
     Activity,
+    Attachment,
     ChannelId,
     Channels,
     DeliveryModes,
@@ -519,3 +520,74 @@ async def test_feedback_loop_type_without_enable_does_not_emit_feedback_loop_obj
 
     assert not streaminfo.feedback_loop
     assert not streaminfo.feedback_loop_enabled
+
+
+@pytest.mark.asyncio
+async def test_add_attachment_included_in_final_message(mocker):
+    context = _create_turn_context(
+        mocker,
+        delivery_mode=DeliveryModes.stream,
+        return_value=ResourceResponse(id="stream-att-1"),
+    )
+    response = StreamingResponse(context)
+    attachment = Attachment(content_type="text/plain", name="test.txt", content="hello")
+
+    response.add_attachment(attachment)
+    response.queue_text_chunk("with attachment")
+    await response.end_stream()
+
+    final = context.send_activity.await_args_list[-1].args[0]
+    assert final.attachments is not None
+    assert len(final.attachments) == 1
+    assert final.attachments[0] is attachment
+
+
+def test_add_attachment_raises_on_none(mocker):
+    context = _create_turn_context(mocker, delivery_mode=DeliveryModes.stream)
+    response = StreamingResponse(context)
+
+    with pytest.raises(ValueError, match="attachment cannot be None"):
+        response.add_attachment(None)
+
+
+@pytest.mark.asyncio
+async def test_add_attachment_cleared_on_reset(mocker):
+    context = _create_turn_context(
+        mocker,
+        delivery_mode=DeliveryModes.stream,
+        return_value=[
+            ResourceResponse(id="stream-att-2"),
+            ResourceResponse(id="stream-att-2"),
+        ],
+    )
+    response = StreamingResponse(context)
+
+    response.add_attachment(Attachment(content_type="text/plain", content="data"))
+    response.queue_text_chunk("first stream")
+    await response.end_stream()
+
+    await response.reset()
+
+    response.queue_text_chunk("second stream")
+    await response.end_stream()
+
+    post_reset_final = context.send_activity.await_args_list[-1].args[0]
+    assert not post_reset_final.attachments or len(post_reset_final.attachments) == 0
+
+
+@pytest.mark.asyncio
+async def test_add_attachment_accumulates_multiple(mocker):
+    context = _create_turn_context(
+        mocker,
+        delivery_mode=DeliveryModes.stream,
+        return_value=ResourceResponse(id="stream-att-3"),
+    )
+    response = StreamingResponse(context)
+
+    response.add_attachment(Attachment(content_type="text/plain", content="one"))
+    response.add_attachment(Attachment(content_type="image/png", content="two"))
+    response.queue_text_chunk("multiple attachments")
+    await response.end_stream()
+
+    final = context.send_activity.await_args_list[-1].args[0]
+    assert len(final.attachments) == 2