diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/__init__.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/__init__.py index 1dee26a02..4d31384b6 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/__init__.py +++ b/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", diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_adapter.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_adapter.py index d8768d551..7d00fc3a1 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_adapter.py +++ b/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: diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/middleware_set.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/middleware_set.py index 5f6f5e584..902c70649 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/middleware_set.py +++ b/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) diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/storage/transcript_logger.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/storage/transcript_logger.py index 93b78baea..3986e4792 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/storage/transcript_logger.py +++ b/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(): diff --git a/test_samples/app_style/empty_agent.py b/test_samples/app_style/empty_agent.py index c68083c90..76c56c132 100644 --- a/test_samples/app_style/empty_agent.py +++ b/test_samples/app_style/empty_agent.py @@ -14,12 +14,15 @@ 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) @@ -27,6 +30,8 @@ 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]( diff --git a/tests/hosting_core/storage/test_transcript_logger_middleware.py b/tests/hosting_core/storage/test_transcript_logger_middleware.py index 30ed29286..744f0df1e 100644 --- a/tests/hosting_core/storage/test_transcript_logger_middleware.py +++ b/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" diff --git a/tests/hosting_core/test_middleware_set.py b/tests/hosting_core/test_middleware_set.py new file mode 100644 index 000000000..56282cdc1 --- /dev/null +++ b/tests/hosting_core/test_middleware_set.py @@ -0,0 +1,338 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from collections.abc import Awaitable, Callable +from typing import Any + +import pytest + +from microsoft_agents.activity import Activity, ActivityTypes +from microsoft_agents.hosting.core import TurnContext +from microsoft_agents.hosting.core.middleware_set import MiddlewareSet + + +def create_turn_context(text: str) -> TurnContext: + return TurnContext(object(), Activity(type=ActivityTypes.message, text=text)) + + +def context_text(context: TurnContext) -> str: + return context.activity.text + + +class RecordingMiddleware: + def __init__( + self, + name: str, + events: list[str], + *, + call_next: bool = True, + next_context: Any | None = None, + result: Any = None, + ): + self.name = name + self.events = events + self.call_next = call_next + self.next_context = next_context + self.result = result + + async def on_turn( + self, context: TurnContext, logic: Callable[[TurnContext], Awaitable[Any]] + ): + self.events.append(f"{self.name}:before:{context_text(context)}") + + if not self.call_next: + self.events.append(f"{self.name}:short-circuit") + return self.result + + downstream_result = await logic( + self.next_context if self.next_context is not None else context + ) + self.events.append(f"{self.name}:after:{context_text(context)}") + return downstream_result + + +@pytest.mark.asyncio +async def test_use_registers_all_middleware_and_returns_self(): + events: list[str] = [] + middleware_set = MiddlewareSet() + first = RecordingMiddleware("first", events) + second = RecordingMiddleware("second", events) + context = create_turn_context("context") + + result = middleware_set.use(first, second) + + assert result is middleware_set + + async def callback(context: TurnContext): + events.append(f"callback:{context_text(context)}") + + await middleware_set.on_turn(context, callback) + + assert events == [ + "first:before:context", + "second:before:context", + "second:after:context", + "first:after:context", + "callback:context", + ] + + +def test_use_rejects_invalid_middleware(): + middleware_set = MiddlewareSet() + + with pytest.raises(TypeError) as error: + middleware_set.use(object()) + + assert ( + str(error.value) + == 'MiddlewareSet.use(): invalid middleware at index "0" being added.' + ) + + +@pytest.mark.asyncio +async def test_on_turn_does_not_return_logic_result(): + events: list[str] = [] + context = create_turn_context("context") + middleware_set = MiddlewareSet().use(RecordingMiddleware("middleware", events)) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + return "logic-result" + + result = await middleware_set.on_turn(context, logic) + + assert result is None + assert events == [ + "middleware:before:context", + "middleware:after:context", + "logic:context", + ] + + +@pytest.mark.asyncio +async def test_middleware_can_short_circuit_pipeline(): + events: list[str] = [] + context = create_turn_context("context") + middleware_set = MiddlewareSet().use( + RecordingMiddleware( + "first", events, call_next=False, result="short-circuit-result" + ), + RecordingMiddleware("second", events), + ) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + return "logic-result" + + result = await middleware_set.on_turn(context, logic) + + assert result is None + assert events == [ + "first:before:context", + "first:short-circuit", + "logic:context", + ] + + +@pytest.mark.asyncio +async def test_middleware_can_pass_updated_context_to_next_step(): + events: list[str] = [] + original_context = create_turn_context("original-context") + updated_context = create_turn_context("updated-context") + middleware_set = MiddlewareSet().use( + RecordingMiddleware("first", events, next_context=updated_context), + RecordingMiddleware("second", events), + ) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + + await middleware_set.on_turn(original_context, logic) + + assert events == [ + "first:before:original-context", + "second:before:updated-context", + "second:after:updated-context", + "first:after:original-context", + "logic:original-context", + ] + + +@pytest.mark.asyncio +async def test_on_turn_runs_middleware_before_noop_logic(): + events: list[str] = [] + context = create_turn_context("context") + middleware_set = MiddlewareSet().use(RecordingMiddleware("middleware", events)) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + + result = await middleware_set.on_turn(context, logic) + + assert result is None + assert events == [ + "middleware:before:context", + "middleware:after:context", + "logic:context", + ] + + +@pytest.mark.asyncio +async def test_on_turn_runs_logic_after_middleware_pipeline_unwinds(): + events: list[str] = [] + context = create_turn_context("context") + middleware_set = MiddlewareSet().use( + RecordingMiddleware("first", events), + RecordingMiddleware("second", events), + ) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + return "logic-result" + + result = await middleware_set.on_turn(context, logic) + + assert result is None + assert events == [ + "first:before:context", + "second:before:context", + "second:after:context", + "first:after:context", + "logic:context", + ] + + +@pytest.mark.asyncio +async def test_middleware_set_can_be_nested_as_middleware(): + events: list[str] = [] + context = create_turn_context("context") + inner_set = MiddlewareSet().use(RecordingMiddleware("inner", events)) + outer_set = MiddlewareSet().use( + RecordingMiddleware("outer-before", events), + inner_set, + RecordingMiddleware("outer-after", events), + ) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + return "logic-result" + + result = await outer_set.on_turn(context, logic) + + assert result is None + assert events == [ + "outer-before:before:context", + "inner:before:context", + "inner:after:context", + "outer-after:before:context", + "outer-after:after:context", + "outer-before:after:context", + "logic:context", + ] + + +@pytest.mark.asyncio +async def test_receive_activity_with_status_runs_logic_inside_middleware_pipeline(): + events: list[str] = [] + context = create_turn_context("context") + middleware_set = MiddlewareSet().use( + RecordingMiddleware("first", events), + RecordingMiddleware("second", events), + ) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + return "logic-result" + + result = await middleware_set.receive_activity_with_status(context, logic) + + assert result is None + assert events == [ + "first:before:context", + "second:before:context", + "logic:context", + "second:after:context", + "first:after:context", + ] + + +@pytest.mark.asyncio +async def test_receive_activity_with_status_honors_short_circuit(): + events: list[str] = [] + context = create_turn_context("context") + middleware_set = MiddlewareSet().use( + RecordingMiddleware( + "first", events, call_next=False, result="short-circuit-result" + ), + RecordingMiddleware("second", events), + ) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + return "logic-result" + + result = await middleware_set.receive_activity_with_status(context, logic) + + assert result is None + assert events == [ + "first:before:context", + "first:short-circuit", + ] + + +@pytest.mark.asyncio +async def test_receive_activity_with_status_can_pass_updated_context_to_logic(): + events: list[str] = [] + original_context = create_turn_context("original-context") + updated_context = create_turn_context("updated-context") + middleware_set = MiddlewareSet().use( + RecordingMiddleware("first", events, next_context=updated_context), + RecordingMiddleware("second", events), + ) + + async def logic(context: TurnContext): + events.append(f"logic:{context_text(context)}") + + result = await middleware_set.receive_activity_with_status(original_context, logic) + + assert result is None + assert events == [ + "first:before:original-context", + "second:before:updated-context", + "logic:updated-context", + "second:after:updated-context", + "first:after:original-context", + ] + + +@pytest.mark.asyncio +async def test_receive_activity_with_status_runs_without_final_logic(): + events: list[str] = [] + context = create_turn_context("context") + middleware_set = MiddlewareSet().use(RecordingMiddleware("middleware", events)) + + result = await middleware_set.receive_activity_with_status(context, None) + + assert result is None + assert events == [ + "middleware:before:context", + "middleware:after:context", + ] + + +@pytest.mark.asyncio +async def test_on_turn_propagates_middleware_exceptions(): + class FailingMiddleware: + async def on_turn( + self, context: TurnContext, logic: Callable[[TurnContext], Awaitable[Any]] + ): + raise RuntimeError("middleware failed") + + context = create_turn_context("context") + middleware_set = MiddlewareSet().use(FailingMiddleware()) + + async def logic(context: TurnContext): + return "logic-result" + + with pytest.raises(RuntimeError, match="middleware failed"): + await middleware_set.on_turn(context, logic)