feat: add App.get_app_graph() and App._get_graph_token()

packages/apps/src/microsoft_teams/apps/app.py

@@ -7,7 +7,7 @@
 import importlib.metadata
 import logging
 import os
-from typing import Any, Awaitable, Callable, List, Optional, TypeVar, Union, Unpack, cast, overload
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Optional, TypeVar, Union, Unpack, cast, overload
 
 from dependency_injector import providers
 from dotenv import find_dotenv, load_dotenv
@@ -28,6 +28,9 @@
 from microsoft_teams.cards import AdaptiveCard
 from microsoft_teams.common import Client, ClientOptions, EventEmitter, LocalStorage
 
+if TYPE_CHECKING:
+    from msgraph.graph_service_client import GraphServiceClient
+
 from .activity_sender import ActivitySender
 from .app_events import EventManager
 from .app_oauth import OauthHandlers
@@ -52,6 +55,7 @@
 from .routing import ActivityHandlerMixin, ActivityRouter
 from .routing.activity_context import ActivityContext
 from .token_manager import TokenManager
+from .utils import create_graph_client
 
 version = importlib.metadata.version("microsoft-teams-apps")
 
@@ -519,3 +523,22 @@ async def _stop_plugins(self) -> None:
 
     async def _get_bot_token(self):
         return await self._token_manager.get_bot_token()
+
+    async def _get_graph_token(self, tenant_id: Optional[str] = None):
+        return await self._token_manager.get_graph_token(tenant_id)
+
+    def get_app_graph(self, tenant_id: Optional[str] = None) -> "GraphServiceClient":
+        """
+        Get a Microsoft Graph client configured with the app's token.
+
+        This client can be used for app-only operations that don't require user context.
+        For multi-tenant apps, pass a tenant_id to get a tenant-specific token.
+
+        Args:
+            tenant_id: Optional tenant ID. If not provided, uses the app's default tenant.
+
+        Raises:
+            ImportError: If the graph dependencies are not installed.
+
+        """
+        return create_graph_client(lambda: self._get_graph_token(tenant_id))

packages/apps/src/microsoft_teams/apps/routing/activity_context.py

@@ -37,6 +37,7 @@
 from microsoft_teams.common.http.client_token import Token
 
 from ..activity_sender import ActivitySender
+from ..utils import create_graph_client
 
 if TYPE_CHECKING:
     from msgraph.graph_service_client import GraphServiceClient
@@ -45,18 +46,6 @@
 logger = logging.getLogger(__name__)
 
 
-def _get_graph_client(token: Token):
-    """Lazy import and call get_graph_client when needed."""
-    try:
-        from microsoft_teams.graph import get_graph_client
-
-        return get_graph_client(token)
-    except ImportError as exc:
-        raise ImportError(
-            "Graph functionality not available. Install with 'pip install microsoft-teams-apps[graph]'"
-        ) from exc
-
-
 @dataclass
 class SignInOptions:
     """Options for the signin method."""
@@ -134,7 +123,7 @@ def user_graph(self) -> "GraphServiceClient":
         if self._user_graph is None:
             try:
                 user_token = JsonWebToken(self.user_token)
-                self._user_graph = _get_graph_client(user_token)
+                self._user_graph = create_graph_client(user_token)
             except Exception as e:
                 self.logger.error(f"Failed to create user graph client: {e}")
                 raise RuntimeError(f"Failed to create user graph client: {e}") from e
@@ -156,7 +145,7 @@ def app_graph(self) -> "GraphServiceClient":
         """
         if self._app_graph is None:
             try:
-                self._app_graph = _get_graph_client(self._app_token)
+                self._app_graph = create_graph_client(self._app_token)
             except Exception as e:
                 self.logger.error(f"Failed to create app graph client: {e}")
                 raise RuntimeError(f"Failed to create app graph client: {e}") from e

packages/apps/src/microsoft_teams/apps/utils/__init__.py

@@ -4,6 +4,7 @@
 """
 
 from .activity_utils import extract_tenant_id
+from .graph import create_graph_client
 from .retry import RetryOptions, retry
 
-__all__ = ["extract_tenant_id", "retry", "RetryOptions"]
+__all__ = ["create_graph_client", "extract_tenant_id", "retry", "RetryOptions"]

packages/apps/src/microsoft_teams/apps/utils/graph.py

@@ -0,0 +1,18 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from microsoft_teams.common.http.client_token import Token
+
+
+def create_graph_client(token: Token):
+    """Lazy import and create a Graph client with the given token."""
+    try:
+        from microsoft_teams.graph import get_graph_client
+
+        return get_graph_client(token)
+    except ImportError as exc:
+        raise ImportError(
+            "Graph functionality not available. Install with 'pip install microsoft-teams-apps[graph]'"
+        ) from exc

packages/apps/tests/test_activity_context.py

