Native agentic token for UserManagedIdentity via fmi_path exchange

libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py

@@ -388,10 +388,73 @@ async def get_agentic_application_token(
 
             return None
 
+        if (
+            self._msal_configuration.AUTH_TYPE == AuthTypes.user_managed_identity
+            and isinstance(msal_auth_client, ManagedIdentityClient)
+        ):
+            return await self._acquire_agentic_token_via_managed_identity(
+                agent_app_instance_id
+            )
+
         raise RuntimeError(
-            "Agentic token acquisition supports ConfidentialClientApplication, or ManagedIdentityClient when AUTH_TYPE is AuthTypes.identity_proxy_manager."
+            "Agentic token acquisition supports ConfidentialClientApplication, or ManagedIdentityClient "
+            "when AUTH_TYPE is AuthTypes.identity_proxy_manager or AuthTypes.user_managed_identity."
+        )
+
+    async def _acquire_agentic_token_via_managed_identity(
+        self, agent_app_instance_id: str
+    ) -> Optional[str]:
+        """Acquire an agentic application token for a user-assigned managed identity.
+
+        MSAL's ``ManagedIdentityClient`` does not support the federated ``fmi_path``
+        token exchange required for agentic application tokens. This uses
+        azure-identity's ``DefaultAzureCredential``, which accepts
+        ``identity_config={"fmi_path": ...}`` and handles the managed-identity
+        (IMDS / App Service) endpoints, to perform the federated exchange.
+
+        :param agent_app_instance_id: The agent application instance ID, used as the
+            federated identity ``fmi_path``.
+        :type agent_app_instance_id: str
+        :return: The agentic application token, or None if acquisition failed.
+        :rtype: Optional[str]
+        """
+        try:
+            from azure.identity.aio import DefaultAzureCredential
+        except ImportError as exc:
+            raise ImportError(
+                "Acquiring an agentic application token for AuthTypes.user_managed_identity "
+                "requires the 'azure-identity' package. Install it with: pip install azure-identity."
+            ) from exc
+
+        credential_kwargs: dict = {
+            "identity_config": {"fmi_path": agent_app_instance_id},
+        }
+        client_id = getattr(self._msal_configuration, "CLIENT_ID", None)
+        if client_id:
+            credential_kwargs["managed_identity_client_id"] = client_id
+
+        logger.info(
+            "Acquiring agentic application token via DefaultAzureCredential for agent_app_instance_id=%s",
+            agent_app_instance_id,
         )
 
+        credential = DefaultAzureCredential(**credential_kwargs)
+        try:
+            token = await credential.get_token("api://AzureAdTokenExchange/.default")
+            return token.token
+        except Exception:  # noqa: BLE001 - failures are logged and surfaced as None
+            logger.exception(
+                "Failed to acquire agentic application token via DefaultAzureCredential "
+                "for agent_app_instance_id=%s",
+                agent_app_instance_id,
+            )
+            return None
+        finally:
+            try:
+                await credential.close()
+            except Exception:  # noqa: BLE001 - best-effort cleanup
+                logger.debug("Error closing DefaultAzureCredential", exc_info=True)
+
     async def get_agentic_instance_token(
         self, tenant_id: str, agent_app_instance_id: str
     ) -> tuple[str, str]:

tests/authentication_msal/test_msal_auth.py

@@ -335,12 +335,62 @@ async def test_get_agentic_application_token_identity_proxy_manager(self, mocker
             resource="https://custom-resource/.default"
         )
 
+    @pytest.mark.asyncio
+    async def test_get_agentic_application_token_user_managed_identity(self, mocker):
+        """UserManagedIdentity acquires the agentic token via DefaultAzureCredential
+        using the federated ``fmi_path`` exchange (no monkey-patch required)."""
+        import sys
+        import types
+
+        config = AgentAuthConfiguration(
+            auth_type=AuthTypes.user_managed_identity,
+            client_id="test-client-id",
+        )
+        auth = MsalAuth(config)
+
+        mock_client = mocker.Mock(spec=ManagedIdentityClient)
+        mocker.patch.object(auth, "_get_client", return_value=mock_client)
+
+        # Inject a fake azure.identity.aio.DefaultAzureCredential so the test does
+        # not require azure-identity to be installed.
+        captured = {}
+
+        class _FakeToken:
+            token = "umi-fmi-token"
+
+        class _FakeCredential:
+            def __init__(self, **kwargs):
+                captured["kwargs"] = kwargs
+
+            async def get_token(self, scope):
+                captured["scope"] = scope
+                return _FakeToken()
+
+            async def close(self):
+                captured["closed"] = True
+
+        fake_module = types.ModuleType("azure.identity.aio")
+        fake_module.DefaultAzureCredential = _FakeCredential
+        mocker.patch.dict(sys.modules, {"azure.identity.aio": fake_module})
+
+        token = await auth.get_agentic_application_token(
+            "test-tenant-id", "test-agent-app-instance-id"
+        )
+
+        assert token == "umi-fmi-token"
+        assert captured["kwargs"]["identity_config"] == {
+            "fmi_path": "test-agent-app-instance-id"
+        }
+        assert captured["kwargs"]["managed_identity_client_id"] == "test-client-id"
+        assert captured["scope"] == "api://AzureAdTokenExchange/.default"
+        assert captured.get("closed") is True
+
     @pytest.mark.asyncio
     async def test_get_agentic_application_token_unsupported_client_raises(
         self, mocker
     ):
         config = AgentAuthConfiguration(
-            auth_type=AuthTypes.user_managed_identity,
+            auth_type=AuthTypes.system_managed_identity,
             client_id="test-client-id",
         )
         auth = MsalAuth(config)