get_token_provider_from_activity added to Connections and ConnectionManager

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/connection_manager.py

@@ -6,6 +6,8 @@
 
 from collections.abc import Callable
 
+from microsoft_agents.activity import Activity
+
 from .agent_auth_configuration import AgentAuthConfiguration
 from .access_token_provider_base import AccessTokenProviderBase
 from .claims_identity import ClaimsIdentity
@@ -190,6 +192,23 @@ def get_token_provider(
             f"No connection found for audience '{aud}' and serviceUrl '{service_url}'."
         )
 
+    def get_token_provider_from_activity(
+        self,
+        claims_identity: ClaimsIdentity,
+        activity: Activity,
+    ) -> AccessTokenProviderBase:
+        """
+        Get the OAuth token provider for the agent from an activity.
+
+        :param claims_identity: The claims identity of the agent.
+        :param activity: The activity from which to get the token provider.
+        :return: The access token provider for the agent.
+        """
+        provider = self.get_token_provider(claims_identity, activity.service_url)
+        if activity.is_agentic_request() and provider.configuration.ALT_BLUEPRINT_ID:
+            provider = self.get_connection(provider.configuration.ALT_BLUEPRINT_ID)
+        return provider
+
     def get_default_connection_configuration(self) -> AgentAuthConfiguration:
         """
         Get the default connection configuration for the agent.

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/connections.py

@@ -4,6 +4,8 @@
 from abc import abstractmethod
 from typing import Protocol
 
+from microsoft_agents.activity import Activity
+
 from .agent_auth_configuration import AgentAuthConfiguration
 from .access_token_provider_base import AccessTokenProviderBase
 from .claims_identity import ClaimsIdentity
@@ -43,6 +45,19 @@ def get_token_provider(
         """
         raise NotImplementedError()
 
+    @abstractmethod
+    def get_token_provider_from_activity(
+        self, claims_identity: ClaimsIdentity, activity: Activity
+    ) -> AccessTokenProviderBase:
+        """
+        Get the OAuth token provider for the agent from an activity.
+
+        :param claims_identity: The claims identity of the agent.
+        :param activity: The activity from which to get the token provider.
+        :return: The access token provider for the agent.
+        """
+        raise NotImplementedError()
+
     @abstractmethod
     def get_default_connection_configuration(self) -> AgentAuthConfiguration:
         """

libraries/microsoft-agents-hosting-msteams/microsoft_agents/hosting/msteams/_graph.py

@@ -165,8 +165,8 @@ def _common_get_app_graph_client(
     """
     if not context.identity:
         raise ValueError("TurnContext.identity is required to get a Graph client.")
