Bug: [FEATURE] Integrate Stripe Connect for paying out



### Feature Description

Stripe Connect platform enables merchants to onboard and pay to their customers, partners, vendors etc. This is to be added as a Payout Connector.

### Possible Implementation

Integrate Stripe Connect under Stripe connector integration for payout actions

- PoCreate
- PoRecipient
- PoRecipientAccount
- PoFulfill

### Have you spent some time checking if this feature request has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!
diff --git a/config/config.example.toml b/config/config.example.toml
index 1076cbdc196..b2e1fe3c458 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -487,7 +487,7 @@ debit = { currency = "USD" }
 
 [connector_customer]
 connector_list = "gocardless,stax,stripe"
-payout_connector_list = "wise"
+payout_connector_list = "stripe,wise"
 
 [bank_config.online_banking_fpx]
 adyen.banks = "affin_bank,agro_bank,alliance_bank,am_bank,bank_islam,bank_muamalat,bank_rakyat,bank_simpanan_nasional,cimb_bank,hong_leong_bank,hsbc_bank,kuwait_finance_house,maybank,ocbc_bank,public_bank,rhb_bank,standard_chartered_bank,uob_bank"
diff --git a/config/development.toml b/config/development.toml
index 4062ae6d9e9..92e13f24fec 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -474,7 +474,7 @@ payme = { payment_method = "card" }
 
 [connector_customer]
 connector_list = "gocardless,stax,stripe"
-payout_connector_list = "wise"
+payout_connector_list = "stripe,wise"
 
 [dummy_connector]
 enabled = true
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index 27e5878de6d..d99ab473501 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -395,7 +395,7 @@ connector_list = "stripe,adyen,cybersource"
 
 [connector_customer]
 connector_list = "gocardless,stax,stripe"
-payout_connector_list = "wise"
+payout_connector_list = "stripe,wise"
 
 [multiple_api_version_supported_connectors]
 supported_connectors = "braintree"
diff --git a/crates/api_models/src/enums.rs b/crates/api_models/src/enums.rs
index 6bdb51281fb..452ad6332fb 100644
--- a/crates/api_models/src/enums.rs
+++ b/crates/api_models/src/enums.rs
@@ -154,6 +154,10 @@ impl Connector {
     pub fn supports_access_token_for_payout(&self, payout_method: PayoutType) -> bool {
         matches!((self, payout_method), (Self::Paypal, _))
     }
+    #[cfg(feature = "payouts")]
+    pub fn supports_vendor_disburse_account_create_for_payout(&self) -> bool {
+        matches!(self, Self::Stripe)
+    }
     pub fn supports_access_token(&self, payment_method: PaymentMethod) -> bool {
         matches!(
             (self, payment_method),
@@ -338,6 +342,7 @@ pub enum AuthenticationConnectors {
 #[strum(serialize_all = "snake_case")]
 pub enum PayoutConnectors {
     Adyen,
+    Stripe,
     Wise,
     Paypal,
 }
@@ -347,6 +352,7 @@ impl From<PayoutConnectors> for RoutableConnectors {
     fn from(value: PayoutConnectors) -> Self {
         match value {
             PayoutConnectors::Adyen => Self::Adyen,
+            PayoutConnectors::Stripe => Self::Stripe,
             PayoutConnectors::Wise => Self::Wise,
             PayoutConnectors::Paypal => Self::Paypal,
         }
@@ -358,6 +364,7 @@ impl From<PayoutConnectors> for Connector {
     fn from(value: PayoutConnectors) -> Self {
         match value {
             PayoutConnectors::Adyen => Self::Adyen,
+            PayoutConnectors::Stripe => Self::Stripe,
             PayoutConnectors::Wise => Self::Wise,
             PayoutConnectors::Paypal => Self::Paypal,
         }
@@ -370,6 +377,7 @@ impl TryFrom<Connector> for PayoutConnectors {
     fn try_from(value: Connector) -> Result<Self, Self::Error> {
         match value {
             Connector::Adyen => Ok(Self::Adyen),
+            Connector::Stripe => Ok(Self::Stripe),
             Connector::Wise => Ok(Self::Wise),
             Connector::Paypal => Ok(Self::Paypal),
             _ => Err(format!("Invalid payout connector {}", value)),
diff --git a/crates/api_models/src/payouts.rs b/crates/api_models/src/payouts.rs
index 4e2d9192554..0951f8c9dfc 100644
--- a/crates/api_models/src/payouts.rs
+++ b/crates/api_models/src/payouts.rs
@@ -446,6 +446,7 @@ pub struct PayoutAttemptResponse {
 #[derive(Default, Debug, Clone, Deserialize, ToSchema)]
 pub struct PayoutRetrieveBody {
     pub force_sync: Option<bool>,
+    pub merchant_id: Option<String>,
 }
 
 #[derive(Default, Debug, Serialize, ToSchema, Clone, Deserialize)]
@@ -464,6 +465,9 @@ pub struct PayoutRetrieveRequest {
     /// (defaults to false)
     #[schema(value_type = Option<bool>, default = false, example = true)]
     pub force_sync: Option<bool>,
+
+    /// The identifier for the Merchant Account.
+    pub merchant_id: Option<String>,
 }
 
 #[derive(Default, Debug, Serialize, ToSchema, Clone, Deserialize)]
@@ -479,6 +483,43 @@ pub struct PayoutActionRequest {
     pub payout_id: String,
 }
 
+#[derive(Default, Debug, ToSchema, Clone, Deserialize)]
+pub struct PayoutVendorAccountDetails {
+    pub vendor_details: PayoutVendorDetails,
+    pub individual_details: PayoutIndividualDetails,
+}
+
+#[derive(Default, Debug, Serialize, ToSchema, Clone, Deserialize)]
+pub struct PayoutVendorDetails {
+    pub account_type: String,
+    pub business_type: String,
+    pub business_profile_mcc: Option<i32>,
+    pub business_profile_url: Option<String>,
+    pub business_profile_name: Option<Secret<String>>,
+    pub company_address_line1: Option<Secret<String>>,
+    pub company_address_line2: Option<Secret<String>>,
+    pub company_address_postal_code: Option<Secret<String>>,
+    pub company_address_city: Option<Secret<String>>,
+    pub company_address_state: Option<Secret<String>>,
+    pub company_phone: Option<Secret<String>>,
+    pub company_tax_id: Option<Secret<String>>,
+    pub company_owners_provided: Option<bool>,
+    pub capabilities_card_payments: Option<bool>,
+    pub capabilities_transfers: Option<bool>,
+}
+
+#[derive(Default, Debug, Serialize, ToSchema, Clone, Deserialize)]
+pub struct PayoutIndividualDetails {
+    pub tos_acceptance_date: Option<i64>,
+    pub tos_acceptance_ip: Option<Secret<String>>,
+    pub individual_dob_day: Option<Secret<String>>,
+    pub individual_dob_month: Option<Secret<String>>,
+    pub individual_dob_year: Option<Secret<String>>,
+    pub individual_id_number: Option<Secret<String>>,
+    pub individual_ssn_last_4: Option<Secret<String>>,
+    pub external_account_account_holder_type: Option<String>,
+}
+
 #[derive(Clone, Debug, serde::Deserialize, ToSchema, serde::Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct PayoutListConstraints {
diff --git a/crates/common_enums/src/enums.rs b/crates/common_enums/src/enums.rs
index 0681edbb15a..0cbb1a03d5b 100644
--- a/crates/common_enums/src/enums.rs
+++ b/crates/common_enums/src/enums.rs
@@ -2113,6 +2113,7 @@ pub enum PayoutStatus {
     RequiresCreation,
     RequiresPayoutMethodData,
     RequiresFulfillment,
+    RequiresVendorAccountCreation,
 }
 
 #[derive(
diff --git a/crates/common_utils/src/consts.rs b/crates/common_utils/src/consts.rs
index 0b988a195fa..d9d1f18c347 100644
--- a/crates/common_utils/src/consts.rs
+++ b/crates/common_utils/src/consts.rs
@@ -28,6 +28,9 @@ pub fn default_payments_list_limit() -> u32 {
     10
 }
 
+/// Average delay (in seconds) between account onboarding's API response and the changes to actually reflect at Stripe's end
+pub const STRIPE_ACCOUNT_ONBOARDING_DELAY_IN_SECONDS: i64 = 15;
+
 /// Maximum limit for payment link list get api
 pub const PAYMENTS_LINK_LIST_LIMIT: u32 = 100;
 
diff --git a/crates/connector_configs/src/connector.rs b/crates/connector_configs/src/connector.rs
index d7ae7b8a01e..497accc3346 100644
--- a/crates/connector_configs/src/connector.rs
+++ b/crates/connector_configs/src/connector.rs
@@ -161,6 +161,8 @@ pub struct ConnectorConfig {
     pub rapyd: Option<ConnectorTomlConfig>,
     pub shift4: Option<ConnectorTomlConfig>,
     pub stripe: Option<ConnectorTomlConfig>,
+    #[cfg(feature = "payouts")]
+    pub stripe_payout: Option<ConnectorTomlConfig>,
     pub signifyd: Option<ConnectorTomlConfig>,
     pub trustpay: Option<ConnectorTomlConfig>,
     pub threedsecureio: Option<ConnectorTomlConfig>,
@@ -214,6 +216,7 @@ impl ConnectorConfig {
         let connector_data = Self::new()?;
         match connector {
             PayoutConnectors::Adyen => Ok(connector_data.adyen_payout),
+            PayoutConnectors::Stripe => Ok(connector_data.stripe_payout),
             PayoutConnectors::Wise => Ok(connector_data.wise_payout),
             PayoutConnectors::Paypal => Ok(connector_data.paypal),
         }
diff --git a/crates/connector_configs/toml/development.toml b/crates/connector_configs/toml/development.toml
index 9a231f0dc6d..a81fb3657ab 100644
--- a/crates/connector_configs/toml/development.toml
+++ b/crates/connector_configs/toml/development.toml
@@ -2580,6 +2580,12 @@ api_key = "Adyen API Key (Payout creation)"
 api_secret = "Adyen Key (Payout submission)"
 key1 = "Adyen Account Id"
 
+[stripe_payout]
+[[stripe_payout.bank_transfer]]
+  payment_method_type = "ach"
+[stripe_payout.connector_auth.HeaderKey]
+api_key = "Stripe API Key"
+
 [wise_payout]
 [[wise_payout.bank_transfer]]
   payment_method_type = "ach"
diff --git a/crates/connector_configs/toml/sandbox.toml b/crates/connector_configs/toml/sandbox.toml
index 5bb61ec2320..1d0fbb71538 100644
--- a/crates/connector_configs/toml/sandbox.toml
+++ b/crates/connector_configs/toml/sandbox.toml
@@ -2582,6 +2582,12 @@ api_key = "Adyen API Key (Payout creation)"
 api_secret = "Adyen Key (Payout submission)"
 key1 = "Adyen Account Id"
 
+[stripe_payout]
+[[stripe_payout.bank_transfer]]
+  payment_method_type = "ach"
+[stripe_payout.connector_auth.HeaderKey]
+api_key = "Stripe API Key"
+
 [wise_payout]
 [[wise_payout.bank_transfer]]
   payment_method_type = "ach"
diff --git a/crates/data_models/src/payouts/payouts.rs b/crates/data_models/src/payouts/payouts.rs
index 28a8e4d4278..679e4488f7b 100644
--- a/crates/data_models/src/payouts/payouts.rs
+++ b/crates/data_models/src/payouts/payouts.rs
@@ -89,6 +89,7 @@ pub struct Payouts {
     pub attempt_count: i16,
     pub profile_id: String,
     pub status: storage_enums::PayoutStatus,
+    pub confirm: Option<bool>,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -110,9 +111,10 @@ pub struct PayoutsNew {
     pub metadata: Option<pii::SecretSerdeValue>,
     pub created_at: Option<PrimitiveDateTime>,
     pub last_modified_at: Option<PrimitiveDateTime>,
+    pub attempt_count: i16,
     pub profile_id: String,
     pub status: storage_enums::PayoutStatus,
-    pub attempt_count: i16,
+    pub confirm: Option<bool>,
 }
 
 impl Default for PayoutsNew {
@@ -137,9 +139,10 @@ impl Default for PayoutsNew {
             metadata: Option::default(),
             created_at: Some(now),
             last_modified_at: Some(now),
+            attempt_count: 1,
             profile_id: String::default(),
             status: storage_enums::PayoutStatus::default(),
-            attempt_count: 1,
+            confirm: None,
         }
     }
 }
@@ -158,6 +161,7 @@ pub enum PayoutsUpdate {
         metadata: Option<pii::SecretSerdeValue>,
         profile_id: Option<String>,
         status: Option<storage_enums::PayoutStatus>,
+        confirm: Option<bool>,
     },
     PayoutMethodIdUpdate {
         payout_method_id: String,
@@ -188,6 +192,7 @@ pub struct PayoutsUpdateInternal {
     pub profile_id: Option<String>,
     pub status: Option<storage_enums::PayoutStatus>,
     pub attempt_count: Option<i16>,
+    pub confirm: Option<bool>,
 }
 
 impl From<PayoutsUpdate> for PayoutsUpdateInternal {
@@ -205,6 +210,7 @@ impl From<PayoutsUpdate> for PayoutsUpdateInternal {
                 metadata,
                 profile_id,
                 status,
+                confirm,
             } => Self {
                 amount: Some(amount),
                 destination_currency: Some(destination_currency),
@@ -217,6 +223,7 @@ impl From<PayoutsUpdate> for PayoutsUpdateInternal {
                 metadata,
                 profile_id,
                 status,
+                confirm,
                 ..Default::default()
             },
             PayoutsUpdate::PayoutMethodIdUpdate { payout_method_id } => Self {
diff --git a/crates/diesel_models/src/payouts.rs b/crates/diesel_models/src/payouts.rs
index 7cc348d445d..4c6fe40c399 100644
--- a/crates/diesel_models/src/payouts.rs
+++ b/crates/diesel_models/src/payouts.rs
@@ -32,6 +32,7 @@ pub struct Payouts {
     pub attempt_count: i16,
     pub profile_id: String,
     pub status: storage_enums::PayoutStatus,
+    pub confirm: Option<bool>,
 }
 
 #[derive(
@@ -67,9 +68,10 @@ pub struct PayoutsNew {
     pub created_at: Option<PrimitiveDateTime>,
     #[serde(default, with = "common_utils::custom_serde::iso8601::option")]
     pub last_modified_at: Option<PrimitiveDateTime>,
+    pub attempt_count: i16,
     pub profile_id: String,
     pub status: storage_enums::PayoutStatus,
-    pub attempt_count: i16,
+    pub confirm: Option<bool>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -86,6 +88,7 @@ pub enum PayoutsUpdate {
         metadata: Option<pii::SecretSerdeValue>,
         profile_id: Option<String>,
         status: Option<storage_enums::PayoutStatus>,
+        confirm: Option<bool>,
     },
     PayoutMethodIdUpdate {
         payout_method_id: String,
@@ -118,6 +121,7 @@ pub struct PayoutsUpdateInternal {
     pub status: Option<storage_enums::PayoutStatus>,
     pub last_modified_at: PrimitiveDateTime,
     pub attempt_count: Option<i16>,
+    pub confirm: Option<bool>,
 }
 
 impl Default for PayoutsUpdateInternal {
@@ -137,6 +141,7 @@ impl Default for PayoutsUpdateInternal {
             status: None,
             last_modified_at: common_utils::date_time::now(),
             attempt_count: None,
+            confirm: None,
         }
     }
 }
@@ -156,6 +161,7 @@ impl From<PayoutsUpdate> for PayoutsUpdateInternal {
                 metadata,
                 profile_id,
                 status,
+                confirm,
             } => Self {
                 amount: Some(amount),
                 destination_currency: Some(destination_currency),
@@ -168,6 +174,7 @@ impl From<PayoutsUpdate> for PayoutsUpdateInternal {
                 metadata,
                 profile_id,
                 status,
+                confirm,
                 ..Default::default()
             },
             PayoutsUpdate::PayoutMethodIdUpdate { payout_method_id } => Self {
@@ -207,6 +214,7 @@ impl PayoutsUpdate {
             status,
             last_modified_at,
             attempt_count,
+            confirm,
         } = self.into();
         Payouts {
             amount: amount.unwrap_or(source.amount),
@@ -223,6 +231,7 @@ impl PayoutsUpdate {
             status: status.unwrap_or(source.status),
             last_modified_at,
             attempt_count: attempt_count.unwrap_or(source.attempt_count),
+            confirm: confirm.or(source.confirm),
             ..source
         }
     }
diff --git a/crates/diesel_models/src/process_tracker.rs b/crates/diesel_models/src/process_tracker.rs
index 0c6aaa5922f..cd0dbed2c64 100644
--- a/crates/diesel_models/src/process_tracker.rs
+++ b/crates/diesel_models/src/process_tracker.rs
@@ -209,6 +209,7 @@ pub enum ProcessTrackerRunner {
     DeleteTokenizeDataWorkflow,
     ApiKeyExpiryWorkflow,
     OutgoingWebhookRetryWorkflow,
+    AttachPayoutAccountWorkflow,
 }
 
 #[cfg(test)]
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index bbe8a8060fc..ff0539cc05e 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -999,6 +999,7 @@ diesel::table! {
         #[max_length = 64]
         profile_id -> Varchar,
         status -> PayoutStatus,
+        confirm -> Nullable<Bool>,
     }
 }
 
diff --git a/crates/router/src/bin/scheduler.rs b/crates/router/src/bin/scheduler.rs
index 1df37e9f6d1..47f41d8700f 100644
--- a/crates/router/src/bin/scheduler.rs
+++ b/crates/router/src/bin/scheduler.rs
@@ -263,6 +263,23 @@ impl ProcessTrackerWorkflows<routes::AppState> for WorkflowRunner {
                 storage::ProcessTrackerRunner::OutgoingWebhookRetryWorkflow => Ok(Box::new(
                     workflows::outgoing_webhook_retry::OutgoingWebhookRetryWorkflow,
                 )),
+                storage::ProcessTrackerRunner::AttachPayoutAccountWorkflow => {
+                    #[cfg(feature = "payouts")]
+                    {
+                        Ok(Box::new(
+                            workflows::attach_payout_account_workflow::AttachPayoutAccountWorkflow,
+                        ))
+                    }
+                    #[cfg(not(feature = "payouts"))]
+                    {
+                        Err(
+                            error_stack::report!(ProcessTrackerError::UnexpectedFlow),
+                        )
+                        .attach_printable(
+                            "Cannot run Stripe external account workflow when payouts feature is disabled",
+                        )
+                    }
+                }
             }
         };
 
diff --git a/crates/router/src/connector/adyen/transformers.rs b/crates/router/src/connector/adyen/transformers.rs
index d02078702e8..bde04877e65 100644
--- a/crates/router/src/connector/adyen/transformers.rs
+++ b/crates/router/src/connector/adyen/transformers.rs
@@ -4768,10 +4768,8 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, AdyenPayoutResponse>>
         let status = payout_eligible.map_or(
             {
                 response.result_code.map_or(
-                    response
-                        .response
-                        .map(storage_enums::PayoutStatus::foreign_from),
-                    |rc| Some(storage_enums::PayoutStatus::foreign_from(rc)),
+                    response.response.map(storage_enums::PayoutStatus::from),
+                    |rc| Some(storage_enums::PayoutStatus::from(rc)),
                 )
             },
             |pe| {
@@ -4788,6 +4786,7 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, AdyenPayoutResponse>>
                 status,
                 connector_payout_id: response.psp_reference,
                 payout_eligible,
+                should_add_next_step_to_process_tracker: false,
             }),
             ..item.data
         })
@@ -4795,8 +4794,8 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, AdyenPayoutResponse>>
 }
 
 #[cfg(feature = "payouts")]
-impl ForeignFrom<AdyenStatus> for storage_enums::PayoutStatus {
-    fn foreign_from(adyen_status: AdyenStatus) -> Self {
+impl From<AdyenStatus> for storage_enums::PayoutStatus {
+    fn from(adyen_status: AdyenStatus) -> Self {
         match adyen_status {
             AdyenStatus::Authorised | AdyenStatus::PayoutConfirmReceived => Self::Success,
             AdyenStatus::Cancelled | AdyenStatus::PayoutDeclineReceived => Self::Cancelled,
diff --git a/crates/router/src/connector/stripe.rs b/crates/router/src/connector/stripe.rs
index e499cdda12c..64789e176fc 100644
--- a/crates/router/src/connector/stripe.rs
+++ b/crates/router/src/connector/stripe.rs
@@ -11,6 +11,8 @@ use stripe::auth_headers;
 
 use self::transformers as stripe;
 use super::utils::{self as connector_utils, RefundsRequestData};
+#[cfg(feature = "payouts")]
+use super::utils::{PayoutsData, RouterData};
 use crate::{
     configs::settings,
     consts,
@@ -27,7 +29,7 @@ use crate::{
     },
     types::{
         self,
-        api::{self, ConnectorCommon},
+        api::{self, ConnectorCommon, ConnectorCommonExt},
         domain,
     },
     utils::{crypto, ByteSliceExt, BytesExt, OptionExt},
@@ -36,6 +38,25 @@ use crate::{
 #[derive(Debug, Clone)]
 pub struct Stripe;
 
+impl<Flow, Request, Response> ConnectorCommonExt<Flow, Request, Response> for Stripe
+where
+    Self: services::ConnectorIntegration<Flow, Request, Response>,
+{
+    fn build_headers(
+        &self,
+        req: &types::RouterData<Flow, Request, Response>,
+        _connectors: &settings::Connectors,
+    ) -> CustomResult<Vec<(String, request::Maskable<String>)>, errors::ConnectorError> {
+        let mut header = vec![(
+            headers::CONTENT_TYPE.to_string(),
+            Self::common_get_content_type(self).to_string().into(),
+        )];
+        let mut api_key = self.get_auth_header(&req.connector_auth_type)?;
+        header.append(&mut api_key);
+        Ok(header)
+    }
+}
+
 impl ConnectorCommon for Stripe {
     fn id(&self) -> &'static str {
         "stripe"
@@ -67,6 +88,34 @@ impl ConnectorCommon for Stripe {
             ),
         ])
     }
+
+    #[cfg(feature = "payouts")]
+    fn build_error_response(
+        &self,
+        res: types::Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<types::ErrorResponse, errors::ConnectorError> {
+        let response: stripe::StripeConnectErrorResponse = res
+            .response
+            .parse_struct("StripeConnectErrorResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_error_response_body(&response));
+        Ok(types::ErrorResponse {
+            status_code: res.status_code,
+            code: response
+                .error
+                .code
+                .clone()
+                .unwrap_or_else(|| consts::NO_ERROR_CODE.to_string()),
+            message: response
+                .error
+                .code
+                .unwrap_or_else(|| consts::NO_ERROR_MESSAGE.to_string()),
+            reason: response.error.message,
+            attempt_status: None,
+            connector_transaction_id: response.error.payment_intent.map(|pi| pi.id),
+        })
+    }
 }
 
 impl ConnectorValidation for Stripe {
@@ -2235,3 +2284,434 @@ impl services::ConnectorRedirectResponse for Stripe {
         }
     }
 }
+
+impl api::Payouts for Stripe {}
+#[cfg(feature = "payouts")]
+impl api::PayoutCancel for Stripe {}
+#[cfg(feature = "payouts")]
+impl api::PayoutCreate for Stripe {}
+#[cfg(feature = "payouts")]
+impl api::PayoutFulfill for Stripe {}
+#[cfg(feature = "payouts")]
+impl api::PayoutRecipient for Stripe {}
+#[cfg(feature = "payouts")]
+impl api::PayoutRecipientAccount for Stripe {}
+
+#[cfg(feature = "payouts")]
+impl services::ConnectorIntegration<api::PoCancel, types::PayoutsData, types::PayoutsResponseData>
+    for Stripe
+{
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        req: &types::PayoutsRouterData<api::PoCancel>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        let transfer_id = req.request.get_transfer_id()?;
+        Ok(format!(
+            "{}v1/transfers/{}/reversals",
+            connectors.stripe.base_url, transfer_id
+        ))
+    }
+
+    fn get_headers(
+        &self,
+        req: &types::PayoutsRouterData<api::PoCancel>,
+        _connectors: &settings::Connectors,
+    ) -> CustomResult<Vec<(String, request::Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, _connectors)
+    }
+
+    fn get_request_body(
+        &self,
+        req: &types::RouterData<api::PoCancel, types::PayoutsData, types::PayoutsResponseData>,
+        _connectors: &settings::Connectors,
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
+        let connector_req = stripe::StripeConnectReversalRequest::try_from(req)?;
+        Ok(RequestContent::FormUrlEncoded(Box::new(connector_req)))
+    }
+
+    fn build_request(
+        &self,
+        req: &types::PayoutsRouterData<api::PoCancel>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Option<services::Request>, errors::ConnectorError> {
+        let request = services::RequestBuilder::new()
+            .method(services::Method::Post)
+            .url(&types::PayoutCancelType::get_url(self, req, connectors)?)
+            .attach_default_headers()
+            .headers(types::PayoutCancelType::get_headers(self, req, connectors)?)
+            .set_body(types::PayoutCancelType::get_request_body(
+                self, req, connectors,
+            )?)
+            .build();
+
+        Ok(Some(request))
+    }
+
+    fn handle_response(
+        &self,
+        data: &types::PayoutsRouterData<api::PoCancel>,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: types::Response,
+    ) -> CustomResult<types::PayoutsRouterData<api::PoCancel>, errors::ConnectorError> {
+        let response: stripe::StripeConnectReversalResponse = res
+            .response
+            .parse_struct("StripeConnectReversalResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_error_response_body(&response));
+        types::RouterData::try_from(types::ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: types::Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }
+}
+
+#[cfg(feature = "payouts")]
+impl services::ConnectorIntegration<api::PoCreate, types::PayoutsData, types::PayoutsResponseData>
+    for Stripe
+{
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        _req: &types::PayoutsRouterData<api::PoCreate>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        Ok(format!("{}v1/transfers", connectors.stripe.base_url))
+    }
+
+    fn get_headers(
+        &self,
+        req: &types::PayoutsRouterData<api::PoCreate>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Vec<(String, request::Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_request_body(
+        &self,
+        req: &types::PayoutsRouterData<api::PoCreate>,
+        _connectors: &settings::Connectors,
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
+        let connector_req = stripe::StripeConnectPayoutCreateRequest::try_from(req)?;
+        Ok(RequestContent::FormUrlEncoded(Box::new(connector_req)))
+    }
+
+    fn build_request(
+        &self,
+        req: &types::PayoutsRouterData<api::PoCreate>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Option<services::Request>, errors::ConnectorError> {
+        let request = services::RequestBuilder::new()
+            .method(services::Method::Post)
+            .url(&types::PayoutCreateType::get_url(self, req, connectors)?)
+            .attach_default_headers()
+            .headers(types::PayoutCreateType::get_headers(self, req, connectors)?)
+            .set_body(types::PayoutCreateType::get_request_body(
+                self, req, connectors,
+            )?)
+            .build();
+
+        Ok(Some(request))
+    }
+
+    fn handle_response(
+        &self,
+        data: &types::PayoutsRouterData<api::PoCreate>,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: types::Response,
+    ) -> CustomResult<types::PayoutsRouterData<api::PoCreate>, errors::ConnectorError> {
+        let response: stripe::StripeConnectPayoutCreateResponse = res
+            .response
+            .parse_struct("StripeConnectPayoutCreateResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_error_response_body(&response));
+        types::RouterData::try_from(types::ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: types::Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }
+}
+
+#[cfg(feature = "payouts")]
+impl services::ConnectorIntegration<api::PoFulfill, types::PayoutsData, types::PayoutsResponseData>
+    for Stripe
+{
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        _req: &types::PayoutsRouterData<api::PoFulfill>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        Ok(format!("{}v1/payouts", connectors.stripe.base_url,))
+    }
+
+    fn get_headers(
+        &self,
+        req: &types::PayoutsRouterData<api::PoFulfill>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Vec<(String, request::Maskable<String>)>, errors::ConnectorError> {
+        let mut headers = self.build_headers(req, connectors)?;
+        let customer_account = req.get_connector_customer_id()?;
+        let mut customer_account_header = vec![(
+            headers::STRIPE_COMPATIBLE_CONNECT_ACCOUNT.to_string(),
+            customer_account.into_masked(),
+        )];
+        headers.append(&mut customer_account_header);
+        Ok(headers)
+    }
+
+    fn get_request_body(
+        &self,
+        req: &types::PayoutsRouterData<api::PoFulfill>,
+        _connectors: &settings::Connectors,
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
+        let connector_req = stripe::StripeConnectPayoutFulfillRequest::try_from(req)?;
+        Ok(RequestContent::FormUrlEncoded(Box::new(connector_req)))
+    }
+
+    fn build_request(
+        &self,
+        req: &types::PayoutsRouterData<api::PoFulfill>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Option<services::Request>, errors::ConnectorError> {
+        let request = services::RequestBuilder::new()
+            .method(services::Method::Post)
+            .url(&types::PayoutFulfillType::get_url(self, req, connectors)?)
+            .attach_default_headers()
+            .headers(types::PayoutFulfillType::get_headers(
+                self, req, connectors,
+            )?)
+            .set_body(types::PayoutFulfillType::get_request_body(
+                self, req, connectors,
+            )?)
+            .build();
+
+        Ok(Some(request))
+    }
+
+    fn handle_response(
+        &self,
+        data: &types::PayoutsRouterData<api::PoFulfill>,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: types::Response,
+    ) -> CustomResult<types::PayoutsRouterData<api::PoFulfill>, errors::ConnectorError> {
+        let response: stripe::StripeConnectPayoutFulfillResponse = res
+            .response
+            .parse_struct("StripeConnectPayoutFulfillResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_error_response_body(&response));
+        types::RouterData::try_from(types::ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: types::Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }
+}
+
+#[cfg(feature = "payouts")]
+impl
+    services::ConnectorIntegration<api::PoRecipient, types::PayoutsData, types::PayoutsResponseData>
+    for Stripe
+{
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        _req: &types::PayoutsRouterData<api::PoRecipient>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        Ok(format!("{}v1/accounts", connectors.stripe.base_url))
+    }
+
+    fn get_headers(
+        &self,
+        req: &types::PayoutsRouterData<api::PoRecipient>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Vec<(String, request::Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_request_body(
+        &self,
+        req: &types::PayoutsRouterData<api::PoRecipient>,
+        _connectors: &settings::Connectors,
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
+        let connector_req = stripe::StripeConnectRecipientCreateRequest::try_from(req)?;
+        Ok(RequestContent::FormUrlEncoded(Box::new(connector_req)))
+    }
+
+    fn build_request(
+        &self,
+        req: &types::PayoutsRouterData<api::PoRecipient>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Option<services::Request>, errors::ConnectorError> {
+        let request = services::RequestBuilder::new()
+            .method(services::Method::Post)
+            .url(&types::PayoutRecipientType::get_url(self, req, connectors)?)
+            .attach_default_headers()
+            .headers(types::PayoutRecipientType::get_headers(
+                self, req, connectors,
+            )?)
+            .set_body(types::PayoutRecipientType::get_request_body(
+                self, req, connectors,
+            )?)
+            .build();
+
+        Ok(Some(request))
+    }
+
+    fn handle_response(
+        &self,
+        data: &types::PayoutsRouterData<api::PoRecipient>,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: types::Response,
+    ) -> CustomResult<types::PayoutsRouterData<api::PoRecipient>, errors::ConnectorError> {
+        let response: stripe::StripeConnectRecipientCreateResponse = res
+            .response
+            .parse_struct("StripeConnectRecipientCreateResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_error_response_body(&response));
+        types::RouterData::try_from(types::ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: types::Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }
+}
+
+#[cfg(feature = "payouts")]
+impl
+    services::ConnectorIntegration<
+        api::PoRecipientAccount,
+        types::PayoutsData,
+        types::PayoutsResponseData,
+    > for Stripe
+{
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        req: &types::PayoutsRouterData<api::PoRecipientAccount>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        let connector_customer_id = req.get_connector_customer_id()?;
+        Ok(format!(
+            "{}v1/accounts/{}/external_accounts",
+            connectors.stripe.base_url, connector_customer_id
+        ))
+    }
+
+    fn get_headers(
+        &self,
+        req: &types::PayoutsRouterData<api::PoRecipientAccount>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Vec<(String, request::Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_request_body(
+        &self,
+        req: &types::PayoutsRouterData<api::PoRecipientAccount>,
+        _connectors: &settings::Connectors,
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
+        let connector_req = stripe::StripeConnectRecipientAccountCreateRequest::try_from(req)?;
+        Ok(RequestContent::FormUrlEncoded(Box::new(connector_req)))
+    }
+
+    fn build_request(
+        &self,
+        req: &types::PayoutsRouterData<api::PoRecipientAccount>,
+        connectors: &settings::Connectors,
+    ) -> CustomResult<Option<services::Request>, errors::ConnectorError> {
+        let request = services::RequestBuilder::new()
+            .method(services::Method::Post)
+            .url(&types::PayoutRecipientAccountType::get_url(
+                self, req, connectors,
+            )?)
+            .attach_default_headers()
+            .headers(types::PayoutRecipientAccountType::get_headers(
+                self, req, connectors,
+            )?)
+            .set_body(types::PayoutRecipientAccountType::get_request_body(
+                self, req, connectors,
+            )?)
+            .build();
+
+        Ok(Some(request))
+    }
+
+    fn handle_response(
+        &self,
+        data: &types::PayoutsRouterData<api::PoRecipientAccount>,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: types::Response,
+    ) -> CustomResult<types::PayoutsRouterData<api::PoRecipientAccount>, errors::ConnectorError>
+    {
+        let response: stripe::StripeConnectRecipientAccountCreateResponse = res
+            .response
+            .parse_struct("StripeConnectRecipientAccountCreateResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_error_response_body(&response));
+        types::RouterData::try_from(types::ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: types::Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }
+}
diff --git a/crates/router/src/connector/stripe/transformers.rs b/crates/router/src/connector/stripe/transformers.rs
index e0801eea284..c561b26bcfd 100644
--- a/crates/router/src/connector/stripe/transformers.rs
+++ b/crates/router/src/connector/stripe/transformers.rs
@@ -15,6 +15,10 @@ use serde_json::Value;
 use time::PrimitiveDateTime;
 use url::Url;
 
+#[cfg(feature = "payouts")]
+pub mod connect;
+#[cfg(feature = "payouts")]
+pub use self::connect::*;
 use crate::{
     collect_missing_value_keys,
     connector::utils::{
diff --git a/crates/router/src/connector/stripe/transformers/connect.rs b/crates/router/src/connector/stripe/transformers/connect.rs
new file mode 100644
index 00000000000..df4376b7174
--- /dev/null
+++ b/crates/router/src/connector/stripe/transformers/connect.rs
@@ -0,0 +1,475 @@
+use api_models;
+use common_utils::pii::Email;
+use error_stack::ResultExt;
+use masking::Secret;
+use serde::{Deserialize, Serialize};
+
+use super::ErrorDetails;
+use crate::{
+    connector::utils::{PayoutsData, RouterData},
+    core::{errors, payments::CustomerDetailsExt},
+    types::{self, storage::enums, PayoutIndividualDetailsExt},
+    utils::OptionExt,
+};
+
+type Error = error_stack::Report<errors::ConnectorError>;
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum StripeConnectPayoutStatus {
+    Canceled,
+    Failed,
+    InTransit,
+    Paid,
+    Pending,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct StripeConnectErrorResponse {
+    pub error: ErrorDetails,
+}
+
+// Payouts
+#[derive(Clone, Debug, Serialize)]
+pub struct StripeConnectPayoutCreateRequest {
+    amount: i64,
+    currency: enums::Currency,
+    destination: String,
+    transfer_group: String,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct StripeConnectPayoutCreateResponse {
+    id: String,
+    description: Option<String>,
+    source_transaction: Option<String>,
+}
+
+#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+pub struct TransferReversals {
+    object: String,
+    has_more: bool,
+    total_count: i32,
+    url: String,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct StripeConnectPayoutFulfillRequest {
+    amount: i64,
+    currency: enums::Currency,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct StripeConnectPayoutFulfillResponse {
+    id: String,
+    currency: String,
+    description: Option<String>,
+    failure_balance_transaction: Option<String>,
+    failure_code: Option<String>,
+    failure_message: Option<String>,
+    original_payout: Option<String>,
+    reversed_by: Option<String>,
+    statement_descriptor: Option<String>,
+    status: StripeConnectPayoutStatus,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct StripeConnectReversalRequest {
+    amount: i64,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct StripeConnectReversalResponse {
+    id: String,
+    source_refund: Option<String>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct StripeConnectRecipientCreateRequest {
+    #[serde(rename = "type")]
+    account_type: String,
+    country: Option<enums::CountryAlpha2>,
+    email: Option<Email>,
+    #[serde(rename = "capabilities[card_payments][requested]")]
+    capabilities_card_payments: Option<bool>,
+    #[serde(rename = "capabilities[transfers][requested]")]
+    capabilities_transfers: Option<bool>,
+    #[serde(rename = "tos_acceptance[date]")]
+    tos_acceptance_date: Option<i64>,
+    #[serde(rename = "tos_acceptance[ip]")]
+    tos_acceptance_ip: Option<Secret<String>>,
+    business_type: String,
+    #[serde(rename = "business_profile[mcc]")]
+    business_profile_mcc: Option<i32>,
+    #[serde(rename = "business_profile[url]")]
+    business_profile_url: Option<String>,
+    #[serde(rename = "business_profile[name]")]
+    business_profile_name: Option<Secret<String>>,
+    #[serde(rename = "company[name]")]
+    company_name: Option<Secret<String>>,
+    #[serde(rename = "company[address][line1]")]
+    company_address_line1: Option<Secret<String>>,
+    #[serde(rename = "company[address][line2]")]
+    company_address_line2: Option<Secret<String>>,
+    #[serde(rename = "company[address][postal_code]")]
+    company_address_postal_code: Option<Secret<String>>,
+    #[serde(rename = "company[address][city]")]
+    company_address_city: Option<Secret<String>>,
+    #[serde(rename = "company[address][state]")]
+    company_address_state: Option<Secret<String>>,
+    #[serde(rename = "company[phone]")]
+    company_phone: Option<Secret<String>>,
+    #[serde(rename = "company[tax_id]")]
+    company_tax_id: Option<Secret<String>>,
+    #[serde(rename = "company[owners_provided]")]
+    company_owners_provided: Option<bool>,
+    #[serde(rename = "individual[first_name]")]
+    individual_first_name: Option<Secret<String>>,
+    #[serde(rename = "individual[last_name]")]
+    individual_last_name: Option<Secret<String>>,
+    #[serde(rename = "individual[dob][day]")]
+    individual_dob_day: Option<Secret<String>>,
+    #[serde(rename = "individual[dob][month]")]
+    individual_dob_month: Option<Secret<String>>,
+    #[serde(rename = "individual[dob][year]")]
+    individual_dob_year: Option<Secret<String>>,
+    #[serde(rename = "individual[address][line1]")]
+    individual_address_line1: Option<Secret<String>>,
+    #[serde(rename = "individual[address][line2]")]
+    individual_address_line2: Option<Secret<String>>,
+    #[serde(rename = "individual[address][postal_code]")]
+    individual_address_postal_code: Option<Secret<String>>,
+    #[serde(rename = "individual[address][city]")]
+    individual_address_city: Option<String>,
+    #[serde(rename = "individual[address][state]")]
+    individual_address_state: Option<Secret<String>>,
+    #[serde(rename = "individual[email]")]
+    individual_email: Option<Email>,
+    #[serde(rename = "individual[phone]")]
+    individual_phone: Option<Secret<String>>,
+    #[serde(rename = "individual[id_number]")]
+    individual_id_number: Option<Secret<String>>,
+    #[serde(rename = "individual[ssn_last_4]")]
+    individual_ssn_last_4: Option<Secret<String>>,
+}
+
+#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+pub struct StripeConnectRecipientCreateResponse {
+    id: String,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(untagged)]
+pub enum StripeConnectRecipientAccountCreateRequest {
+    Bank(RecipientBankAccountRequest),
+    Card(RecipientCardAccountRequest),
+    Token(RecipientTokenRequest),
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct RecipientTokenRequest {
+    external_account: String,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct RecipientCardAccountRequest {
+    #[serde(rename = "external_account[object]")]
+    external_account_object: String,
+    #[serde(rename = "external_account[number]")]
+    external_account_number: Secret<String>,
+    #[serde(rename = "external_account[exp_month]")]
+    external_account_exp_month: Secret<String>,
+    #[serde(rename = "external_account[exp_year]")]
+    external_account_exp_year: Secret<String>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct RecipientBankAccountRequest {
+    #[serde(rename = "external_account[object]")]
+    external_account_object: String,
+    #[serde(rename = "external_account[country]")]
+    external_account_country: enums::CountryAlpha2,
+    #[serde(rename = "external_account[currency]")]
+    external_account_currency: enums::Currency,
+    #[serde(rename = "external_account[account_holder_name]")]
+    external_account_account_holder_name: Secret<String>,
+    #[serde(rename = "external_account[account_number]")]
+    external_account_account_number: Secret<String>,
+    #[serde(rename = "external_account[account_holder_type]")]
+    external_account_account_holder_type: String,
+    #[serde(rename = "external_account[routing_number]")]
+    external_account_routing_number: Secret<String>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct StripeConnectRecipientAccountCreateResponse {
+    id: String,
+}
+
+// Payouts create/transfer request transform
+impl<F> TryFrom<&types::PayoutsRouterData<F>> for StripeConnectPayoutCreateRequest {
+    type Error = Error;
+    fn try_from(item: &types::PayoutsRouterData<F>) -> Result<Self, Self::Error> {
+        let request = item.request.to_owned();
+        let connector_customer_id = item.get_connector_customer_id()?;
+        Ok(Self {
+            amount: request.amount,
+            currency: request.destination_currency,
+            destination: connector_customer_id,
+            transfer_group: request.payout_id,
+        })
+    }
+}
+
+// Payouts create response transform
+impl<F> TryFrom<types::PayoutsResponseRouterData<F, StripeConnectPayoutCreateResponse>>
+    for types::PayoutsRouterData<F>
+{
+    type Error = Error;
+    fn try_from(
+        item: types::PayoutsResponseRouterData<F, StripeConnectPayoutCreateResponse>,
+    ) -> Result<Self, Self::Error> {
+        let response: StripeConnectPayoutCreateResponse = item.response;
+
+        Ok(Self {
+            response: Ok(types::PayoutsResponseData {
+                status: Some(enums::PayoutStatus::RequiresFulfillment),
+                connector_payout_id: response.id,
+                payout_eligible: None,
+                should_add_next_step_to_process_tracker: false,
+            }),
+            ..item.data
+        })
+    }
+}
+
+// Payouts fulfill request transform
+impl<F> TryFrom<&types::PayoutsRouterData<F>> for StripeConnectPayoutFulfillRequest {
+    type Error = Error;
+    fn try_from(item: &types::PayoutsRouterData<F>) -> Result<Self, Self::Error> {
+        let request = item.request.to_owned();
+        Ok(Self {
+            amount: request.amount,
+            currency: request.destination_currency,
+        })
+    }
+}
+
+// Payouts fulfill response transform
+impl<F> TryFrom<types::PayoutsResponseRouterData<F, StripeConnectPayoutFulfillResponse>>
+    for types::PayoutsRouterData<F>
+{
+    type Error = Error;
+    fn try_from(
+        item: types::PayoutsResponseRouterData<F, StripeConnectPayoutFulfillResponse>,
+    ) -> Result<Self, Self::Error> {
+        let response: StripeConnectPayoutFulfillResponse = item.response;
+
+        Ok(Self {
+            response: Ok(types::PayoutsResponseData {
+                status: Some(enums::PayoutStatus::from(response.status)),
+                connector_payout_id: response.id,
+                payout_eligible: None,
+                should_add_next_step_to_process_tracker: false,
+            }),
+            ..item.data
+        })
+    }
+}
+
+// Payouts reversal request transform
+impl<F> TryFrom<&types::PayoutsRouterData<F>> for StripeConnectReversalRequest {
+    type Error = Error;
+    fn try_from(item: &types::PayoutsRouterData<F>) -> Result<Self, Self::Error> {
+        Ok(Self {
+            amount: item.request.amount,
+        })
+    }
+}
+
+// Payouts reversal response transform
+impl<F> TryFrom<types::PayoutsResponseRouterData<F, StripeConnectReversalResponse>>
+    for types::PayoutsRouterData<F>
+{
+    type Error = Error;
+    fn try_from(
+        item: types::PayoutsResponseRouterData<F, StripeConnectReversalResponse>,
+    ) -> Result<Self, Self::Error> {
+        let response: StripeConnectReversalResponse = item.response;
+
+        Ok(Self {
+            response: Ok(types::PayoutsResponseData {
+                status: Some(enums::PayoutStatus::Cancelled),
+                connector_payout_id: response.id,
+                payout_eligible: None,
+                should_add_next_step_to_process_tracker: false,
+            }),
+            ..item.data
+        })
+    }
+}
+
+// Recipient creation request transform
+impl<F> TryFrom<&types::PayoutsRouterData<F>> for StripeConnectRecipientCreateRequest {
+    type Error = Error;
+    fn try_from(item: &types::PayoutsRouterData<F>) -> Result<Self, Self::Error> {
+        let request = item.request.to_owned();
+        let customer_details = request.get_customer_details()?;
+        let customer_email = customer_details.get_email()?;
+        let address = item.get_billing_address()?.clone();
+        let payout_vendor_details = request.get_vendor_details()?;
+        let (vendor_details, individual_details) = (
+            payout_vendor_details.vendor_details,
+            payout_vendor_details.individual_details,
+        );
+        Ok(Self {
+            account_type: vendor_details.account_type,
+            country: address.country,
+            email: Some(customer_email.clone()),
+            capabilities_card_payments: vendor_details.capabilities_card_payments,
+            capabilities_transfers: vendor_details.capabilities_transfers,
+            tos_acceptance_date: individual_details.tos_acceptance_date,
+            tos_acceptance_ip: individual_details.tos_acceptance_ip,
+            business_type: vendor_details.business_type,
+            business_profile_mcc: vendor_details.business_profile_mcc,
+            business_profile_url: vendor_details.business_profile_url,
+            business_profile_name: vendor_details.business_profile_name.clone(),
+            company_name: vendor_details.business_profile_name,
+            company_address_line1: vendor_details.company_address_line1,
+            company_address_line2: vendor_details.company_address_line2,
+            company_address_postal_code: vendor_details.company_address_postal_code,
+            company_address_city: vendor_details.company_address_city,
+            company_address_state: vendor_details.company_address_state,
+            company_phone: vendor_details.company_phone,
+            company_tax_id: vendor_details.company_tax_id,
+            company_owners_provided: vendor_details.company_owners_provided,
+            individual_first_name: address.first_name,
+            individual_last_name: address.last_name,
+            individual_dob_day: individual_details.individual_dob_day,
+            individual_dob_month: individual_details.individual_dob_month,
+            individual_dob_year: individual_details.individual_dob_year,
+            individual_address_line1: address.line1,
+            individual_address_line2: address.line2,
+            individual_address_postal_code: address.zip,
+            individual_address_city: address.city,
+            individual_address_state: address.state,
+            individual_email: Some(customer_email),
+            individual_phone: customer_details.phone,
+            individual_id_number: individual_details.individual_id_number,
+            individual_ssn_last_4: individual_details.individual_ssn_last_4,
+        })
+    }
+}
+
+// Recipient creation response transform
+impl<F> TryFrom<types::PayoutsResponseRouterData<F, StripeConnectRecipientCreateResponse>>
+    for types::PayoutsRouterData<F>
+{
+    type Error = Error;
+    fn try_from(
+        item: types::PayoutsResponseRouterData<F, StripeConnectRecipientCreateResponse>,
+    ) -> Result<Self, Self::Error> {
+        let response: StripeConnectRecipientCreateResponse = item.response;
+
+        Ok(Self {
+            response: Ok(types::PayoutsResponseData {
+                status: Some(enums::PayoutStatus::RequiresVendorAccountCreation),
+                connector_payout_id: response.id,
+                payout_eligible: None,
+                should_add_next_step_to_process_tracker: true,
+            }),
+            ..item.data
+        })
+    }
+}
+
+// Recipient account's creation request
+impl<F> TryFrom<&types::PayoutsRouterData<F>> for StripeConnectRecipientAccountCreateRequest {
+    type Error = Error;
+    fn try_from(item: &types::PayoutsRouterData<F>) -> Result<Self, Self::Error> {
+        let request = item.request.to_owned();
+        let payout_method_data = item.get_payout_method_data()?;
+        let customer_details = request.get_customer_details()?;
+        let customer_name = customer_details.get_name()?;
+        let payout_vendor_details = request.get_vendor_details()?;
+        match payout_method_data {
+            api_models::payouts::PayoutMethodData::Card(_) => {
+                Ok(Self::Token(RecipientTokenRequest {
+                    external_account: "tok_visa_debit".to_string(),
+                }))
+            }
+            api_models::payouts::PayoutMethodData::Bank(bank) => match bank {
+                api_models::payouts::Bank::Ach(bank_details) => {
+                    Ok(Self::Bank(RecipientBankAccountRequest {
+                        external_account_object: "bank_account".to_string(),
+                        external_account_country: bank_details
+                            .bank_country_code
+                            .get_required_value("bank_country_code")
+                            .change_context(errors::ConnectorError::MissingRequiredField {
+                                field_name: "bank_country_code",
+                            })?,
+                        external_account_currency: request.destination_currency.to_owned(),
+                        external_account_account_holder_name: customer_name,
+                        external_account_account_holder_type: payout_vendor_details
+                            .individual_details
+                            .get_external_account_account_holder_type()?,
+                        external_account_account_number: bank_details.bank_account_number,
+                        external_account_routing_number: bank_details.bank_routing_number,
+                    }))
+                }
+                api_models::payouts::Bank::Bacs(_) => Err(errors::ConnectorError::NotSupported {
+                    message: "BACS payouts are not supported".to_string(),
+                    connector: "stripe",
+                }
+                .into()),
+                api_models::payouts::Bank::Sepa(_) => Err(errors::ConnectorError::NotSupported {
+                    message: "SEPA payouts are not supported".to_string(),
+                    connector: "stripe",
+                }
+                .into()),
+            },
+            api_models::payouts::PayoutMethodData::Wallet(_) => {
+                Err(errors::ConnectorError::NotSupported {
+                    message: "Payouts via wallets are not supported".to_string(),
+                    connector: "stripe",
+                }
+                .into())
+            }
+        }
+    }
+}
+
+// Recipient account's creation response
+impl<F> TryFrom<types::PayoutsResponseRouterData<F, StripeConnectRecipientAccountCreateResponse>>
+    for types::PayoutsRouterData<F>
+{
+    type Error = Error;
+    fn try_from(
+        item: types::PayoutsResponseRouterData<F, StripeConnectRecipientAccountCreateResponse>,
+    ) -> Result<Self, Self::Error> {
+        let response: StripeConnectRecipientAccountCreateResponse = item.response;
+
+        Ok(Self {
+            response: Ok(types::PayoutsResponseData {
+                status: Some(enums::PayoutStatus::RequiresCreation),
+                connector_payout_id: response.id,
+                payout_eligible: None,
+                should_add_next_step_to_process_tracker: false,
+            }),
+            ..item.data
+        })
+    }
+}
+
+impl From<StripeConnectPayoutStatus> for enums::PayoutStatus {
+    fn from(stripe_connect_status: StripeConnectPayoutStatus) -> Self {
+        match stripe_connect_status {
+            StripeConnectPayoutStatus::Paid => Self::Success,
+            StripeConnectPayoutStatus::Failed => Self::Failed,
+            StripeConnectPayoutStatus::Canceled => Self::Cancelled,
+            StripeConnectPayoutStatus::Pending | StripeConnectPayoutStatus::InTransit => {
+                Self::Pending
+            }
+        }
+    }
+}
diff --git a/crates/router/src/connector/utils.rs b/crates/router/src/connector/utils.rs
index 6d427c54198..aa4307661e7 100644
--- a/crates/router/src/connector/utils.rs
+++ b/crates/router/src/connector/utils.rs
@@ -1,5 +1,7 @@
 use std::collections::HashMap;
 
+#[cfg(feature = "payouts")]
+use api_models::payouts::PayoutVendorAccountDetails;
 use api_models::{
     enums::{CanadaStatesAbbreviation, UsStatesAbbreviation},
     payments::{self, OrderDetailsWithAmount},
@@ -18,6 +20,8 @@ use once_cell::sync::Lazy;
 use regex::Regex;
 use serde::Serializer;
 use time::PrimitiveDateTime;
+#[cfg(feature = "payouts")]
+use types::CustomerDetails;
 
 #[cfg(feature = "frm")]
 use crate::types::{fraud_check, storage::enums as storage_enums};
@@ -898,6 +902,32 @@ impl RefundsRequestData for types::RefundsData {
     }
 }
 
+#[cfg(feature = "payouts")]
+pub trait PayoutsData {
+    fn get_transfer_id(&self) -> Result<String, Error>;
+    fn get_customer_details(&self) -> Result<CustomerDetails, Error>;
+    fn get_vendor_details(&self) -> Result<PayoutVendorAccountDetails, Error>;
+}
+
+#[cfg(feature = "payouts")]
+impl PayoutsData for types::PayoutsData {
+    fn get_transfer_id(&self) -> Result<String, Error> {
+        self.connector_payout_id
+            .clone()
+            .ok_or_else(missing_field_err("transfer_id"))
+    }
+    fn get_customer_details(&self) -> Result<CustomerDetails, Error> {
+        self.customer_details
+            .clone()
+            .ok_or_else(missing_field_err("customer_details"))
+    }
+    fn get_vendor_details(&self) -> Result<PayoutVendorAccountDetails, Error> {
+        self.vendor_details
+            .clone()
+            .ok_or_else(missing_field_err("vendor_details"))
+    }
+}
+
 #[derive(Clone, Debug, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GooglePayWalletData {
diff --git a/crates/router/src/connector/wise/transformers.rs b/crates/router/src/connector/wise/transformers.rs
index 5416db2d439..c3347fd9387 100644
--- a/crates/router/src/connector/wise/transformers.rs
+++ b/crates/router/src/connector/wise/transformers.rs
@@ -13,7 +13,6 @@ use crate::{
     types::{
         api::payouts,
         storage::enums::{self as storage_enums, PayoutEntityType},
-        transformers::ForeignFrom,
     },
 };
 use crate::{core::errors, types};
@@ -320,7 +319,7 @@ fn get_payout_bank_details(
     }?;
     match payout_method_data {
         PayoutMethodData::Bank(payouts::BankPayout::Ach(b)) => Ok(WiseBankDetails {
-            legal_type: LegalType::foreign_from(entity_type),
+            legal_type: LegalType::from(entity_type),
             address: Some(wise_address_details),
             account_number: Some(b.bank_account_number.to_owned()),
             abartn: Some(b.bank_routing_number),
@@ -328,14 +327,14 @@ fn get_payout_bank_details(
             ..WiseBankDetails::default()
         }),
         PayoutMethodData::Bank(payouts::BankPayout::Bacs(b)) => Ok(WiseBankDetails {
-            legal_type: LegalType::foreign_from(entity_type),
+            legal_type: LegalType::from(entity_type),
             address: Some(wise_address_details),
             account_number: Some(b.bank_account_number.to_owned()),
             sort_code: Some(b.bank_sort_code),
             ..WiseBankDetails::default()
         }),
         PayoutMethodData::Bank(payouts::BankPayout::Sepa(b)) => Ok(WiseBankDetails {
-            legal_type: LegalType::foreign_from(entity_type),
+            legal_type: LegalType::from(entity_type),
             address: Some(wise_address_details),
             iban: Some(b.iban.to_owned()),
             bic: b.bic,
@@ -409,6 +408,7 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, WiseRecipientCreateResponse>
                 status: Some(storage_enums::PayoutStatus::RequiresCreation),
                 connector_payout_id: response.id.to_string(),
                 payout_eligible: None,
+                should_add_next_step_to_process_tracker: false,
             }),
             ..item.data
         })
@@ -454,6 +454,7 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, WisePayoutQuoteResponse>>
                 status: Some(storage_enums::PayoutStatus::RequiresCreation),
                 connector_payout_id: response.id,
                 payout_eligible: None,
+                should_add_next_step_to_process_tracker: false,
             }),
             ..item.data
         })
@@ -506,7 +507,7 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, WisePayoutResponse>>
         item: types::PayoutsResponseRouterData<F, WisePayoutResponse>,
     ) -> Result<Self, Self::Error> {
         let response: WisePayoutResponse = item.response;
-        let status = match storage_enums::PayoutStatus::foreign_from(response.status) {
+        let status = match storage_enums::PayoutStatus::from(response.status) {
             storage_enums::PayoutStatus::Cancelled => storage_enums::PayoutStatus::Cancelled,
             _ => storage_enums::PayoutStatus::RequiresFulfillment,
         };
@@ -516,6 +517,7 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, WisePayoutResponse>>
                 status: Some(status),
                 connector_payout_id: response.id.to_string(),
                 payout_eligible: None,
+                should_add_next_step_to_process_tracker: false,
             }),
             ..item.data
         })
@@ -554,9 +556,10 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, WiseFulfillResponse>>
 
         Ok(Self {
             response: Ok(types::PayoutsResponseData {
-                status: Some(storage_enums::PayoutStatus::foreign_from(response.status)),
+                status: Some(storage_enums::PayoutStatus::from(response.status)),
                 connector_payout_id: "".to_string(),
                 payout_eligible: None,
+                should_add_next_step_to_process_tracker: false,
             }),
             ..item.data
         })
@@ -564,8 +567,8 @@ impl<F> TryFrom<types::PayoutsResponseRouterData<F, WiseFulfillResponse>>
 }
 
 #[cfg(feature = "payouts")]
-impl ForeignFrom<WiseStatus> for storage_enums::PayoutStatus {
-    fn foreign_from(wise_status: WiseStatus) -> Self {
+impl From<WiseStatus> for storage_enums::PayoutStatus {
+    fn from(wise_status: WiseStatus) -> Self {
         match wise_status {
             WiseStatus::Completed => Self::Success,
             WiseStatus::Rejected => Self::Failed,
@@ -578,8 +581,8 @@ impl ForeignFrom<WiseStatus> for storage_enums::PayoutStatus {
 }
 
 #[cfg(feature = "payouts")]
-impl ForeignFrom<PayoutEntityType> for LegalType {
-    fn foreign_from(entity_type: PayoutEntityType) -> Self {
+impl From<PayoutEntityType> for LegalType {
+    fn from(entity_type: PayoutEntityType) -> Self {
         match entity_type {
             PayoutEntityType::Individual
             | PayoutEntityType::Personal
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 2eb080f952c..b3828e441d6 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -59,6 +59,7 @@ use super::{
 use crate::core::fraud_check as frm_core;
 use crate::{
     configs::settings::{ApplePayPreDecryptFlow, PaymentMethodTypeTokenFilter},
+    connector::utils::missing_field_err,
     core::{
         authentication as authentication_core,
         errors::{self, CustomResult, RouterResponse, RouterResult},
@@ -2549,6 +2550,22 @@ pub struct CustomerDetails {
     pub phone_country_code: Option<String>,
 }
 
+pub trait CustomerDetailsExt {
+    type Error;
+    fn get_name(&self) -> Result<Secret<String, masking::WithType>, Self::Error>;
+    fn get_email(&self) -> Result<pii::Email, Self::Error>;
+}
+
+impl CustomerDetailsExt for CustomerDetails {
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn get_name(&self) -> Result<Secret<String, masking::WithType>, Self::Error> {
+        self.name.clone().ok_or_else(missing_field_err("name"))
+    }
+    fn get_email(&self) -> Result<pii::Email, Self::Error> {
+        self.email.clone().ok_or_else(missing_field_err("email"))
+    }
+}
+
 pub fn if_not_create_change_operation<'a, Op, F, Ctx>(
     status: storage_enums::IntentStatus,
     confirm: Option<bool>,
diff --git a/crates/router/src/core/payments/flows.rs b/crates/router/src/core/payments/flows.rs
index 59c0b687d64..e2c17a37d92 100644
--- a/crates/router/src/core/payments/flows.rs
+++ b/crates/router/src/core/payments/flows.rs
@@ -1008,7 +1008,6 @@ default_imp_for_payouts!(
     connector::Signifyd,
     connector::Square,
     connector::Stax,
-    connector::Stripe,
     connector::Shift4,
     connector::Threedsecureio,
     connector::Trustpay,
@@ -1095,7 +1094,6 @@ default_imp_for_payouts_create!(
     connector::Signifyd,
     connector::Square,
     connector::Stax,
-    connector::Stripe,
     connector::Shift4,
     connector::Threedsecureio,
     connector::Trustpay,
@@ -1272,7 +1270,6 @@ default_imp_for_payouts_fulfill!(
     connector::Signifyd,
     connector::Square,
     connector::Stax,
-    connector::Stripe,
     connector::Shift4,
     connector::Threedsecureio,
     connector::Trustpay,
@@ -1359,7 +1356,6 @@ default_imp_for_payouts_cancel!(
     connector::Signifyd,
     connector::Square,
     connector::Stax,
-    connector::Stripe,
     connector::Shift4,
     connector::Threedsecureio,
     connector::Trustpay,
@@ -1535,7 +1531,6 @@ default_imp_for_payouts_recipient!(
     connector::Signifyd,
     connector::Square,
     connector::Stax,
-    connector::Stripe,
     connector::Shift4,
     connector::Threedsecureio,
     connector::Trustpay,
@@ -1547,6 +1542,97 @@ default_imp_for_payouts_recipient!(
     connector::Zsl
 );
 
+#[cfg(feature = "payouts")]
+macro_rules! default_imp_for_payouts_recipient_account {
+    ($($path:ident::$connector:ident),*) => {
+        $(
+            impl api::PayoutRecipientAccount for $path::$connector {}
+            impl
+            services::ConnectorIntegration<
+            api::PoRecipientAccount,
+            types::PayoutsData,
+            types::PayoutsResponseData,
+        > for $path::$connector
+        {}
+    )*
+    };
+}
+
+#[cfg(feature = "payouts")]
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8> api::PayoutRecipientAccount for connector::DummyConnector<T> {}
+#[cfg(feature = "payouts")]
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8>
+    services::ConnectorIntegration<
+        api::PoRecipientAccount,
+        types::PayoutsData,
+        types::PayoutsResponseData,
+    > for connector::DummyConnector<T>
+{
+}
+
+#[cfg(feature = "payouts")]
+default_imp_for_payouts_recipient_account!(
+    connector::Aci,
+    connector::Adyen,
+    connector::Airwallex,
+    connector::Authorizedotnet,
+    connector::Bambora,
+    connector::Bankofamerica,
+    connector::Billwerk,
+    connector::Bitpay,
+    connector::Bluesnap,
+    connector::Boku,
+    connector::Braintree,
+    connector::Cashtocode,
+    connector::Checkout,
+    connector::Cryptopay,
+    connector::Cybersource,
+    connector::Coinbase,
+    connector::Dlocal,
+    connector::Ebanx,
+    connector::Fiserv,
+    connector::Forte,
+    connector::Globalpay,
+    connector::Globepay,
+    connector::Gocardless,
+    connector::Helcim,
+    connector::Iatapay,
+    connector::Klarna,
+    connector::Mollie,
+    connector::Multisafepay,
+    connector::Netcetera,
+    connector::Nexinets,
+    connector::Nmi,
+    connector::Noon,
+    connector::Nuvei,
+    connector::Opayo,
+    connector::Opennode,
+    connector::Payeezy,
+    connector::Payme,
+    connector::Paypal,
+    connector::Payu,
+    connector::Placetopay,
+    connector::Powertranz,
+    connector::Prophetpay,
+    connector::Rapyd,
+    connector::Riskified,
+    connector::Signifyd,
+    connector::Square,
+    connector::Stax,
+    connector::Shift4,
+    connector::Threedsecureio,
+    connector::Trustpay,
+    connector::Tsys,
+    connector::Volt,
+    connector::Wise,
+    connector::Worldline,
+    connector::Worldpay,
+    connector::Zen,
+    connector::Zsl
+);
+
 macro_rules! default_imp_for_approve {
     ($($path:ident::$connector:ident),*) => {
         $(
diff --git a/crates/router/src/core/payouts.rs b/crates/router/src/core/payouts.rs
index 1d791e78677..bdaa0c01a10 100644
--- a/crates/router/src/core/payouts.rs
+++ b/crates/router/src/core/payouts.rs
@@ -3,31 +3,37 @@ pub mod helpers;
 #[cfg(feature = "payout_retry")]
 pub mod retry;
 pub mod validator;
-
 use std::vec::IntoIter;
 
 use api_models::enums as api_enums;
-use common_utils::{crypto::Encryptable, ext_traits::ValueExt, pii};
+use common_utils::{consts, crypto::Encryptable, ext_traits::ValueExt, pii};
 #[cfg(feature = "olap")]
 use data_models::errors::StorageError;
 use diesel_models::enums as storage_enums;
 use error_stack::{report, ResultExt};
 #[cfg(feature = "olap")]
 use futures::future::join_all;
+#[cfg(feature = "payout_retry")]
+use retry::GsmValidation;
 #[cfg(feature = "olap")]
 use router_env::logger;
 use router_env::{instrument, tracing};
+use scheduler::utils as pt_utils;
 use serde_json;
 
-use super::errors::{ConnectorErrorExt, StorageErrorExt};
+use super::{
+    errors::{ConnectorErrorExt, StorageErrorExt},
+    payments::customers,
+};
 #[cfg(feature = "olap")]
 use crate::types::{domain::behaviour::Conversion, transformers::ForeignFrom};
 use crate::{
     core::{
-        errors::{self, RouterResponse, RouterResult},
+        errors::{self, CustomResult, RouterResponse, RouterResult},
         payments::{self, helpers as payment_helpers},
         utils as core_utils,
     },
+    db::StorageInterface,
     routes::AppState,
     services,
     types::{
@@ -50,6 +56,7 @@ pub struct PayoutData {
     pub payout_attempt: storage::PayoutAttempt,
     pub payout_method_data: Option<payouts::PayoutMethodData>,
     pub profile_id: String,
+    pub should_terminate: bool,
 }
 
 // ********************************************** CORE FLOWS **********************************************
@@ -143,23 +150,20 @@ pub async fn get_connector_choice(
     }
 }
 
-#[cfg(feature = "payouts")]
 #[instrument(skip_all)]
 pub async fn make_connector_decision(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutCreateRequest,
     connector_call_type: api::ConnectorCallType,
-    mut payout_data: PayoutData,
-) -> RouterResult<PayoutData> {
+    payout_data: &mut PayoutData,
+) -> RouterResult<()> {
     match connector_call_type {
         api::ConnectorCallType::PreDetermined(connector_data) => {
-            payout_data = call_connector_payout(
+            call_connector_payout(
                 state,
                 merchant_account,
                 key_store,
-                req,
                 &connector_data,
                 payout_data,
             )
@@ -167,7 +171,6 @@ pub async fn make_connector_decision(
 
             #[cfg(feature = "payout_retry")]
             {
-                use crate::core::payouts::retry::GsmValidation;
                 let config_bool = retry::config_should_call_gsm_payout(
                     &*state.store,
                     &merchant_account.merchant_id,
@@ -176,30 +179,28 @@ pub async fn make_connector_decision(
                 .await;
 
                 if config_bool && payout_data.should_call_gsm() {
-                    payout_data = Box::pin(retry::do_gsm_single_connector_actions(
+                    Box::pin(retry::do_gsm_single_connector_actions(
                         state,
                         connector_data,
                         payout_data,
                         merchant_account,
                         key_store,
-                        req,
                     ))
                     .await?;
                 }
             }
 
-            Ok(payout_data)
+            Ok(())
         }
         api::ConnectorCallType::Retryable(connectors) => {
             let mut connectors = connectors.into_iter();
 
             let connector_data = get_next_connector(&mut connectors)?;
 
-            payout_data = call_connector_payout(
+            call_connector_payout(
                 state,
                 merchant_account,
                 key_store,
-                req,
                 &connector_data,
                 payout_data,
             )
@@ -207,7 +208,6 @@ pub async fn make_connector_decision(
 
             #[cfg(feature = "payout_retry")]
             {
-                use crate::core::payouts::retry::GsmValidation;
                 let config_multiple_connector_bool = retry::config_should_call_gsm_payout(
                     &*state.store,
                     &merchant_account.merchant_id,
@@ -216,14 +216,13 @@ pub async fn make_connector_decision(
                 .await;
 
                 if config_multiple_connector_bool && payout_data.should_call_gsm() {
-                    payout_data = Box::pin(retry::do_gsm_multiple_connector_actions(
+                    Box::pin(retry::do_gsm_multiple_connector_actions(
                         state,
                         connectors,
                         connector_data.clone(),
                         payout_data,
                         merchant_account,
                         key_store,
-                        req,
                     ))
                     .await?;
                 }
@@ -236,24 +235,57 @@ pub async fn make_connector_decision(
                 .await;
 
                 if config_single_connector_bool && payout_data.should_call_gsm() {
-                    payout_data = Box::pin(retry::do_gsm_single_connector_actions(
+                    Box::pin(retry::do_gsm_single_connector_actions(
                         state,
                         connector_data,
                         payout_data,
                         merchant_account,
                         key_store,
-                        req,
                     ))
                     .await?;
                 }
             }
 
-            Ok(payout_data)
+            Ok(())
         }
         _ => Err(errors::ApiErrorResponse::InternalServerError)?,
     }
 }
 
+#[instrument(skip_all)]
+pub async fn payouts_core(
+    state: &AppState,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    payout_data: &mut PayoutData,
+    routing_algorithm: Option<serde_json::Value>,
+    eligible_connectors: Option<Vec<api_models::enums::PayoutConnectors>>,
+) -> RouterResult<()> {
+    let payout_attempt = &payout_data.payout_attempt;
+
+    // Form connector data
+    let connector_call_type = get_connector_choice(
+        state,
+        merchant_account,
+        key_store,
+        payout_attempt.connector.clone(),
+        routing_algorithm,
+        payout_data,
+        eligible_connectors,
+    )
+    .await?;
+
+    // Call connector steps
+    Box::pin(make_connector_decision(
+        state,
+        merchant_account,
+        key_store,
+        connector_call_type,
+        payout_data,
+    ))
+    .await
+}
+
 #[instrument(skip_all)]
 pub async fn payouts_create_core(
     state: AppState,
@@ -277,34 +309,33 @@ pub async fn payouts_create_core(
     )
     .await?;
 
-    let connector_call_type = get_connector_choice(
+    let payout_attempt = payout_data.payout_attempt.to_owned();
+
+    // Persist payout method data in temp locker
+    payout_data.payout_method_data = helpers::make_payout_method_data(
         &state,
-        &merchant_account,
+        req.payout_method_data.as_ref(),
+        payout_attempt.payout_token.as_deref(),
+        &payout_attempt.customer_id,
+        &payout_attempt.merchant_id,
+        Some(&payout_data.payouts.payout_type.clone()),
         &key_store,
-        None,
-        req.routing.clone(),
-        &mut payout_data,
-        req.connector.clone(),
+        Some(&mut payout_data),
+        merchant_account.storage_scheme,
     )
     .await?;
 
-    payout_data = Box::pin(make_connector_decision(
+    payouts_core(
         &state,
         &merchant_account,
         &key_store,
-        &req,
-        connector_call_type,
-        payout_data,
-    ))
+        &mut payout_data,
+        req.routing.clone(),
+        req.connector.clone(),
+    )
     .await?;
 
-    response_handler(
-        &state,
-        &merchant_account,
-        &payouts::PayoutRequest::PayoutCreateRequest(req.to_owned()),
-        &payout_data,
-    )
-    .await
+    response_handler(&merchant_account, &payout_data).await
 }
 
 pub async fn payouts_update_core(
@@ -349,6 +380,7 @@ pub async fn payouts_update_core(
         metadata: req.metadata.clone().or(payouts.metadata.clone()),
         status: Some(status),
         profile_id: Some(payout_attempt.profile_id.clone()),
+        confirm: req.confirm,
     };
 
     let db = &*state.store;
@@ -400,61 +432,39 @@ pub async fn payouts_update_core(
         }
     }
 
-    if (
-        req.connector.is_none(),
-        payout_data.payout_attempt.connector.is_some(),
-    ) != (true, true)
-    {
+    let payout_attempt = payout_data.payout_attempt.to_owned();
+
+    if (req.connector.is_none(), payout_attempt.connector.is_some()) != (true, true) {
         // if the connector is not updated but was provided during payout create
         payout_data.payout_attempt.connector = None;
         payout_data.payout_attempt.routing_info = None;
-
-        //fetch payout_method_data
-        payout_data.payout_method_data = Some(
-            helpers::make_payout_method_data(
-                &state,
-                req.payout_method_data.as_ref(),
-                payout_data.payout_attempt.payout_token.clone().as_deref(),
-                &payout_data.payout_attempt.customer_id.clone(),
-                &payout_data.payout_attempt.merchant_id.clone(),
-                Some(&payouts.payout_type),
-                &key_store,
-                Some(&mut payout_data),
-                merchant_account.storage_scheme,
-            )
-            .await?
-            .get_required_value("payout_method_data")?,
-        );
     };
 
-    let connector_call_type = get_connector_choice(
+    // Update payout method data in temp locker
+    payout_data.payout_method_data = helpers::make_payout_method_data(
         &state,
-        &merchant_account,
+        req.payout_method_data.as_ref(),
+        payout_attempt.payout_token.as_deref(),
+        &payout_attempt.customer_id,
+        &payout_attempt.merchant_id,
+        Some(&payout_data.payouts.payout_type.clone()),
         &key_store,
-        None,
-        req.routing.clone(),
-        &mut payout_data,
-        req.connector.clone(),
+        Some(&mut payout_data),
+        merchant_account.storage_scheme,
     )
     .await?;
 
-    payout_data = Box::pin(make_connector_decision(
+    payouts_core(
         &state,
         &merchant_account,
         &key_store,
-        &req,
-        connector_call_type,
-        payout_data,
-    ))
+        &mut payout_data,
+        req.routing.clone(),
+        req.connector.clone(),
+    )
     .await?;
 
-    response_handler(
-        &state,
-        &merchant_account,
-        &payouts::PayoutRequest::PayoutCreateRequest(req.to_owned()),
-        &payout_data,
-    )
-    .await
+    response_handler(&merchant_account, &payout_data).await
 }
 
 #[instrument(skip_all)]
@@ -472,13 +482,7 @@ pub async fn payouts_retrieve_core(
     )
     .await?;
 
-    response_handler(
-        &state,
-        &merchant_account,
-        &payouts::PayoutRequest::PayoutRetrieveRequest(req.to_owned()),
-        &payout_data,
-    )
-    .await
+    response_handler(&merchant_account, &payout_data).await
 }
 
 #[instrument(skip_all)]
@@ -563,11 +567,10 @@ pub async fn payouts_cancel_core(
             .attach_printable("Connector not found for payout cancellation")?,
         };
 
-        payout_data = cancel_payout(
+        cancel_payout(
             &state,
             &merchant_account,
             &key_store,
-            &payouts::PayoutRequest::PayoutActionRequest(req.to_owned()),
             &connector_data,
             &mut payout_data,
         )
@@ -575,13 +578,7 @@ pub async fn payouts_cancel_core(
         .attach_printable("Payout cancellation failed for given Payout request")?;
     }
 
-    response_handler(
-        &state,
-        &merchant_account,
-        &payouts::PayoutRequest::PayoutActionRequest(req.to_owned()),
-        &payout_data,
-    )
-    .await
+    response_handler(&merchant_account, &payout_data).await
 }
 
 #[instrument(skip_all)]
@@ -649,11 +646,10 @@ pub async fn payouts_fulfill_core(
         .await?
         .get_required_value("payout_method_data")?,
     );
-    payout_data = fulfill_payout(
+    fulfill_payout(
         &state,
         &merchant_account,
         &key_store,
-        &payouts::PayoutRequest::PayoutActionRequest(req.to_owned()),
         &connector_data,
         &mut payout_data,
     )
@@ -668,13 +664,7 @@ pub async fn payouts_fulfill_core(
         }));
     }
 
-    response_handler(
-        &state,
-        &merchant_account,
-        &payouts::PayoutRequest::PayoutActionRequest(req.to_owned()),
-        &payout_data,
-    )
-    .await
+    response_handler(&merchant_account, &payout_data).await
 }
 
 #[cfg(feature = "olap")]
@@ -857,10 +847,9 @@ pub async fn call_connector_payout(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutCreateRequest,
     connector_data: &api::ConnectorData,
-    mut payout_data: PayoutData,
-) -> RouterResult<PayoutData> {
+    payout_data: &mut PayoutData,
+) -> RouterResult<()> {
     let payout_attempt = &payout_data.payout_attempt.to_owned();
     let payouts = &payout_data.payouts.to_owned();
 
@@ -891,13 +880,13 @@ pub async fn call_connector_payout(
         payout_data.payout_method_data = Some(
             helpers::make_payout_method_data(
                 state,
-                req.payout_method_data.as_ref(),
+                payout_data.payout_method_data.to_owned().as_ref(),
                 payout_attempt.payout_token.as_deref(),
                 &payout_attempt.customer_id,
                 &payout_attempt.merchant_id,
                 Some(&payouts.payout_type),
                 key_store,
-                Some(&mut payout_data),
+                Some(payout_data),
                 merchant_account.storage_scheme,
             )
             .await?
@@ -905,35 +894,42 @@ pub async fn call_connector_payout(
         );
     }
 
-    if let Some(true) = req.confirm {
+    if let Some(true) = payouts.confirm {
         // Eligibility flow
-        payout_data = complete_payout_eligibility(
+        complete_payout_eligibility(
             state,
             merchant_account,
             key_store,
-            req,
             connector_data,
             payout_data,
         )
         .await?;
 
         // Create customer flow
-        payout_data = complete_create_recipient(
+        complete_create_recipient(
+            state,
+            merchant_account,
+            key_store,
+            connector_data,
+            payout_data,
+        )
+        .await?;
+
+        // Create customer's disbursement account flow
+        complete_create_recipient_disburse_account(
             state,
             merchant_account,
             key_store,
-            req,
             connector_data,
             payout_data,
         )
         .await?;
 
         // Payout creation flow
-        payout_data = complete_create_payout(
+        complete_create_payout(
             state,
             merchant_account,
             key_store,
-            req,
             connector_data,
             payout_data,
         )
@@ -943,57 +939,54 @@ pub async fn call_connector_payout(
     // Auto fulfillment flow
     let status = payout_data.payout_attempt.status;
     if payouts.auto_fulfill && status == storage_enums::PayoutStatus::RequiresFulfillment {
-        payout_data = fulfill_payout(
+        fulfill_payout(
             state,
             merchant_account,
             key_store,
-            &payouts::PayoutRequest::PayoutCreateRequest(req.to_owned()),
             connector_data,
-            &mut payout_data,
+            payout_data,
         )
         .await
         .attach_printable("Payout fulfillment failed for given Payout request")?;
     }
 
-    Ok(payout_data)
+    Ok(())
 }
 
 pub async fn complete_create_recipient(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutCreateRequest,
     connector_data: &api::ConnectorData,
-    mut payout_data: PayoutData,
-) -> RouterResult<PayoutData> {
-    if payout_data.payout_attempt.status == storage_enums::PayoutStatus::RequiresCreation
+    payout_data: &mut PayoutData,
+) -> RouterResult<()> {
+    if !payout_data.should_terminate
+        && payout_data.payout_attempt.status == storage_enums::PayoutStatus::RequiresCreation
         && connector_data
             .connector_name
             .supports_create_recipient(payout_data.payouts.payout_type)
     {
-        payout_data = create_recipient(
+        create_recipient(
             state,
             merchant_account,
             key_store,
-            req,
             connector_data,
-            &mut payout_data,
+            payout_data,
         )
         .await
         .attach_printable("Creation of customer failed")?;
     }
 
-    Ok(payout_data)
+    Ok(())
 }
 
 pub async fn create_recipient(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutCreateRequest,
     connector_data: &api::ConnectorData,
     payout_data: &mut PayoutData,
-) -> RouterResult<PayoutData> {
+) -> RouterResult<()> {
     let customer_details = payout_data.customer_details.to_owned();
     let connector_name = connector_data.connector_name.to_string();
 
@@ -1009,12 +1002,11 @@ pub async fn create_recipient(
         );
     if should_call_connector {
         // 1. Form router data
-        let customer_router_data = core_utils::construct_payout_router_data(
+        let router_data = core_utils::construct_payout_router_data(
             state,
-            &connector_name,
+            &connector_data.connector_name,
             merchant_account,
             key_store,
-            &payouts::PayoutRequest::PayoutCreateRequest(req.to_owned()),
             payout_data,
         )
         .await?;
@@ -1030,8 +1022,8 @@ pub async fn create_recipient(
         // 3. Call connector service
         let router_resp = services::execute_connector_processing_step(
             state,
-            connector_integration,
-            &customer_router_data,
+            connector_integration.to_owned(),
+            &router_data,
             payments::CallConnectorAction::Trigger,
             None,
         )
@@ -1040,28 +1032,79 @@ pub async fn create_recipient(
 
         match router_resp.response {
             Ok(recipient_create_data) => {
+                let db = &*state.store;
                 if let Some(customer) = customer_details {
-                    let db = &*state.store;
                     let customer_id = customer.customer_id.to_owned();
                     let merchant_id = merchant_account.merchant_id.to_owned();
-                    let updated_customer = storage::CustomerUpdate::ConnectorCustomer {
-                        connector_customer: Some(
-                            serde_json::json!({connector_label: recipient_create_data.connector_payout_id}),
-                        ),
+                    if let Some(updated_customer) =
+                        customers::update_connector_customer_in_customers(
+                            &connector_label,
+                            Some(&customer),
+                            &Some(recipient_create_data.connector_payout_id.clone()),
+                        )
+                        .await
+                    {
+                        payout_data.customer_details = Some(
+                            db.update_customer_by_customer_id_merchant_id(
+                                customer_id,
+                                merchant_id,
+                                customer,
+                                updated_customer,
+                                key_store,
+                                merchant_account.storage_scheme,
+                            )
+                            .await
+                            .change_context(errors::ApiErrorResponse::InternalServerError)
+                            .attach_printable("Error updating customers in db")?,
+                        )
+                    }
+                }
+
+                // Add next step to ProcessTracker
+                if recipient_create_data.should_add_next_step_to_process_tracker {
+                    add_external_account_addition_task(
+                        &*state.store,
+                        payout_data,
+                        common_utils::date_time::now().saturating_add(time::Duration::seconds(consts::STRIPE_ACCOUNT_ONBOARDING_DELAY_IN_SECONDS)),
+                    )
+                    .await
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("Failed while adding attach_payout_account_workflow workflow to process tracker")?;
+
+                    // Update payout status in DB
+                    let status = recipient_create_data
+                        .status
+                        .unwrap_or(api_enums::PayoutStatus::RequiresVendorAccountCreation);
+                    let updated_payout_attempt = storage::PayoutAttemptUpdate::StatusUpdate {
+                        connector_payout_id: recipient_create_data.connector_payout_id,
+                        status,
+                        error_code: None,
+                        error_message: None,
+                        is_eligible: recipient_create_data.payout_eligible,
                     };
-                    payout_data.customer_details = Some(
-                        db.update_customer_by_customer_id_merchant_id(
-                            customer_id,
-                            merchant_id,
-                            customer,
-                            updated_customer,
-                            key_store,
+                    payout_data.payout_attempt = db
+                        .update_payout_attempt(
+                            &payout_data.payout_attempt,
+                            updated_payout_attempt,
+                            &payout_data.payouts,
                             merchant_account.storage_scheme,
                         )
                         .await
                         .change_context(errors::ApiErrorResponse::InternalServerError)
-                        .attach_printable("Error updating customers in db")?,
-                    )
+                        .attach_printable("Error updating payout_attempt in db")?;
+                    payout_data.payouts = db
+                        .update_payout(
+                            &payout_data.payouts,
+                            storage::PayoutsUpdate::StatusUpdate { status },
+                            &payout_data.payout_attempt,
+                            merchant_account.storage_scheme,
+                        )
+                        .await
+                        .change_context(errors::ApiErrorResponse::InternalServerError)
+                        .attach_printable("Error updating payouts in db")?;
+
+                    // Helps callee functions skip the execution
+                    payout_data.should_terminate = true;
                 }
             }
             Err(err) => Err(errors::ApiErrorResponse::PayoutFailed {
@@ -1069,31 +1112,30 @@ pub async fn create_recipient(
             })?,
         }
     }
-    Ok(payout_data.clone())
+    Ok(())
 }
 
 pub async fn complete_payout_eligibility(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutCreateRequest,
     connector_data: &api::ConnectorData,
-    mut payout_data: PayoutData,
-) -> RouterResult<PayoutData> {
+    payout_data: &mut PayoutData,
+) -> RouterResult<()> {
     let payout_attempt = &payout_data.payout_attempt.to_owned();
 
-    if payout_attempt.is_eligible.is_none()
+    if !payout_data.should_terminate
+        && payout_attempt.is_eligible.is_none()
         && connector_data
             .connector_name
             .supports_payout_eligibility(payout_data.payouts.payout_type)
     {
-        payout_data = check_payout_eligibility(
+        check_payout_eligibility(
             state,
             merchant_account,
             key_store,
-            req,
             connector_data,
-            &mut payout_data,
+            payout_data,
         )
         .await
         .attach_printable("Eligibility failed for given Payout request")?;
@@ -1113,24 +1155,22 @@ pub async fn complete_payout_eligibility(
         },
     )?;
 
-    Ok(payout_data)
+    Ok(())
 }
 
 pub async fn check_payout_eligibility(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutCreateRequest,
     connector_data: &api::ConnectorData,
     payout_data: &mut PayoutData,
-) -> RouterResult<PayoutData> {
+) -> RouterResult<()> {
     // 1. Form Router data
     let router_data = core_utils::construct_payout_router_data(
         state,
-        &connector_data.connector_name.to_string(),
+        &connector_data.connector_name,
         merchant_account,
         key_store,
-        &payouts::PayoutRequest::PayoutCreateRequest(req.to_owned()),
         payout_data,
     )
     .await?;
@@ -1229,18 +1269,19 @@ pub async fn check_payout_eligibility(
         }
     };
 
-    Ok(payout_data.clone())
+    Ok(())
 }
 
 pub async fn complete_create_payout(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutCreateRequest,
     connector_data: &api::ConnectorData,
-    mut payout_data: PayoutData,
-) -> RouterResult<PayoutData> {
-    if payout_data.payout_attempt.status == storage_enums::PayoutStatus::RequiresCreation {
+    payout_data: &mut PayoutData,
+) -> RouterResult<()> {
+    if !payout_data.should_terminate
+        && payout_data.payout_attempt.status == storage_enums::PayoutStatus::RequiresCreation
+    {
         if connector_data
             .connector_name
             .supports_instant_payout(payout_data.payouts.payout_type)
@@ -1279,36 +1320,33 @@ pub async fn complete_create_payout(
                 .attach_printable("Error updating payouts in db")?;
         } else {
             // create payout_object in connector as well as router
-            payout_data = create_payout(
+            create_payout(
                 state,
                 merchant_account,
                 key_store,
-                req,
                 connector_data,
-                &mut payout_data,
+                payout_data,
             )
             .await
             .attach_printable("Payout creation failed for given Payout request")?;
         }
     }
-    Ok(payout_data)
+    Ok(())
 }
 
 pub async fn create_payout(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutCreateRequest,
     connector_data: &api::ConnectorData,
     payout_data: &mut PayoutData,
-) -> RouterResult<PayoutData> {
+) -> RouterResult<()> {
     // 1. Form Router data
     let mut router_data = core_utils::construct_payout_router_data(
         state,
-        &connector_data.connector_name.to_string(),
+        &connector_data.connector_name,
         merchant_account,
         key_store,
-        &payouts::PayoutRequest::PayoutCreateRequest(req.to_owned()),
         payout_data,
     )
     .await?;
@@ -1423,24 +1461,135 @@ pub async fn create_payout(
         }
     };
 
-    Ok(payout_data.clone())
+    Ok(())
+}
+
+pub async fn complete_create_recipient_disburse_account(
+    state: &AppState,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    connector_data: &api::ConnectorData,
+    payout_data: &mut PayoutData,
+) -> RouterResult<()> {
+    if !payout_data.should_terminate
+        && payout_data.payout_attempt.status
+            == storage_enums::PayoutStatus::RequiresVendorAccountCreation
+        && connector_data
+            .connector_name
+            .supports_vendor_disburse_account_create_for_payout()
+    {
+        create_recipient_disburse_account(
+            state,
+            merchant_account,
+            key_store,
+            connector_data,
+            payout_data,
+        )
+        .await
+        .attach_printable("Creation of customer failed")?;
+    }
+    Ok(())
+}
+
+pub async fn create_recipient_disburse_account(
+    state: &AppState,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    connector_data: &api::ConnectorData,
+    payout_data: &mut PayoutData,
+) -> RouterResult<()> {
+    // 1. Form Router data
+    let router_data = core_utils::construct_payout_router_data(
+        state,
+        &connector_data.connector_name,
+        merchant_account,
+        key_store,
+        payout_data,
+    )
+    .await?;
+
+    // 2. Fetch connector integration details
+    let connector_integration: services::BoxedConnectorIntegration<
+        '_,
+        api::PoRecipientAccount,
+        types::PayoutsData,
+        types::PayoutsResponseData,
+    > = connector_data.connector.get_connector_integration();
+
+    // 3. Call connector service
+    let router_data_resp = services::execute_connector_processing_step(
+        state,
+        connector_integration,
+        &router_data,
+        payments::CallConnectorAction::Trigger,
+        None,
+    )
+    .await
+    .to_payout_failed_response()?;
+
+    // 4. Process data returned by the connector
+    let db = &*state.store;
+    match router_data_resp.response {
+        Ok(payout_response_data) => {
+            let payout_attempt = &payout_data.payout_attempt;
+            let status = payout_response_data
+                .status
+                .unwrap_or(payout_attempt.status.to_owned());
+            let updated_payout_attempt = storage::PayoutAttemptUpdate::StatusUpdate {
+                connector_payout_id: payout_response_data.connector_payout_id,
+                status,
+                error_code: None,
+                error_message: None,
+                is_eligible: payout_response_data.payout_eligible,
+            };
+            payout_data.payout_attempt = db
+                .update_payout_attempt(
+                    payout_attempt,
+                    updated_payout_attempt,
+                    &payout_data.payouts,
+                    merchant_account.storage_scheme,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Error updating payout_attempt in db")?;
+        }
+        Err(err) => {
+            let updated_payout_attempt = storage::PayoutAttemptUpdate::StatusUpdate {
+                connector_payout_id: String::default(),
+                status: storage_enums::PayoutStatus::Failed,
+                error_code: Some(err.code),
+                error_message: Some(err.message),
+                is_eligible: None,
+            };
+            payout_data.payout_attempt = db
+                .update_payout_attempt(
+                    &payout_data.payout_attempt,
+                    updated_payout_attempt,
+                    &payout_data.payouts,
+                    merchant_account.storage_scheme,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Error updating payout_attempt in db")?;
+        }
+    };
+
+    Ok(())
 }
 
 pub async fn cancel_payout(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutRequest,
     connector_data: &api::ConnectorData,
     payout_data: &mut PayoutData,
-) -> RouterResult<PayoutData> {
+) -> RouterResult<()> {
     // 1. Form Router data
     let router_data = core_utils::construct_payout_router_data(
         state,
-        &connector_data.connector_name.to_string(),
+        &connector_data.connector_name,
         merchant_account,
         key_store,
-        req,
         payout_data,
     )
     .await?;
@@ -1531,24 +1680,22 @@ pub async fn cancel_payout(
         }
     };
 
-    Ok(payout_data.clone())
+    Ok(())
 }
 
 pub async fn fulfill_payout(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &payouts::PayoutRequest,
     connector_data: &api::ConnectorData,
     payout_data: &mut PayoutData,
-) -> RouterResult<PayoutData> {
+) -> RouterResult<()> {
     // 1. Form Router data
     let mut router_data = core_utils::construct_payout_router_data(
         state,
-        &connector_data.connector_name.to_string(),
+        &connector_data.connector_name,
         merchant_account,
         key_store,
-        req,
         payout_data,
     )
     .await?;
@@ -1673,13 +1820,11 @@ pub async fn fulfill_payout(
         }
     };
 
-    Ok(payout_data.clone())
+    Ok(())
 }
 
 pub async fn response_handler(
-    _state: &AppState,
     merchant_account: &domain::MerchantAccount,
-    _req: &payouts::PayoutRequest,
     payout_data: &PayoutData,
 ) -> RouterResponse<payouts::PayoutCreateResponse> {
     let payout_attempt = payout_data.payout_attempt.to_owned();
@@ -1837,6 +1982,8 @@ pub async fn payout_create_db_entries(
         payout_method_id,
         profile_id: profile_id.to_string(),
         attempt_count: 1,
+        metadata: req.metadata.clone(),
+        confirm: req.confirm,
         ..Default::default()
     };
     let payouts = db
@@ -1900,6 +2047,7 @@ pub async fn payout_create_db_entries(
             .as_ref()
             .cloned()
             .or(stored_payout_method_data.cloned()),
+        should_terminate: false,
         profile_id: profile_id.to_owned(),
     })
 }
@@ -1974,10 +2122,44 @@ pub async fn make_payout_data(
         payout_attempt,
         payout_method_data: None,
         merchant_connector_account: None,
+        should_terminate: false,
         profile_id,
     })
 }
 
+pub async fn add_external_account_addition_task(
+    db: &dyn StorageInterface,
+    payout_data: &PayoutData,
+    schedule_time: time::PrimitiveDateTime,
+) -> CustomResult<(), errors::StorageError> {
+    let runner = storage::ProcessTrackerRunner::AttachPayoutAccountWorkflow;
+    let task = "STRPE_ATTACH_EXTERNAL_ACCOUNT";
+    let tag = ["PAYOUTS", "STRIPE", "ACCOUNT", "CREATE"];
+    let process_tracker_id = pt_utils::get_process_tracker_id(
+        runner,
+        task,
+        &payout_data.payout_attempt.payout_attempt_id,
+        &payout_data.payout_attempt.merchant_id,
+    );
+    let tracking_data = api::PayoutRetrieveRequest {
+        payout_id: payout_data.payouts.payout_id.to_owned(),
+        force_sync: None,
+        merchant_id: Some(payout_data.payouts.merchant_id.to_owned()),
+    };
+    let process_tracker_entry = storage::ProcessTrackerNew::new(
+        process_tracker_id,
+        task,
+        runner,
+        tag,
+        tracking_data,
+        schedule_time,
+    )
+    .map_err(errors::StorageError::from)?;
+
+    db.insert_process(process_tracker_entry).await?;
+    Ok(())
+}
+
 async fn validate_and_get_business_profile(
     state: &AppState,
     profile_id: &String,
diff --git a/crates/router/src/core/payouts/retry.rs b/crates/router/src/core/payouts/retry.rs
index af8b016d884..6bbee9a4cec 100644
--- a/crates/router/src/core/payouts/retry.rs
+++ b/crates/router/src/core/payouts/retry.rs
@@ -1,6 +1,5 @@
 use std::{cmp::Ordering, str::FromStr, vec::IntoIter};
 
-use api_models::payouts::PayoutCreateRequest;
 use error_stack::{report, ResultExt};
 use router_env::{
     logger,
@@ -32,11 +31,10 @@ pub async fn do_gsm_multiple_connector_actions(
     state: &app::AppState,
     mut connectors: IntoIter<api::ConnectorData>,
     original_connector_data: api::ConnectorData,
-    mut payout_data: PayoutData,
+    payout_data: &mut PayoutData,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &PayoutCreateRequest,
-) -> RouterResult<PayoutData> {
+) -> RouterResult<()> {
     let mut retries = None;
 
     metrics::AUTO_PAYOUT_RETRY_ELIGIBLE_REQUEST_COUNT.add(&metrics::CONTEXT, 1, &[]);
@@ -44,7 +42,7 @@ pub async fn do_gsm_multiple_connector_actions(
     let mut connector = original_connector_data;
 
     loop {
-        let gsm = get_gsm(state, &connector, &payout_data).await?;
+        let gsm = get_gsm(state, &connector, payout_data).await?;
 
         match get_gsm_decision(gsm) {
             api_models::gsm::GsmDecision::Retry => {
@@ -70,13 +68,12 @@ pub async fn do_gsm_multiple_connector_actions(
 
                 connector = super::get_next_connector(&mut connectors)?;
 
-                payout_data = Box::pin(do_retry(
+                Box::pin(do_retry(
                     &state.clone(),
                     connector.to_owned(),
                     merchant_account,
                     key_store,
                     payout_data,
-                    req,
                 ))
                 .await?;
 
@@ -92,7 +89,7 @@ pub async fn do_gsm_multiple_connector_actions(
             api_models::gsm::GsmDecision::DoDefault => break,
         }
     }
-    Ok(payout_data)
+    Ok(())
 }
 
 #[instrument(skip_all)]
@@ -100,11 +97,10 @@ pub async fn do_gsm_multiple_connector_actions(
 pub async fn do_gsm_single_connector_actions(
     state: &app::AppState,
     original_connector_data: api::ConnectorData,
-    mut payout_data: PayoutData,
+    payout_data: &mut PayoutData,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    req: &PayoutCreateRequest,
-) -> RouterResult<PayoutData> {
+) -> RouterResult<()> {
     let mut retries = None;
 
     metrics::AUTO_PAYOUT_RETRY_ELIGIBLE_REQUEST_COUNT.add(&metrics::CONTEXT, 1, &[]);
@@ -112,7 +108,7 @@ pub async fn do_gsm_single_connector_actions(
     let mut previous_gsm = None; // to compare previous status
 
     loop {
-        let gsm = get_gsm(state, &original_connector_data, &payout_data).await?;
+        let gsm = get_gsm(state, &original_connector_data, payout_data).await?;
 
         // if the error config is same as previous, we break out of the loop
         if let Ordering::Equal = gsm.cmp(&previous_gsm) {
@@ -136,13 +132,12 @@ pub async fn do_gsm_single_connector_actions(
                     break;
                 }
 
-                payout_data = Box::pin(do_retry(
+                Box::pin(do_retry(
                     &state.clone(),
                     original_connector_data.to_owned(),
                     merchant_account,
                     key_store,
                     payout_data,
-                    req,
                 ))
                 .await?;
 
@@ -158,7 +153,7 @@ pub async fn do_gsm_single_connector_actions(
             api_models::gsm::GsmDecision::DoDefault => break,
         }
     }
-    Ok(payout_data)
+    Ok(())
 }
 
 #[instrument(skip_all)]
@@ -245,22 +240,13 @@ pub async fn do_retry(
     connector: api::ConnectorData,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    mut payout_data: PayoutData,
-    req: &PayoutCreateRequest,
-) -> RouterResult<PayoutData> {
+    payout_data: &mut PayoutData,
+) -> RouterResult<()> {
     metrics::AUTO_RETRY_PAYOUT_COUNT.add(&metrics::CONTEXT, 1, &[]);
 
-    modify_trackers(state, &connector, merchant_account, &mut payout_data).await?;
+    modify_trackers(state, &connector, merchant_account, payout_data).await?;
 
-    call_connector_payout(
-        state,
-        merchant_account,
-        key_store,
-        req,
-        &connector,
-        payout_data,
-    )
-    .await
+    call_connector_payout(state, merchant_account, key_store, &connector, payout_data).await
 }
 
 #[instrument(skip_all)]
@@ -363,6 +349,7 @@ impl GsmValidation for PayoutData {
             | common_enums::PayoutStatus::Ineligible
             | common_enums::PayoutStatus::RequiresCreation
             | common_enums::PayoutStatus::RequiresPayoutMethodData
+            | common_enums::PayoutStatus::RequiresVendorAccountCreation
             | common_enums::PayoutStatus::RequiresFulfillment => false,
             common_enums::PayoutStatus::Failed => true,
         }
diff --git a/crates/router/src/core/utils.rs b/crates/router/src/core/utils.rs
index 568e34c04a6..c67aafe938f 100644
--- a/crates/router/src/core/utils.rs
+++ b/crates/router/src/core/utils.rs
@@ -1,11 +1,15 @@
 use std::{marker::PhantomData, str::FromStr};
 
 use api_models::enums::{DisputeStage, DisputeStatus};
+#[cfg(feature = "payouts")]
+use api_models::payouts::PayoutVendorAccountDetails;
 use common_enums::{IntentStatus, RequestIncrementalAuthorization};
 #[cfg(feature = "payouts")]
 use common_utils::{crypto::Encryptable, pii::Email};
 use common_utils::{errors::CustomResult, ext_traits::AsyncExt};
 use error_stack::{report, ResultExt};
+#[cfg(feature = "payouts")]
+use masking::PeekInterface;
 use maud::{html, PreEscaped};
 use router_env::{instrument, tracing};
 use uuid::Uuid;
@@ -31,6 +35,9 @@ use crate::{
 
 pub const IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_DISPUTE_FLOW: &str =
     "irrelevant_connector_request_reference_id_in_dispute_flow";
+#[cfg(feature = "payouts")]
+pub const IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_PAYOUTS_FLOW: &str =
+    "irrelevant_connector_request_reference_id_in_payouts_flow";
 const IRRELEVANT_PAYMENT_ID_IN_DISPUTE_FLOW: &str = "irrelevant_payment_id_in_dispute_flow";
 const IRRELEVANT_ATTEMPT_ID_IN_DISPUTE_FLOW: &str = "irrelevant_attempt_id_in_dispute_flow";
 
@@ -65,15 +72,14 @@ pub async fn get_mca_for_payout<'a>(
 #[instrument(skip_all)]
 pub async fn construct_payout_router_data<'a, F>(
     state: &'a AppState,
-    connector_id: &str,
+    connector_name: &api_models::enums::Connector,
     merchant_account: &domain::MerchantAccount,
     key_store: &domain::MerchantKeyStore,
-    _request: &api_models::payouts::PayoutRequest,
     payout_data: &mut PayoutData,
 ) -> RouterResult<types::PayoutsRouterData<F>> {
     let merchant_connector_account = get_mca_for_payout(
         state,
-        connector_id,
+        &connector_name.to_string(),
         merchant_account,
         key_store,
         payout_data,
@@ -117,25 +123,36 @@ pub async fn construct_payout_router_data<'a, F>(
     let payouts = &payout_data.payouts;
     let payout_attempt = &payout_data.payout_attempt;
     let customer_details = &payout_data.customer_details;
-    let connector_name = payout_attempt
-        .connector
-        .clone()
-        .get_required_value("connector")
-        .change_context(errors::ApiErrorResponse::InvalidRequestData {
-            message: "Could not decide to route the connector".to_string(),
-        })?;
     let connector_label = format!("{}_{}", payout_data.profile_id, connector_name);
     let connector_customer_id = customer_details
         .as_ref()
         .and_then(|c| c.connector_customer.as_ref())
         .and_then(|cc| cc.get(connector_label))
         .and_then(|id| serde_json::from_value::<String>(id.to_owned()).ok());
+
+    let vendor_details: Option<PayoutVendorAccountDetails> =
+        match api_models::enums::PayoutConnectors::try_from(connector_name.to_owned()).map_err(
+            |err| report!(errors::ApiErrorResponse::InternalServerError).attach_printable(err),
+        )? {
+            api_models::enums::PayoutConnectors::Stripe => {
+                payout_data.payouts.metadata.to_owned().and_then(|meta| {
+                    let val = meta
+                        .peek()
+                        .to_owned()
+                        .parse_value("PayoutVendorAccountDetails")
+                        .ok();
+                    val
+                })
+            }
+            _ => None,
+        };
+
     let router_data = types::RouterData {
         flow: PhantomData,
         merchant_id: merchant_account.merchant_id.to_owned(),
         customer_id: None,
         connector_customer: connector_customer_id,
-        connector: connector_id.to_string(),
+        connector: connector_name.to_string(),
         payment_id: "".to_string(),
         attempt_id: "".to_string(),
         status: enums::AttemptStatus::Failure,
@@ -157,6 +174,7 @@ pub async fn construct_payout_router_data<'a, F>(
             source_currency: payouts.source_currency,
             entity_type: payouts.entity_type.to_owned(),
             payout_type: payouts.payout_type,
+            vendor_details,
             customer_details: customer_details
                 .to_owned()
                 .map(|c| payments::CustomerDetails {
@@ -174,7 +192,7 @@ pub async fn construct_payout_router_data<'a, F>(
         payment_method_token: None,
         recurring_mandate_payment_data: None,
         preprocessing_id: None,
-        connector_request_reference_id: IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_DISPUTE_FLOW
+        connector_request_reference_id: IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_PAYOUTS_FLOW
             .to_string(),
         payout_method_data: payout_data.payout_method_data.to_owned(),
         quote_id: None,
diff --git a/crates/router/src/lib.rs b/crates/router/src/lib.rs
index feffb7f1e8e..fdabd07fa01 100644
--- a/crates/router/src/lib.rs
+++ b/crates/router/src/lib.rs
@@ -71,6 +71,7 @@ pub mod headers {
     pub const X_WEBHOOK_SIGNATURE: &str = "X-Webhook-Signature-512";
     pub const X_REQUEST_ID: &str = "X-Request-Id";
     pub const STRIPE_COMPATIBLE_WEBHOOK_SIGNATURE: &str = "Stripe-Signature";
+    pub const STRIPE_COMPATIBLE_CONNECT_ACCOUNT: &str = "Stripe-Account";
 }
 
 pub mod pii {
diff --git a/crates/router/src/routes/payouts.rs b/crates/router/src/routes/payouts.rs
index f3b6a7cd477..315c6e6bd0a 100644
--- a/crates/router/src/routes/payouts.rs
+++ b/crates/router/src/routes/payouts.rs
@@ -70,7 +70,8 @@ pub async fn payouts_retrieve(
 ) -> HttpResponse {
     let payout_retrieve_request = payout_types::PayoutRetrieveRequest {
         payout_id: path.into_inner(),
-        force_sync: query_params.force_sync,
+        force_sync: query_params.force_sync.to_owned(),
+        merchant_id: query_params.merchant_id.to_owned(),
     };
     let flow = Flow::PayoutsRetrieve;
     Box::pin(api::server_wrap(
diff --git a/crates/router/src/types.rs b/crates/router/src/types.rs
index 6004131be40..a0a085440c8 100644
--- a/crates/router/src/types.rs
+++ b/crates/router/src/types.rs
@@ -32,7 +32,10 @@ use serde::Serialize;
 use self::storage::enums as storage_enums;
 pub use crate::core::payments::{payment_address::PaymentAddress, CustomerDetails};
 #[cfg(feature = "payouts")]
-use crate::core::utils::IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_DISPUTE_FLOW;
+use crate::{
+    connector::utils::missing_field_err,
+    core::utils::IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_PAYOUTS_FLOW,
+};
 use crate::{
     consts,
     core::{
@@ -190,6 +193,9 @@ pub type PayoutFulfillType =
 pub type PayoutRecipientType =
     dyn services::ConnectorIntegration<api::PoRecipient, PayoutsData, PayoutsResponseData>;
 #[cfg(feature = "payouts")]
+pub type PayoutRecipientAccountType =
+    dyn services::ConnectorIntegration<api::PoRecipientAccount, PayoutsData, PayoutsResponseData>;
+#[cfg(feature = "payouts")]
 pub type PayoutQuoteType =
     dyn services::ConnectorIntegration<api::PoQuote, PayoutsData, PayoutsResponseData>;
 
@@ -308,7 +314,7 @@ pub struct RouterData<Flow, Request, Response> {
     pub payout_method_data: Option<api::PayoutMethodData>,
 
     #[cfg(feature = "payouts")]
-    /// Contains payout method data
+    /// Contains payout's quote ID
     pub quote_id: Option<String>,
 
     pub test_mode: Option<bool>,
@@ -394,6 +400,23 @@ pub struct PayoutsData {
     pub payout_type: storage_enums::PayoutType,
     pub entity_type: storage_enums::PayoutEntityType,
     pub customer_details: Option<CustomerDetails>,
+    pub vendor_details: Option<api_models::payouts::PayoutVendorAccountDetails>,
+}
+
+#[cfg(feature = "payouts")]
+pub trait PayoutIndividualDetailsExt {
+    type Error;
+    fn get_external_account_account_holder_type(&self) -> Result<String, Self::Error>;
+}
+
+#[cfg(feature = "payouts")]
+impl PayoutIndividualDetailsExt for api_models::payouts::PayoutIndividualDetails {
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn get_external_account_account_holder_type(&self) -> Result<String, Self::Error> {
+        self.external_account_account_holder_type
+            .clone()
+            .ok_or_else(missing_field_err("external_account_account_holder_type"))
+    }
 }
 
 #[cfg(feature = "payouts")]
@@ -402,12 +425,7 @@ pub struct PayoutsResponseData {
     pub status: Option<storage_enums::PayoutStatus>,
     pub connector_payout_id: String,
     pub payout_eligible: Option<bool>,
-}
-
-#[derive(Clone, Debug, Default)]
-pub struct PayoutsFulfillResponseData {
-    pub status: Option<storage_enums::PayoutStatus>,
-    pub reference_id: Option<String>,
+    pub should_add_next_step_to_process_tracker: bool,
 }
 
 #[derive(Debug, Clone)]
@@ -1587,7 +1605,7 @@ impl<F1, F2>
             preprocessing_id: None,
             connector_customer: data.connector_customer.clone(),
             connector_request_reference_id:
-                IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_DISPUTE_FLOW.to_string(),
+                IRRELEVANT_CONNECTOR_REQUEST_REFERENCE_ID_IN_PAYOUTS_FLOW.to_string(),
             payout_method_data: data.payout_method_data.clone(),
             quote_id: data.quote_id.clone(),
             test_mode: data.test_mode,
diff --git a/crates/router/src/types/api.rs b/crates/router/src/types/api.rs
index 01f4db9ea90..648c7a224eb 100644
--- a/crates/router/src/types/api.rs
+++ b/crates/router/src/types/api.rs
@@ -427,6 +427,7 @@ pub trait Payouts:
     + PayoutFulfill
     + PayoutQuote
     + PayoutRecipient
+    + PayoutRecipientAccount
 {
 }
 #[cfg(not(feature = "payouts"))]
diff --git a/crates/router/src/types/api/payouts.rs b/crates/router/src/types/api/payouts.rs
index bed87173b84..a04c2fd095b 100644
--- a/crates/router/src/types/api/payouts.rs
+++ b/crates/router/src/types/api/payouts.rs
@@ -25,6 +25,9 @@ pub struct PoQuote;
 #[derive(Debug, Clone)]
 pub struct PoRecipient;
 
+#[derive(Debug, Clone)]
+pub struct PoRecipientAccount;
+
 pub trait PayoutCancel:
     api::ConnectorIntegration<PoCancel, types::PayoutsData, types::PayoutsResponseData>
 {
@@ -54,3 +57,8 @@ pub trait PayoutRecipient:
     api::ConnectorIntegration<PoRecipient, types::PayoutsData, types::PayoutsResponseData>
 {
 }
+
+pub trait PayoutRecipientAccount:
+    api::ConnectorIntegration<PoRecipientAccount, types::PayoutsData, types::PayoutsResponseData>
+{
+}
diff --git a/crates/router/src/workflows.rs b/crates/router/src/workflows.rs
index 6158031079c..7b29ded5185 100644
--- a/crates/router/src/workflows.rs
+++ b/crates/router/src/workflows.rs
@@ -1,5 +1,7 @@
 #[cfg(feature = "email")]
 pub mod api_key_expiry;
+#[cfg(feature = "payouts")]
+pub mod attach_payout_account_workflow;
 pub mod outgoing_webhook_retry;
 pub mod payment_sync;
 pub mod refund_router;
diff --git a/crates/router/src/workflows/attach_payout_account_workflow.rs b/crates/router/src/workflows/attach_payout_account_workflow.rs
new file mode 100644
index 00000000000..98d3f7844b4
--- /dev/null
+++ b/crates/router/src/workflows/attach_payout_account_workflow.rs
@@ -0,0 +1,72 @@
+use common_utils::ext_traits::{OptionExt, ValueExt};
+use scheduler::{
+    consumer::{self, workflows::ProcessTrackerWorkflow},
+    errors,
+};
+
+use crate::{
+    core::payouts,
+    errors as core_errors,
+    routes::AppState,
+    types::{api, storage},
+};
+
+pub struct AttachPayoutAccountWorkflow;
+
+#[async_trait::async_trait]
+impl ProcessTrackerWorkflow<AppState> for AttachPayoutAccountWorkflow {
+    async fn execute_workflow<'a>(
+        &'a self,
+        state: &'a AppState,
+        process: storage::ProcessTracker,
+    ) -> Result<(), errors::ProcessTrackerError> {
+        // Gather context
+        let db = &*state.store;
+        let tracking_data: api::PayoutRetrieveRequest = process
+            .tracking_data
+            .clone()
+            .parse_value("PayoutRetrieveRequest")?;
+
+        let merchant_id = tracking_data
+            .merchant_id
+            .clone()
+            .get_required_value("merchant_id")?;
+
+        let key_store = db
+            .get_merchant_key_store_by_merchant_id(
+                merchant_id.as_ref(),
+                &db.get_master_key().to_vec().into(),
+            )
+            .await?;
+
+        let merchant_account = db
+            .find_merchant_account_by_merchant_id(&merchant_id, &key_store)
+            .await?;
+
+        let request = api::payouts::PayoutRequest::PayoutRetrieveRequest(tracking_data);
+
+        let mut payout_data =
+            payouts::make_payout_data(state, &merchant_account, &key_store, &request).await?;
+
+        payouts::payouts_core(
+            state,
+            &merchant_account,
+            &key_store,
+            &mut payout_data,
+            None,
+            None,
+        )
+        .await?;
+
+        Ok(())
+    }
+
+    async fn error_handler<'a>(
+        &'a self,
+        state: &'a AppState,
+        process: storage::ProcessTracker,
+        error: errors::ProcessTrackerError,
+    ) -> core_errors::CustomResult<(), errors::ProcessTrackerError> {
+        consumer::consumer_error_handler(state.store.as_scheduler(), process, error).await
+    }
+}
diff --git a/crates/router/tests/connectors/utils.rs b/crates/router/tests/connectors/utils.rs
index 47ec71cc94e..7f1cd31f305 100644
--- a/crates/router/tests/connectors/utils.rs
+++ b/crates/router/tests/connectors/utils.rs
@@ -484,6 +484,7 @@ pub trait ConnectorActions: Connector {
                     phone: Some(Secret::new("620874518".to_string())),
                     phone_country_code: Some("+31".to_string()),
                 }),
+                vendor_details: None,
             },
             payment_info,
         )
diff --git a/crates/storage_impl/src/payouts/payouts.rs b/crates/storage_impl/src/payouts/payouts.rs
index 0f72aa592bf..0b166946758 100644
--- a/crates/storage_impl/src/payouts/payouts.rs
+++ b/crates/storage_impl/src/payouts/payouts.rs
@@ -86,6 +86,7 @@ impl<T: DatabaseStore> PayoutsInterface for KVRouterStore<T> {
                     profile_id: new.profile_id.clone(),
                     status: new.status,
                     attempt_count: new.attempt_count,
+                    confirm: new.confirm,
                 };
 
                 let redis_entry = kv::TypedSql {
@@ -674,6 +675,7 @@ impl DataModelExt for Payouts {
             profile_id: self.profile_id,
             status: self.status,
             attempt_count: self.attempt_count,
+            confirm: self.confirm,
         }
     }
 
@@ -699,6 +701,7 @@ impl DataModelExt for Payouts {
             profile_id: storage_model.profile_id,
             status: storage_model.status,
             attempt_count: storage_model.attempt_count,
+            confirm: storage_model.confirm,
         }
     }
 }
@@ -727,6 +730,7 @@ impl DataModelExt for PayoutsNew {
             profile_id: self.profile_id,
             status: self.status,
             attempt_count: self.attempt_count,
+            confirm: self.confirm,
         }
     }
 
@@ -752,6 +756,7 @@ impl DataModelExt for PayoutsNew {
             profile_id: storage_model.profile_id,
             status: storage_model.status,
             attempt_count: storage_model.attempt_count,
+            confirm: storage_model.confirm,
         }
     }
 }
@@ -771,6 +776,7 @@ impl DataModelExt for PayoutsUpdate {
                 metadata,
                 profile_id,
                 status,
+                confirm,
             } => DieselPayoutsUpdate::Update {
                 amount,
                 destination_currency,
@@ -783,6 +789,7 @@ impl DataModelExt for PayoutsUpdate {
                 metadata,
                 profile_id,
                 status,
+                confirm,
             },
             Self::PayoutMethodIdUpdate { payout_method_id } => {
                 DieselPayoutsUpdate::PayoutMethodIdUpdate { payout_method_id }
diff --git a/migrations/2024-04-09-202926_add_confirm_to_payouts/down.sql b/migrations/2024-04-09-202926_add_confirm_to_payouts/down.sql
new file mode 100644
index 00000000000..ae191c474f0
--- /dev/null
+++ b/migrations/2024-04-09-202926_add_confirm_to_payouts/down.sql
@@ -0,0 +1 @@
+ALTER TABLE payouts DROP COLUMN IF EXISTS confirm;
\ No newline at end of file
diff --git a/migrations/2024-04-09-202926_add_confirm_to_payouts/up.sql b/migrations/2024-04-09-202926_add_confirm_to_payouts/up.sql
new file mode 100644
index 00000000000..0eccb099525
--- /dev/null
+++ b/migrations/2024-04-09-202926_add_confirm_to_payouts/up.sql
@@ -0,0 +1 @@
+ALTER TABLE payouts ADD COLUMN IF NOT EXISTS confirm bool;
\ No newline at end of file
diff --git a/migrations/2024-04-10-034442_alter_payout_status/down.sql b/migrations/2024-04-10-034442_alter_payout_status/down.sql
new file mode 100644
index 00000000000..027b7d63fdb
--- /dev/null
+++ b/migrations/2024-04-10-034442_alter_payout_status/down.sql
@@ -0,0 +1 @@
+SELECT 1;
\ No newline at end of file
diff --git a/migrations/2024-04-10-034442_alter_payout_status/up.sql b/migrations/2024-04-10-034442_alter_payout_status/up.sql
new file mode 100644
index 00000000000..80177f5324b
--- /dev/null
+++ b/migrations/2024-04-10-034442_alter_payout_status/up.sql
@@ -0,0 +1 @@
+ALTER TYPE "PayoutStatus" ADD VALUE IF NOT EXISTS 'requires_vendor_account_creation';
\ No newline at end of file
diff --git a/openapi/openapi_spec.json b/openapi/openapi_spec.json
index b64756fc434..b46a76eb5aa 100644
--- a/openapi/openapi_spec.json
+++ b/openapi/openapi_spec.json
@@ -15607,6 +15607,7 @@
         "type": "string",
         "enum": [
           "adyen",
+          "stripe",
           "wise",
           "paypal"
         ]
@@ -16211,6 +16212,10 @@
           "force_sync": {
             "type": "boolean",
             "nullable": true
+          },
+          "merchant_id": {
+            "type": "string",
+            "nullable": true
           }
         }
       },
@@ -16233,6 +16238,11 @@
             "default": false,
             "example": true,
             "nullable": true
+          },
+          "merchant_id": {
+            "type": "string",
+            "description": "The identifier for the Merchant Account.",
+            "nullable": true
           }
         }
       },
@@ -16246,7 +16256,8 @@
           "ineligible",
           "requires_creation",
           "requires_payout_method_data",
-          "requires_fulfillment"
+          "requires_fulfillment",
+          "requires_vendor_account_creation"
         ]
       },
       "PayoutType": {

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
This includes integration of StripeConnect APIs in Stripe connector for payouts functionality. For doing the same, there is an addition of a new Payout connector action (PayoutRecipientAccountCreation). This helps in creation of external accounts (bank / card) for Stripe Connect users.

### Additional Changes

- [x] This PR modifies the API contract
- [x] This PR modifies the database schema
- [x] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
Stripe Connect is a virtual marketplace for onboarding and paying out to platform's affiliates / vendors / resellers. Integrating Stripe Connect offers payout abstraction for such use-cases.

## How did you test it?

- Tested existing payout functionalities using postman test suite.
- Tested Stripe Connect integration using raw PMD + SPMs
- Postman suite - https://galactic-capsule-229427.postman.co/workspace/My-Workspace~2b563e0d-bad3-420f-8c0b-0fd5b278a4fe/collection/9906252-8f51a4d2-86d6-4f72-8198-3fff5a4e71af?action=share&creator=9906252

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [x] I added unit tests for my changes where possible
- [x] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
