MiddlewareSet fix

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/__init__.py

@@ -6,7 +6,7 @@
 from .channel_service_adapter import ChannelServiceAdapter
 from .channel_service_client_factory_base import ChannelServiceClientFactoryBase
 from .message_factory import MessageFactory
-from .middleware_set import Middleware
+from .middleware_set import Middleware, MiddlewareSet
 from .rest_channel_service_client_factory import RestChannelServiceClientFactory
 from .turn_context import TurnContext
 
@@ -174,6 +174,7 @@
     "MemoryStorage",
     "AgenticUserAuthorization",
     "Authorization",
+    "MiddlewareSet",
     "error_resources",
     "ErrorMessage",
     "ErrorResources",

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_adapter.py

@@ -232,16 +232,14 @@ async def run_pipeline(
         :param context: The context object for the turn.
         :type context: :class:`microsoft_agents.hosting.core.turn_context.TurnContext`
         :param callback: A callback method to run at the end of the pipeline.
-        :type callback: Callable[[TurnContext], Awaitable]
-        :return: Result produced by the middleware pipeline.
-        :rtype: typing.Any
+        :type callback: Callable[[TurnContext], Awaitable] | None
         """
         if context is None:
             raise TypeError(context.__class__.__name__)
 
         if context.activity is not None:
             try:
-                return await self.middleware_set.receive_activity_with_status(
+                await self.middleware_set.receive_activity_with_status(
                     context, callback
                 )
             except Exception as error:

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/middleware_set.py

@@ -3,15 +3,23 @@
 
 from abc import abstractmethod
 from typing import Awaitable, Callable, Protocol
+from typing_extensions import Self
 
 from .turn_context import TurnContext
 
 
 class Middleware(Protocol):
+
     @abstractmethod
     async def on_turn(
         self, context: TurnContext, logic: Callable[[TurnContext], Awaitable]
     ):
+        """
+        Called for each turn of the conversation.
+
+        :param context: The turn context.
+        :param logic: The next middleware in the pipeline or the final logic to be executed.
+        """
         pass
 
 
@@ -23,56 +31,69 @@ class MiddlewareSet(Middleware):
     """
 
     def __init__(self):
-        super(MiddlewareSet, self).__init__()
+        super().__init__()
         self._middleware: list[Middleware] = []
 
-    def use(self, *middleware: Middleware):
+    def use(self, *middleware: Middleware) -> Self:
         """
         Registers middleware plugin(s) with the agent or set.
-        :param middleware :
-        :return:
+        :param middleware : The middleware plugin(s) to register.
+        :return: The `MiddlewareSet` instance to allow chaining of `use` calls.
+        :rtype: Self
         """
         for idx, mid in enumerate(middleware):
             if hasattr(mid, "on_turn") and callable(mid.on_turn):
                 self._middleware.append(mid)
-                return self
-            raise TypeError(
-                'MiddlewareSet.use(): invalid middleware at index "%s" being added.'
-                % idx
-            )
-
-    async def receive_activity(self, context: TurnContext):
-        await self.receive_activity_internal(context, None)
-
-    async def on_turn(
-        self, context: TurnContext, logic: Callable[[TurnContext], Awaitable]
-    ):
-        await self.receive_activity_internal(context, None)
-        await logic()
+            else:
+                raise TypeError(
+                    'MiddlewareSet.use(): invalid middleware at index "%s" being added.'
+                    % idx
+                )
+        return self
 
-    async def receive_activity_with_status(
-        self, context: TurnContext, callback: Callable[[TurnContext], Awaitable]
-    ):
-        return await self.receive_activity_internal(context, callback)
-
-    async def receive_activity_internal(
+    async def _receive_activity_internal(
         self,
         context: TurnContext,
-        callback: Callable[[TurnContext], Awaitable],
+        callback: Callable[[TurnContext], Awaitable] | None,
         next_middleware_index: int = 0,
     ):
+        """Recursively invokes middleware in the set, passing the context and callback through the pipeline.
+
+        :param context: The turn context.
+        :param callback: The final logic to be executed after the middleware pipeline.
+        :param next_middleware_index: The index of the next middleware to invoke.
+        """
         if next_middleware_index == len(self._middleware):
             if callback is not None:
-                return await callback(context)
-            return None
+                await callback(context)
+            return
+
         next_middleware = self._middleware[next_middleware_index]
 
-        async def call_next_middleware():
-            return await self.receive_activity_internal(
-                context, callback, next_middleware_index + 1
+        async def call_next_middleware(ctx: TurnContext):
+            await self._receive_activity_internal(
+                ctx, callback, next_middleware_index + 1
             )
 
-        try:
-            return await next_middleware.on_turn(context, call_next_middleware)
-        except Exception as error:
-            raise error
+        await next_middleware.on_turn(context, call_next_middleware)
+
+    async def receive_activity_with_status(
+        self, context: TurnContext, logic: Callable[[TurnContext], Awaitable] | None
+    ):
+        """Handles an incoming activity by passing it through the middleware pipeline and then to the final logic.
+
+        :param context: The turn context.
+        :param logic: The final logic to be executed after the middleware pipeline.
+        """
+        await self._receive_activity_internal(context, logic)
+
+    async def on_turn(
+        self, context: TurnContext, logic: Callable[[TurnContext], Awaitable]
+    ):
+        """Handles an incoming activity by passing it through the middleware pipeline and then to the final logic.
+
+        :param context: The turn context.
+        :param logic: The final logic to be executed after the middleware pipeline.
+        """
+        await self._receive_activity_internal(context, None)
+        await logic(context)

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/storage/transcript_logger.py

@@ -112,7 +112,7 @@ def __init__(self, logger: TranscriptLogger):
         self.logger = logger
 
     async def on_turn(
-        self, context: TurnContext, logic: Callable[[TurnContext], Awaitable]
+        self, context: TurnContext, logic: Callable[[TurnContext], Awaitable] | None
     ):
         """Initialization for middleware.
         :param context: Context for the current turn of conversation with the user.
@@ -201,7 +201,7 @@ async def delete_activity_handler(
         context.on_delete_activity(delete_activity_handler)
 
         if logic:
-            await logic()
+            await logic(context)
 
         # Flush transcript at end of turn
         while not transcript.empty():

test_samples/app_style/empty_agent.py

@@ -14,19 +14,24 @@
     TurnContext,
     MemoryStorage,
 )
+from microsoft_agents.hosting.core.storage import (
+    ConsoleTranscriptLogger,
+    TranscriptLoggerMiddleware,
+)
 from microsoft_agents.hosting.aiohttp import CloudAdapter
 from microsoft_agents.hosting.core.app.oauth.authorization import Authorization
 
 from shared import start_server
 
-logging.basicConfig(level=logging.INFO)
 load_dotenv(path.join(path.dirname(__file__), ".env"))
 
 agents_sdk_config = load_configuration_from_env(environ)
 
 STORAGE = MemoryStorage()
 CONNECTION_MANAGER = MsalConnectionManager(**agents_sdk_config)
 ADAPTER = CloudAdapter(connection_manager=CONNECTION_MANAGER)
+ADAPTER.use(TranscriptLoggerMiddleware(ConsoleTranscriptLogger()))
+
 AUTHORIZATION = Authorization(STORAGE, CONNECTION_MANAGER, **agents_sdk_config)
 
 AGENT_APP = AgentApplication[TurnState](

tests/hosting_core/storage/test_transcript_logger_middleware.py

@@ -52,6 +52,37 @@ async def callback(tc):
     assert pagedResult.continuation_token is None
 
 
+@pytest.mark.asyncio
+async def test_should_log_outgoing_activity_sent_by_callback():
+    transcript_store = TranscriptMemoryStore()
+    conversation_id = "id.1"
+    transcript_middleware = TranscriptLoggerMiddleware(transcript_store)
+    channelName = "Channel1"
+
+    adapter = MockTestingAdapter(channelName)
+    adapter.use(transcript_middleware)
+    id = ClaimsIdentity({}, True)
+
+    async def callback(tc):
+        await tc.send_activity("bot response")
+
+    a1 = adapter.make_activity("user message")
+    a1.conversation.id = conversation_id
+
+    await adapter.process_activity(id, a1, callback)
+
+    pagedResult = await transcript_store.get_transcript_activities(
+        channelName, conversation_id
+    )
+
+    assert len(pagedResult.items) == 2
+    transcript_by_text = {activity.text: activity for activity in pagedResult.items}
+    assert "user message" in transcript_by_text
+    assert "bot response" in transcript_by_text
+    assert transcript_by_text["bot response"].from_property.id == "agent"
+    assert pagedResult.continuation_token is None
+
+
 @pytest.mark.asyncio
 async def test_should_write_to_file():
     fileName = "test_transcript.log"