feat(oauth): add configurable token_refresh_request_type for GET-based OAuth APIs

Add token_refresh_request_type property to OAuthAuthenticator supporting:
- body_data (default): POST with form-encoded body (standard OAuth2)
- body_json: POST with JSON body
- query_params: GET with query parameters (e.g., Marketo)

This unblocks Connector Builder users from building connectors for APIs
like Marketo that require GET requests to their OAuth token endpoint.

Changes:
- AbstractOauth2Authenticator: add get_token_refresh_request_type() and
  update _make_handled_request() to dispatch on request type
- DeclarativeOauth2Authenticator: thread through token_refresh_request_type
- declarative_component_schema.yaml: add token_refresh_request_type enum
- declarative_component_schema.py: add Pydantic field
- model_to_component_factory.py: wire through the new property

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/auth/oauth.py

@@ -72,6 +72,7 @@ class DeclarativeOauth2Authenticator(AbstractOauth2Authenticator, DeclarativeAut
     refresh_request_headers: Optional[Mapping[str, Any]] = None
     grant_type_name: Union[InterpolatedString, str] = "grant_type"
     grant_type: Union[InterpolatedString, str] = "refresh_token"
+    token_refresh_request_type: str = "body_data"
     message_repository: MessageRepository = NoopMessageRepository()
     profile_assertion: Optional[DeclarativeAuthenticator] = None
     use_profile_assertion: Optional[Union[InterpolatedBoolean, str, bool]] = False
@@ -247,6 +248,9 @@ def get_refresh_request_body(self) -> Mapping[str, Any]:
     def get_refresh_request_headers(self) -> Mapping[str, Any]:
         return self._refresh_request_headers.eval(self.config)
 
+    def get_token_refresh_request_type(self) -> str:
+        return self.token_refresh_request_type
+
     def get_token_expiry_date(self) -> AirbyteDateTime:
         if not self._has_access_token_been_initialized():
             return AirbyteDateTime.from_datetime(datetime.min)

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1400,6 +1400,15 @@ definitions:
         examples:
           - refresh_token
           - client_credentials
+      token_refresh_request_type:
+        title: Token Refresh Request Type
+        description: Configures how the token refresh request is sent. Use body_data (default) for POST with form-encoded body, body_json for POST with JSON body, or query_params for GET with query parameters (required by some APIs like Marketo).
+        type: string
+        default: "body_data"
+        enum:
+          - body_data
+          - body_json
+          - query_params
       refresh_request_body:
         title: Refresh Request Body
         description: Body of the request sent to get a new access token.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1871,6 +1871,12 @@ class OAuthAuthenticator(BaseModel):
         examples=["refresh_token", "client_credentials"],
         title="Grant Type",
     )
+    token_refresh_request_type: Optional[str] = Field(
+        "body_data",
+        description="Configures how the token refresh request is sent. Use body_data (default) for POST with form-encoded body, body_json for POST with JSON body, or query_params for GET with query parameters (required by some APIs like Marketo).",
+        enum=["body_data", "body_json", "query_params"],
+        title="Token Refresh Request Type",
+    )
     refresh_request_body: Optional[Dict[str, Any]] = Field(
         None,
         description="Body of the request sent to get a new access token.",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2837,6 +2837,7 @@ def create_oauth_authenticator(
             expires_in_name=model.expires_in_name or "expires_in",
             grant_type_name=model.grant_type_name or "grant_type",
             grant_type=model.grant_type or "refresh_token",
+            token_refresh_request_type=model.token_refresh_request_type or "body_data",
             refresh_request_body=model.refresh_request_body,
             refresh_request_headers=model.refresh_request_headers,
             refresh_token_name=model.refresh_token_name or "refresh_token",

airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py

@@ -239,9 +239,12 @@ def _make_handled_request(self) -> Any:
         """
         Makes a handled HTTP request to refresh an OAuth token.
 
-        This method sends a POST request to the token refresh endpoint with the necessary
-        headers and body to obtain a new access token. It handles various exceptions that
-        may occur during the request and logs the response for troubleshooting purposes.
+        This method sends an HTTP request to the token refresh endpoint with the necessary
+        headers and body/params to obtain a new access token. The HTTP method and parameter
+        encoding are determined by get_token_refresh_request_type():
+        - "body_data" (default): POST with form-encoded body
+        - "body_json": POST with JSON body
+        - "query_params": GET with query parameters
 
         Returns:
             Mapping[str, Any]: The JSON response from the token refresh endpoint.
@@ -254,12 +257,32 @@ def _make_handled_request(self) -> Any:
             Exception: For any other exceptions that occur during the request.
         """
         try:
-            response = requests.request(
-                method="POST",
-                url=self.get_token_refresh_endpoint(),  # type: ignore # returns None, if not provided, but str | bytes is expected.
-                data=self.build_refresh_request_body(),
-                headers=self.build_refresh_request_headers(),
-            )
+            request_type = self.get_token_refresh_request_type()
+            headers = self.build_refresh_request_headers()
+            body = self.build_refresh_request_body()
+            url = self.get_token_refresh_endpoint()
+
+            if request_type == "query_params":
+                response = requests.request(
+                    method="GET",
+                    url=url,  # type: ignore[arg-type]  # returns None if not provided, but str | bytes is expected
+                    params=body,
+                    headers=headers,
+                )
+            elif request_type == "body_json":
+                response = requests.request(
+                    method="POST",
+                    url=url,  # type: ignore[arg-type]  # returns None if not provided, but str | bytes is expected
+                    json=body,
+                    headers=headers,
+                )
+            else:
+                response = requests.request(
+                    method="POST",
+                    url=url,  # type: ignore[arg-type]  # returns None if not provided, but str | bytes is expected
+                    data=body,
+                    headers=headers,
+                )
 
             if not response.ok:
                 # log the response even if the request failed for troubleshooting purposes
@@ -543,6 +566,16 @@ def get_grant_type(self) -> str:
     def get_grant_type_name(self) -> str:
         """Returns grant_type specified name for requesting access_token"""
 
+    def get_token_refresh_request_type(self) -> str:
+        """Returns the request type for the token refresh request.
+
+        Supported values:
+        - "body_data": POST with form-encoded body (default, standard OAuth2)
+        - "body_json": POST with JSON-encoded body
+        - "query_params": GET with query parameters (e.g., Marketo)
+        """
+        return "body_data"
+
     @property
     @abstractmethod
     def access_token(self) -> str: