diff --git a/python/samples/getting_started_with_agents/multi_agent_orchestration/step2b_sequential_streaming_agent_response_callback.py b/python/samples/getting_started_with_agents/multi_agent_orchestration/step2b_sequential_streaming_agent_response_callback.py
new file mode 100644
index 000000000000..3fc8c4bb5c26
--- /dev/null
+++ b/python/samples/getting_started_with_agents/multi_agent_orchestration/step2b_sequential_streaming_agent_response_callback.py
@@ -0,0 +1,159 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+
+from semantic_kernel.agents import Agent, ChatCompletionAgent, SequentialOrchestration
+from semantic_kernel.agents.runtime import InProcessRuntime
+from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
+
+"""
+The following sample demonstrates how to create a sequential orchestration for
+executing multiple agents in sequence, i.e. the output of one agent is the input
+to the next agent.
+
+This sample demonstrates the basic steps of creating and starting a runtime, creating
+a sequential orchestration, invoking the orchestration, and finally waiting for the
+results.
+"""
+
+
+def get_agents() -> list[Agent]:
+    """Return a list of agents that will participate in the sequential orchestration.
+
+    Feel free to add or remove agents.
+    """
+    concept_extractor_agent = ChatCompletionAgent(
+        name="ConceptExtractorAgent",
+        instructions=(
+            "You are a marketing analyst. Given a product description, identify:\n"
+            "- Key features\n"
+            "- Target audience\n"
+            "- Unique selling points\n\n"
+        ),
+        service=AzureChatCompletion(),
+    )
+    writer_agent = ChatCompletionAgent(
+        name="WriterAgent",
+        instructions=(
+            "You are a marketing copywriter. Given a block of text describing features, audience, and USPs, "
+            "compose a compelling marketing copy (like a newsletter section) that highlights these points. "
+            "Output should be short (around 150 words), output just the copy as a single text block."
+        ),
+        service=AzureChatCompletion(),
+    )
+    format_proof_agent = ChatCompletionAgent(
+        name="FormatProofAgent",
+        instructions=(
+            "You are an editor. Given the draft copy, correct grammar, improve clarity, ensure consistent tone, "
+            "give format and make it polished. Output the final improved copy as a single text block."
+        ),
+        service=AzureChatCompletion(),
+    )
+
+    # The order of the agents in the list will be the order in which they are executed
+    return [concept_extractor_agent, writer_agent, format_proof_agent]
+
+
+# Flag to indicate if a new message is being received
+is_new_message = True
+
+
+def streaming_agent_response_callback(message: StreamingChatMessageContent, is_final: bool) -> None:
+    """Observer function to print the messages from the agents.
+
+    Args:
+        message (StreamingChatMessageContent): The streaming message content from the agent.
+        is_final (bool): Indicates if this is the final part of the message.
+    """
+    global is_new_message
+    if is_new_message:
+        print(f"# {message.name}")
+        is_new_message = False
+    print(message.content, end="", flush=True)
+    if is_final:
+        print()
+        is_new_message = True
+
+
+async def main():
+    """Main function to run the agents."""
+    # 1. Create a sequential orchestration with multiple agents and an agent
+    #    response callback to observe the output from each agent as they stream
+    #    their responses.
+    agents = get_agents()
+    sequential_orchestration = SequentialOrchestration(
+        members=agents,
+        streaming_agent_response_callback=streaming_agent_response_callback,
+    )
+
+    # 2. Create a runtime and start it
+    runtime = InProcessRuntime()
+    runtime.start()
+
+    # 3. Invoke the orchestration with a task and the runtime
+    orchestration_result = await sequential_orchestration.invoke(
+        task="An eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours",
+        runtime=runtime,
+    )
+
+    # 4. Wait for the results
+    value = await orchestration_result.get(timeout=20)
+    print(f"***** Final Result *****\n{value}")
+
+    # 5. Stop the runtime when idle
+    await runtime.stop_when_idle()
+
+    """
+    Sample output:
+    # ConceptExtractorAgent
+    **Key Features:**
+    - Made from eco-friendly stainless steel
+    - Insulation technology that keeps drinks cold for up to 24 hours
+    - Reusable design, promoting sustainability
+    - Possible variations in sizes and colors
+
+    **Target Audience:**
+    - Environmentally conscious consumers
+    - Active individuals and outdoor enthusiasts
+    - Health-conscious individuals looking to stay hydrated
+    - Students and professionals looking for stylish and functional drinkware
+
+    **Unique Selling Points:**
+    - Combines eco-friendliness with high performance in temperature retention
+    - Durable and reusable, reducing reliance on single-use plastics
+    - Sleek design that appeals to modern aesthetics while being functional
+    - Supporting sustainability initiatives through responsible manufacturing practices
+    # WriterAgent
+    Sip sustainably with our eco-friendly stainless steel water bottles, designed for the conscious consumer who values
+    both performance and aesthetics. Our bottles feature advanced insulation technology that keeps your drinks icy cold
+    for up to 24 hours, making them perfect for outdoor adventures, gym sessions, or a busy day at the office. Choose
+    from various sizes and stunning colors to match your personal style while making a positive impact on the planet.
+    Each reusable bottle helps reduce single-use plastics, supporting a cleaner, greener world. Join the movement toward
+    sustainability without compromising on style or functionality. Stay hydrated, look great, and make a difference—get
+    your eco-friendly water bottle today!
+    # FormatProofAgent
+    Sip sustainably with our eco-friendly stainless steel water bottles, designed for the conscious consumer who values
+    both performance and aesthetics. Our bottles utilize advanced insulation technology to keep your beverages icy cold
+    for up to 24 hours, making them perfect for outdoor adventures, gym sessions, or a busy day at the office. 
+
+    Choose from a variety of sizes and stunning colors to match your personal style while positively impacting the
+    planet. Each reusable bottle helps reduce single-use plastics, supporting a cleaner, greener world. 
+
+    Join the movement towards sustainability without compromising on style or functionality. Stay hydrated, look great,
+    and make a difference—get your eco-friendly water bottle today!
+    ***** Final Result *****
+    Sip sustainably with our eco-friendly stainless steel water bottles, designed for the conscious consumer who values
+    both performance and aesthetics. Our bottles utilize advanced insulation technology to keep your beverages icy cold
+    for up to 24 hours, making them perfect for outdoor adventures, gym sessions, or a busy day at the office.
+
+    Choose from a variety of sizes and stunning colors to match your personal style while positively impacting the
+    planet. Each reusable bottle helps reduce single-use plastics, supporting a cleaner, greener world.
+
+    Join the movement towards sustainability without compromising on style or functionality. Stay hydrated, look great,
+    and make a difference—get your eco-friendly water bottle today!
+    """
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/python/semantic_kernel/agents/orchestration/agent_actor_base.py b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
index 6760584eb62a..31f44cf84772 100644
--- a/python/semantic_kernel/agents/orchestration/agent_actor_base.py
+++ b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
@@ -10,7 +10,7 @@
 from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias
 from semantic_kernel.agents.runtime.core.message_context import MessageContext
 from semantic_kernel.agents.runtime.core.routed_agent import RoutedAgent
-from semantic_kernel.contents.chat_history import ChatHistory
+from semantic_kernel.contents import ChatHistory, ChatMessageContent, StreamingChatMessageContent
 from semantic_kernel.utils.feature_stage_decorator import experimental
 
 if sys.version_info >= (3, 12):
@@ -44,18 +44,23 @@ def __init__(
         agent: Agent,
         internal_topic_type: str,
         agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
+        streaming_agent_response_callback: Callable[[StreamingChatMessageContent, bool], Awaitable[None] | None]
+        | None = None,
     ) -> None:
         """Initialize the agent container.
 
         Args:
             agent (Agent): An agent to be run in the container.
             internal_topic_type (str): The topic type of the internal topic.
-            agent_response_callback (Callable | None): A function that is called when a response is produced
+            agent_response_callback (Callable | None): A function that is called when a full response is produced
                 by the agents.
+            streaming_agent_response_callback (Callable | None): A function that is called when a streaming response
+                is produced by the agents.
         """
         self._agent = agent
         self._internal_topic_type = internal_topic_type
         self._agent_response_callback = agent_response_callback
+        self._streaming_agent_response_callback = streaming_agent_response_callback
 
         self._agent_thread: AgentThread | None = None
         # Chat history to temporarily store messages before the agent thread is created
@@ -69,9 +74,78 @@ async def _call_agent_response_callback(self, message: DefaultTypeAlias) -> None
         Args:
             message (DefaultTypeAlias): The message to be sent to the agent_response_callback.
         """
-        # TODO(@taochen): Support streaming
         if self._agent_response_callback:
             if inspect.iscoroutinefunction(self._agent_response_callback):
                 await self._agent_response_callback(message)
             else:
                 self._agent_response_callback(message)
+
+    async def _call_streaming_agent_response_callback(
+        self,
+        message_chunk: StreamingChatMessageContent,
+        is_final: bool,
+    ) -> None:
+        """Call the streaming_agent_response_callback function if it is set.
+
+        Args:
+            message_chunk (StreamingChatMessageContent): The message chunk.
+            is_final (bool): Whether this is the final chunk of the response.
+        """
+        if self._streaming_agent_response_callback:
+            if inspect.iscoroutinefunction(self._streaming_agent_response_callback):
+                await self._streaming_agent_response_callback(message_chunk, is_final)
+            else:
+                self._streaming_agent_response_callback(message_chunk, is_final)
+
+    async def _invoke_agent(self, additional_messages: DefaultTypeAlias | None = None, **kwargs) -> ChatMessageContent:
+        """Invoke the agent with the current chat history or thread and optionally additional messages.
+
+        Args:
+            additional_messages (DefaultTypeAlias | None): Additional messages to be sent to the agent.
+            **kwargs: Additional keyword arguments to be passed to the agent's invoke method:
+                - kernel: The kernel to use for the agent invocation.
+
+        Returns:
+            DefaultTypeAlias: The response from the agent.
+        """
+        streaming_message_buffer: list[StreamingChatMessageContent] = []
+        messages = self._create_messages(additional_messages)
+
+        async for response_item in self._agent.invoke_stream(messages=messages, thread=self._agent_thread, **kwargs):  # type: ignore[arg-type]
+            # Buffer message chunks and stream them with correct is_final flag.
+            streaming_message_buffer.append(response_item.message)
+            if len(streaming_message_buffer) > 1:
+                await self._call_streaming_agent_response_callback(streaming_message_buffer[-2], is_final=False)
+            if self._agent_thread is None:
+                self._agent_thread = response_item.thread
+
+        if streaming_message_buffer:
+            # Call the callback for the last message chunk with is_final=True.
+            await self._call_streaming_agent_response_callback(streaming_message_buffer[-1], is_final=True)
+
+        if not streaming_message_buffer:
+            raise RuntimeError(f'Agent "{self._agent.name}" did not return any response.')
+
+        # Build the full response from the streaming messages
+        full_response = sum(streaming_message_buffer[1:], streaming_message_buffer[0])
+        await self._call_agent_response_callback(full_response)
+
+        return full_response
+
+    def _create_messages(self, additional_messages: DefaultTypeAlias | None = None) -> list[ChatMessageContent]:
+        """Create a list of messages to be sent to the agent along with a potential thread.
+
+        Args:
+            additional_messages (DefaultTypeAlias | None): Additional messages to be sent to the agent.
+
+        Returns:
+            list[ChatMessageContent]: A list of messages to be sent to the agent.
+        """
+        base_messages = self._chat_history.messages[:] if self._agent_thread is None else []
+
+        if additional_messages is None:
+            return base_messages
+
+        if isinstance(additional_messages, list):
+            return base_messages + additional_messages
+        return [*base_messages, additional_messages]
diff --git a/python/semantic_kernel/agents/orchestration/concurrent.py b/python/semantic_kernel/agents/orchestration/concurrent.py
index c46b466872dc..c669114ca7d3 100644
--- a/python/semantic_kernel/agents/orchestration/concurrent.py
+++ b/python/semantic_kernel/agents/orchestration/concurrent.py
@@ -20,6 +20,7 @@
 from semantic_kernel.agents.runtime.core.routed_agent import message_handler
 from semantic_kernel.agents.runtime.core.topic import TopicId
 from semantic_kernel.agents.runtime.in_process.type_subscription import TypeSubscription
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.kernel_pydantic import KernelBaseModel
 from semantic_kernel.utils.feature_stage_decorator import experimental
 
@@ -56,6 +57,8 @@ def __init__(
         internal_topic_type: str,
         collection_agent_type: str,
         agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
+        streaming_agent_response_callback: Callable[[StreamingChatMessageContent, bool], Awaitable[None] | None]
+        | None = None,
     ) -> None:
         """Initialize the agent actor.
 
