Add async routine

stytch/consumer/api/idp.py

@@ -37,7 +37,21 @@ def introspect_idp_access_token(
         ) or self.introspect_idp_access_token_network(
             access_token, client_id, client_secret, token_type_hint
         )
-
+
+    async def introspect_idp_access_token_async(
+        self,
+        access_token: str,
+        client_id: str,
+        client_secret: Optional[str] = None,
+        token_type_hint: str = "access_token",
+    ) -> Optional[AccessTokenJWTClaims]:
+        local_introspection_response = self.introspect_idp_access_token_local(access_token, client_id)
+        if local_introspection_response is not None:
+            return local_introspection_response
+        return self.introspect_idp_access_token_network_async(
+            access_token, client_id, client_secret, token_type_hint
+        )
+
     def introspect_idp_access_token_network(
         self,
         access_token: str,
@@ -73,6 +87,42 @@ def introspect_idp_access_token_network(
             issuer=jwtResponse.iss,
             not_before=jwtResponse.nbf,
         )
+
+    async def introspect_idp_access_token_network_async(
+        self,
+        access_token: str,
+        client_id: str,
+        client_secret: Optional[str] = None,
+        token_type_hint: str = "access_token",
+    ) -> Optional[AccessTokenJWTClaims]:
+        headers: Dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
+        data: Dict[str, Any] = {
+            "token": access_token,
+            "client_id": client_id,
+            "token_type_hint": token_type_hint,
+        }
+        if client_secret is not None:
+            data["client_secret"] = client_secret
+
+        url = self.api_base.url_for(
+            f"/v1/public/{self.project_id}/oauth2/introspect", data
+        )
+        res = await self.async_client.postForm(url, data, headers)
+        jwtResponse = AccessTokenJWTResponse.from_json(
+            res.response.status, res.json
+        )
+        if not jwtResponse.active:
+            return None
+        return AccessTokenJWTClaims(
+            subject=jwtResponse.sub,
+            scope=jwtResponse.scope,
+            custom_claims={},
+            audience=jwtResponse.aud,
+            expires_at=jwtResponse.exp,
+            issued_at=jwtResponse.iat,
+            issuer=jwtResponse.iss,
+            not_before=jwtResponse.nbf,
+        )
 
     def introspect_idp_access_token_local(
         self,

stytch/core/http/client.py

@@ -8,6 +8,8 @@
 import requests
 import requests.auth
 
+import json
+
 from stytch.version import __version__
 
 HEADERS = {
@@ -138,6 +140,16 @@ async def _response_from_request(
         except Exception:
             resp_json = {}
         return ResponseWithJson(response=r, json=resp_json)
+
+    async def _response_from_post_form_request(
+        cls, r: aiohttp.ClientResponse
+    ) -> ResponseWithJson:
+        try:
+            content = await r.content.read()
+            resp_json = json.loads(content.decode())
+        except Exception as e:
+            resp_json = {}
+        return ResponseWithJson(response=r, json=resp_json)
 
     async def get(
         self,
@@ -164,6 +176,19 @@ async def post(
             url, json=json, headers=final_headers, auth=self.auth
         )
         return await self._response_from_request(resp)
+
+    async def postForm(
+        self,
+        url: str,
+        form: Optional[Dict[str, Any]],
+        headers: Optional[Dict[str, str]] = None,
+    ) -> ResponseWithJson:
+        final_headers = self.headers.copy()
+        final_headers.update(headers or {})
+        resp = await self._session.post(
+            url, data=form, headers=final_headers, auth=self.auth
+        )
+        return await self._response_from_post_form_request(resp)
 
     async def put(
         self,