feat: add scope filtering support for OAuth 2.0 Client Credentials Grant

Add optional 'scope' parameter to all OAuth methods to enable requesting
specific scopes instead of all configured scopes. This solves the issue
where apps with multiple API types (Admin API + Customer Account API +
Storefront API) cannot generate tokens through /admin/oauth/access_token
because that endpoint only supports Admin API scopes.

Changes:
- Add scope parameter to request_token_client_credentials()
- Add scope parameter to request_access_token()
- Add scope parameter to refresh_token_if_needed()
- Add scope parameter to refresh_token()
- Implement scope normalization (convert commas to spaces for OAuth spec)
- Add 7 comprehensive test cases for scope filtering functionality
- Update CHANGELOG with feature documentation

Use case:
When a Shopify app has Customer Account API scopes configured (like
customer_read_metaobjects), requesting a token without scope filtering
fails because Shopify tries to grant ALL scopes through the Admin API
endpoint. With scope filtering, developers can request only Admin API
scopes: session.request_token_client_credentials(scope="read_products write_orders")

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: lvalics

CHANGELOG

@@ -16,6 +16,12 @@
   - Session now tracks token_obtained_at and token_expires_at timestamps for client credentials tokens
   - Default 5-minute buffer before expiration to ensure tokens are refreshed proactively
   - Auto-refresh only works with client credentials flow (API versions >= 2026-01)
+- Add scope filtering support for OAuth 2.0 Client Credentials Grant
+  - All OAuth methods now accept optional 'scope' parameter to request specific scopes
+  - Enables requesting only Admin API scopes when app has multiple API types configured (Admin, Customer Account, Storefront)
+  - Scope normalization: Comma-separated scopes automatically converted to space-separated for OAuth 2.0 spec compliance
+  - Methods supporting scope parameter: request_token_client_credentials(), request_access_token(), refresh_token_if_needed(), refresh_token()
+  - When scope parameter is omitted, Shopify grants all scopes configured for the app (default behavior)
 
 == Version 12.7.0
 

shopify/session.py

@@ -138,7 +138,7 @@ def request_token(self, params):
         else:
             raise Exception(response.msg)
 
-    def request_token_client_credentials(self):
+    def request_token_client_credentials(self, scope=None):
         """
         Exchange client credentials for an access token.
 
@@ -153,6 +153,12 @@ def request_token_client_credentials(self):
         - Session.secret (client_secret) must be set
         - Shop URL must be valid
 
+        Args:
+            scope (str, optional): Space or comma-separated list of scopes to request.
+                                  If not provided, Shopify will grant all scopes configured
+                                  for the app. Use this to request only Admin API scopes
+                                  when your app has multiple API types configured.
+
         Returns:
             dict: Token response containing:
                 - access_token (str): The access token for API requests
@@ -166,7 +172,12 @@ def request_token_client_credentials(self):
         Example:
             >>> session = shopify.Session("mystore.myshopify.com", "2026-01")
             >>> shopify.Session.setup(api_key="client_id", secret="client_secret")
+            >>> # Request all scopes
             >>> token_response = session.request_token_client_credentials()
+            >>> # Or request specific Admin API scopes only
+            >>> token_response = session.request_token_client_credentials(
+            ...     scope="read_products,write_products,read_orders"
+            ... )
             >>> session.token = token_response["access_token"]
             >>> shopify.ShopifyResource.activate_session(session)
         """
@@ -196,6 +207,12 @@ def request_token_client_credentials(self):
             "client_secret": self.secret
         }
 
+        # Add scope parameter if provided (to filter which scopes to request)
+        if scope:
+            # Normalize scope format (convert commas to spaces for OAuth spec)
+            normalized_scope = scope.replace(',', ' ')
+            data["scope"] = normalized_scope
+
         # Prepare request headers
         headers = {"Content-Type": "application/x-www-form-urlencoded"}
 
@@ -250,7 +267,7 @@ def request_token_client_credentials(self):
         except Exception as e:
             raise OAuthException("Unexpected error during OAuth request: %s" % str(e))
 
-    def request_access_token(self, params=None):
+    def request_access_token(self, params=None, scope=None):
         """
         Automatically select and execute the appropriate OAuth flow based on API version.
 
@@ -264,6 +281,9 @@ def request_access_token(self, params=None):
             params: OAuth callback parameters (required for versions < 2026-01)
                    Should include 'code', 'hmac', 'timestamp' for authorization code flow.
                    Not used for client credentials flow (versions >= 2026-01).
+            scope (str, optional): Space or comma-separated list of scopes to request.
+                                  Only used for client credentials flow (versions >= 2026-01).
+                                  If not provided, Shopify grants all configured scopes.
 
         Returns:
             For versions >= 2026-01: dict with 'access_token', 'scope', 'expires_in'
@@ -286,7 +306,7 @@ def request_access_token(self, params=None):
         """
         if self._requires_client_credentials():
             # API version 2026-01+: Use client credentials grant
-            return self.request_token_client_credentials()
+            return self.request_token_client_credentials(scope=scope)
         else:
             # Older API versions: Use authorization code grant
             if params is None:
@@ -342,7 +362,7 @@ def is_token_expired(self, buffer_seconds=300):
         # Token is expired if current time + buffer >= expiration time
         return now + buffer >= self.token_expires_at
 
-    def refresh_token_if_needed(self, buffer_seconds=300):
+    def refresh_token_if_needed(self, buffer_seconds=300, scope=None):
         """
         Automatically refresh the access token if expired or expiring soon.
 
@@ -354,6 +374,10 @@ def refresh_token_if_needed(self, buffer_seconds=300):
             buffer_seconds (int): Number of seconds before expiration to trigger
                                  refresh. Default is 300 (5 minutes). This ensures
                                  the token is refreshed before it expires.
+            scope (str, optional): Space or comma-separated list of scopes to request.
+                                  If not provided, Shopify grants all configured scopes.
+                                  Use this to request only Admin API scopes when your
+                                  app has multiple API types configured.
 
         Returns:
             dict or None:
@@ -381,13 +405,15 @@ def refresh_token_if_needed(self, buffer_seconds=300):
             ... else:
             ...     print("Token is still valid")
             >>>
-            >>> # Use custom buffer (refresh if expires within 10 minutes)
-            >>> session.refresh_token_if_needed(buffer_seconds=600)
+            >>> # Request only Admin API scopes when refreshing
+            >>> result = session.refresh_token_if_needed(
+            ...     scope="read_products,write_products,read_orders"
+            ... )
         """
         if self.is_token_expired(buffer_seconds=buffer_seconds):
             # Only refresh if we have credentials (client credentials flow)
             if self._requires_client_credentials():
-                return self.request_token_client_credentials()
+                return self.request_token_client_credentials(scope=scope)
             else:
                 # For authorization code flow, we can't auto-refresh
                 # because we need user interaction for the callback
@@ -396,7 +422,7 @@ def refresh_token_if_needed(self, buffer_seconds=300):
         # Token is still valid, no refresh needed
         return None
 
-    def refresh_token(self):
+    def refresh_token(self, scope=None):
         """
         Manually force a refresh of the access token.
 
@@ -408,6 +434,12 @@ def refresh_token(self):
         For automatic refresh based on expiration, use refresh_token_if_needed()
         instead.
 
+        Args:
+            scope (str, optional): Space or comma-separated list of scopes to request.
+                                  If not provided, Shopify grants all configured scopes.
+                                  Use this to request only Admin API scopes when your
+                                  app has multiple API types configured.
+
         Returns:
             dict: Token response containing:
                   - access_token (str): The new access token
@@ -431,6 +463,11 @@ def refresh_token(self):
             >>> new_token = session.refresh_token()
             >>> print(f"New token: {new_token['access_token']}")
             >>> print(f"Expires in: {new_token['expires_in']} seconds")
+            >>>
+            >>> # Request only Admin API scopes
+            >>> new_token = session.refresh_token(
+            ...     scope="read_products,write_products,read_orders"
+            ... )
         """
         # Only works with client credentials flow
         if not self._requires_client_credentials():