@@ -63,13 +66,15 @@ def __init__(
             agent: The agent to be executed.
             internal_topic_type: The internal topic type for the actor.
             collection_agent_type: The collection agent type for the actor.
-            agent_response_callback: A callback function to handle the response from the agent.
+            agent_response_callback: A callback function to handle the full response from the agent.
+            streaming_agent_response_callback: A callback function to handle streaming responses from the agent.
         """
         self._collection_agent_type = collection_agent_type
         super().__init__(
             agent=agent,
             internal_topic_type=internal_topic_type,
             agent_response_callback=agent_response_callback,
+            streaming_agent_response_callback=streaming_agent_response_callback,
         )
 
     @message_handler
@@ -77,17 +82,13 @@ async def _handle_message(self, message: ConcurrentRequestMessage, ctx: MessageC
         """Handle a message."""
         logger.debug(f"Concurrent actor (Actor ID: {self.id}; Agent name: {self._agent.name}) started processing...")
 
-        response = await self._agent.get_response(
-            messages=message.body,  # type: ignore[arg-type]
-        )
+        response = await self._invoke_agent(additional_messages=message.body)
 
         logger.debug(f"Concurrent actor (Actor ID: {self.id}; Agent name: {self._agent.name}) finished processing.")
 
-        await self._call_agent_response_callback(response.message)
-
         target_actor_id = await self.runtime.get(self._collection_agent_type)
         await self.send_message(
-            ConcurrentResponseMessage(body=response.message),
+            ConcurrentResponseMessage(body=response),
             target_actor_id,
             cancellation_token=ctx.cancellation_token,
         )
@@ -181,6 +182,7 @@ async def _internal_helper(agent: Agent) -> None:
                     internal_topic_type,
                     collection_agent_type=self._get_collection_actor_type(internal_topic_type),
                     agent_response_callback=self._agent_response_callback,
+                    streaming_agent_response_callback=self._streaming_agent_response_callback,
                 ),
             )
 
diff --git a/python/semantic_kernel/agents/orchestration/group_chat.py b/python/semantic_kernel/agents/orchestration/group_chat.py
index f197809004c2..a070f380810b 100644
--- a/python/semantic_kernel/agents/orchestration/group_chat.py
+++ b/python/semantic_kernel/agents/orchestration/group_chat.py
@@ -19,6 +19,7 @@
 from semantic_kernel.agents.runtime.in_process.type_subscription import TypeSubscription
 from semantic_kernel.contents.chat_history import ChatHistory
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
 from semantic_kernel.kernel_pydantic import KernelBaseModel
 from semantic_kernel.utils.feature_stage_decorator import experimental
@@ -148,31 +149,17 @@ async def _handle_request_message(self, message: GroupChatRequestMessage, ctx: M
             return
 
         logger.debug(f"{self.id}: Received group chat request message.")
-        if self._agent_thread is None:
-            # Add a user message to steer the agent to respond more closely to the instructions.
-            self._chat_history.add_message(
-                ChatMessageContent(
-                    role=AuthorRole.USER,
-                    content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
-                )
-            )
-            response_item = await self._agent.get_response(
-                messages=self._chat_history.messages,  # type: ignore[arg-type]
-            )
-            self._agent_thread = response_item.thread
-        else:
-            # Add a user message to steer the agent to respond more closely to the instructions.
-            new_message = ChatMessageContent(
-                role=AuthorRole.USER,
-                content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
-            )
-            response_item = await self._agent.get_response(messages=new_message, thread=self._agent_thread)
 
-        logger.debug(f"{self.id} responded with {response_item.message.content}.")
-        await self._call_agent_response_callback(response_item.message)
+        persona_adoption_message = ChatMessageContent(
+            role=AuthorRole.USER,
+            content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
+        )
+        response = await self._invoke_agent(additional_messages=persona_adoption_message)
+
+        logger.debug(f"{self.id} responded with {response}.")
 
         await self.publish_message(
-            GroupChatResponseMessage(body=response_item.message),
+            GroupChatResponseMessage(body=response),
             TopicId(self._internal_topic_type, self.id.key),
             cancellation_token=ctx.cancellation_token,
         )
@@ -415,6 +402,8 @@ def __init__(
         input_transform: Callable[[TIn], Awaitable[DefaultTypeAlias] | DefaultTypeAlias] | None = None,
         output_transform: Callable[[DefaultTypeAlias], Awaitable[TOut] | TOut] | None = None,
         agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
+        streaming_agent_response_callback: Callable[[StreamingChatMessageContent, bool], Awaitable[None] | None]
+        | None = None,
     ) -> None:
         """Initialize the group chat orchestration.
 
@@ -426,8 +415,10 @@ def __init__(
             description (str | None): The description of the orchestration.
             input_transform (Callable | None): A function that transforms the external input message.
             output_transform (Callable | None): A function that transforms the internal output message.
-            agent_response_callback (Callable | None): A function that is called when a response is produced
+            agent_response_callback (Callable | None): A function that is called when a full response is produced
                 by the agents.
+            streaming_agent_response_callback (Callable | None): A function that is called when a streaming response
+                is produced by the agents.
         """
         self._manager = manager
 
@@ -442,6 +433,7 @@ def __init__(
             input_transform=input_transform,
             output_transform=output_transform,
             agent_response_callback=agent_response_callback,
+            streaming_agent_response_callback=streaming_agent_response_callback,
         )
 
     @override
@@ -497,7 +489,12 @@ async def _register_members(self, runtime: CoreRuntime, internal_topic_type: str
             GroupChatAgentActor.register(
                 runtime,
                 self._get_agent_actor_type(agent, internal_topic_type),
-                lambda agent=agent: GroupChatAgentActor(agent, internal_topic_type, self._agent_response_callback),  # type: ignore[misc]
+                lambda agent=agent: GroupChatAgentActor(  # type: ignore[misc]
+                    agent,
+                    internal_topic_type,
+                    agent_response_callback=self._agent_response_callback,
+                    streaming_agent_response_callback=self._streaming_agent_response_callback,
+                ),
             )
             for agent in self._members
         ])
diff --git a/python/semantic_kernel/agents/orchestration/handoffs.py b/python/semantic_kernel/agents/orchestration/handoffs.py
index 839e1692d7e0..9cc58637f082 100644
--- a/python/semantic_kernel/agents/orchestration/handoffs.py
+++ b/python/semantic_kernel/agents/orchestration/handoffs.py
@@ -18,6 +18,7 @@
 from semantic_kernel.agents.runtime.core.topic import TopicId
 from semantic_kernel.agents.runtime.in_process.type_subscription import TypeSubscription
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
 from semantic_kernel.filters.auto_function_invocation.auto_function_invocation_context import (
     AutoFunctionInvocationContext,
@@ -162,6 +163,8 @@ def __init__(
         handoff_connections: AgentHandoffs,
         result_callback: Callable[[DefaultTypeAlias], Awaitable[None]] | None = None,
         agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
+        streaming_agent_response_callback: Callable[[StreamingChatMessageContent, bool], Awaitable[None] | None]
+        | None = None,
         human_response_function: Callable[[], Awaitable[ChatMessageContent] | ChatMessageContent] | None = None,
     ) -> None:
         """Initialize the handoff agent actor."""
@@ -179,6 +182,7 @@ def __init__(
             agent=agent,
             internal_topic_type=internal_topic_type,
             agent_response_callback=agent_response_callback,
+            streaming_agent_response_callback=streaming_agent_response_callback,
         )
 
     def _add_handoff_functions(self) -> None:
@@ -275,45 +279,16 @@ async def _handle_request_message(self, message: HandoffRequestMessage, cts: Mes
             return
         logger.debug(f"{self.id}: Received handoff request message.")
 
-        if self._agent_thread is None:
-            self._chat_history.add_message(
-                ChatMessageContent(
-                    role=AuthorRole.USER,
-                    content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
-                )
-            )
-            response_item = await self._agent.get_response(
-                messages=self._chat_history.messages,  # type: ignore[arg-type]
-                kernel=self._kernel,
-            )
-        else:
-            response_item = await self._agent.get_response(
-                messages=ChatMessageContent(
-                    role=AuthorRole.USER,
-                    content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
-                ),
-                thread=self._agent_thread,
-                kernel=self._kernel,
-            )
-
-        if self._agent_thread is None:
-            self._agent_thread = response_item.thread
+        persona_adoption_message = ChatMessageContent(
+            role=AuthorRole.USER,
+            content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
+        )
+        response = await self._invoke_agent_with_potentially_no_response(
+            additional_messages=persona_adoption_message,
+            kernel=self._kernel,
+        )
 
         while not self._task_completed:
-            if response_item.message.role == AuthorRole.ASSISTANT:
-                # The response can potentially be a TOOL message from the Handoff plugin
-                # since we have added a filter which will terminate the conversation when
-                # a function from the handoff plugin is called. And we don't want to publish
-                # that message. So we only publish if the response is an ASSISTANT message.
-                logger.debug(f"{self.id} responded with: {response_item.message.content}")
-                await self._call_agent_response_callback(response_item.message)
-
-                await self.publish_message(
-                    HandoffResponseMessage(body=response_item.message),
-                    TopicId(self._internal_topic_type, self.id.key),
-                    cancellation_token=cts.cancellation_token,
-                )
-
             if self._handoff_agent_name:
                 await self.publish_message(
                     HandoffRequestMessage(agent_name=self._handoff_agent_name),
@@ -321,6 +296,18 @@ async def _handle_request_message(self, message: HandoffRequestMessage, cts: Mes
                 )
                 self._handoff_agent_name = None
                 break
+
+            if response is None:
+                raise RuntimeError(
+                    f'Agent "{self._agent.name}" did not return any response nor did not set a handoff agent name.'
+                )
+
+            await self.publish_message(
+                HandoffResponseMessage(body=response),
+                TopicId(self._internal_topic_type, self.id.key),
+                cancellation_token=cts.cancellation_token,
+            )
+
             if self._human_response_function:
                 human_response = await self._call_human_response_function()
                 await self.publish_message(
@@ -328,9 +315,8 @@ async def _handle_request_message(self, message: HandoffRequestMessage, cts: Mes
                     TopicId(self._internal_topic_type, self.id.key),
                     cancellation_token=cts.cancellation_token,
                 )
-                response_item = await self._agent.get_response(
-                    messages=human_response,
-                    thread=self._agent_thread,
+                response = await self._invoke_agent_with_potentially_no_response(
+                    additional_messages=human_response,
                     kernel=self._kernel,
                 )
             else:
@@ -346,6 +332,43 @@ async def _call_human_response_function(self) -> ChatMessageContent:
             return await self._human_response_function()
         return self._human_response_function()  # type: ignore[return-value]
 
+    async def _invoke_agent_with_potentially_no_response(
+        self, additional_messages: DefaultTypeAlias | None = None, **kwargs
+    ) -> ChatMessageContent | None:
+        """Invoke the agent with the current chat history or thread and optionally additional messages.
+
+        This method differs from `_invoke_agent` in that it handles the case where no response is returned
+        from the agent gracefully, returning `None` instead of raising an error.
+
+        The reason for this is that agents in a handoff group chat may not always produce a response when
+        a handoff function is invoked, where the `_handoff_function_filter` will terminate the auto function
+        invocation loop before a response is produced. In such cases, this method will return `None`
+        instead of raising an error.
+        """
+        streaming_message_buffer: list[StreamingChatMessageContent] = []
+        messages = self._create_messages(additional_messages)
+
+        async for response_item in self._agent.invoke_stream(messages=messages, thread=self._agent_thread, **kwargs):  # type: ignore[arg-type]
+            # Buffer message chunks and stream them with correct is_final flag.
+            streaming_message_buffer.append(response_item.message)
+            if len(streaming_message_buffer) > 1:
+                await self._call_streaming_agent_response_callback(streaming_message_buffer[-2], is_final=False)
+            if self._agent_thread is None:
+                self._agent_thread = response_item.thread
+
+        if streaming_message_buffer:
+            # Call the callback for the last message chunk with is_final=True.
+            await self._call_streaming_agent_response_callback(streaming_message_buffer[-1], is_final=True)
+
+        if not streaming_message_buffer:
+            return None
+
+        # Build the full response from the streaming messages
+        full_response = sum(streaming_message_buffer[1:], streaming_message_buffer[0])
+        await self._call_agent_response_callback(full_response)
+
+        return full_response
+
 
 # endregion HandoffAgentActor
 
@@ -365,6 +388,8 @@ def __init__(
         input_transform: Callable[[TIn], Awaitable[DefaultTypeAlias] | DefaultTypeAlias] | None = None,
         output_transform: Callable[[DefaultTypeAlias], Awaitable[TOut] | TOut] | None = None,
         agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
+        streaming_agent_response_callback: Callable[[StreamingChatMessageContent, bool], Awaitable[None] | None]
+        | None = None,
         human_response_function: Callable[[], Awaitable[ChatMessageContent] | ChatMessageContent] | None = None,
     ) -> None:
         """Initialize the handoff orchestration.
@@ -377,8 +402,10 @@ def __init__(
             description (str | None): The description of the orchestration.
             input_transform (Callable | None): A function that transforms the external input message.
             output_transform (Callable | None): A function that transforms the internal output message.
-            agent_response_callback (Callable | None): A function that is called when a response is produced
+            agent_response_callback (Callable | None): A function that is called when a full response is produced
                 by the agents.
+            streaming_agent_response_callback (Callable | None): A function that is called when a streaming response
+                is produced by the agents.
             human_response_function (Callable | None): A function that is called when a human response is
                 needed.
         """
@@ -392,6 +419,7 @@ def __init__(
             input_transform=input_transform,
             output_transform=output_transform,
             agent_response_callback=agent_response_callback,
+            streaming_agent_response_callback=streaming_agent_response_callback,
         )
 
         self._validate_handoffs()
@@ -461,6 +489,7 @@ async def _register_helper(agent: Agent) -> None:
                     handoff_connections,
                     result_callback=result_callback,
                     agent_response_callback=self._agent_response_callback,
+                    streaming_agent_response_callback=self._streaming_agent_response_callback,
                     human_response_function=self._human_response_function,
                 ),
             )
diff --git a/python/semantic_kernel/agents/orchestration/magentic.py b/python/semantic_kernel/agents/orchestration/magentic.py
index 9970ae3506d7..1ced5aad9dbd 100644
--- a/python/semantic_kernel/agents/orchestration/magentic.py
+++ b/python/semantic_kernel/agents/orchestration/magentic.py
@@ -31,6 +31,7 @@
 from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
 from semantic_kernel.contents.chat_history import ChatHistory
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
 from semantic_kernel.functions.kernel_arguments import KernelArguments
 from semantic_kernel.kernel import Kernel
@@ -701,29 +702,17 @@ async def _handle_request_message(self, message: MagenticRequestMessage, ctx: Me
             return
 
         logger.debug(f"{self.id}: Received request message.")
-        if self._agent_thread is None:
-            # Add a user message to steer the agent to respond more closely to the instructions.
-            self._chat_history.add_message(
-                ChatMessageContent(
-                    role=AuthorRole.USER,
-                    content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
-                )
-            )
-            response_item = await self._agent.get_response(messages=self._chat_history.messages)  # type: ignore[arg-type]
-            self._agent_thread = response_item.thread
-        else:
-            # Add a user message to steer the agent to respond more closely to the instructions.
-            new_message = ChatMessageContent(
-                role=AuthorRole.USER,
-                content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
-            )
-            response_item = await self._agent.get_response(messages=new_message, thread=self._agent_thread)
 
-        logger.debug(f"{self.id} responded with {response_item.message.content}.")
-        await self._call_agent_response_callback(response_item.message)
+        persona_adoption_message = ChatMessageContent(
+            role=AuthorRole.USER,
+            content=f"Transferred to {self._agent.name}, adopt the persona immediately.",
+        )
+        response = await self._invoke_agent(additional_messages=persona_adoption_message)
+
+        logger.debug(f"{self.id} responded with {response}.")
 
         await self.publish_message(
-            MagenticResponseMessage(body=response_item.message),
+            MagenticResponseMessage(body=response),
             TopicId(self._internal_topic_type, self.id.key),
             cancellation_token=ctx.cancellation_token,
         )
@@ -756,6 +745,8 @@ def __init__(
         input_transform: Callable[[TIn], Awaitable[DefaultTypeAlias] | DefaultTypeAlias] | None = None,
         output_transform: Callable[[DefaultTypeAlias], Awaitable[TOut] | TOut] | None = None,
         agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
+        streaming_agent_response_callback: Callable[[StreamingChatMessageContent, bool], Awaitable[None] | None]
+        | None = None,
     ) -> None:
         """Initialize the Magentic One orchestration.
 
@@ -768,6 +759,8 @@ def __init__(
             output_transform (Callable | None): A function that transforms the internal output message.
             agent_response_callback (Callable | None): A function that is called when a response is produced
                 by the agents.
+            streaming_agent_response_callback (Callable | None): A function that is called when a streaming response
+                is produced by the agents.
         """
         self._manager = manager
 
@@ -782,6 +775,7 @@ def __init__(
             input_transform=input_transform,
             output_transform=output_transform,
             agent_response_callback=agent_response_callback,
+            streaming_agent_response_callback=streaming_agent_response_callback,
         )
 
     @override
@@ -826,6 +820,7 @@ async def _register_members(self, runtime: CoreRuntime, internal_topic_type: str
                     agent,
                     internal_topic_type,
                     self._agent_response_callback,
+                    self._streaming_agent_response_callback,
                 ),
             )
             for agent in self._members
diff --git a/python/semantic_kernel/agents/orchestration/orchestration_base.py b/python/semantic_kernel/agents/orchestration/orchestration_base.py
index abdc3833b320..ed547828ba0f 100644
--- a/python/semantic_kernel/agents/orchestration/orchestration_base.py
+++ b/python/semantic_kernel/agents/orchestration/orchestration_base.py
@@ -16,6 +16,7 @@
 from semantic_kernel.agents.runtime.core.cancellation_token import CancellationToken
 from semantic_kernel.agents.runtime.core.core_runtime import CoreRuntime
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
 from semantic_kernel.kernel_pydantic import KernelBaseModel
 from semantic_kernel.utils.feature_stage_decorator import experimental
@@ -94,6 +95,8 @@ def __init__(
         input_transform: Callable[[TIn], Awaitable[DefaultTypeAlias] | DefaultTypeAlias] | None = None,
         output_transform: Callable[[DefaultTypeAlias], Awaitable[TOut] | TOut] | None = None,
         agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
+        streaming_agent_response_callback: Callable[[StreamingChatMessageContent, bool], Awaitable[None] | None]
+        | None = None,
     ) -> None:
         """Initialize the orchestration base.
 
@@ -103,8 +106,10 @@ def __init__(
             description (str | None): The description of the orchestration. If None, use a default description.
             input_transform (Callable | None): A function that transforms the external input message.
             output_transform (Callable | None): A function that transforms the internal output message.
-            agent_response_callback (Callable | None): A function that is called when a response is produced
+            agent_response_callback (Callable | None): A function that is called when a full response is produced
                 by the agents.
+            streaming_agent_response_callback (Callable | None): A function that is called when a streaming response
+                is produced by the agents.
         """
         if not members:
             raise ValueError("The members list cannot be empty.")
@@ -117,6 +122,7 @@ def __init__(
         self._output_transform = output_transform or self._default_output_transform
 
         self._agent_response_callback = agent_response_callback
+        self._streaming_agent_response_callback = streaming_agent_response_callback
 
     def _set_types(self) -> None:
         """Set the external input and output types from the class arguments.
diff --git a/python/semantic_kernel/agents/orchestration/sequential.py b/python/semantic_kernel/agents/orchestration/sequential.py
index 07095460f284..d04369d88c02 100644
--- a/python/semantic_kernel/agents/orchestration/sequential.py
+++ b/python/semantic_kernel/agents/orchestration/sequential.py
@@ -12,6 +12,7 @@
 from semantic_kernel.agents.runtime.core.message_context import MessageContext
 from semantic_kernel.agents.runtime.core.routed_agent import message_handler
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.kernel_pydantic import KernelBaseModel
 from semantic_kernel.utils.feature_stage_decorator import experimental
 
@@ -48,6 +49,8 @@ def __init__(
         internal_topic_type: str,
         next_agent_type: str,
         agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
+        streaming_agent_response_callback: Callable[[StreamingChatMessageContent, bool], Awaitable[None] | None]
+        | None = None,
     ) -> None:
         """Initialize the agent actor."""
         self._next_agent_type = next_agent_type
@@ -55,6 +58,7 @@ def __init__(
             agent=agent,
             internal_topic_type=internal_topic_type,
             agent_response_callback=agent_response_callback,
+            streaming_agent_response_callback=streaming_agent_response_callback,
         )
 
     @message_handler
@@ -62,15 +66,13 @@ async def _handle_message(self, message: SequentialRequestMessage, ctx: MessageC
         """Handle a message."""
         logger.debug(f"Sequential actor (Actor ID: {self.id}; Agent name: {self._agent.name}) started processing...")
 
-        response = await self._agent.get_response(messages=message.body)  # type: ignore[arg-type]
+        response = await self._invoke_agent(additional_messages=message.body)
 
         logger.debug(f"Sequential actor (Actor ID: {self.id}; Agent name: {self._agent.name}) finished processing.")
 
-        await self._call_agent_response_callback(response.message)
-
         target_actor_id = await self.runtime.get(self._next_agent_type)
         await self.send_message(
-            SequentialRequestMessage(body=response.message),
+            SequentialRequestMessage(body=response),
             target_actor_id,
             cancellation_token=ctx.cancellation_token,
         )
@@ -155,6 +157,7 @@ async def _register_members(
                     internal_topic_type,
                     next_agent_type=next_actor_type,
                     agent_response_callback=self._agent_response_callback,
+                    streaming_agent_response_callback=self._streaming_agent_response_callback,
                 ),
             )
             logger.debug(f"Registered agent actor of type {self._get_agent_actor_type(agent, internal_topic_type)}")
diff --git a/python/tests/unit/agents/orchestration/conftest.py b/python/tests/unit/agents/orchestration/conftest.py
index e5943dede427..43597e0f84b3 100644
--- a/python/tests/unit/agents/orchestration/conftest.py
+++ b/python/tests/unit/agents/orchestration/conftest.py
@@ -43,15 +43,7 @@ async def get_response(
         thread: AgentThread | None = None,
         **kwargs,
     ) -> AgentResponseItem[ChatMessageContent]:
-        # Simulate some processing time
-        await asyncio.sleep(0.1)
-        return AgentResponseItem[ChatMessageContent](
-            message=ChatMessageContent(
-                role=AuthorRole.ASSISTANT,
-                content="mock_response",
-            ),
-            thread=thread or MockAgentThread(),
-        )
+        pass
 
     @override
     async def invoke(
@@ -73,7 +65,29 @@ async def invoke_stream(
         on_intermediate_message: Callable[[ChatMessageContent], Awaitable[None]] | None = None,
         **kwargs,
     ) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]:
-        pass
+        """Simulate streaming response from the agent."""
+        # Simulate some processing time
+        await asyncio.sleep(0.05)
+        yield AgentResponseItem[StreamingChatMessageContent](
+            message=StreamingChatMessageContent(
+                role=AuthorRole.ASSISTANT,
+                name=self.name,
+                content="mock",
+                choice_index=0,
+            ),
+            thread=thread or MockAgentThread(),
+        )
+        # Simulate some processing time before sending the next part of the response
+        await asyncio.sleep(0.05)
+        yield AgentResponseItem[StreamingChatMessageContent](
+            message=StreamingChatMessageContent(
+                role=AuthorRole.ASSISTANT,
+                name=self.name,
+                content="_response",
+                choice_index=0,
+            ),
+            thread=thread or MockAgentThread(),
+        )
 
 
 class MockRuntime(CoreRuntime):
diff --git a/python/tests/unit/agents/orchestration/test_concurrent.py b/python/tests/unit/agents/orchestration/test_concurrent.py
index d88e7e221df9..17f3f9a9391d 100644
--- a/python/tests/unit/agents/orchestration/test_concurrent.py
+++ b/python/tests/unit/agents/orchestration/test_concurrent.py
@@ -10,6 +10,7 @@
 from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias, OrchestrationResult
 from semantic_kernel.agents.runtime.in_process.in_process_runtime import InProcessRuntime
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from tests.unit.agents.orchestration.conftest import MockAgent, MockRuntime
 
 
@@ -80,6 +81,37 @@ async def test_invoke_with_response_callback():
         await runtime.stop_when_idle()
 
 
+async def test_invoke_with_streaming_response_callback():
+    """Test the invoke method of the ConcurrentOrchestration with a streaming response callback."""
+    agent_a = MockAgent()
+    agent_b = MockAgent()
+
+    runtime = InProcessRuntime()
+    runtime.start()
+
+    responses: dict[str, list[StreamingChatMessageContent]] = {}
+    try:
+        orchestration = ConcurrentOrchestration(
+            members=[agent_a, agent_b],
+            streaming_agent_response_callback=lambda x, _: responses.setdefault(x.name, []).append(x),
+        )
+        orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
+        await orchestration_result.get(1.0)
+
+        assert len(responses[agent_a.name]) == 2
+        assert len(responses[agent_b.name]) == 2
+
+        assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_a.name])
+        assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_b.name])
+
+        agent_a_response = sum(responses[agent_a.name][1:], responses[agent_a.name][0])
+        assert agent_a_response.content == "mock_response"
+        agent_b_response = sum(responses[agent_b.name][1:], responses[agent_b.name][0])
+        assert agent_b_response.content == "mock_response"
+    finally:
+        await runtime.stop_when_idle()
+
+
 @pytest.mark.skipif(
     sys.version_info < (3, 11),
     reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.",
@@ -87,7 +119,7 @@ async def test_invoke_with_response_callback():
 async def test_invoke_cancel_before_completion():
     """Test the invoke method of the ConcurrentOrchestration with cancellation before completion."""
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
     ):
         agent_a = MockAgent()
         agent_b = MockAgent()
@@ -105,7 +137,7 @@ async def test_invoke_cancel_before_completion():
         finally:
             await runtime.stop_when_idle()
 
-        assert mock_get_response.call_count == 2
+        assert mock_invoke_stream.call_count == 2
 
 
 async def test_invoke_cancel_after_completion():
diff --git a/python/tests/unit/agents/orchestration/test_group_chat.py b/python/tests/unit/agents/orchestration/test_group_chat.py
index 18270ff61453..c256bc519138 100644
--- a/python/tests/unit/agents/orchestration/test_group_chat.py
+++ b/python/tests/unit/agents/orchestration/test_group_chat.py
@@ -6,14 +6,35 @@
 
 import pytest
 
-from semantic_kernel.agents.orchestration.group_chat import GroupChatOrchestration, RoundRobinGroupChatManager
+from semantic_kernel.agents.orchestration.group_chat import (
+    BooleanResult,
+    GroupChatOrchestration,
+    RoundRobinGroupChatManager,
+)
 from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias, OrchestrationResult
 from semantic_kernel.agents.runtime.in_process.in_process_runtime import InProcessRuntime
 from semantic_kernel.contents.chat_history import ChatHistory
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
 from tests.unit.agents.orchestration.conftest import MockAgent, MockRuntime
 
+if sys.version_info >= (3, 12):
+    from typing import override  # pragma: no cover
+else:
+    from typing_extensions import override  # pragma: no cover
+
+
+class RoundRobinGroupChatManagerWithUserInput(RoundRobinGroupChatManager):
+    @override
+    async def should_request_user_input(self, chat_history: ChatHistory) -> BooleanResult:
+        """Check if the group chat should request user input."""
+        return BooleanResult(
+            result=True,
+            reason="Allow user input for testing purposes.",
+        )
+
+
 # region GroupChatOrchestration
 
 
@@ -55,7 +76,7 @@ async def test_prepare():
 async def test_invoke():
     """Test the invoke method of the GroupChatOrchestration."""
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
     ):
         agent_a = MockAgent(description="test agent")
         agent_b = MockAgent(description="test agent")
