From b8f00855e9a10565e3bd99275c7a3a0ebf353789 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Fri, 30 May 2025 09:31:30 -0700
Subject: [PATCH 01/10] Experimental: streaming agent response callback

---
 .../step2_sequential.py                       | 18 ++++-
 .../agents/orchestration/agent_actor_base.py  | 74 ++++++++++++++++++-
 .../orchestration/orchestration_base.py       |  6 ++
 .../agents/orchestration/sequential.py        | 10 ++-
 4 files changed, 102 insertions(+), 6 deletions(-)

diff --git a/python/samples/getting_started_with_agents/multi_agent_orchestration/step2_sequential.py b/python/samples/getting_started_with_agents/multi_agent_orchestration/step2_sequential.py
index 28390f0643b8..7fdbccefd9b7 100644
--- a/python/samples/getting_started_with_agents/multi_agent_orchestration/step2_sequential.py
+++ b/python/samples/getting_started_with_agents/multi_agent_orchestration/step2_sequential.py
@@ -6,6 +6,7 @@
 from semantic_kernel.agents.runtime import InProcessRuntime
 from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
 from semantic_kernel.contents import ChatMessageContent
+from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 
 """
 The following sample demonstrates how to create a sequential orchestration for
@@ -60,6 +61,21 @@ def agent_response_callback(message: ChatMessageContent) -> None:
     print(f"# {message.name}\n{message.content}")
 
 
+is_new_message = True
+
+
+def streaming_agent_response_callback(message: StreamingChatMessageContent, is_final: bool) -> None:
+    """Observer function to print the messages from the agents."""
+    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
@@ -67,7 +83,7 @@ async def main():
     agents = get_agents()
     sequential_orchestration = SequentialOrchestration(
         members=agents,
-        agent_response_callback=agent_response_callback,
+        streaming_agent_response_callback=streaming_agent_response_callback,
     )
 
     # 2. Create a runtime and start it
diff --git a/python/semantic_kernel/agents/orchestration/agent_actor_base.py b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
index 6760584eb62a..d5d7d0cd0414 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,6 +44,8 @@ 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.
 
@@ -52,10 +54,13 @@ def __init__(
             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
                 by the agents.
+            streaming_agent_response_callback (Callable | None): A function that is called when a response is produced
+                by the agents. The response will be a streaming chunk in this case.
         """
         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
@@ -75,3 +80,70 @@ async def _call_agent_response_callback(self, message: DefaultTypeAlias) -> None
                 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) -> 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.
+
+        Returns:
+            DefaultTypeAlias: The response from the agent.
+        """
+        streaming_message_buffer: list[StreamingChatMessageContent] = []
+
+        if self._agent_thread is None:
+            messages = self._chat_history.messages[:]
+            if additional_messages:
+                messages.extend(additional_messages if isinstance(additional_messages, list) else [additional_messages])
+
+            async for response_item in self._agent.invoke_stream(messages=messages):
+                # 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:
+                await self._call_streaming_agent_response_callback(streaming_message_buffer[-1], is_final=True)
+        else:
+            messages = (
+                []
+                if additional_messages is None
+                else additional_messages
+                if isinstance(additional_messages, list)
+                else [additional_messages]
+            )
+
+            async for response_item in self._agent.invoke_stream(
+                messages=messages,
+                thread=self._agent_thread,
+            ):
+                # 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 streaming_message_buffer:
+                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.')
+
+        return sum(streaming_message_buffer[1:], streaming_message_buffer[0])
diff --git a/python/semantic_kernel/agents/orchestration/orchestration_base.py b/python/semantic_kernel/agents/orchestration/orchestration_base.py
index abdc3833b320..19b2ff5271cb 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.
 
@@ -105,6 +108,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 response is produced
+                by the agents. The response will be a streaming chunk in this case.
         """
         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..ef2b13331121 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,
         )

From f4ab7144f6a5cfdbc63ad4abd55f33d0050ca85c Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Fri, 30 May 2025 10:36:08 -0700
Subject: [PATCH 02/10] Add streaming agent response callback to sequential and
 concurrent

---
 .../agents/orchestration/agent_actor_base.py       |  6 +++---
 .../agents/orchestration/concurrent.py             | 14 ++++++++------
 .../agents/orchestration/orchestration_base.py     |  6 +++---
 .../agents/orchestration/sequential.py             |  1 +
 4 files changed, 15 insertions(+), 12 deletions(-)

