diff --git a/changelog.md b/changelog.md index 45e71dbaf..000d3f7c5 100644 --- a/changelog.md +++ b/changelog.md @@ -7,6 +7,7 @@ - **Attachment Downloaders**: Added `AttachmentDownloader` and `M365AttachmentDownloader` for downloading standard, Microsoft Teams, and Microsoft 365 Copilot attachments into `TurnState.temp.input_files`. - **Teams API Dependency**: Updated `microsoft-agents-hosting-msteams` to use `microsoft-teams-api` 2.1.0 +- **Teams API Client Performance**: Reused a cached SSL context when constructing per-turn Teams API clients, reducing repeated CA certificate loading and client initialization overhead ## Samples 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 6499eefbe..716713834 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 @@ -114,7 +114,11 @@ async def continue_conversation( :rtype: typing.Any """ context = TurnContext(self, reference.get_continuation_activity()) - return await self.run_pipeline(context, callback) + try: + res = await self.run_pipeline(context, callback) + finally: + await context._aclose() + return res async def continue_conversation_with_claims( self, @@ -220,7 +224,11 @@ async def create_conversation( context.turn_state["ConversationParameters"] = conversation_parameters # Process the activity through the middleware pipeline - return await self.run_pipeline(context, callback) + try: + res = await self.run_pipeline(context, callback) + finally: + await context._aclose() + return res async def run_pipeline( self, diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py index cd57a5c82..4e11c86ec 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py @@ -4,6 +4,7 @@ from __future__ import annotations from abc import ABC +from contextlib import AsyncExitStack from http import HTTPStatus from typing import Awaitable, Callable, Optional, cast from uuid import uuid4 @@ -308,11 +309,12 @@ async def create_conversation( # pylint: disable=arguments-differ user_token_client # for back-compat ) - # Run the pipeline - await self.run_pipeline(context, callback) - - await connector_client.close() - await user_token_client.close() + # Run the pipeline and release resources in reverse registration order. + async with AsyncExitStack() as cleanup: + cleanup.push_async_callback(context._aclose) + cleanup.push_async_callback(user_token_client.close) + cleanup.push_async_callback(connector_client.close) + await self.run_pipeline(context, callback) async def process_proactive( self, @@ -356,11 +358,12 @@ async def process_proactive( connector_client # for back-compat ) - # Run the pipeline - await self.run_pipeline(context, callback) - - await connector_client.close() - await user_token_client.close() + # Run the pipeline and release resources in reverse registration order. + async with AsyncExitStack() as cleanup: + cleanup.push_async_callback(context._aclose) + cleanup.push_async_callback(user_token_client.close) + cleanup.push_async_callback(connector_client.close) + await self.run_pipeline(context, callback) def _resolve_if_connector_client_is_needed(self, activity: Activity) -> bool: """Determine if a connector client is needed based on the activity's delivery mode and service URL. @@ -449,11 +452,12 @@ async def process_activity( connector_client # for back-compat ) - await self.run_pipeline(context, callback) - - if connector_client: - await connector_client.close() - await user_token_client.close() + async with AsyncExitStack() as cleanup: + cleanup.push_async_callback(context._aclose) + cleanup.push_async_callback(user_token_client.close) + if connector_client: + cleanup.push_async_callback(connector_client.close) + await self.run_pipeline(context, callback) # If there are any results they will have been left on the TurnContext. return self._process_turn_results(context) diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/turn_context.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/turn_context.py index f98d9d3bd..db43341bd 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/turn_context.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/turn_context.py @@ -51,6 +51,7 @@ class TurnContext(TurnContextProtocol): _on_send_activities: list[OnSendActivitiesHandler] _on_update_activity: list[OnUpdateActivityHandler] _on_delete_activity: list[OnDeleteActivityHandler] + _on_aclose_list: list[Callable[["TurnContext"], Awaitable[None]]] def __init__( self, @@ -77,6 +78,7 @@ def __init__( self._on_delete_activity = [] self._responded: bool = False self._identity = identity + self._on_aclose_list = [] if self.adapter is None: raise TypeError("TurnContext must be instantiated with an adapter.") @@ -109,6 +111,7 @@ def copy_to(self, context: "TurnContext") -> None: "_on_send_activities", "_on_update_activity", "_on_delete_activity", + "_on_aclose_list", ]: setattr(context, attribute, getattr(self, attribute)) @@ -416,3 +419,21 @@ def get_mentions(activity: Activity) -> list[Mention]: :return: A list of Mention objects representing all mentions in the activity. """ return activity.get_mentions() + + async def _aclose(self) -> None: + """ + Asynchronously close any resources associated with the turn context. + + :return: None + """ + for handler in self._on_aclose_list: + await handler(self) + + def _on_aclose(self, handler: Callable[[TurnContext], Awaitable[None]]) -> None: + """ + Register a handler to be called when the turn context is asynchronously closed. + + :param handler: A callable that takes the turn context and returns an awaitable. + :return: None + """ + self._on_aclose_list.append(handler) diff --git a/libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/_teams_api_client.py b/libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/_teams_api_client.py index dbbd0c7c8..ea6cdc115 100644 --- a/libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/_teams_api_client.py +++ b/libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/_teams_api_client.py @@ -8,7 +8,14 @@ available. """ -from microsoft_teams.common import ClientOptions +import ssl +import certifi + +from typing import Callable, Awaitable + +import httpx + +from microsoft_teams.common import Client, ClientOptions from microsoft_teams.api import ApiClient from microsoft_agents.hosting.core import ( @@ -16,6 +23,55 @@ TurnContext, ) +_ssl_context: ssl.SSLContext | None = None + + +def _get_ssl_context() -> ssl.SSLContext: + """Get or create the SSL context for verifying HTTPS requests. + + httpx by default creates a new SSL context for each new client instance. This can be + inefficient. For example, at the time of writing this, without caching the SSL context, + the hosting_msteams unit tests took 32 seconds to complete. With caching, the total + time for the unit tests dropped to 7 seconds. For an agent with lots of traffic, + caching the SSL context may significantly improve performance and avoid extra io operations. + """ + global _ssl_context + + if _ssl_context is None: + _ssl_context = ssl.create_default_context(cafile=certifi.where()) + return _ssl_context + + +def _client( + base_url: str, + headers: dict, + token_factory: Callable[[], Awaitable[str]] | None = None, +) -> Client: + """ + Create a new Client instance configured with the given base URL, headers, and token factory. + + :param base_url: The base URL for the client. + :param headers: The headers to include in the client requests. + :param token_factory: A callable that returns an access token asynchronously, or None if no token is required. + :return: A configured Client instance. + """ + + options = ClientOptions( + base_url=base_url, + headers=headers, + token=token_factory, + ) + + return Client( + options, + _http=httpx.AsyncClient( + base_url=base_url, + headers=headers, + timeout=options.timeout, + verify=_get_ssl_context(), + ), + ) + def _get_teams_api_client(context: TurnContext) -> ApiClient: """ @@ -49,28 +105,28 @@ def _set_teams_api_client( "Content-Type": "application/json", } - options: ClientOptions + token_factory: Callable[[], Awaitable[str]] | None = None if context.identity: provider = connection_manager.get_token_provider( context.identity, context.activity.service_url ) - async def token_factory() -> str: + async def _token_factory() -> str: return await provider.get_access_token( "https://api.botframework.com", ["https://api.botframework.com/.default"], ) - options = ClientOptions( - base_url=context.activity.service_url, headers=headers, token=token_factory - ) - else: - options = ClientOptions(base_url=context.activity.service_url, headers=headers) + token_factory = _token_factory api_client = ApiClient( context.activity.service_url, - options, + options=_client( + base_url=context.activity.service_url, + headers=headers, + token_factory=token_factory, + ), ) context.services.set(ApiClient, api_client) diff --git a/libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/teams_agent_extension.py b/libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/teams_agent_extension.py index 0b6cfd0e5..8d48690ce 100644 --- a/libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/teams_agent_extension.py +++ b/libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/teams_agent_extension.py @@ -123,9 +123,17 @@ def __init__(self, app: AgentApplication[StateT]) -> None: def _configure_app(self): """Configure the underlying AgentApplication with Teams-specific routes.""" + async def on_close_turn(context: TurnContext) -> None: + if context.activity.channel_id == Channels.ms_teams: + api_client = context.services.get(ApiClient) + if api_client is not None: + await api_client.http.http.aclose() + async def on_before_turn(context: TurnContext, state: StateT) -> bool: if context.activity.channel_id == Channels.ms_teams: _set_teams_api_client(context, self._app.connection_manager) + context._on_aclose(on_close_turn) + # caches the deserialized version of ChannelData context.activity.channel_data = _try_get_channel_data(context.activity) return True diff --git a/scripts/dev_setup.ps1 b/scripts/dev_setup.ps1 index 2b3f8d35d..5d473ec3c 100644 --- a/scripts/dev_setup.ps1 +++ b/scripts/dev_setup.ps1 @@ -9,6 +9,7 @@ pip install -e ./libraries/microsoft-agents-copilotstudio-client/ --config-setti pip install -e ./libraries/microsoft-agents-hosting-aiohttp/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-hosting-core/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-hosting-teams/ --config-settings editable_mode=compat +pip install -e ./libraries/microsoft-agents-hosting-msteams/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-hosting-dialogs/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-storage-blob/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-storage-cosmos/ --config-settings editable_mode=compat diff --git a/scripts/dev_setup.sh b/scripts/dev_setup.sh index b013fced4..eeb075f84 100644 --- a/scripts/dev_setup.sh +++ b/scripts/dev_setup.sh @@ -9,6 +9,7 @@ pip install -e ./libraries/microsoft-agents-copilotstudio-client/ --config-setti pip install -e ./libraries/microsoft-agents-hosting-aiohttp/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-hosting-core/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-hosting-teams/ --config-settings editable_mode=compat +pip install -e ./libraries/microsoft-agents-hosting-msteams/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-hosting-dialogs/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-storage-blob/ --config-settings editable_mode=compat pip install -e ./libraries/microsoft-agents-storage-cosmos/ --config-settings editable_mode=compat diff --git a/tests/hosting_core/test_channel_service_adapter.py b/tests/hosting_core/test_channel_service_adapter.py index 4dd678f6f..85097efb4 100644 --- a/tests/hosting_core/test_channel_service_adapter.py +++ b/tests/hosting_core/test_channel_service_adapter.py @@ -263,6 +263,38 @@ async def callback(context: TurnContext): assert context_arg.services.get(UserTokenClientBase) is user_token_client assert context_arg.services.get(ConnectorClientBase) is connector_client + @pytest.mark.asyncio + async def test_process_proactive_attempts_all_cleanup_when_connector_close_fails( + self, mocker, user_token_client, connector_client, adapter + ): + connector_error = RuntimeError("connector close failed") + connector_client.close = mocker.AsyncMock(side_effect=connector_error) + user_token_client.close = mocker.AsyncMock() + context_cleanup = mocker.AsyncMock() + + async def run_pipeline(context, callback): + context._on_aclose(context_cleanup) + + adapter.run_pipeline = mocker.AsyncMock(side_effect=run_pipeline) + activity = Activity( + type="message", + conversation={"id": "conversation123"}, + channel_id="channel_id", + service_url="service_url", + ) + + with pytest.raises(RuntimeError, match="connector close failed"): + await adapter.process_proactive( + ClaimsIdentity(), + activity, + "audience", + mocker.AsyncMock(), + ) + + connector_client.close.assert_awaited_once_with() + user_token_client.close.assert_awaited_once_with() + context_cleanup.assert_awaited_once() + @pytest.mark.asyncio async def test_process_proactive_uses_anonymous_clients(self, mocker): factory = mocker.Mock(spec=ChannelServiceClientFactoryBase) diff --git a/tests/hosting_msteams/helpers.py b/tests/hosting_msteams/helpers.py index 9bcd83c49..3708d08c0 100644 --- a/tests/hosting_msteams/helpers.py +++ b/tests/hosting_msteams/helpers.py @@ -92,7 +92,8 @@ def _make_context( def _cache_teams_api_client(context: TurnContext) -> None: - context.services.set(ApiClient, object.__new__(ApiClient)) + pass + # context.services.set(ApiClient, object.__new__(ApiClient)) def _make_teams_context() -> "TeamsTurnContext": diff --git a/tests/hosting_msteams/test_teams_agent_extension.py b/tests/hosting_msteams/test_teams_agent_extension.py index c8da5a822..df47a80d3 100644 --- a/tests/hosting_msteams/test_teams_agent_extension.py +++ b/tests/hosting_msteams/test_teams_agent_extension.py @@ -35,6 +35,9 @@ def __init__(self): def has(self, key): return key in self._state + def get(self, key): + return self._state.get(key) + def set(self, key, value): self._state[key] = value @@ -46,6 +49,14 @@ def __init__(self, activity, identity=None): self.activity = activity self.identity = identity self.services = _FakeServiceSet() + self.on_aclose_handlers = [] + + def _on_aclose(self, handler): + self.on_aclose_handlers.append(handler) + + async def _aclose(self): + for handler in self.on_aclose_handlers: + await handler(self) class TestTeamsAgentExtensionProperties: @@ -109,9 +120,10 @@ async def test_non_teams_channel_is_untouched(self): # channel_data left as the raw dict; no Teams API client cached assert activity.channel_data == {"channel": {"id": "c"}} assert not ctx.services.has(ApiClient) + assert ctx.on_aclose_handlers == [] @pytest.mark.asyncio - async def test_teams_channel_deserializes_channel_data(self): + async def test_teams_channel_deserializes_channel_data(self, mocker): activity = Activity( type="conversationUpdate", channel_id=Channels.ms_teams, @@ -126,9 +138,16 @@ async def test_teams_channel_deserializes_channel_data(self): assert isinstance(activity.channel_data, ChannelData) assert activity.channel_data.channel.id == "c1" assert ctx.services.has(ApiClient) + assert len(ctx.on_aclose_handlers) == 1 + async_client = ctx.services.get(ApiClient).http.http + close_spy = mocker.spy(async_client, "aclose") + + await ctx._aclose() + + close_spy.assert_awaited_once_with() @pytest.mark.asyncio - async def test_teams_channel_without_channel_data_sets_none(self): + async def test_teams_channel_without_channel_data_sets_none(self, mocker): activity = Activity( type="conversationUpdate", channel_id=Channels.ms_teams, @@ -140,3 +159,10 @@ async def test_teams_channel_without_channel_data_sets_none(self): assert result is True assert activity.channel_data is None + assert len(ctx.on_aclose_handlers) == 1 + async_client = ctx.services.get(ApiClient).http.http + close_spy = mocker.spy(async_client, "aclose") + + await ctx._aclose() + + close_spy.assert_awaited_once_with()