@@ -302,11 +302,11 @@ def test_user_graph_raises_when_no_user_token(self) -> None:
             _ = ctx.user_graph
 
     def test_user_graph_raises_runtime_error_when_graph_import_fails(self) -> None:
-        """user_graph raises RuntimeError when _get_graph_client raises ImportError."""
+        """user_graph raises RuntimeError when create_graph_client raises ImportError."""
         ctx, _ = _create_activity_context(is_signed_in=True, user_token="some.jwt.token")
 
         with patch(
-            "microsoft_teams.apps.routing.activity_context._get_graph_client",
+            "microsoft_teams.apps.routing.activity_context.create_graph_client",
             side_effect=ImportError("graph not installed"),
         ):
             with pytest.raises(RuntimeError, match="Failed to create user graph client"):
@@ -317,11 +317,11 @@ class TestActivityContextAppGraph:
     """Tests for ActivityContext.app_graph property."""
 
     def test_app_graph_raises_runtime_error_when_graph_import_fails(self) -> None:
-        """app_graph raises RuntimeError when _get_graph_client raises ImportError."""
+        """app_graph raises RuntimeError when create_graph_client raises ImportError."""
         ctx, _ = _create_activity_context()
 
         with patch(
-            "microsoft_teams.apps.routing.activity_context._get_graph_client",
+            "microsoft_teams.apps.routing.activity_context.create_graph_client",
             side_effect=ImportError("graph not installed"),
         ):
             with pytest.raises(RuntimeError, match="Failed to create app graph client"):
@@ -333,14 +333,14 @@ def test_app_graph_returns_cached_client_on_second_access(self) -> None:
         ctx, _ = _create_activity_context()
 
         with patch(
-            "microsoft_teams.apps.routing.activity_context._get_graph_client",
+            "microsoft_teams.apps.routing.activity_context.create_graph_client",
             return_value=mock_graph_client,
         ):
             first = ctx.app_graph
             second = ctx.app_graph
 
         assert first is second
-        # _get_graph_client should only have been called once (caching)
+        # create_graph_client should only have been called once (caching)
         assert ctx._app_graph is mock_graph_client
 
 

packages/apps/tests/test_optional_graph_dependencies.py

@@ -5,7 +5,7 @@
 
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from microsoft_teams.apps.routing.activity_context import ActivityContext
@@ -126,3 +126,70 @@ def test_app_graph_property_no_token(self) -> None:
         # app_graph should raise ValueError when no app token is available
         with pytest.raises(RuntimeError, match="Token cannot be None"):
             _ = activity_context.app_graph
+
+
+class TestAppGetAppGraph:
+    """Test App.get_app_graph method."""
+
+    def _create_app(self):
+        from microsoft_teams.apps import App, AppOptions
+
+        return App(**AppOptions(client_id="test-id", client_secret="test-secret"))
+
+    def test_get_app_graph_raises_import_error_when_graph_not_installed(self) -> None:
+        """get_app_graph raises ImportError when graph dependencies are not available."""
+        app = self._create_app()
+
+        with patch(
+            "microsoft_teams.apps.app.create_graph_client",
+            side_effect=ImportError("graph not installed"),
+        ):
+            with pytest.raises(ImportError):
+                _ = app.get_app_graph()
+
+    def test_get_app_graph_returns_new_client_each_call(self) -> None:
+        """get_app_graph returns a new client on every call (no caching)."""
+        app = self._create_app()
+
+        mock_client_1 = MagicMock()
+        mock_client_2 = MagicMock()
+        side_effects = [mock_client_1, mock_client_2]
+
+        with patch(
+            "microsoft_teams.apps.app.create_graph_client",
+            side_effect=side_effects,
+        ):
+            first = app.get_app_graph()
+            second = app.get_app_graph()
+
+        assert first is mock_client_1
+        assert second is mock_client_2
+        assert first is not second
+
+    def test_get_app_graph_passes_tenant_id(self) -> None:
+        """get_app_graph passes the tenant_id through to the token factory callable."""
+        app = self._create_app()
+
+        mock_client = MagicMock()
+        captured_token_arg = []
+
+        def capture_token(token):
+            captured_token_arg.append(token)
+            return mock_client
+
+        with patch(
+            "microsoft_teams.apps.app.create_graph_client",
+            side_effect=capture_token,
+        ):
+            app.get_app_graph(tenant_id="my-tenant-id")
+
+        assert len(captured_token_arg) == 1
+        # token arg should be a callable (lambda)
+        assert callable(captured_token_arg[0])
+
+        # Verify the lambda invokes _get_graph_token with the correct tenant_id
+        with patch.object(app, "_get_graph_token", new=AsyncMock(return_value=None)) as mock_get_token:
+            import asyncio
+
+            asyncio.run(captured_token_arg[0]())
+            mock_get_token.assert_called_once_with("my-tenant-id")