@@ -78,7 +99,7 @@ async def test_invoke():
         assert result.role == AuthorRole.ASSISTANT
         assert result.content == "mock_response"
 
-        assert mock_get_response.call_count == 3
+        assert mock_invoke_stream.call_count == 3
 
 
 @pytest.mark.skipif(
@@ -88,7 +109,7 @@ async def test_invoke():
 async def test_invoke_with_list():
     """Test the invoke method of the GroupChatOrchestration with a list of messages."""
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
     ):
         agent_a = MockAgent(description="test agent")
         agent_b = MockAgent(description="test agent")
@@ -111,11 +132,11 @@ async def test_invoke_with_list():
         finally:
             await runtime.stop_when_idle()
 
-        assert mock_get_response.call_count == 2
+        assert mock_invoke_stream.call_count == 2
         # Two messages + one message added internally to steer the conversation
-        assert len(mock_get_response.call_args_list[0][1]["messages"]) == 3
+        assert len(mock_invoke_stream.call_args_list[0][1]["messages"]) == 3
         # Two messages + two message added internally to steer the conversation + response from agent A
-        assert len(mock_get_response.call_args_list[1][1]["messages"]) == 5
+        assert len(mock_invoke_stream.call_args_list[1][1]["messages"]) == 5
 
 
 async def test_invoke_with_response_callback():
