fix: route app.get_app_graph + ctx.*_graph to sovereign endpoints via cloud.graph_scope

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

@@ -146,6 +146,7 @@ def __init__(self, **options: Unpack[AppOptions]):
             self._token_manager,
             self.options.api_client_settings,
             self.activity_sender,
+            self.cloud,
         )
         self.event_manager = EventManager(self._events)
         self.activity_processor.event_manager = self.event_manager
@@ -608,4 +609,4 @@ def get_app_graph(self, tenant_id: Optional[str] = None) -> "GraphServiceClient"
             ImportError: If the graph dependencies are not installed.
 
         """
-        return create_graph_client(lambda: self._get_graph_token(tenant_id))
+        return create_graph_client(lambda: self._get_graph_token(tenant_id), cloud=self.cloud)

packages/apps/src/microsoft_teams/apps/app_process.py

@@ -18,6 +18,7 @@
     TokenProtocol,
     is_invoke_response,
 )
+from microsoft_teams.api.auth.cloud_environment import PUBLIC, CloudEnvironment
 from microsoft_teams.api.clients.user.params import GetUserTokenParams
 from microsoft_teams.cards import AdaptiveCard
 from microsoft_teams.common import Client, ClientOptions, LocalStorage, Storage
@@ -49,6 +50,7 @@ def __init__(
         token_manager: TokenManager,
         api_client_settings: Optional[ApiClientSettings],
         activity_sender: ActivitySender,
+        cloud: CloudEnvironment = PUBLIC,
     ) -> None:
         self.router = router
         self.id = id
@@ -58,6 +60,7 @@ def __init__(
         self.token_manager = token_manager
         self.api_client_settings = api_client_settings
         self.activity_sender = activity_sender
+        self.cloud = cloud
 
         # This will be set after the EventManager is initialized due to
         # a circular dependency
@@ -126,6 +129,7 @@ async def _build_context(
             self.default_connection_name,
             activity_sender=self.activity_sender,
             app_token=lambda: self.token_manager.get_graph_token(tenant_id),
+            cloud=self.cloud,
         )
 
         send = activityCtx.send

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

@@ -27,6 +27,7 @@
     TokenExchangeState,
     TokenPostResource,
 )
+from microsoft_teams.api.auth.cloud_environment import PUBLIC, CloudEnvironment
 from microsoft_teams.api.models.attachment.card_attachment import (
     OAuthCardAttachment,
     card_attachment,
@@ -83,6 +84,7 @@ def __init__(
         connection_name: str,
         activity_sender: ActivitySender,
         app_token: Token,
+        cloud: CloudEnvironment = PUBLIC,
     ):
         self.activity = activity
         self.app_id = app_id
@@ -93,6 +95,7 @@ def __init__(
         self.user_token = user_token
         self.connection_name = connection_name
         self.is_signed_in = is_signed_in
+        self.cloud = cloud
         self._activity_sender = activity_sender
         self._app_token = app_token
         self.stream = activity_sender.create_stream(conversation_ref)
@@ -123,7 +126,7 @@ def user_graph(self) -> "GraphServiceClient":
         if self._user_graph is None:
             try:
                 user_token = JsonWebToken(self.user_token)
-                self._user_graph = create_graph_client(user_token)
+                self._user_graph = create_graph_client(user_token, cloud=self.cloud)
             except ImportError:
                 raise
             except Exception as e:
@@ -147,7 +150,7 @@ def app_graph(self) -> "GraphServiceClient":
         """
         if self._app_graph is None:
             try:
-                self._app_graph = create_graph_client(self._app_token)
+                self._app_graph = create_graph_client(self._app_token, cloud=self.cloud)
             except ImportError:
                 raise
             except Exception as e:

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

@@ -3,15 +3,51 @@
 Licensed under the MIT License.
 """
 
+import logging
+import re
+from typing import Optional
+
+from microsoft_teams.api.auth.cloud_environment import CloudEnvironment
 from microsoft_teams.common.http.client_token import Token
 
+logger = logging.getLogger(__name__)
+
+# Extracts scheme + host (+ optional port) from a URL-like scope such as
+# "https://graph.microsoft.us/.default" -> "https://graph.microsoft.us".
+_GRAPH_BASE_URL_RE = re.compile(r"^(https?://[^/]+)", re.IGNORECASE)
+
+
+def _derive_graph_base_url(cloud: Optional[CloudEnvironment]) -> Optional[str]:
+    """Derive the Graph API base URL from a cloud's graph_scope, or None if unavailable."""
+    if cloud is None:
+        return None
+    scope = (cloud.graph_scope or "").strip()
+    if not scope:
+        return None
+    match = _GRAPH_BASE_URL_RE.match(scope)
+    if match is None:
+        logger.warning(
+            "graph_scope %r is not a URL; Graph calls will route to the public cloud. "
+            "Set graph_scope to an 'https://<host>/.default' value to route to the correct Graph endpoint.",
+            scope,
+        )
+        return None
+    return match.group(1)
+
+
+def create_graph_client(token: Token, cloud: Optional[CloudEnvironment] = None):
+    """Lazy import and create a Graph client with the given token.
 
-def create_graph_client(token: Token):
-    """Lazy import and create a Graph client with the given token."""
+    Args:
+        token: The token used to authenticate Graph requests.
+        cloud: Optional cloud environment. When provided (and non-None), the Graph client
+            routes HTTP calls to the cloud's Graph endpoint derived from ``graph_scope``.
+            When ``None``, the public Graph endpoint is used.
+    """
     try:
         from microsoft_teams.graph import get_graph_client
 