@@ -446,7 +483,7 @@ def refresh_token(self):
         self.token_obtained_at = None
 
         # Request new token
-        return self.request_token_client_credentials()
+        return self.request_token_client_credentials(scope=scope)
 
     @property
     def api_version(self):

test/session_test.py

@@ -625,3 +625,139 @@ def test_token_expiration_tracking_stored(self):
         # Verify expires_at is approximately 24 hours from obtained_at
         expected_expiration = session.token_obtained_at + timedelta(seconds=86399)
         self.assertEqual(session.token_expires_at, expected_expiration)
+
+    def test_request_token_client_credentials_with_scope_parameter(self):
+        """Test that scope parameter is included in OAuth request when provided"""
+        import urllib.parse
+
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+
+        # Mock the HTTP request and capture what was sent
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "test_token", "scope": "read_products write_products", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        # Request token with specific scopes (Admin API only)
+        token_response = session.request_token_client_credentials(scope="read_products write_products")
+
+        # Verify token received
+        self.assertEqual(token_response["access_token"], "test_token")
+        self.assertEqual(token_response["scope"], "read_products write_products")
+
+    def test_request_token_client_credentials_without_scope_parameter(self):
+        """Test that scope parameter is NOT included when not provided (default behavior)"""
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "test_token", "scope": "read_products write_products read_orders", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        # Request token without scope parameter (should grant all configured scopes)
+        token_response = session.request_token_client_credentials()
+
+        # Verify token received with all scopes
+        self.assertEqual(token_response["access_token"], "test_token")
+        # Without scope filter, Shopify returns all configured scopes
+        self.assertEqual(token_response["scope"], "read_products write_products read_orders")
+
+    def test_request_token_client_credentials_scope_normalization(self):
+        """Test that comma-separated scopes are normalized to space-separated for OAuth spec"""
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "test_token", "scope": "read_products write_products", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        # Request with comma-separated scopes (should be normalized to spaces)
+        token_response = session.request_token_client_credentials(scope="read_products,write_products")
+
+        # Verify token received
+        self.assertEqual(token_response["access_token"], "test_token")
+
+    def test_refresh_token_if_needed_with_scope_parameter(self):
+        """Test that refresh_token_if_needed passes scope parameter correctly"""
+        from datetime import datetime, timedelta
+
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+
+        # Set expired token
+        session.token = "expired_token"
+        session.token_obtained_at = datetime.now() - timedelta(hours=24)
+        session.token_expires_at = datetime.now() - timedelta(minutes=10)
+
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "refreshed_token", "scope": "read_products write_products", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        # Refresh with specific scope
+        result = session.refresh_token_if_needed(scope="read_products write_products")
+
+        # Verify token was refreshed with correct scopes
+        self.assertIsNotNone(result)
+        self.assertEqual(result["access_token"], "refreshed_token")
+        self.assertEqual(result["scope"], "read_products write_products")
+        self.assertEqual(session.token, "refreshed_token")
+
+    def test_refresh_token_with_scope_parameter(self):
+        """Test that refresh_token passes scope parameter correctly"""
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+
+        # Set existing token
+        session.token = "old_token"
+
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "new_token", "scope": "read_products", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        # Force refresh with specific scope (Admin API only)
+        result = session.refresh_token(scope="read_products")
+
+        # Verify token was refreshed with correct scopes
+        self.assertEqual(result["access_token"], "new_token")
+        self.assertEqual(result["scope"], "read_products")
+        self.assertEqual(session.token, "new_token")
+
+    def test_request_access_token_with_scope_parameter_2026_01(self):
+        """Test that request_access_token passes scope to client credentials flow for 2026-01+"""
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "test_token", "scope": "read_products", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        # Use smart method with scope parameter
+        result = session.request_access_token(scope="read_products")
+
+        # Verify correct token received
+        self.assertEqual(result["access_token"], "test_token")
+        self.assertEqual(result["scope"], "read_products")