diff --git a/python/semantic_kernel/agents/orchestration/agent_actor_base.py b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
index d5d7d0cd0414..a1e6cf08ddea 100644
--- a/python/semantic_kernel/agents/orchestration/agent_actor_base.py
+++ b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
@@ -52,10 +52,10 @@ def __init__(
         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 response is produced
-                by the agents. The response will be a streaming chunk in this case.
+            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
diff --git a/python/semantic_kernel/agents/orchestration/concurrent.py b/python/semantic_kernel/agents/orchestration/concurrent.py
index c46b466872dc..33203c269291 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,14 +82,10 @@ 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),
@@ -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/orchestration_base.py b/python/semantic_kernel/agents/orchestration/orchestration_base.py
index 19b2ff5271cb..ed547828ba0f 100644
--- a/python/semantic_kernel/agents/orchestration/orchestration_base.py
+++ b/python/semantic_kernel/agents/orchestration/orchestration_base.py
@@ -106,10 +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 response is produced
-                by the agents. The response will be a streaming chunk in this case.
+            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.")
diff --git a/python/semantic_kernel/agents/orchestration/sequential.py b/python/semantic_kernel/agents/orchestration/sequential.py
index ef2b13331121..d04369d88c02 100644
--- a/python/semantic_kernel/agents/orchestration/sequential.py
+++ b/python/semantic_kernel/agents/orchestration/sequential.py
@@ -157,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)}")

From d56aabd1bd63cf98fd2c61816bf1612984118797 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Fri, 30 May 2025 10:52:50 -0700
Subject: [PATCH 03/10] Add streaming agent response callback to group chat

---
 .../step2_sequential.py                       |  18 +-
 ...ntial_streaming_agent_response_callback.py | 159 ++++++++++++++++++
 .../agents/orchestration/agent_actor_base.py  |   5 +-
 .../agents/orchestration/group_chat.py        |  45 +++--
 4 files changed, 185 insertions(+), 42 deletions(-)
 create mode 100644 python/samples/getting_started_with_agents/multi_agent_orchestration/step2b_sequential_streaming_agent_response_callback.py

diff --git a/python/samples/getting_started_with_agents/multi_agent_orchestration/step2_sequential.py b/python/samples/getting_started_with_agents/multi_agent_orchestration/step2_sequential.py
index 7fdbccefd9b7..28390f0643b8 100644
--- a/python/samples/getting_started_with_agents/multi_agent_orchestration/step2_sequential.py
+++ b/python/samples/getting_started_with_agents/multi_agent_orchestration/step2_sequential.py
@@ -6,7 +6,6 @@
 from semantic_kernel.agents.runtime import InProcessRuntime
 from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
 from semantic_kernel.contents import ChatMessageContent
-from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 
 """
 The following sample demonstrates how to create a sequential orchestration for
@@ -61,21 +60,6 @@ def agent_response_callback(message: ChatMessageContent) -> None:
     print(f"# {message.name}\n{message.content}")
 
 
-is_new_message = True
-
-
-def streaming_agent_response_callback(message: StreamingChatMessageContent, is_final: bool) -> None:
-    """Observer function to print the messages from the agents."""
-    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
@@ -83,7 +67,7 @@ async def main():
     agents = get_agents()
     sequential_orchestration = SequentialOrchestration(
         members=agents,
-        streaming_agent_response_callback=streaming_agent_response_callback,
+        agent_response_callback=agent_response_callback,
     )
 
     # 2. Create a runtime and start it
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..7f2ea6f5bae3
--- /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 a1e6cf08ddea..6eada7bca213 100644
--- a/python/semantic_kernel/agents/orchestration/agent_actor_base.py
+++ b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
@@ -146,4 +146,7 @@ async def _invoke_agent(self, additional_messages: DefaultTypeAlias | None = Non
         if not streaming_message_buffer:
             raise RuntimeError(f'Agent "{self._agent.name}" did not return any response.')
 
-        return sum(streaming_message_buffer[1:], streaming_message_buffer[0])
+        full_response = sum(streaming_message_buffer[1:], streaming_message_buffer[0])
+        await self._call_agent_response_callback(full_response)
+
+        return full_response
diff --git a/python/semantic_kernel/agents/orchestration/group_chat.py b/python/semantic_kernel/agents/orchestration/group_chat.py
index f197809004c2..4270bbef1f84 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(
+                    agent,
+                    internal_topic_type,
+                    agent_response_callback=self._agent_response_callback,
+                    streaming_agent_response_callback=self._streaming_agent_response_callback,
+                ),  # type: ignore[misc]
             )
             for agent in self._members
         ])

From b181150d41fd4b4918d7dd2e9c73ad86c6c6a392 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Tue, 3 Jun 2025 10:21:05 -0700
Subject: [PATCH 04/10] Add streaming agent response callback to handoff

---
 .../agents/orchestration/agent_actor_base.py  |  68 +++++------
 .../agents/orchestration/handoffs.py          | 111 +++++++++++-------
 2 files changed, 104 insertions(+), 75 deletions(-)

diff --git a/python/semantic_kernel/agents/orchestration/agent_actor_base.py b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
index 6eada7bca213..1a6e698efb5d 100644
--- a/python/semantic_kernel/agents/orchestration/agent_actor_base.py
+++ b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
@@ -98,55 +98,55 @@ async def _call_streaming_agent_response_callback(
             else:
                 self._streaming_agent_response_callback(message_chunk, is_final)
 
-    async def _invoke_agent(self, additional_messages: DefaultTypeAlias | None = None) -> ChatMessageContent:
+    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)
 
-        if self._agent_thread is None:
-            messages = self._chat_history.messages[:]
-            if additional_messages:
-                messages.extend(additional_messages if isinstance(additional_messages, list) else [additional_messages])
-
-            async for response_item in self._agent.invoke_stream(messages=messages):
-                # 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:
-                await self._call_streaming_agent_response_callback(streaming_message_buffer[-1], is_final=True)
-        else:
-            messages = (
-                []
-                if additional_messages is None
-                else additional_messages
-                if isinstance(additional_messages, list)
-                else [additional_messages]
-            )
-
-            async for response_item in self._agent.invoke_stream(
-                messages=messages,
-                thread=self._agent_thread,
-            ):
-                # 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 streaming_message_buffer:
-                await self._call_streaming_agent_response_callback(streaming_message_buffer[-1], is_final=True)
+        async for response_item in self._agent.invoke_stream(messages=messages, thread=self._agent_thread, **kwargs):
+            # 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/handoffs.py b/python/semantic_kernel/agents/orchestration/handoffs.py
index 839e1692d7e0..ce9383f982a0 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):
+            # 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,
                 ),
             )

From 2775f2638cbb514492e6c5e70aabd02a58807d28 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Tue, 3 Jun 2025 10:31:08 -0700
Subject: [PATCH 05/10] Add streaming agent response callback to magentic

---
 .../agents/orchestration/magentic.py          | 35 ++++++++-----------
 1 file changed, 15 insertions(+), 20 deletions(-)

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

From c64e56143601a098f89af3c6a3770fe19e741da8 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Tue, 3 Jun 2025 10:36:35 -0700
Subject: [PATCH 06/10] Fix sample output

---
 ...ntial_streaming_agent_response_callback.py | 90 +++++++++----------
 1 file changed, 45 insertions(+), 45 deletions(-)

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
index 7f2ea6f5bae3..3fc8c4bb5c26 100644
--- 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
@@ -107,51 +107,51 @@ async def main():
     """
     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!
+    **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!
     """
 
 

From 1df89135be3f90320356090dc80c06b2f913fc26 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Tue, 3 Jun 2025 11:20:29 -0700
Subject: [PATCH 07/10] Fix unit tests

---
 .../agents/orchestration/concurrent.py        |  2 +-
 .../unit/agents/orchestration/conftest.py     | 32 ++++++---
 .../agents/orchestration/test_concurrent.py   |  4 +-
 .../agents/orchestration/test_group_chat.py   | 16 ++---
 .../unit/agents/orchestration/test_handoff.py | 65 ++++++++-----------
 .../agents/orchestration/test_magentic.py     | 16 ++---
 .../agents/orchestration/test_sequential.py   |  4 +-
 7 files changed, 71 insertions(+), 68 deletions(-)

diff --git a/python/semantic_kernel/agents/orchestration/concurrent.py b/python/semantic_kernel/agents/orchestration/concurrent.py
index 33203c269291..c669114ca7d3 100644
--- a/python/semantic_kernel/agents/orchestration/concurrent.py
+++ b/python/semantic_kernel/agents/orchestration/concurrent.py
@@ -88,7 +88,7 @@ async def _handle_message(self, message: ConcurrentRequestMessage, ctx: MessageC
 
         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,
         )
diff --git a/python/tests/unit/agents/orchestration/conftest.py b/python/tests/unit/agents/orchestration/conftest.py
index e5943dede427..e48020e2e89b 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,27 @@ 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,
+                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,
+                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..5bc7558dc050 100644
--- a/python/tests/unit/agents/orchestration/test_concurrent.py
+++ b/python/tests/unit/agents/orchestration/test_concurrent.py
@@ -87,7 +87,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 +105,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..08b775912ace 100644
--- a/python/tests/unit/agents/orchestration/test_group_chat.py
+++ b/python/tests/unit/agents/orchestration/test_group_chat.py
@@ -55,7 +55,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 +78,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 +88,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 +111,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():
@@ -150,7 +150,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 +171,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..f86421931750 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,32 +47,7 @@ async def get_response(
         kernel: Kernel | None = None,
         **kwargs,
     ) -> AgentResponseItem[ChatMessageContent]:
-        # Simulate some processing time
-        await asyncio.sleep(0.1)
-        await kernel.invoke_function_call(
-            function_call=FunctionCallContent(
-                function_name=f"transfer_to_{self.target_agent.name}",
-                plugin_name=HANDOFF_PLUGIN_NAME,
-                call_id="test_call_id",
-                id="test_id",
-            ),
-            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(),
-        )
+        pass
 
     @override
     async def invoke(
@@ -93,9 +67,26 @@ 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=f"transfer_to_{self.target_agent.name}",
+                plugin_name=HANDOFF_PLUGIN_NAME,
+                call_id="test_call_id",
+                id="test_id",
+            ),
+            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 +281,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 +302,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
@@ -328,7 +319,7 @@ async def test_invoke():
 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 +342,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
@@ -424,7 +415,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 +436,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..b5452fbe0c0a 100644
--- a/python/tests/unit/agents/orchestration/test_magentic.py
+++ b/python/tests/unit/agents/orchestration/test_magentic.py
@@ -173,7 +173,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 +208,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
 
 
@@ -298,7 +298,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 +328,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 +344,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 +375,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 +390,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 +422,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
 
diff --git a/python/tests/unit/agents/orchestration/test_sequential.py b/python/tests/unit/agents/orchestration/test_sequential.py
index d29edc97792a..ed255d2e2c11 100644
--- a/python/tests/unit/agents/orchestration/test_sequential.py
+++ b/python/tests/unit/agents/orchestration/test_sequential.py
@@ -85,7 +85,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 +103,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():

From ba626a0e3065add670b69b134f9b97d5619ac7b1 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Tue, 3 Jun 2025 14:11:56 -0700
Subject: [PATCH 08/10] Add unit tests

---
 .../agents/orchestration/agent_actor_base.py  |   1 -
 .../unit/agents/orchestration/conftest.py     |   2 +
 .../agents/orchestration/test_concurrent.py   |  32 +++++
 .../agents/orchestration/test_group_chat.py   |  92 +++++++++++-
 .../unit/agents/orchestration/test_handoff.py | 131 ++++++++++++++++++
 .../agents/orchestration/test_magentic.py     |  54 +++++++-
 .../agents/orchestration/test_sequential.py   |  32 +++++
 7 files changed, 340 insertions(+), 4 deletions(-)

diff --git a/python/semantic_kernel/agents/orchestration/agent_actor_base.py b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
index 1a6e698efb5d..02215f53ae9c 100644
--- a/python/semantic_kernel/agents/orchestration/agent_actor_base.py
+++ b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
@@ -74,7 +74,6 @@ 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)
diff --git a/python/tests/unit/agents/orchestration/conftest.py b/python/tests/unit/agents/orchestration/conftest.py
index e48020e2e89b..43597e0f84b3 100644
--- a/python/tests/unit/agents/orchestration/conftest.py
+++ b/python/tests/unit/agents/orchestration/conftest.py
@@ -71,6 +71,7 @@ async def invoke_stream(
         yield AgentResponseItem[StreamingChatMessageContent](
             message=StreamingChatMessageContent(
                 role=AuthorRole.ASSISTANT,
+                name=self.name,
                 content="mock",
                 choice_index=0,
             ),
@@ -81,6 +82,7 @@ async def invoke_stream(
         yield AgentResponseItem[StreamingChatMessageContent](
             message=StreamingChatMessageContent(
                 role=AuthorRole.ASSISTANT,
+                name=self.name,
                 content="_response",
                 choice_index=0,
             ),
diff --git a/python/tests/unit/agents/orchestration/test_concurrent.py b/python/tests/unit/agents/orchestration/test_concurrent.py
index 5bc7558dc050..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.",
diff --git a/python/tests/unit/agents/orchestration/test_group_chat.py b/python/tests/unit/agents/orchestration/test_group_chat.py
index 08b775912ace..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
 
 
@@ -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.",
diff --git a/python/tests/unit/agents/orchestration/test_handoff.py b/python/tests/unit/agents/orchestration/test_handoff.py
index f86421931750..56021ca3d765 100644
--- a/python/tests/unit/agents/orchestration/test_handoff.py
+++ b/python/tests/unit/agents/orchestration/test_handoff.py
@@ -89,6 +89,61 @@ async def invoke_stream(
             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(
+        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(
+            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
 
 
@@ -378,6 +433,82 @@ 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
+
+
 async def test_invoke_with_handoff_function_call():
     """Test the invoke method of the HandoffOrchestration with a handoff function call."""
     agent_b = MockAgent()
diff --git a/python/tests/unit/agents/orchestration/test_magentic.py b/python/tests/unit/agents/orchestration/test_magentic.py
index b5452fbe0c0a..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
 
@@ -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.",
@@ -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 ed255d2e2c11..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.",

From 5ce7ccdcc1c79cc4f1962129dd9432696439b3a6 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Tue, 3 Jun 2025 14:39:58 -0700
Subject: [PATCH 09/10] Fix mypy

---
 .../semantic_kernel/agents/orchestration/agent_actor_base.py  | 2 +-
 python/semantic_kernel/agents/orchestration/group_chat.py     | 4 ++--
 python/semantic_kernel/agents/orchestration/handoffs.py       | 2 +-
 3 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/python/semantic_kernel/agents/orchestration/agent_actor_base.py b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
index 02215f53ae9c..31f44cf84772 100644
--- a/python/semantic_kernel/agents/orchestration/agent_actor_base.py
+++ b/python/semantic_kernel/agents/orchestration/agent_actor_base.py
@@ -111,7 +111,7 @@ async def _invoke_agent(self, additional_messages: DefaultTypeAlias | None = Non
         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):
+        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:
diff --git a/python/semantic_kernel/agents/orchestration/group_chat.py b/python/semantic_kernel/agents/orchestration/group_chat.py
index 4270bbef1f84..a070f380810b 100644
--- a/python/semantic_kernel/agents/orchestration/group_chat.py
+++ b/python/semantic_kernel/agents/orchestration/group_chat.py
@@ -489,12 +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(
+                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,
-                ),  # type: ignore[misc]
+                ),
             )
             for agent in self._members
         ])
diff --git a/python/semantic_kernel/agents/orchestration/handoffs.py b/python/semantic_kernel/agents/orchestration/handoffs.py
index ce9383f982a0..9cc58637f082 100644
--- a/python/semantic_kernel/agents/orchestration/handoffs.py
+++ b/python/semantic_kernel/agents/orchestration/handoffs.py
@@ -348,7 +348,7 @@ async def _invoke_agent_with_potentially_no_response(
         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):
+        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:

From e109de54b99f873c15d4d5792876b8d281d62150 Mon Sep 17 00:00:00 2001
From: Tao Chen <taochen@microsoft.com>
Date: Tue, 3 Jun 2025 14:52:29 -0700
Subject: [PATCH 10/10] Skip tests in 3.10

---
 python/tests/unit/agents/orchestration/test_handoff.py | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/python/tests/unit/agents/orchestration/test_handoff.py b/python/tests/unit/agents/orchestration/test_handoff.py
index 56021ca3d765..ebf289661304 100644
--- a/python/tests/unit/agents/orchestration/test_handoff.py
+++ b/python/tests/unit/agents/orchestration/test_handoff.py
@@ -371,6 +371,10 @@ 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 (
@@ -509,6 +513,10 @@ def human_response_function() -> ChatMessageContent:
         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()