@@ -143,6 +164,75 @@ async def test_invoke_with_response_callback():
     assert all(item.content == "mock_response" for item in responses)
 
 
+async def test_invoke_with_streaming_response_callback():
+    """Test the invoke method of the GroupChatOrchestration with a streaming_response callback."""
+    agent_a = MockAgent(description="test agent")
+    agent_b = MockAgent(description="test agent")
+
+    runtime = InProcessRuntime()
+    runtime.start()
+
+    responses: dict[str, list[StreamingChatMessageContent]] = {}
+    try:
+        orchestration = GroupChatOrchestration(
+            members=[agent_a, agent_b],
+            manager=RoundRobinGroupChatManager(max_rounds=3),
+            streaming_agent_response_callback=lambda x, _: responses.setdefault(x.name, []).append(x),
+        )
+        orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
+        await orchestration_result.get(1.0)
+    finally:
+        await runtime.stop_when_idle()
+
+    assert len(responses[agent_a.name]) == 4  # Invoke twice, each with 2 chunks
+    assert len(responses[agent_b.name]) == 2  # Invoke once, with 2 chunks
+
+    assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_a.name])
+    assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_b.name])
+
+    agent_a_response = sum(responses[agent_a.name][1:2], responses[agent_a.name][0])
+    assert agent_a_response.content == "mock_response"
+    agent_b_response = sum(responses[agent_b.name][1:], responses[agent_b.name][0])
+    assert agent_b_response.content == "mock_response"
+
+
+async def test_invoke_with_human_response_function():
+    """Test the invoke method of the GroupChatOrchestration with a human response function."""
+    agent_a = MockAgent(description="test agent")
+    agent_b = MockAgent(description="test agent")
+
+    user_input_count = 0
+
+    def human_response_function(chat_history: ChatHistory) -> ChatMessageContent:
+        # Simulate user input
+        nonlocal user_input_count
+        user_input_count += 1
+        return ChatMessageContent(
+            role=AuthorRole.USER,
+            content=f"user_input_{user_input_count}",
+        )
+
+    orchestration_manager = RoundRobinGroupChatManagerWithUserInput(
+        max_rounds=3,
+        human_response_function=human_response_function,
+    )
+
+    runtime = InProcessRuntime()
+    runtime.start()
+
+    try:
+        orchestration = GroupChatOrchestration(
+            members=[agent_a, agent_b],
+            manager=orchestration_manager,
+        )
+        orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
+        await orchestration_result.get(1.0)
+    finally:
+        await runtime.stop_when_idle()
+
+    assert user_input_count == 4  # 3 rounds + 1 initial user input
+
+
 @pytest.mark.skipif(
     sys.version_info < (3, 11),
     reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.",
@@ -150,7 +240,7 @@ async def test_invoke_with_response_callback():
 async def test_invoke_cancel_before_completion():
     """Test the invoke method of the GroupChatOrchestration with cancellation before completion."""
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
     ):
         agent_a = MockAgent(description="test agent")
         agent_b = MockAgent(description="test agent")