-    token_provider = app.connection_manager.get_token_provider(
-        context.identity, context.activity.service_url
+    token_provider = app.connection_manager.get_token_provider_from_activity(
+        context.identity, context.activity
     )
     return _create_app_graph_service_client(token_provider, graph_base_url)
 

tests/_common/testing_objects/testing_connection_manager.py

@@ -1,3 +1,4 @@
+from microsoft_agents.activity import Activity
 from microsoft_agents.hosting.core import (
     Connections,
     AccessTokenProviderBase,
@@ -57,6 +58,17 @@ def get_token_provider(
         """
         return self.get_default_connection()
 
+    def get_token_provider_from_activity(
+        self, claims_identity: ClaimsIdentity, activity: Activity
+    ) -> AccessTokenProviderBase:
+        """
+        Get a token provider based on claims identity and activity.
+
+        In this test implementation, returns the default connection regardless
+        of the provided parameters.
+        """
+        return self.get_token_provider(claims_identity, activity.service_url)
+
     def get_default_connection_configuration(self) -> AgentAuthConfiguration:
         """
         Get the default authentication configuration.

tests/hosting_core/test_connection_manager.py

@@ -3,6 +3,7 @@
 
 import pytest
 
+from microsoft_agents.activity import Activity, ChannelAccount, RoleTypes
 from microsoft_agents.hosting.core import (
     AgentAuthConfiguration,
     AccessTokenProviderBase,
@@ -48,11 +49,37 @@ async def get_access_token(self, resource_url, scopes, force_refresh=False):
     ],
 }
 
+ALT_BLUEPRINT_CONFIG = {
+    "CONNECTIONS": {
+        "SERVICE_CONNECTION": {
+            "SETTINGS": {
+                "CLIENTID": "client-service",
+                "ALTERNATEBLUEPRINTCONNECTIONNAME": "ALT_BLUEPRINT",
+            }
+        },
+        "ALT_BLUEPRINT": {"SETTINGS": {"CLIENTID": "client-alt-blueprint"}},
+    }
+}
+
 
 class TestGenericConnectionManager:
     def _make(self, **kwargs):
         return ConnectionManager(provider_factory=FakeProvider, **kwargs)
 
+    def _activity(
+        self,
+        service_url: str = "https://service.url/",
+        recipient_role: RoleTypes | None = RoleTypes.agent,
+    ) -> Activity:
+        activity = {
+            "type": "message",
+            "channel_id": "msteams",
+            "service_url": service_url,
+        }
+        if recipient_role is not None:
+            activity["recipient"] = ChannelAccount(id="bot1", role=recipient_role)
+        return Activity(**activity)
+
     def test_uses_provider_factory(self):
         cm = self._make(**ENV_CONFIG)
         provider = cm.get_default_connection()
@@ -124,6 +151,64 @@ def test_token_provider_no_map_returns_default(self):
             is cm.get_default_connection()
         )
 
+    def test_token_provider_from_activity_uses_activity_service_url(self):
+        cm = self._make(**ENV_CONFIG)
+        claims = ClaimsIdentity(claims={}, is_authenticated=False)
+        activity = self._activity(service_url="https://host/agentic/path")
+
+        assert cm.get_token_provider_from_activity(
+            claims, activity
+        ) is cm.get_connection("AGENTIC")
+
+    def test_token_provider_from_activity_regular_role_ignores_alternate_blueprint(
+        self,
+    ):
+        cm = self._make(**ALT_BLUEPRINT_CONFIG)
+        claims = ClaimsIdentity(claims={}, is_authenticated=False)
+        activity = self._activity(recipient_role=RoleTypes.agent)
+
+        assert cm.get_token_provider_from_activity(
+            claims, activity
+        ) is cm.get_connection("SERVICE_CONNECTION")
+
+    @pytest.mark.parametrize(
+        "recipient_role",
+        [RoleTypes.agentic_identity, RoleTypes.agentic_user],
+    )
+    def test_token_provider_from_activity_agentic_role_uses_alternate_blueprint(
+        self, recipient_role
+    ):
+        cm = self._make(**ALT_BLUEPRINT_CONFIG)
+        claims = ClaimsIdentity(claims={}, is_authenticated=False)
+        activity = self._activity(recipient_role=recipient_role)
+
+        assert cm.get_token_provider_from_activity(
+            claims, activity
+        ) is cm.get_connection("ALT_BLUEPRINT")
+
+    def test_token_provider_from_activity_agentic_without_alternate_uses_mapped_provider(
+        self,
+    ):
+        cm = self._make(**ENV_CONFIG)
+        claims = ClaimsIdentity(claims={}, is_authenticated=False)
+        activity = self._activity(
+            service_url="https://host/agentic/path",
+            recipient_role=RoleTypes.agentic_identity,
+        )
+
+        assert cm.get_token_provider_from_activity(
+            claims, activity
+        ) is cm.get_connection("AGENTIC")
+
+    def test_token_provider_from_activity_without_recipient_is_not_agentic(self):
+        cm = self._make(**ALT_BLUEPRINT_CONFIG)
+        claims = ClaimsIdentity(claims={}, is_authenticated=False)
+        activity = self._activity(recipient_role=None)
+
+        assert cm.get_token_provider_from_activity(
+            claims, activity
+        ) is cm.get_connection("SERVICE_CONNECTION")
+
     @pytest.mark.parametrize(
         "claims, service_url",
         [

tests/hosting_msteams/test_graph_clients.py

@@ -55,8 +55,9 @@ def __init__(self):
         self.calls = []
 
     def get_token_provider(self, identity, service_url):
-        self.calls.append(("get_token_provider", identity, service_url))
-        return self.turn_provider
+        raise AssertionError(
+            "Graph clients should resolve context providers from the full activity"
+        )
 
     def get_connection(self, connection_name):
         self.calls.append(("get_connection", connection_name))
@@ -66,6 +67,10 @@ def get_default_connection(self):
         self.calls.append(("get_default_connection",))
         return self.default_provider
 
+    def get_token_provider_from_activity(self, identity, activity):
+        self.calls.append(("get_token_provider_from_activity", identity, activity))
+        return self.turn_provider
+
 
 @pytest.mark.asyncio
 async def test_delegated_graph_client_gets_token_from_authorization_handler():
@@ -119,7 +124,7 @@ async def test_app_graph_client_uses_default_scope_for_custom_graph_cloud():
     assert "authorization" in native_request.headers
 
 
-def test_context_app_graph_client_resolves_connection_from_turn_identity_and_service_url():
+def test_context_app_graph_client_resolves_connection_from_turn_identity_and_activity():
     connection_manager = _RecordingConnectionManager()
     app = SimpleNamespace(connection_manager=connection_manager)
     identity = SimpleNamespace()
@@ -133,9 +138,9 @@ def test_context_app_graph_client_resolves_connection_from_turn_identity_and_ser
     assert graph.request_adapter.base_url == "https://graph.microsoft.com/v1.0/"
     assert connection_manager.calls == [
         (
-            "get_token_provider",
+            "get_token_provider_from_activity",
             identity,
-            "https://smba.trafficmanager.net/teams/",
+            context.activity,
         )
     ]