-        return get_graph_client(token)
+        return get_graph_client(token, base_url=_derive_graph_base_url(cloud))
     except ImportError as exc:
         raise ImportError(
             "Graph functionality not available. Install with 'pip install microsoft-teams-apps[graph]'"

packages/apps/tests/test_activity_context.py

@@ -10,6 +10,7 @@
 
 import pytest
 from microsoft_teams.api import Account, MessageActivityInput, SentActivity
+from microsoft_teams.api.auth.cloud_environment import PUBLIC
 from microsoft_teams.apps.routing.activity_context import ActivityContext
 
 
@@ -40,6 +41,7 @@ def _create_activity_context(
         connection_name="test-connection",
         activity_sender=mock_activity_sender,
         app_token=MagicMock(),
+        cloud=PUBLIC,
     )
     return ctx, mock_activity_sender
 

packages/apps/tests/test_app_oauth.py

@@ -16,6 +16,7 @@
     SignInVerifyStateInvokeActivity,
     TokenExchangeInvokeResponse,
 )
+from microsoft_teams.api.auth.cloud_environment import PUBLIC
 from microsoft_teams.api.models import (
     Account,
     ConversationAccount,
@@ -446,6 +447,7 @@ def processor(self, router):
             token_manager=MagicMock(),
             api_client_settings=None,
             activity_sender=MagicMock(),
+            cloud=PUBLIC,
         )
 
     @staticmethod
@@ -462,6 +464,7 @@ def _make_ctx(activity):
             connection_name="graph",
             activity_sender=MagicMock(),
             app_token=MagicMock(),
+            cloud=PUBLIC,
         )
 
     @pytest.mark.asyncio

packages/apps/tests/test_app_process.py

@@ -15,6 +15,7 @@
     InvokeResponse,
     TokenProtocol,
 )
+from microsoft_teams.api.auth.cloud_environment import PUBLIC
 from microsoft_teams.apps import ActivityContext, ActivityEvent
 from microsoft_teams.apps.activity_sender import ActivitySender
 from microsoft_teams.apps.app_events import EventManager
@@ -56,6 +57,7 @@ def activity_processor(self, mock_http_client):
             mock_token_manager,
             None,
             mock_activity_sender,
+            PUBLIC,
         )
 
     @pytest.mark.asyncio
@@ -83,6 +85,7 @@ async def test_execute_middleware_chain_with_two_handlers(self, activity_process
             connection_name="default_connection",
             activity_sender=mock_activity_sender,
             app_token=lambda: None,
+            cloud=PUBLIC,
         )
 
         handler_one = AsyncMock(spec=ActivityHandler)

packages/apps/tests/test_optional_graph_dependencies.py

@@ -8,7 +8,51 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from microsoft_teams.api.auth.cloud_environment import (
+    CHINA,
+    PUBLIC,
+    US_GOV,
+    US_GOV_DOD,
+    CloudEnvironment,
+)
 from microsoft_teams.apps.routing.activity_context import ActivityContext
+from microsoft_teams.apps.utils.graph import _derive_graph_base_url
+
+
+class TestDeriveGraphBaseUrl:
+    """Tests for the cloud -> Graph base URL derivation used by create_graph_client."""
+
+    def test_none_cloud_returns_none(self) -> None:
+        assert _derive_graph_base_url(None) is None
+
+    @pytest.mark.parametrize(
+        "cloud,expected",
+        [
+            (PUBLIC, "https://graph.microsoft.com"),
+            (US_GOV, "https://graph.microsoft.us"),
+            (US_GOV_DOD, "https://dod-graph.microsoft.us"),
+            (CHINA, "https://microsoftgraph.chinacloudapi.cn"),
+        ],
+    )
+    def test_preset_cloud_derives_base_url(self, cloud: CloudEnvironment, expected: str) -> None:
+        assert _derive_graph_base_url(cloud) == expected
+
+    def test_non_url_scope_returns_none(self, caplog: pytest.LogCaptureFixture) -> None:
+        # Construct a cloud whose graph_scope isn't a URL.
+        class _FakeCloud:
+            graph_scope = "user.read"
+
+        with caplog.at_level("WARNING"):
+            assert _derive_graph_base_url(_FakeCloud()) is None  # type: ignore[arg-type]
+        assert any("not a URL" in record.message for record in caplog.records)
+
+    def test_empty_scope_returns_none_no_warning(self, caplog: pytest.LogCaptureFixture) -> None:
+        class _FakeCloud:
+            graph_scope = ""
+
+        with caplog.at_level("WARNING"):
+            assert _derive_graph_base_url(_FakeCloud()) is None  # type: ignore[arg-type]
+        assert not any("not a URL" in record.message for record in caplog.records)
 
 
 class TestOptionalGraphDependencies:
@@ -36,6 +80,7 @@ def _create_activity_context(self) -> ActivityContext[Any]:
             connection_name="test-connection",
             activity_sender=mock_activity_sender,
             app_token=mock_app_token,  # This is needed for app_graph to work
+            cloud=PUBLIC,
         )
 
     def test_app_graph_property_without_graph_available(self) -> None:
@@ -61,7 +106,7 @@ def mock_import(name: str, *args: Any, **kwargs: Any) -> Any:
             if name == "microsoft_teams.graph":
                 # Create a mock module with get_graph_client
                 mock_module = SimpleNamespace()
-                mock_module.get_graph_client = lambda x: "MockGraphClient"  # type: ignore
+                mock_module.get_graph_client = lambda x, base_url=None: "MockGraphClient"  # type: ignore
                 return mock_module
             return __import__(name, *args, **kwargs)
 
@@ -83,6 +128,7 @@ def test_user_graph_property_not_signed_in(self) -> None:
             connection_name="test-connection",
             activity_sender=MagicMock(),
             app_token=None,
+            cloud=PUBLIC,
         )
 
         # user_graph should raise ValueError when user is not signed in
@@ -102,6 +148,7 @@ def test_user_graph_property_no_token(self) -> None:
             connection_name="test-connection",
             activity_sender=MagicMock(),
             app_token=None,
+            cloud=PUBLIC,
         )
 
         # user_graph should raise ValueError when no user token is available
@@ -121,6 +168,7 @@ def test_app_graph_property_no_token(self) -> None:
             connection_name="test-connection",
             activity_sender=MagicMock(),
             app_token=None,  # No app token
+            cloud=PUBLIC,
         )
 
         # app_graph should raise RuntimeError when no app token is available
@@ -173,7 +221,7 @@ def test_get_app_graph_passes_tenant_id(self) -> None:
         mock_client = MagicMock()
         captured_token_arg = []
 
-        def capture_token(token):
+        def capture_token(token, cloud=None):
             captured_token_arg.append(token)
             return mock_client
 

packages/graph/src/microsoft_teams/graph/graph.py

@@ -6,19 +6,30 @@
 from typing import Optional
 
 from azure.core.exceptions import ClientAuthenticationError
+from kiota_authentication_azure.azure_identity_authentication_provider import (
+    AzureIdentityAuthenticationProvider,
+)
 from microsoft_teams.common.http.client_token import Token
+from msgraph.graph_request_adapter import GraphRequestAdapter
 from msgraph.graph_service_client import GraphServiceClient
 
 from .auth_provider import AuthProvider
 
 
-def get_graph_client(token: Optional[Token] = None) -> GraphServiceClient:
+def get_graph_client(
+    token: Optional[Token] = None,
+    base_url: Optional[str] = None,
+) -> GraphServiceClient:
     """
     Get a configured Microsoft Graph client using a Token.
 
     Args:
         token: Token data (string, StringLike, callable, or None). If None,
                will raise ClientAuthenticationError with a clear message.
+        base_url: Optional Graph API base URL override for sovereign clouds
+               (e.g. "https://graph.microsoft.us" for GCCH). When provided,
+               the client routes HTTP calls to this endpoint + "/v1.0/". When
+               None, the public Graph endpoint is used.
 
     Returns:
         GraphServiceClient: A configured client ready for Microsoft Graph API calls
@@ -28,20 +39,11 @@ def get_graph_client(token: Optional[Token] = None) -> GraphServiceClient:
 
     Example:
         ```python
-        # Using a string token
-        graph = get_graph_client("eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIs...")
+        # Public cloud (default)
+        graph = get_graph_client("eyJ0eXAiOiJKV1Qi...")
 
-
-        # Using a callable that returns a string
-        def get_token():
-            return "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIs..."
-
-
-        graph = get_graph_client(get_token)
-
-        # Make Graph API calls
-        me = await graph.me.get()
-        messages = await graph.me.messages.get()
+        # Sovereign cloud (GCCH)
+        graph = get_graph_client("eyJ0eXAiOiJKV1Qi...", base_url="https://graph.microsoft.us")
         ```
     """
     try:
@@ -53,8 +55,17 @@ def get_token():
             )
 
         credential = AuthProvider(token)
-        client = GraphServiceClient(credentials=credential)
-        return client
+
+        if base_url is None:
+            return GraphServiceClient(credentials=credential)
+
+        # Build a custom request adapter with the sovereign base URL.
+        # Normalize: strip any trailing slash on caller's input, then append "/v1.0/"
+        # to match the msgraph-sdk default shape ("https://graph.microsoft.com/v1.0/").
+        auth_provider = AzureIdentityAuthenticationProvider(credential)
+        adapter = GraphRequestAdapter(auth_provider)
+        adapter.base_url = f"{base_url.rstrip('/')}/v1.0/"
+        return GraphServiceClient(request_adapter=adapter)
 
     except Exception as e:
         if isinstance(e, ClientAuthenticationError):