@@ -171,7 +261,7 @@ async def test_invoke_cancel_before_completion():
         finally:
             await runtime.stop_when_idle()
 
-        assert mock_get_response.call_count == 2
+        assert mock_invoke_stream.call_count == 2
 
 
 async def test_invoke_cancel_after_completion():
diff --git a/python/tests/unit/agents/orchestration/test_handoff.py b/python/tests/unit/agents/orchestration/test_handoff.py
index 4d4326928131..ebf289661304 100644
--- a/python/tests/unit/agents/orchestration/test_handoff.py
+++ b/python/tests/unit/agents/orchestration/test_handoff.py
@@ -19,11 +19,10 @@
 from semantic_kernel.contents.chat_history import ChatHistory
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
 from semantic_kernel.contents.function_call_content import FunctionCallContent
-from semantic_kernel.contents.function_result_content import FunctionResultContent
 from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
 from semantic_kernel.kernel import Kernel
-from tests.unit.agents.orchestration.conftest import MockAgent, MockAgentThread, MockRuntime
+from tests.unit.agents.orchestration.conftest import MockAgent, MockRuntime
 
 if sys.version_info >= (3, 12):
     from typing import override  # pragma: no cover
@@ -48,6 +47,30 @@ async def get_response(
         kernel: Kernel | None = None,
         **kwargs,
     ) -> AgentResponseItem[ChatMessageContent]:
+        pass
+
+    @override
+    async def invoke(
+        self,
+        *,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
+        thread: AgentThread | None = None,
+        on_intermediate_message: Callable[[ChatMessageContent], Awaitable[None]] | None = None,
+        **kwargs,
+    ) -> AgentResponseItem[ChatMessageContent]:
+        pass
+
+    @override
+    async def invoke_stream(
+        self,
+        *,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
+        thread: AgentThread | None = None,
+        on_intermediate_message: Callable[[ChatMessageContent], Awaitable[None]] | None = None,
+        kernel: Kernel | None = None,
+        **kwargs,
+    ) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]:
+        """Simulate streaming response from the agent."""
         # Simulate some processing time
         await asyncio.sleep(0.1)
         await kernel.invoke_function_call(
@@ -60,20 +83,25 @@ async def get_response(
             chat_history=ChatHistory(),
         )
 
-        return AgentResponseItem[ChatMessageContent](
-            message=ChatMessageContent(
-                role=AuthorRole.TOOL,
-                items=[
-                    FunctionResultContent(
-                        call_id="test_call_id",
-                        id="test_id",
-                        function_name=f"transfer_to_{self.target_agent.name}",
-                        plugin_name=HANDOFF_PLUGIN_NAME,
-                    )
-                ],
-            ),
-            thread=thread or MockAgentThread(),
-        )
+        # Do not yield any messages, as the agent doesn't yield any tool related messages from the streaming API.
+        # Nevertheless, the method needs have a `yield` code path to satisfy the AsyncIterable interface.
+        if False:
+            yield
+
+
+class MockAgentWithCompleteTaskFunctionCall(Agent):
+    """A mock agent with complete_task function call for testing purposes."""
+
+    @override
+    async def get_response(
+        self,
+        *,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
+        thread: AgentThread | None = None,
+        kernel: Kernel | None = None,
+        **kwargs,
+    ) -> AgentResponseItem[ChatMessageContent]:
+        pass
 
     @override
     async def invoke(
@@ -93,9 +121,27 @@ async def invoke_stream(
         messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
         thread: AgentThread | None = None,
         on_intermediate_message: Callable[[ChatMessageContent], Awaitable[None]] | None = None,
+        kernel: Kernel | None = None,
         **kwargs,
     ) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]:
-        pass
+        """Simulate streaming response from the agent."""
+        # Simulate some processing time
+        await asyncio.sleep(0.1)
+        await kernel.invoke_function_call(
+            function_call=FunctionCallContent(
+                function_name="complete_task",
+                plugin_name=HANDOFF_PLUGIN_NAME,
+                call_id="test_call_id",
+                id="test_id",
+                arguments={"task_summary": "test_summary"},
+            ),
+            chat_history=ChatHistory(),
+        )
+
+        # Do not yield any messages, as the agent doesn't yield any tool related messages from the streaming API.
+        # Nevertheless, the method needs have a `yield` code path to satisfy the AsyncIterable interface.
+        if False:
+            yield
 
 
 # region HandoffOrchestration
@@ -290,7 +336,7 @@ async def test_invoke():
     """Test the prepare method of the HandoffOrchestration."""
     with (
         patch.object(HandoffAgentActor, "__init__", wraps=HandoffAgentActor.__init__, autospec=True) as mock_init,
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
     ):
         agent_a = MockAgent()
         agent_b = MockAgent()
@@ -311,8 +357,8 @@ async def test_invoke():
             await orchestration_result.get()
 
             assert mock_init.call_args_list[0][0][3] == {agent_b.name: "test", agent_c.name: "test"}
-            assert isinstance(mock_get_response.call_args_list[0][1]["kernel"], Kernel)
-            kernel = mock_get_response.call_args_list[0][1]["kernel"]
+            assert isinstance(mock_invoke_stream.call_args_list[0][1]["kernel"], Kernel)
+            kernel = mock_invoke_stream.call_args_list[0][1]["kernel"]
             assert HANDOFF_PLUGIN_NAME in kernel.plugins
             assert (
                 len(kernel.plugins[HANDOFF_PLUGIN_NAME].functions) == 3
@@ -325,10 +371,14 @@ async def test_invoke():
             await runtime.stop_when_idle()
 
 
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.",
+)
 async def test_invoke_with_list():
     """Test the invoke method of the HandoffOrchestration with a list of messages."""
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
     ):
         agent_a = MockAgent()
         agent_b = MockAgent()
@@ -351,9 +401,9 @@ async def test_invoke_with_list():
         finally:
             await runtime.stop_when_idle()
 
-        assert mock_get_response.call_count == 1
+        assert mock_invoke_stream.call_count == 1
         # Two messages + one message added internally to steer the conversation
-        assert len(mock_get_response.call_args_list[0][1]["messages"]) == 3
+        assert len(mock_invoke_stream.call_args_list[0][1]["messages"]) == 3
         # The kernel in the agent should not be modified
         assert len(agent_a.kernel.plugins) == 0
         assert len(agent_b.kernel.plugins) == 0
@@ -387,6 +437,86 @@ async def test_invoke_with_response_callback():
     assert len(agent_b.kernel.plugins) == 0
 
 
+async def test_invoke_with_streaming_response_callback():
+    """Test the invoke method of the HandoffOrchestration with a streaming response callback."""
+    agent_a = MockAgent()
+    agent_b = MockAgent()
+
+    runtime = InProcessRuntime()
+    runtime.start()
+
+    responses: dict[str, list[StreamingChatMessageContent]] = {}
+    try:
+        orchestration = HandoffOrchestration(
+            members=[agent_a, agent_b],
+            handoffs={agent_a.name: {agent_b.name: "test"}},
+            streaming_agent_response_callback=lambda x, _: responses.setdefault(x.name, []).append(x),
+        )
+        orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
+        await orchestration_result.get(1.0)
+    finally:
+        await runtime.stop_when_idle()
+
+    assert len(responses[agent_a.name]) == 2
+    assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_a.name])
+    agent_a_response = sum(responses[agent_a.name][1:], responses[agent_a.name][0])
+    assert agent_a_response.content == "mock_response"
+
+    # Agent B was not invoked, so it should not have any responses
+    assert agent_b.name not in responses or len(responses[agent_b.name]) == 0
+
+    # The kernel in the agent should not be modified
+    assert len(agent_a.kernel.plugins) == 0
+    assert len(agent_b.kernel.plugins) == 0
+
+
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.",
+)
+async def test_invoke_with_human_response_function():
+    """Test the invoke method of the HandoffOrchestration with a human response function."""
+    complete_task_agent_instance = MockAgentWithCompleteTaskFunctionCall()
+    normal_agent_instance = MockAgent()
+    call_sequence = iter([normal_agent_instance.invoke_stream, complete_task_agent_instance.invoke_stream])
+
+    user_input_count = 0
+
+    def human_response_function() -> ChatMessageContent:
+        # Simulate user input
+        nonlocal user_input_count
+        user_input_count += 1
+        return ChatMessageContent(role=AuthorRole.USER, content="user_input")
+
+    with (
+        patch.object(MockAgent, "invoke_stream") as mock_invoke_stream,
+    ):
+        mock_invoke_stream.side_effect = lambda *args, **kwargs: next(call_sequence)(*args, **kwargs)
+
+        agent_a = MockAgent(name="agent_a")
+        agent_b = MockAgent(name="agent_b")
+
+        runtime = InProcessRuntime()
+        runtime.start()
+
+        try:
+            orchestration = HandoffOrchestration(
+                members=[agent_a, agent_b],
+                handoffs={agent_a.name: {agent_b.name: "test"}},
+                human_response_function=human_response_function,
+            )
+            orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
+            await orchestration_result.get(1.0)
+        finally:
+            await runtime.stop_when_idle()
+
+        assert user_input_count == 1
+
+
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.",
+)
 async def test_invoke_with_handoff_function_call():
     """Test the invoke method of the HandoffOrchestration with a handoff function call."""
     agent_b = MockAgent()
@@ -424,7 +554,7 @@ async def test_invoke_with_handoff_function_call():
 async def test_invoke_cancel_before_completion():
     """Test the invoke method of the HandoffOrchestration with cancellation before completion."""
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
     ):
         agent_a = MockAgent()
         agent_b = MockAgent()
@@ -445,7 +575,7 @@ async def test_invoke_cancel_before_completion():
         finally:
             await runtime.stop_when_idle()
 
-        assert mock_get_response.call_count == 1
+        assert mock_invoke_stream.call_count == 1
 
 
 async def test_invoke_cancel_after_completion():
