feat(outbound): add token/api-key upload and connection-status list methods

README.md

@@ -1644,6 +1644,48 @@ latest_tenant_token = descope_client.mgmt.outbound_application.fetch_tenant_toke
     "tenant-id",
     {"forceRefresh": True}  # Optional
 )
+
+# List the IDs of the outbound apps a user currently holds a valid token for.
+# Use this for connection-status views instead of calling fetch_token once per app.
+connected = descope_client.mgmt.outbound_application.list_apps_with_user_token(
+    "user-id",
+    tenant_id="tenant-id",  # Optional
+)
+# connected => {"appIds": ["app-1", "app-2"]}
+
+# Store a static API key for a user / tenant on an apikey-type outbound app
+descope_client.mgmt.outbound_application.upload_user_api_key(
+    "my-app-id",
+    "user-id",
+    "the-users-api-key",
+    tenant_id="tenant-id",  # Optional
+)
+descope_client.mgmt.outbound_application.upload_tenant_api_key(
+    "my-app-id",
+    "tenant-id",
+    "the-tenants-api-key",
+)
+
+# Upload (migrate) an existing OAuth token for a user / tenant on an oauth-type outbound app,
+# without requiring the user to re-run the OAuth flow.
+descope_client.mgmt.outbound_application.upload_user_token(
+    "my-app-id",
+    "user-id",
+    refresh_token="the-refresh-token",
+    scopes=["read", "write"],
+)
+descope_client.mgmt.outbound_application.upload_tenant_token(
+    "my-app-id",
+    "tenant-id",
+    access_token="the-access-token",
+)
+
+# Batch upload OAuth tokens (all-or-nothing): inspect "failures" to see rejected items
+batch_resp = descope_client.mgmt.outbound_application.batch_upload_user_tokens([
+    {"appId": "my-app-id", "userId": "user-1", "accessToken": "token-1"},
+    {"appId": "my-app-id", "userId": "user-2", "accessToken": "token-2"},
+])
+# batch_resp => {"failures": [{"appId": ..., "userId": ..., "errorCode": ..., "reason": ...}, ...]}
 ```
 
 Fetch outbound application tokens using an inbound application token that includes the "outbound.token.fetch" scope (no management key required)

descope/management/_outbound_application_base.py

@@ -67,3 +67,37 @@
         if prompt is not None:
             body["prompt"] = [p.value for p in prompt]
         return body
+
+    @staticmethod
+    def _compose_oauth_upload_body(
+        base: dict,
+        refresh_token: Optional[str] = None,
+        access_token: Optional[str] = None,
+        access_token_expiry: Optional[int] = None,
+        access_token_type: Optional[str] = None,
+        scopes: Optional[List[str]] = None,
+        external_identifier: Optional[str] = None,
+        id_token: Optional[str] = None,
+        granted_by: Optional[str] = None,
+        verify_refresh: Optional[bool] = None,
+    ) -> dict:
+        body = dict(base)
+        if refresh_token is not None:
+            body["refreshToken"] = refresh_token
+        if access_token is not None:
+            body["accessToken"] = access_token
+        if access_token_expiry is not None:
+            body["accessTokenExpiry"] = access_token_expiry
+        if access_token_type is not None:
+            body["accessTokenType"] = access_token_type
+        if scopes is not None:
+            body["scopes"] = scopes
+        if external_identifier is not None:
+            body["externalIdentifier"] = external_identifier
+        if id_token is not None:
+            body["idToken"] = id_token
+        if granted_by is not None:
+            body["grantedBy"] = granted_by
+        if verify_refresh is not None:
+            body["verifyRefresh"] = verify_refresh
+        return body

descope/management/common.py

@@ -120,6 +120,13 @@ class MgmtV1:
     outbound_application_fetch_tenant_token_path = "/v1/mgmt/outbound/app/tenant/token/latest"
     outbound_application_delete_user_tokens_path = "/v1/mgmt/outbound/user/tokens"
     outbound_application_delete_token_path = "/v1/mgmt/outbound/token"
+    outbound_application_list_apps_with_user_token_path = "/v1/mgmt/outbound/apps-with-user-token"
+    outbound_application_upload_user_api_key_path = "/v1/mgmt/outbound/app/user/apikey/upload"
+    outbound_application_upload_tenant_api_key_path = "/v1/mgmt/outbound/app/tenant/apikey/upload"
+    outbound_application_upload_user_token_path = "/v1/mgmt/outbound/app/user/oauthtoken/upload"
+    outbound_application_upload_tenant_token_path = "/v1/mgmt/outbound/app/tenant/oauthtoken/upload"
+    outbound_application_batch_upload_user_tokens_path = "/v1/mgmt/outbound/app/user/oauthtoken/batch/upload"
+    outbound_application_batch_upload_tenant_tokens_path = "/v1/mgmt/outbound/app/tenant/oauthtoken/batch/upload"
 
     # user
     user_create_path = "/v1/mgmt/user/create"

descope/management/outbound_application.py

@@ -489,6 +489,236 @@
         uri = MgmtV1.outbound_application_delete_token_path
         self._http.delete(uri, params={"id": token_id})
 
+    def list_apps_with_user_token(
+        self,
+        user_id: str,
+        tenant_id: Optional[str] = None,
+    ) -> dict:
+        """
+        List the IDs of the outbound applications the given user currently holds a valid token for.
+        Use this for connection-status views instead of calling fetch_token once per application.
+
+        Args:
+        user_id (str): The ID of the user.
+        tenant_id (str): Optional tenant ID to scope the lookup to.
+
+        Return value (dict):
+        Return dict in the format
+             {"appIds": [<app_id>, ...]}
+
+        Raise:
+        AuthException: raised if the operation fails
+        """
+        params = {"userId": user_id}
+        if tenant_id:
+            params["tenantId"] = tenant_id
+        response = self._http.get(
+            MgmtV1.outbound_application_list_apps_with_user_token_path,
+            params=params,
+        )
+        return response.json()
+
+    def upload_user_api_key(
+        self,
+        app_id: str,
+        user_id: str,
+        api_key: str,
+        tenant_id: Optional[str] = None,
+    ) -> dict:
+        """
+        Upload (set) a static API key for a user on an apikey-type outbound application.
+
+        Args:
+        app_id (str): The ID of the outbound application.
+        user_id (str): The ID of the user.
+        api_key (str): The API key to store for the user.
+        tenant_id (str): Optional tenant ID, used to associate different keys for the
+            same user in different tenants.
+
+        Raise:
+        AuthException: raised if the operation fails
+        """
+        body: dict = {"appId": app_id, "userId": user_id, "apiKey": api_key}
+        if tenant_id is not None:
+            body["tenantId"] = tenant_id
+        response = self._http.post(
+            MgmtV1.outbound_application_upload_user_api_key_path,
+            body=body,
+        )
+        return response.json()
+
+    def upload_tenant_api_key(
+        self,
+        app_id: str,
+        tenant_id: str,
+        api_key: str,
+    ) -> dict:
+        """
+        Upload (set) a static API key for a tenant on an apikey-type outbound application.
+
+        Args:
+        app_id (str): The ID of the outbound application.
+        tenant_id (str): The ID of the tenant.
+        api_key (str): The API key to store for the tenant.
+
+        Raise:
+        AuthException: raised if the operation fails
+        """
+        response = self._http.post(
+            MgmtV1.outbound_application_upload_tenant_api_key_path,
+            body={"appId": app_id, "tenantId": tenant_id, "apiKey": api_key},
+        )
+        return response.json()
+
+    def upload_user_token(
+        self,
+        app_id: str,
+        user_id: str,
+        tenant_id: Optional[str] = None,
+        refresh_token: Optional[str] = None,
+        access_token: Optional[str] = None,
+        access_token_expiry: Optional[int] = None,
+        access_token_type: Optional[str] = None,
+        scopes: Optional[List[str]] = None,
+        external_identifier: Optional[str] = None,
+        id_token: Optional[str] = None,
+        verify_refresh: Optional[bool] = None,
+        granted_by: Optional[str] = None,
+    ) -> dict:
+        """
+        Upload (migrate) an existing OAuth token for a user on an oauth-type outbound application,
+        without requiring the user to re-run the OAuth flow. At least one of refresh_token or
+        access_token must be provided.
+
+        Args:
+        app_id (str): The ID of the outbound application.
+        user_id (str): The ID of the user.
+        tenant_id (str): Optional tenant ID.
+        refresh_token (str): Optional refresh token.
+        access_token (str): Optional access token.
+        access_token_expiry (int): Optional access token expiry, in epoch seconds.
+        access_token_type (str): Optional token type (defaults to "Bearer").
+        scopes (List[str]): Optional scopes the token was granted for.
+        external_identifier (str): Optional external identifier.
+        id_token (str): Optional id token.
+        verify_refresh (bool): If True, verify the refresh token against the provider before
+            persisting; nothing is written if verification fails.
+        granted_by (str): Optional attribution for who granted the token.
+
+        Raise:
+        AuthException: raised if the operation fails
+        """
+        base: dict = {"appId": app_id, "userId": user_id}
+        if tenant_id is not None:
+            base["tenantId"] = tenant_id
+        body = OutboundApplicationBase._compose_oauth_upload_body(
+            base,
+            refresh_token,
+            access_token,
+            access_token_expiry,
+            access_token_type,
+            scopes,
+            external_identifier,
+            id_token,
+            granted_by,
+            verify_refresh,
+        )
+        response = self._http.post(
+            MgmtV1.outbound_application_upload_user_token_path,
+            body=body,
+        )
+        return response.json()
+
+    def upload_tenant_token(
+        self,
+        app_id: str,
+        tenant_id: str,
+        refresh_token: Optional[str] = None,
+        access_token: Optional[str] = None,
+        access_token_expiry: Optional[int] = None,
+        access_token_type: Optional[str] = None,
+        scopes: Optional[List[str]] = None,
+        external_identifier: Optional[str] = None,
+        id_token: Optional[str] = None,
+        verify_refresh: Optional[bool] = None,
+        granted_by: Optional[str] = None,
+    ) -> dict:
+        """
+        Upload (migrate) an existing OAuth token for a tenant on an oauth-type outbound application.
+        At least one of refresh_token or access_token must be provided.
+
+        Args:
+        app_id (str): The ID of the outbound application.
+        tenant_id (str): The ID of the tenant.
+        (remaining args): see upload_user_token.
+
+        Raise:
+        AuthException: raised if the operation fails
+        """
+        base: dict = {"appId": app_id, "tenantId": tenant_id}
+        body = OutboundApplicationBase._compose_oauth_upload_body(
+            base,
+            refresh_token,
+            access_token,
+            access_token_expiry,
+            access_token_type,
+            scopes,
+            external_identifier,
+            id_token,
+            granted_by,
+            verify_refresh,
+        )
+        response = self._http.post(
+            MgmtV1.outbound_application_upload_tenant_token_path,
+            body=body,
+        )
+        return response.json()
+
+    def batch_upload_user_tokens(self, tokens: List[dict]) -> dict:
+        """
+        Batch upload (migrate) existing OAuth tokens for users. All-or-nothing: if any item fails
+        per-item validation, the returned failures are populated and no tokens are committed.
+
+        Args:
+        tokens (List[dict]): A list of token dicts, each with at least appId and userId plus one of
+            refreshToken / accessToken. See upload_user_token for the full set of fields
+            (verifyRefresh is not supported for batch uploads).
+
+        Return value (dict):
+        Return dict in the format
+             {"failures": [{"appId": <id>, "userId": <id>, "errorCode": <code>, "reason": <reason>}, ...]}
+
+        Raise:
+        AuthException: raised if the operation fails
+        """
+        response = self._http.post(
+            MgmtV1.outbound_application_batch_upload_user_tokens_path,
+            body={"tokens": tokens},
+        )
+        return response.json()
+
+    def batch_upload_tenant_tokens(self, tokens: List[dict]) -> dict:
+        """
+        Batch upload (migrate) existing OAuth tokens for tenants. All-or-nothing: if any item fails
+        per-item validation, the returned failures are populated and no tokens are committed.
+
+        Args:
+        tokens (List[dict]): A list of token dicts, each with at least appId and tenantId plus one of
+            refreshToken / accessToken.
+
+        Return value (dict):
+        Return dict in the format
+             {"failures": [{"appId": <id>, "tenantId": <id>, "errorCode": <code>, "reason": <reason>}, ...]}
+
+        Raise:
+        AuthException: raised if the operation fails
+        """
+        response = self._http.post(
+            MgmtV1.outbound_application_batch_upload_tenant_tokens_path,
+            body={"tokens": tokens},
+        )
+        return response.json()
+
 
 class OutboundApplicationByToken(HTTPBase):
     def __init__(self, http_client: HTTPClient):