diff --git a/python/tests/unit/agents/orchestration/test_magentic.py b/python/tests/unit/agents/orchestration/test_magentic.py
index bb416b327622..f7d03a2e5df1 100644
--- a/python/tests/unit/agents/orchestration/test_magentic.py
+++ b/python/tests/unit/agents/orchestration/test_magentic.py
@@ -29,6 +29,7 @@
 from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
 from semantic_kernel.contents.chat_history import ChatHistory
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
 from tests.unit.agents.orchestration.conftest import MockAgent, MockRuntime
 
@@ -173,7 +174,7 @@ async def test_prepare():
 async def test_invoke():
     """Test the invoke method of the MagenticOrchestration."""
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
         patch.object(
             MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock
         ) as mock_get_chat_message_content,
@@ -208,7 +209,7 @@ async def test_invoke():
         assert result.role == AuthorRole.ASSISTANT
         assert result.content == "mock_response"
 
-        assert mock_get_response.call_count == 2
+        assert mock_invoke_stream.call_count == 2
         assert mock_get_chat_message_content.call_count == 3
 
 
@@ -256,7 +257,6 @@ async def test_invoke_with_response_callback():
 
     responses: list[DefaultTypeAlias] = []
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True),
         patch.object(
             MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock
         ) as mock_get_chat_message_content,
@@ -288,6 +288,57 @@ async def test_invoke_with_response_callback():
         assert all(item.content == "mock_response" for item in responses)
 
 
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.",
+)
+async def test_invoke_with_streaming_response_callback():
+    """Test the invoke method of the MagenticOrchestration with a streaming response callback."""
+
+    runtime = InProcessRuntime()
+    runtime.start()
+
+    responses: dict[str, list[StreamingChatMessageContent]] = {}
+
+    with (
+        patch.object(
+            MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock
+        ) as mock_get_chat_message_content,
+        patch.object(
+            StandardMagenticManager, "create_progress_ledger", new_callable=AsyncMock, side_effect=ManagerProgressList
+        ),
+    ):
+        mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response")
+
+        agent_a = MockAgent(name="agent_a", description="test agent")
+        agent_b = MockAgent(name="agent_b", description="test agent")
+
+        try:
+            orchestration = MagenticOrchestration(
+                members=[agent_a, agent_b],
+                manager=StandardMagenticManager(
+                    chat_completion_service=MockChatCompletionService(ai_model_id="test"),
+                    prompt_execution_settings=MockPromptExecutionSettings(),
+                ),
+                streaming_agent_response_callback=lambda x, _: responses.setdefault(x.name, []).append(x),
+            )
+            orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
+            await orchestration_result.get(1.0)
+        finally:
+            await runtime.stop_when_idle()
+
+        assert len(responses[agent_a.name]) == 2
+        assert len(responses[agent_b.name]) == 2
+
+        assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_a.name])
+        assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_b.name])
+
+        agent_a_response = sum(responses[agent_a.name][1:], responses[agent_a.name][0])
+        assert agent_a_response.content == "mock_response"
+        agent_b_response = sum(responses[agent_b.name][1:], responses[agent_b.name][0])
+        assert agent_b_response.content == "mock_response"
+
+
 @pytest.mark.skipif(
     sys.version_info < (3, 11),
     reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.",
@@ -298,7 +349,7 @@ async def test_invoke_with_max_stall_count_exceeded():
     runtime.start()
 
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
         patch.object(
             MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock
         ) as mock_get_chat_message_content,
@@ -328,7 +379,7 @@ async def test_invoke_with_max_stall_count_exceeded():
         finally:
             await runtime.stop_when_idle()
 
-        assert mock_get_response.call_count == 3
+        assert mock_invoke_stream.call_count == 3
         # Exceeding max stall count will trigger replanning, which will recreate the facts and plan,
         # resulting in two additional calls to get_chat_message_content compared to the `test_invoke` test.
         assert mock_get_chat_message_content.call_count == 5
@@ -344,7 +395,7 @@ async def test_invoke_with_max_round_count_exceeded():
     runtime.start()
 
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
         patch.object(
             MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock
         ) as mock_get_chat_message_content,
@@ -375,7 +426,7 @@ async def test_invoke_with_max_round_count_exceeded():
             await runtime.stop_when_idle()
 
         assert result.content == "Max round count reached."
-        assert mock_get_response.call_count == 1
+        assert mock_invoke_stream.call_count == 1
         # Planning will be called once, so the facts and plan will be created once.
         assert mock_get_chat_message_content.call_count == 2
 
@@ -390,7 +441,7 @@ async def test_invoke_with_max_reset_count_exceeded():
     runtime.start()
 
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
         patch.object(
             MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock
         ) as mock_get_chat_message_content,
@@ -422,7 +473,7 @@ async def test_invoke_with_max_reset_count_exceeded():
             await runtime.stop_when_idle()
 
         assert result.content == "Max reset count reached."
-        assert mock_get_response.call_count == 1
+        assert mock_invoke_stream.call_count == 1
         # Planning and replanning will be each called once, so the facts and plan will be created twice.
         assert mock_get_chat_message_content.call_count == 4
 
@@ -437,7 +488,6 @@ async def test_invoke_with_unknown_speaker():
     runtime.start()
 
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True),
         patch.object(
             MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock
         ) as mock_get_chat_message_content,
diff --git a/python/tests/unit/agents/orchestration/test_sequential.py b/python/tests/unit/agents/orchestration/test_sequential.py
index d29edc97792a..7d4e8a027e2d 100644
--- a/python/tests/unit/agents/orchestration/test_sequential.py
+++ b/python/tests/unit/agents/orchestration/test_sequential.py
@@ -10,6 +10,7 @@
 from semantic_kernel.agents.orchestration.sequential import SequentialOrchestration
 from semantic_kernel.agents.runtime.in_process.in_process_runtime import InProcessRuntime
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from tests.unit.agents.orchestration.conftest import MockAgent, MockRuntime
 
 
@@ -78,6 +79,37 @@ async def test_invoke_with_response_callback():
         await runtime.stop_when_idle()
 
 
+async def test_invoke_with_streaming_response_callback():
+    """Test the invoke method of the SequentialOrchestration with a streaming response callback."""
+    agent_a = MockAgent()
+    agent_b = MockAgent()
+
+    runtime = InProcessRuntime()
+    runtime.start()
+
+    responses: dict[str, list[StreamingChatMessageContent]] = {}
+    try:
+        orchestration = SequentialOrchestration(
+            members=[agent_a, agent_b],
+            streaming_agent_response_callback=lambda x, _: responses.setdefault(x.name, []).append(x),
+        )
+        orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
+        await orchestration_result.get(1.0)
+
+        assert len(responses[agent_a.name]) == 2
+        assert len(responses[agent_b.name]) == 2
+
+        assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_a.name])
+        assert all(isinstance(item, StreamingChatMessageContent) for item in responses[agent_b.name])
+
+        agent_a_response = sum(responses[agent_a.name][1:], responses[agent_a.name][0])
+        assert agent_a_response.content == "mock_response"
+        agent_b_response = sum(responses[agent_b.name][1:], responses[agent_b.name][0])
+        assert agent_b_response.content == "mock_response"
+    finally:
+        await runtime.stop_when_idle()
+
+
 @pytest.mark.skipif(
     sys.version_info < (3, 11),
     reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.",
@@ -85,7 +117,7 @@ async def test_invoke_with_response_callback():
 async def test_invoke_cancel_before_completion():
     """Test the invoke method of the SequentialOrchestration with cancellation before completion."""
     with (
-        patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
+        patch.object(MockAgent, "invoke_stream", wraps=MockAgent.invoke_stream, autospec=True) as mock_invoke_stream,
     ):
         agent_a = MockAgent()
         agent_b = MockAgent()
@@ -103,7 +135,7 @@ async def test_invoke_cancel_before_completion():
         finally:
             await runtime.stop_when_idle()
 
-        assert mock_get_response.call_count == 1
+        assert mock_invoke_stream.call_count == 1
 
 
 async def test_invoke_cancel_after_completion():