diff --git a/api-reference/openapi_spec.json b/api-reference/openapi_spec.json
index 9d8110d076..c4e01dcca1 100644
--- a/api-reference/openapi_spec.json
+++ b/api-reference/openapi_spec.json
@@ -9372,7 +9372,8 @@
           "payments",
           "refunds",
           "disputes",
-          "mandates"
+          "mandates",
+          "payouts"
         ]
       },
       "EventListItemResponse": {
@@ -11933,6 +11934,7 @@
             "nullable": true
           },
           "pm_auth_config": {
+            "type": "object",
             "nullable": true
           },
           "status": {
@@ -12181,6 +12183,7 @@
             "nullable": true
           },
           "pm_auth_config": {
+            "type": "object",
             "nullable": true
           },
           "status": {
@@ -12303,6 +12306,8 @@
             "nullable": true
           },
           "pm_auth_config": {
+            "type": "object",
+            "description": "pm_auth_config will relate MCA records to their respective chosen auth services, based on payment_method and pmt",
             "nullable": true
           },
           "status": {
@@ -20882,7 +20887,8 @@
       "TransactionType": {
         "type": "string",
         "enum": [
-          "payment"
+          "payment",
+          "payout"
         ]
       },
       "UpdateApiKeyRequest": {
diff --git a/crates/api_models/Cargo.toml b/crates/api_models/Cargo.toml
index a2eb9d476e..84f8d2f01e 100644
--- a/crates/api_models/Cargo.toml
+++ b/crates/api_models/Cargo.toml
@@ -12,13 +12,14 @@ default = ["v1"]
 errors = ["dep:actix-web", "dep:reqwest"]
 dummy_connector = ["euclid/dummy_connector", "common_enums/dummy_connector"]
 detailed_errors = []
-payouts = []
+payouts = ["common_enums/payouts"]
 frm = []
 olap = []
 openapi = ["common_enums/openapi", "olap", "recon", "dummy_connector", "olap"]
 recon = []
 v2 = []
 v1 = []
+merchant_connector_account_v2 = []
 customer_v2 = []
 merchant_account_v2 = []
 payment_v2 = []
diff --git a/crates/api_models/src/admin.rs b/crates/api_models/src/admin.rs
index 5ba200bdc9..4f1143efaa 100644
--- a/crates/api_models/src/admin.rs
+++ b/crates/api_models/src/admin.rs
@@ -574,6 +574,137 @@ pub struct MerchantConnectorId {
     pub merchant_connector_id: String,
 }
 
+#[cfg(all(feature = "v2", feature = "merchant_connector_account_v2"))]
+/// Create a new Merchant Connector for the merchant account. The connector could be a payment processor / facilitator / acquirer or specialized services like Fraud / Accounting etc."
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct MerchantConnectorCreate {
+    /// Type of the Connector for the financial use case. Could range from Payments to Accounting to Banking.
+    #[schema(value_type = ConnectorType, example = "payment_processor")]
+    pub connector_type: api_enums::ConnectorType,
+    /// Name of the Connector
+    #[schema(value_type = Connector, example = "stripe")]
+    pub connector_name: api_enums::Connector,
+    /// This is an unique label you can generate and pass in order to identify this connector account on your Hyperswitch dashboard and reports. Eg: if your profile label is `default`, connector label can be `stripe_default`
+    #[schema(example = "stripe_US_travel")]
+    pub connector_label: Option<String>,
+
+    /// Identifier for the business profile, if not provided default will be chosen from merchant account
+    pub profile_id: Option<String>,
+
+    /// An object containing the required details/credentials for a Connector account.
+    #[schema(value_type = Option<MerchantConnectorDetails>,example = json!({ "auth_type": "HeaderKey","api_key": "Basic MyVerySecretApiKey" }))]
+    pub connector_account_details: Option<pii::SecretSerdeValue>,
+
+    /// An object containing the details about the payment methods that need to be enabled under this merchant connector account
+    #[schema(example = json!([
+        {
+            "payment_method": "wallet",
+            "payment_method_types": [
+                "upi_collect",
+                "upi_intent"
+            ],
+            "payment_method_issuers": [
+                "labore magna ipsum",
+                "aute"
+            ],
+            "payment_schemes": [
+                "Discover",
+                "Discover"
+            ],
+            "accepted_currencies": {
+                "type": "enable_only",
+                "list": ["USD", "EUR"]
+            },
+            "accepted_countries": {
+                "type": "disable_only",
+                "list": ["FR", "DE","IN"]
+            },
+            "minimum_amount": 1,
+            "maximum_amount": 68607706,
+            "recurring_enabled": true,
+            "installment_payment_enabled": true
+        }
+    ]))]
+    pub payment_methods_enabled: Option<Vec<PaymentMethodsEnabled>>,
+
+    /// Webhook details of this merchant connector
+    #[schema(example = json!({
+        "connector_webhook_details": {
+            "merchant_secret": "1234567890987654321"
+        }
+    }))]
+    pub connector_webhook_details: Option<MerchantConnectorWebhookDetails>,
+
+    /// You can specify up to 50 keys, with key names up to 40 characters long and values up to 500 characters long. Metadata is useful for storing additional, structured information on an object.
+    #[schema(value_type = Option<Object>,max_length = 255,example = json!({ "city": "NY", "unit": "245" }))]
+    pub metadata: Option<pii::SecretSerdeValue>,
+
+    /// A boolean value to indicate if the connector is disabled. By default, its value is false.
+    #[schema(default = false, example = false)]
+    pub disabled: Option<bool>,
+
+    /// Contains the frm configs for the merchant connector
+    #[schema(example = json!(consts::FRM_CONFIGS_EG))]
+    pub frm_configs: Option<Vec<FrmConfigs>>,
+
+    /// Unique ID of the connector
+    #[schema(example = "mca_5apGeP94tMts6rg3U3kR")]
+    pub merchant_connector_id: Option<String>,
+
+    /// pm_auth_config will relate MCA records to their respective chosen auth services, based on payment_method and pmt
+    #[schema(value_type = Option<Object>)]
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
+
+    #[schema(value_type = Option<ConnectorStatus>, example = "inactive")]
+    // By default the ConnectorStatus is Active
+    pub status: Option<api_enums::ConnectorStatus>,
+
+    /// The identifier for the Merchant Account
+    #[schema(value_type = String, max_length = 64, min_length = 1, example = "y3oqhf46pyzuxjbcn2giaqnb44")]
+    pub merchant_id: id_type::MerchantId,
+
+    /// In case the merchant needs to store any additional sensitive data
+    #[schema(value_type = Option<AdditionalMerchantData>)]
+    pub additional_merchant_data: Option<AdditionalMerchantData>,
+}
+
+#[cfg(all(feature = "v2", feature = "merchant_connector_account_v2"))]
+impl MerchantConnectorCreate {
+    pub fn get_transaction_type(&self) -> api_enums::TransactionType {
+        match self.connector_type {
+            #[cfg(feature = "payouts")]
+            api_enums::ConnectorType::PayoutProcessor => api_enums::TransactionType::Payout,
+            _ => api_enums::TransactionType::Payment,
+        }
+    }
+
+    pub fn get_frm_config_as_secret(&self) -> Option<Vec<Secret<serde_json::Value>>> {
+        match self.frm_configs.as_ref() {
+            Some(frm_value) => {
+                let configs_for_frm_value: Vec<Secret<serde_json::Value>> = frm_value
+                    .iter()
+                    .map(|config| config.encode_to_value().map(Secret::new))
+                    .collect::<Result<Vec<_>, _>>()
+                    .ok()?;
+                Some(configs_for_frm_value)
+            }
+            None => None,
+        }
+    }
+
+    pub fn get_connector_label(&self, profile_name: String) -> String {
+        match self.connector_label.clone() {
+            Some(connector_label) => connector_label,
+            None => format!("{}_{}", self.connector_name, profile_name),
+        }
+    }
+}
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "merchant_connector_account_v2")
+))]
 /// Create a new Merchant Connector for the merchant account. The connector could be a payment processor / facilitator / acquirer or specialized services like Fraud / Accounting etc."
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 #[serde(deny_unknown_fields)]
@@ -666,7 +797,8 @@ pub struct MerchantConnectorCreate {
     #[schema(example = "mca_5apGeP94tMts6rg3U3kR")]
     pub merchant_connector_id: Option<String>,
 
-    pub pm_auth_config: Option<serde_json::Value>,
+    #[schema(value_type = Option<Object>)]
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
 
     #[schema(value_type = Option<ConnectorStatus>, example = "inactive")]
     pub status: Option<api_enums::ConnectorStatus>,
@@ -676,6 +808,34 @@ pub struct MerchantConnectorCreate {
     pub additional_merchant_data: Option<AdditionalMerchantData>,
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "merchant_connector_account_v2")
+))]
+impl MerchantConnectorCreate {
+    pub fn get_transaction_type(&self) -> api_enums::TransactionType {
+        match self.connector_type {
+            #[cfg(feature = "payouts")]
+            api_enums::ConnectorType::PayoutProcessor => api_enums::TransactionType::Payout,
+            _ => api_enums::TransactionType::Payment,
+        }
+    }
+
+    pub fn get_frm_config_as_secret(&self) -> Option<Vec<Secret<serde_json::Value>>> {
+        match self.frm_configs.as_ref() {
+            Some(frm_value) => {
+                let configs_for_frm_value: Vec<Secret<serde_json::Value>> = frm_value
+                    .iter()
+                    .map(|config| config.encode_to_value().map(Secret::new))
+                    .collect::<Result<Vec<_>, _>>()
+                    .ok()?;
+                Some(configs_for_frm_value)
+            }
+            None => None,
+        }
+    }
+}
+
 #[derive(Debug, Clone, serde::Deserialize, serde::Serialize, ToSchema)]
 #[serde(rename_all = "snake_case")]
 pub enum AdditionalMerchantData {
@@ -764,7 +924,124 @@ pub struct MerchantConnectorInfo {
     pub merchant_connector_id: String,
 }
 
+impl MerchantConnectorInfo {
+    pub fn new(connector_label: String, merchant_connector_id: String) -> Self {
+        Self {
+            connector_label,
+            merchant_connector_id,
+        }
+    }
+}
+
 /// Response of creating a new Merchant Connector for the merchant account."
+#[cfg(all(feature = "v2", feature = "merchant_connector_account_v2"))]
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct MerchantConnectorResponse {
+    /// Type of the Connector for the financial use case. Could range from Payments to Accounting to Banking.
+    #[schema(value_type = ConnectorType, example = "payment_processor")]
+    pub connector_type: api_enums::ConnectorType,
+    /// Name of the Connector
+    #[schema(value_type = Connector, example = "stripe")]
+    pub connector_name: String,
+
+    /// A unique label to identify the connector account created under a business profile
+    #[schema(example = "stripe_US_travel")]
+    pub connector_label: Option<String>,
+
+    /// Unique ID of the merchant connector account
+    #[schema(example = "mca_5apGeP94tMts6rg3U3kR")]
+    pub connector_id: String,
+
+    /// Identifier for the business profile, if not provided default will be chosen from merchant account
+    #[schema(max_length = 64)]
+    pub profile_id: Option<String>,
+
+    /// An object containing the required details/credentials for a Connector account.
+    #[schema(value_type = Option<MerchantConnectorDetails>,example = json!({ "auth_type": "HeaderKey","api_key": "Basic MyVerySecretApiKey" }))]
+    pub connector_account_details: pii::SecretSerdeValue,
+
+    /// An object containing the details about the payment methods that need to be enabled under this merchant connector account
+    #[schema(example = json!([
+        {
+            "payment_method": "wallet",
+            "payment_method_types": [
+                "upi_collect",
+                "upi_intent"
+            ],
+            "payment_method_issuers": [
+                "labore magna ipsum",
+                "aute"
+            ],
+            "payment_schemes": [
+                "Discover",
+                "Discover"
+            ],
+            "accepted_currencies": {
+                "type": "enable_only",
+                "list": ["USD", "EUR"]
+            },
+            "accepted_countries": {
+                "type": "disable_only",
+                "list": ["FR", "DE","IN"]
+            },
+            "minimum_amount": 1,
+            "maximum_amount": 68607706,
+            "recurring_enabled": true,
+            "installment_payment_enabled": true
+        }
+    ]))]
+    pub payment_methods_enabled: Option<Vec<PaymentMethodsEnabled>>,
+
+    /// Webhook details of this merchant connector
+    #[schema(example = json!({
+        "connector_webhook_details": {
+            "merchant_secret": "1234567890987654321"
+        }
+    }))]
+    pub connector_webhook_details: Option<MerchantConnectorWebhookDetails>,
+
+    /// You can specify up to 50 keys, with key names up to 40 characters long and values up to 500 characters long. Metadata is useful for storing additional, structured information on an object.
+    #[schema(value_type = Option<Object>,max_length = 255,example = json!({ "city": "NY", "unit": "245" }))]
+    pub metadata: Option<pii::SecretSerdeValue>,
+
+    /// A boolean value to indicate if the connector is disabled. By default, its value is false.
+    #[schema(default = false, example = false)]
+    pub disabled: Option<bool>,
+
+    /// Contains the frm configs for the merchant connector
+    #[schema(example = json!(consts::FRM_CONFIGS_EG))]
+    pub frm_configs: Option<Vec<FrmConfigs>>,
+
+    /// identifier for the verified domains of a particular connector account
+    pub applepay_verified_domains: Option<Vec<String>>,
+
+    /// pm_auth_config will relate MCA records to their respective chosen auth services, based on payment_method and pmt
+    #[schema(value_type = Option<Object>)]
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
+
+    #[schema(value_type = ConnectorStatus, example = "inactive")]
+    pub status: api_enums::ConnectorStatus,
+
+    #[schema(value_type = Option<AdditionalMerchantData>)]
+    pub additional_merchant_data: Option<AdditionalMerchantData>,
+}
+
+#[cfg(all(feature = "v2", feature = "merchant_connector_account_v2"))]
+impl MerchantConnectorResponse {
+    pub fn to_merchant_connector_info(&self, connector_label: &String) -> MerchantConnectorInfo {
+        MerchantConnectorInfo {
+            connector_label: connector_label.to_string(),
+            merchant_connector_id: self.connector_id.clone(),
+        }
+    }
+}
+
+/// Response of creating a new Merchant Connector for the merchant account."
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "merchant_connector_account_v2")
+))]
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 #[serde(deny_unknown_fields)]
 pub struct MerchantConnectorResponse {
@@ -862,7 +1139,8 @@ pub struct MerchantConnectorResponse {
     /// identifier for the verified domains of a particular connector account
     pub applepay_verified_domains: Option<Vec<String>>,
 
-    pub pm_auth_config: Option<serde_json::Value>,
+    #[schema(value_type = Option<Object>)]
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
 
     #[schema(value_type = ConnectorStatus, example = "inactive")]
     pub status: api_enums::ConnectorStatus,
@@ -871,6 +1149,19 @@ pub struct MerchantConnectorResponse {
     pub additional_merchant_data: Option<AdditionalMerchantData>,
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "merchant_connector_account_v2")
+))]
+impl MerchantConnectorResponse {
+    pub fn to_merchant_connector_info(&self, connector_label: &String) -> MerchantConnectorInfo {
+        MerchantConnectorInfo {
+            connector_label: connector_label.to_string(),
+            merchant_connector_id: self.merchant_connector_id.clone(),
+        }
+    }
+}
+
 /// Create a new Merchant Connector for the merchant account. The connector could be a payment processor / facilitator / acquirer or specialized services like Fraud / Accounting etc."
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 #[serde(deny_unknown_fields)]
@@ -943,7 +1234,9 @@ pub struct MerchantConnectorUpdate {
     #[schema(example = json!(consts::FRM_CONFIGS_EG))]
     pub frm_configs: Option<Vec<FrmConfigs>>,
 
-    pub pm_auth_config: Option<serde_json::Value>,
+    /// pm_auth_config will relate MCA records to their respective chosen auth services, based on payment_method and pmt
+    #[schema(value_type = Option<Object>)]
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
 
     #[schema(value_type = ConnectorStatus, example = "inactive")]
     pub status: Option<api_enums::ConnectorStatus>,
diff --git a/crates/api_models/src/enums.rs b/crates/api_models/src/enums.rs
index 85212ec7d0..c87c94084b 100644
--- a/crates/api_models/src/enums.rs
+++ b/crates/api_models/src/enums.rs
@@ -1,6 +1,8 @@
 use std::str::FromStr;
 
 pub use common_enums::*;
+#[cfg(feature = "dummy_connector")]
+use common_utils::errors;
 use utoipa::ToSchema;
 
 #[derive(
@@ -272,6 +274,31 @@ impl Connector {
     pub fn is_pre_processing_required_before_authorize(&self) -> bool {
         matches!(self, Self::Airwallex)
     }
+    #[cfg(feature = "dummy_connector")]
+    pub fn validate_dummy_connector_enabled(
+        &self,
+        is_dummy_connector_enabled: bool,
+    ) -> errors::CustomResult<(), errors::ValidationError> {
+        if !is_dummy_connector_enabled
+            && matches!(
+                self,
+                Self::DummyConnector1
+                    | Self::DummyConnector2
+                    | Self::DummyConnector3
+                    | Self::DummyConnector4
+                    | Self::DummyConnector5
+                    | Self::DummyConnector6
+                    | Self::DummyConnector7
+            )
+        {
+            Err(errors::ValidationError::InvalidValue {
+                message: "Invalid connector name".to_string(),
+            }
+            .into())
+        } else {
+            Ok(())
+        }
+    }
 }
 
 #[derive(
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index 2f50571113..a466288b41 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -8,12 +8,13 @@ use cards::CardNumber;
 use common_utils::{
     consts::default_payments_list_limit,
     crypto,
-    ext_traits::{ConfigExt, Encode},
+    ext_traits::{ConfigExt, Encode, ValueExt},
     hashing::HashedString,
     id_type,
     pii::{self, Email, EmailStrategy},
     types::{keymanager::ToEncryptable, MinorUnit, StringMajorUnit},
 };
+use error_stack::ResultExt;
 use euclid::dssa::graph::euclid_graph_prelude::FxHashMap;
 use masking::{ExposeInterface, PeekInterface, Secret, SwitchStrategy, WithType};
 use router_derive::Setter;
@@ -4521,6 +4522,32 @@ pub struct ConnectorMetadata {
     pub noon: Option<NoonData>,
 }
 
+impl ConnectorMetadata {
+    pub fn from_value(
+        value: pii::SecretSerdeValue,
+    ) -> common_utils::errors::CustomResult<Self, common_utils::errors::ParsingError> {
+        value
+            .parse_value::<Self>("ConnectorMetadata")
+            .change_context(common_utils::errors::ParsingError::StructParseFailure(
+                "Metadata",
+            ))
+    }
+    pub fn get_apple_pay_certificates(self) -> Option<(Secret<String>, Secret<String>)> {
+        self.apple_pay.and_then(|applepay_metadata| {
+            applepay_metadata
+                .session_token_data
+                .map(|session_token_data| {
+                    let SessionTokenInfo {
+                        certificate,
+                        certificate_keys,
+                        ..
+                    } = session_token_data;
+                    (certificate, certificate_keys)
+                })
+        })
+    }
+}
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
 pub struct AirwallexData {
     /// payload required by airwallex
diff --git a/crates/diesel_models/src/merchant_connector_account.rs b/crates/diesel_models/src/merchant_connector_account.rs
index eb561f2f9c..2d12dc9f25 100644
--- a/crates/diesel_models/src/merchant_connector_account.rs
+++ b/crates/diesel_models/src/merchant_connector_account.rs
@@ -41,7 +41,7 @@ pub struct MerchantConnectorAccount {
     pub profile_id: Option<String>,
     #[diesel(deserialize_as = super::OptionalDieselArray<String>)]
     pub applepay_verified_domains: Option<Vec<String>>,
-    pub pm_auth_config: Option<serde_json::Value>,
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
     pub status: storage_enums::ConnectorStatus,
     pub additional_merchant_data: Option<Encryption>,
     pub connector_wallets_details: Option<Encryption>,
@@ -72,7 +72,7 @@ pub struct MerchantConnectorAccountNew {
     pub profile_id: Option<String>,
     #[diesel(deserialize_as = super::OptionalDieselArray<String>)]
     pub applepay_verified_domains: Option<Vec<String>>,
-    pub pm_auth_config: Option<serde_json::Value>,
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
     pub status: storage_enums::ConnectorStatus,
     pub additional_merchant_data: Option<Encryption>,
     pub connector_wallets_details: Option<Encryption>,
@@ -97,7 +97,7 @@ pub struct MerchantConnectorAccountUpdateInternal {
     pub frm_config: Option<Vec<Secret<serde_json::Value>>>,
     #[diesel(deserialize_as = super::OptionalDieselArray<String>)]
     pub applepay_verified_domains: Option<Vec<String>>,
-    pub pm_auth_config: Option<serde_json::Value>,
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
     pub status: Option<storage_enums::ConnectorStatus>,
     pub connector_wallets_details: Option<Encryption>,
 }
diff --git a/crates/hyperswitch_domain_models/src/router_data.rs b/crates/hyperswitch_domain_models/src/router_data.rs
index c2df268644..609014b691 100644
--- a/crates/hyperswitch_domain_models/src/router_data.rs
+++ b/crates/hyperswitch_domain_models/src/router_data.rs
@@ -1,6 +1,7 @@
 use std::{collections::HashMap, marker::PhantomData};
 
-use common_utils::{errors::IntegrityCheckError, id_type, types::MinorUnit};
+use common_utils::{errors::IntegrityCheckError, ext_traits::OptionExt, id_type, types::MinorUnit};
+use error_stack::ResultExt;
 use masking::Secret;
 
 use crate::{payment_address::PaymentAddress, payment_method_data};
@@ -108,6 +109,18 @@ pub enum ConnectorAuthType {
     NoKey,
 }
 
+impl ConnectorAuthType {
+    pub fn from_option_secret_value(
+        value: Option<common_utils::pii::SecretSerdeValue>,
+    ) -> common_utils::errors::CustomResult<Self, common_utils::errors::ParsingError> {
+        value
+            .parse_value::<Self>("ConnectorAuthType")
+            .change_context(common_utils::errors::ParsingError::StructParseFailure(
+                "ConnectorAuthType",
+            ))
+    }
+}
+
 #[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
 pub struct AccessToken {
     pub token: Secret<String>,
diff --git a/crates/kgraph_utils/Cargo.toml b/crates/kgraph_utils/Cargo.toml
index d4110ae618..7dfebc3604 100644
--- a/crates/kgraph_utils/Cargo.toml
+++ b/crates/kgraph_utils/Cargo.toml
@@ -7,8 +7,11 @@ rust-version.workspace = true
 license.workspace = true
 
 [features]
+default = ["v1"]
 dummy_connector = ["api_models/dummy_connector", "euclid/dummy_connector"]
-
+v1 = []
+v2 = []
+merchant_connector_account_v2 = ["api_models/merchant_connector_account_v2"]
 
 [dependencies]
 api_models = { version = "0.1.0", path = "../api_models", package = "api_models" }
diff --git a/crates/kgraph_utils/benches/evaluation.rs b/crates/kgraph_utils/benches/evaluation.rs
index a137a6fd54..a48aefe4c9 100644
--- a/crates/kgraph_utils/benches/evaluation.rs
+++ b/crates/kgraph_utils/benches/evaluation.rs
@@ -52,6 +52,29 @@ fn build_test_data(
         });
     }
 
+    #[cfg(all(feature = "v2", feature = "merchant_connector_account_v2"))]
+    let stripe_account = MerchantConnectorResponse {
+        connector_type: api_enums::ConnectorType::FizOperations,
+        connector_name: "stripe".to_string(),
+        connector_id: "something".to_string(),
+        connector_account_details: masking::Secret::new(serde_json::json!({})),
+        disabled: None,
+        metadata: None,
+        payment_methods_enabled: Some(pms_enabled),
+        connector_label: Some("something".to_string()),
+        frm_configs: None,
+        connector_webhook_details: None,
+        profile_id: None,
+        applepay_verified_domains: None,
+        pm_auth_config: None,
+        status: api_enums::ConnectorStatus::Inactive,
+        additional_merchant_data: None,
+    };
+
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "merchant_connector_account_v2")
+    ))]
     let stripe_account = MerchantConnectorResponse {
         connector_type: api_enums::ConnectorType::FizOperations,
         connector_name: "stripe".to_string(),
diff --git a/crates/kgraph_utils/src/mca.rs b/crates/kgraph_utils/src/mca.rs
index a636c4b3a2..95bcf1a9e0 100644
--- a/crates/kgraph_utils/src/mca.rs
+++ b/crates/kgraph_utils/src/mca.rs
@@ -704,7 +704,64 @@ mod tests {
 
     fn build_test_data() -> ConstraintGraph<dir::DirValue> {
         use api_models::{admin::*, payment_methods::*};
-
+        #[cfg(all(feature = "v2", feature = "merchant_connector_account_v2"))]
+        let stripe_account = MerchantConnectorResponse {
+            connector_type: api_enums::ConnectorType::FizOperations,
+            connector_name: "stripe".to_string(),
+            connector_id: "something".to_string(),
+            connector_label: Some("something".to_string()),
+            connector_account_details: masking::Secret::new(serde_json::json!({})),
+            disabled: None,
+            metadata: None,
+            payment_methods_enabled: Some(vec![PaymentMethodsEnabled {
+                payment_method: api_enums::PaymentMethod::Card,
+                payment_method_types: Some(vec![
+                    RequestPaymentMethodTypes {
+                        payment_method_type: api_enums::PaymentMethodType::Credit,
+                        payment_experience: None,
+                        card_networks: Some(vec![
+                            api_enums::CardNetwork::Visa,
+                            api_enums::CardNetwork::Mastercard,
+                        ]),
+                        accepted_currencies: Some(AcceptedCurrencies::EnableOnly(vec![
+                            api_enums::Currency::INR,
+                        ])),
+                        accepted_countries: None,
+                        minimum_amount: Some(MinorUnit::new(10)),
+                        maximum_amount: Some(MinorUnit::new(1000)),
+                        recurring_enabled: true,
+                        installment_payment_enabled: true,
+                    },
+                    RequestPaymentMethodTypes {
+                        payment_method_type: api_enums::PaymentMethodType::Debit,
+                        payment_experience: None,
+                        card_networks: Some(vec![
+                            api_enums::CardNetwork::Maestro,
+                            api_enums::CardNetwork::JCB,
+                        ]),
+                        accepted_currencies: Some(AcceptedCurrencies::EnableOnly(vec![
+                            api_enums::Currency::GBP,
+                        ])),
+                        accepted_countries: None,
+                        minimum_amount: Some(MinorUnit::new(10)),
+                        maximum_amount: Some(MinorUnit::new(1000)),
+                        recurring_enabled: true,
+                        installment_payment_enabled: true,
+                    },
+                ]),
+            }]),
+            frm_configs: None,
+            connector_webhook_details: None,
+            profile_id: None,
+            applepay_verified_domains: None,
+            pm_auth_config: None,
+            status: api_enums::ConnectorStatus::Inactive,
+            additional_merchant_data: None,
+        };
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "merchant_connector_account_v2")
+        ))]
         let stripe_account = MerchantConnectorResponse {
             connector_type: api_enums::ConnectorType::FizOperations,
             connector_name: "stripe".to_string(),
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 62ddb0067b..e7a0db07e8 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -30,11 +30,12 @@ payouts = ["api_models/payouts", "common_enums/payouts", "hyperswitch_domain_mod
 payout_retry = ["payouts"]
 recon = ["email", "api_models/recon"]
 retry = []
-v2 = ["api_models/v2", "diesel_models/v2", "hyperswitch_domain_models/v2", "storage_impl/v2"]
+v2 = ["api_models/v2", "diesel_models/v2", "hyperswitch_domain_models/v2", "storage_impl/v2", "kgraph_utils/v2"]
 v1 = ["api_models/v1", "diesel_models/v1", "hyperswitch_domain_models/v1", "storage_impl/v1"]
 customer_v2 = ["api_models/customer_v2", "diesel_models/customer_v2", "hyperswitch_domain_models/customer_v2"]
 merchant_account_v2 = ["api_models/merchant_account_v2", "diesel_models/merchant_account_v2", "hyperswitch_domain_models/merchant_account_v2"]
 payment_v2 = ["api_models/payment_v2", "diesel_models/payment_v2", "hyperswitch_domain_models/payment_v2"]
+merchant_connector_account_v2 = ["api_models/merchant_connector_account_v2", "kgraph_utils/merchant_connector_account_v2"]
 
 [dependencies]
 actix-cors = "0.6.5"
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index e6c132b499..51fffea3b5 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -4,27 +4,33 @@ use api_models::{
     admin::{self as admin_types},
     enums as api_enums, routing as routing_types,
 };
+use base64::Engine;
 use common_utils::{
     date_time,
     ext_traits::{AsyncExt, ConfigExt, Encode, ValueExt},
     id_type, pii,
-    types::keymanager as km_types,
+    types::keymanager::{self as km_types, KeyManagerState},
 };
 use diesel_models::configs;
 use error_stack::{report, FutureExt, ResultExt};
 use futures::future::try_join_all;
-use masking::{PeekInterface, Secret};
+use masking::{ExposeInterface, PeekInterface, Secret};
 use pm_auth::{connector::plaid::transformers::PlaidAuthType, types as pm_auth_types};
 use regex::Regex;
 use router_env::metrics::add_attributes;
 use uuid::Uuid;
 
+#[cfg(any(feature = "v1", feature = "v2"))]
+use crate::types::transformers::ForeignFrom;
 use crate::{
-    consts,
+    consts::{self, BASE64_ENGINE},
     core::{
         encryption::transfer_encryption_key,
         errors::{self, RouterResponse, RouterResult, StorageErrorExt},
-        payment_methods::{cards, cards::create_encrypted_data, transformers},
+        payment_methods::{
+            cards::{self, create_encrypted_data},
+            transformers,
+        },
         payments::helpers,
         pm_auth::helpers::PaymentAuthConnectorDataExt,
         routing::helpers as routing_helpers,
@@ -41,9 +47,9 @@ use crate::{
             types::{self as domain_types, AsyncLift},
         },
         storage::{self, enums::MerchantStorageScheme},
-        transformers::{ForeignFrom, ForeignTryFrom},
+        transformers::ForeignTryFrom,
     },
-    utils::{self, OptionExt},
+    utils,
 };
 
 const IBAN_MAX_LENGTH: usize = 34;
@@ -180,10 +186,7 @@ pub async fn create_merchant_account(
     req: api::MerchantAccountCreate,
 ) -> RouterResponse<api::MerchantAccountResponse> {
     #[cfg(feature = "keymanager_create")]
-    use {
-        base64::Engine,
-        common_utils::{keymanager, types::keymanager::EncryptionTransferRequest},
-    };
+    use common_utils::{keymanager, types::keymanager::EncryptionTransferRequest};
 
     let db = state.store.as_ref();
 
@@ -202,7 +205,7 @@ pub async fn create_merchant_account(
             key_manager_state,
             EncryptionTransferRequest {
                 identifier: identifier.clone(),
-                key: consts::BASE64_ENGINE.encode(key),
+                key: BASE64_ENGINE.encode(key),
             },
         )
         .await
@@ -1182,39 +1185,1076 @@ async fn validate_merchant_id(
         .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)
 }
 
-fn validate_certificate_in_mca_metadata(
-    connector_metadata: Secret<serde_json::Value>,
-) -> RouterResult<()> {
-    let parsed_connector_metadata = connector_metadata
-        .parse_value::<api_models::payments::ConnectorMetadata>("ConnectorMetadata")
-        .change_context(errors::ParsingError::StructParseFailure("Metadata"))
+struct ConnectorAuthTypeAndMetadataValidation<'a> {
+    connector_name: &'a api_models::enums::Connector,
+    auth_type: &'a types::ConnectorAuthType,
+    connector_meta_data: &'a Option<pii::SecretSerdeValue>,
+}
+
+impl<'a> ConnectorAuthTypeAndMetadataValidation<'a> {
+    pub fn validate_auth_and_metadata_type(
+        &self,
+    ) -> Result<(), error_stack::Report<errors::ApiErrorResponse>> {
+        let connector_auth_type_validation = ConnectorAuthTypeValidation {
+            auth_type: self.auth_type,
+        };
+        connector_auth_type_validation.validate_connector_auth_type()?;
+        self.validate_auth_and_metadata_type_with_connector()
+            .map_err(|err| match *err.current_context() {
+                errors::ConnectorError::InvalidConnectorName => {
+                    err.change_context(errors::ApiErrorResponse::InvalidRequestData {
+                        message: "The connector name is invalid".to_string(),
+                    })
+                }
+                errors::ConnectorError::InvalidConnectorConfig { config: field_name } => err
+                    .change_context(errors::ApiErrorResponse::InvalidRequestData {
+                        message: format!("The {} is invalid", field_name),
+                    }),
+                errors::ConnectorError::FailedToObtainAuthType => {
+                    err.change_context(errors::ApiErrorResponse::InvalidRequestData {
+                        message: "The auth type is invalid for the connector".to_string(),
+                    })
+                }
+                _ => err.change_context(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "The request body is invalid".to_string(),
+                }),
+            })
+    }
+
+    fn validate_auth_and_metadata_type_with_connector(
+        &self,
+    ) -> Result<(), error_stack::Report<errors::ConnectorError>> {
+        use crate::connector::*;
+
+        match self.connector_name {
+            api_enums::Connector::Adyenplatform => {
+                adyenplatform::transformers::AdyenplatformAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            // api_enums::Connector::Payone => {payone::transformers::PayoneAuthType::try_from(val)?;Ok(())} Added as a template code for future usage
+            #[cfg(feature = "dummy_connector")]
+            api_enums::Connector::DummyConnector1
+            | api_enums::Connector::DummyConnector2
+            | api_enums::Connector::DummyConnector3
+            | api_enums::Connector::DummyConnector4
+            | api_enums::Connector::DummyConnector5
+            | api_enums::Connector::DummyConnector6
+            | api_enums::Connector::DummyConnector7 => {
+                dummyconnector::transformers::DummyConnectorAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Aci => {
+                aci::transformers::AciAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Adyen => {
+                adyen::transformers::AdyenAuthType::try_from(self.auth_type)?;
+                adyen::transformers::AdyenConnectorMetadataObject::try_from(
+                    self.connector_meta_data,
+                )?;
+                Ok(())
+            }
+            api_enums::Connector::Airwallex => {
+                airwallex::transformers::AirwallexAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Authorizedotnet => {
+                authorizedotnet::transformers::AuthorizedotnetAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Bankofamerica => {
+                bankofamerica::transformers::BankOfAmericaAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Billwerk => {
+                billwerk::transformers::BillwerkAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Bitpay => {
+                bitpay::transformers::BitpayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Bambora => {
+                bambora::transformers::BamboraAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Bamboraapac => {
+                bamboraapac::transformers::BamboraapacAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Boku => {
+                boku::transformers::BokuAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Bluesnap => {
+                bluesnap::transformers::BluesnapAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Braintree => {
+                braintree::transformers::BraintreeAuthType::try_from(self.auth_type)?;
+                braintree::braintree_graphql_transformers::BraintreeMeta::try_from(
+                    self.connector_meta_data,
+                )?;
+                Ok(())
+            }
+            api_enums::Connector::Cashtocode => {
+                cashtocode::transformers::CashtocodeAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Checkout => {
+                checkout::transformers::CheckoutAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Coinbase => {
+                coinbase::transformers::CoinbaseAuthType::try_from(self.auth_type)?;
+                coinbase::transformers::CoinbaseConnectorMeta::try_from(self.connector_meta_data)?;
+                Ok(())
+            }
+            api_enums::Connector::Cryptopay => {
+                cryptopay::transformers::CryptopayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Cybersource => {
+                cybersource::transformers::CybersourceAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Datatrans => {
+                datatrans::transformers::DatatransAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Dlocal => {
+                dlocal::transformers::DlocalAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Ebanx => {
+                ebanx::transformers::EbanxAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Fiserv => {
+                fiserv::transformers::FiservAuthType::try_from(self.auth_type)?;
+                fiserv::transformers::FiservSessionObject::try_from(self.connector_meta_data)?;
+                Ok(())
+            }
+            api_enums::Connector::Forte => {
+                forte::transformers::ForteAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Globalpay => {
+                globalpay::transformers::GlobalpayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Globepay => {
+                globepay::transformers::GlobepayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Gocardless => {
+                gocardless::transformers::GocardlessAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Gpayments => {
+                gpayments::transformers::GpaymentsAuthType::try_from(self.auth_type)?;
+                gpayments::transformers::GpaymentsMetaData::try_from(self.connector_meta_data)?;
+                Ok(())
+            }
+            api_enums::Connector::Helcim => {
+                helcim::transformers::HelcimAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Iatapay => {
+                iatapay::transformers::IatapayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Itaubank => {
+                itaubank::transformers::ItaubankAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Klarna => {
+                klarna::transformers::KlarnaAuthType::try_from(self.auth_type)?;
+                klarna::transformers::KlarnaConnectorMetadataObject::try_from(
+                    self.connector_meta_data,
+                )?;
+                Ok(())
+            }
+            api_enums::Connector::Mifinity => {
+                mifinity::transformers::MifinityAuthType::try_from(self.auth_type)?;
+                mifinity::transformers::MifinityConnectorMetadataObject::try_from(
+                    self.connector_meta_data,
+                )?;
+                Ok(())
+            }
+            api_enums::Connector::Mollie => {
+                mollie::transformers::MollieAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Multisafepay => {
+                multisafepay::transformers::MultisafepayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Netcetera => {
+                netcetera::transformers::NetceteraAuthType::try_from(self.auth_type)?;
+                netcetera::transformers::NetceteraMetaData::try_from(self.connector_meta_data)?;
+                Ok(())
+            }
+            api_enums::Connector::Nexinets => {
+                nexinets::transformers::NexinetsAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Nmi => {
+                nmi::transformers::NmiAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Noon => {
+                noon::transformers::NoonAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Nuvei => {
+                nuvei::transformers::NuveiAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Opennode => {
+                opennode::transformers::OpennodeAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Payme => {
+                payme::transformers::PaymeAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Paypal => {
+                paypal::transformers::PaypalAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Payone => {
+                payone::transformers::PayoneAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Payu => {
+                payu::transformers::PayuAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Placetopay => {
+                placetopay::transformers::PlacetopayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Powertranz => {
+                powertranz::transformers::PowertranzAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Prophetpay => {
+                prophetpay::transformers::ProphetpayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Rapyd => {
+                rapyd::transformers::RapydAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Razorpay => {
+                razorpay::transformers::RazorpayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Shift4 => {
+                shift4::transformers::Shift4AuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Square => {
+                square::transformers::SquareAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Stax => {
+                stax::transformers::StaxAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Stripe => {
+                stripe::transformers::StripeAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Trustpay => {
+                trustpay::transformers::TrustpayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Tsys => {
+                tsys::transformers::TsysAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Volt => {
+                volt::transformers::VoltAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            // api_enums::Connector::Wellsfargo => {
+            //     wellsfargo::transformers::WellsfargoAuthType::try_from(self.auth_type)?;
+            //     Ok(())
+            // }
+            api_enums::Connector::Wise => {
+                wise::transformers::WiseAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Worldline => {
+                worldline::transformers::WorldlineAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Worldpay => {
+                worldpay::transformers::WorldpayAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Zen => {
+                zen::transformers::ZenAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Zsl => {
+                zsl::transformers::ZslAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Signifyd => {
+                signifyd::transformers::SignifydAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Riskified => {
+                riskified::transformers::RiskifiedAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Plaid => {
+                PlaidAuthType::foreign_try_from(self.auth_type)?;
+                Ok(())
+            }
+            api_enums::Connector::Threedsecureio => {
+                threedsecureio::transformers::ThreedsecureioAuthType::try_from(self.auth_type)?;
+                Ok(())
+            }
+        }
+    }
+}
+
+struct ConnectorAuthTypeValidation<'a> {
+    auth_type: &'a types::ConnectorAuthType,
+}
+
+impl<'a> ConnectorAuthTypeValidation<'a> {
+    fn validate_connector_auth_type(
+        &self,
+    ) -> Result<(), error_stack::Report<errors::ApiErrorResponse>> {
+        let validate_non_empty_field = |field_value: &str, field_name: &str| {
+            if field_value.trim().is_empty() {
+                Err(errors::ApiErrorResponse::InvalidDataFormat {
+                    field_name: format!("connector_account_details.{}", field_name),
+                    expected_format: "a non empty String".to_string(),
+                }
+                .into())
+            } else {
+                Ok(())
+            }
+        };
+        match self.auth_type {
+            hyperswitch_domain_models::router_data::ConnectorAuthType::TemporaryAuth => Ok(()),
+            hyperswitch_domain_models::router_data::ConnectorAuthType::HeaderKey { api_key } => {
+                validate_non_empty_field(api_key.peek(), "api_key")
+            }
+            hyperswitch_domain_models::router_data::ConnectorAuthType::BodyKey {
+                api_key,
+                key1,
+            } => {
+                validate_non_empty_field(api_key.peek(), "api_key")?;
+                validate_non_empty_field(key1.peek(), "key1")
+            }
+            hyperswitch_domain_models::router_data::ConnectorAuthType::SignatureKey {
+                api_key,
+                key1,
+                api_secret,
+            } => {
+                validate_non_empty_field(api_key.peek(), "api_key")?;
+                validate_non_empty_field(key1.peek(), "key1")?;
+                validate_non_empty_field(api_secret.peek(), "api_secret")
+            }
+            hyperswitch_domain_models::router_data::ConnectorAuthType::MultiAuthKey {
+                api_key,
+                key1,
+                api_secret,
+                key2,
+            } => {
+                validate_non_empty_field(api_key.peek(), "api_key")?;
+                validate_non_empty_field(key1.peek(), "key1")?;
+                validate_non_empty_field(api_secret.peek(), "api_secret")?;
+                validate_non_empty_field(key2.peek(), "key2")
+            }
+            hyperswitch_domain_models::router_data::ConnectorAuthType::CurrencyAuthKey {
+                auth_key_map,
+            } => {
+                if auth_key_map.is_empty() {
+                    Err(errors::ApiErrorResponse::InvalidDataFormat {
+                        field_name: "connector_account_details.auth_key_map".to_string(),
+                        expected_format: "a non empty map".to_string(),
+                    }
+                    .into())
+                } else {
+                    Ok(())
+                }
+            }
+            hyperswitch_domain_models::router_data::ConnectorAuthType::CertificateAuth {
+                certificate,
+                private_key,
+            } => {
+                helpers::create_identity_from_certificate_and_key(
+                    certificate.to_owned(),
+                    private_key.to_owned(),
+                )
+                .change_context(errors::ApiErrorResponse::InvalidDataFormat {
+                    field_name:
+                        "connector_account_details.certificate or connector_account_details.private_key"
+                            .to_string(),
+                    expected_format:
+                        "a valid base64 encoded string of PEM encoded Certificate and Private Key"
+                            .to_string(),
+                })?;
+                Ok(())
+            }
+            hyperswitch_domain_models::router_data::ConnectorAuthType::NoKey => Ok(()),
+        }
+    }
+}
+
+struct ConnectorStatusAndDisabledValidation<'a> {
+    status: &'a Option<api_enums::ConnectorStatus>,
+    disabled: &'a Option<bool>,
+    auth: &'a types::ConnectorAuthType,
+    current_status: &'a api_enums::ConnectorStatus,
+}
+
+impl<'a> ConnectorStatusAndDisabledValidation<'a> {
+    fn validate_status_and_disabled(
+        &self,
+    ) -> RouterResult<(api_enums::ConnectorStatus, Option<bool>)> {
+        let connector_status = match (self.status, self.auth) {
+            (
+                Some(common_enums::ConnectorStatus::Active),
+                types::ConnectorAuthType::TemporaryAuth,
+            ) => {
+                return Err(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "Connector status cannot be active when using TemporaryAuth"
+                        .to_string(),
+                }
+                .into());
+            }
+            (Some(status), _) => status,
+            (None, types::ConnectorAuthType::TemporaryAuth) => {
+                &common_enums::ConnectorStatus::Inactive
+            }
+            (None, _) => self.current_status,
+        };
+
+        let disabled = match (self.disabled, connector_status) {
+            (Some(false), common_enums::ConnectorStatus::Inactive) => {
+                return Err(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "Connector cannot be enabled when connector_status is inactive or when using TemporaryAuth"
+                        .to_string(),
+                }
+                .into());
+            }
+            (Some(disabled), _) => Some(*disabled),
+            (None, common_enums::ConnectorStatus::Inactive) => Some(true),
+            (None, _) => None,
+        };
+
+        Ok((*connector_status, disabled))
+    }
+}
+
+struct PaymentMethodsEnabled<'a> {
+    payment_methods_enabled: &'a Option<Vec<api_models::admin::PaymentMethodsEnabled>>,
+}
+
+impl<'a> PaymentMethodsEnabled<'a> {
+    fn get_payment_methods_enabled(&self) -> RouterResult<Option<Vec<serde_json::Value>>> {
+        let mut vec = Vec::new();
+        let payment_methods_enabled = match self.payment_methods_enabled.clone() {
+            Some(val) => {
+                for pm in val.into_iter() {
+                    let pm_value = pm
+                        .encode_to_value()
+                        .change_context(errors::ApiErrorResponse::InternalServerError)
+                        .attach_printable(
+                            "Failed while encoding to serde_json::Value, PaymentMethod",
+                        )?;
+                    vec.push(pm_value)
+                }
+                Some(vec)
+            }
+            None => None,
+        };
+        Ok(payment_methods_enabled)
+    }
+}
+
+struct CertificateAndCertificateKey<'a> {
+    certificate: &'a Secret<String>,
+    certificate_key: &'a Secret<String>,
+}
+
+impl<'a> CertificateAndCertificateKey<'a> {
+    pub fn create_identity_from_certificate_and_key(
+        &self,
+    ) -> Result<reqwest::Identity, error_stack::Report<errors::ApiClientError>> {
+        let decoded_certificate = BASE64_ENGINE
+            .decode(self.certificate.clone().expose())
+            .change_context(errors::ApiClientError::CertificateDecodeFailed)?;
+
+        let decoded_certificate_key = BASE64_ENGINE
+            .decode(self.certificate_key.clone().expose())
+            .change_context(errors::ApiClientError::CertificateDecodeFailed)?;
+
+        let certificate = String::from_utf8(decoded_certificate)
+            .change_context(errors::ApiClientError::CertificateDecodeFailed)?;
+
+        let certificate_key = String::from_utf8(decoded_certificate_key)
+            .change_context(errors::ApiClientError::CertificateDecodeFailed)?;
+
+        reqwest::Identity::from_pkcs8_pem(certificate.as_bytes(), certificate_key.as_bytes())
+            .change_context(errors::ApiClientError::CertificateDecodeFailed)
+    }
+}
+
+struct ConnectorMetadata<'a> {
+    connector_metadata: &'a Option<pii::SecretSerdeValue>,
+}
+
+impl<'a> ConnectorMetadata<'a> {
+    fn validate_apple_pay_certificates_in_mca_metadata(&self) -> RouterResult<()> {
+        self.connector_metadata
+            .clone()
+            .map(api_models::payments::ConnectorMetadata::from_value)
+            .transpose()
+            .change_context(errors::ApiErrorResponse::InvalidDataFormat {
+                field_name: "metadata".to_string(),
+                expected_format: "connector metadata".to_string(),
+            })?
+            .and_then(|metadata| metadata.get_apple_pay_certificates())
+            .map(|(certificate, certificate_key)| {
+                let certificate_and_certificate_key = CertificateAndCertificateKey {
+                    certificate: &certificate,
+                    certificate_key: &certificate_key,
+                };
+                certificate_and_certificate_key.create_identity_from_certificate_and_key()
+            })
+            .transpose()
+            .change_context(errors::ApiErrorResponse::InvalidDataValue {
+                field_name: "certificate/certificate key",
+            })?;
+        Ok(())
+    }
+}
+
+struct PMAuthConfigValidation<'a> {
+    connector_type: &'a api_enums::ConnectorType,
+    pm_auth_config: &'a Option<pii::SecretSerdeValue>,
+    db: &'a dyn StorageInterface,
+    merchant_id: &'a id_type::MerchantId,
+    profile_id: &'a Option<String>,
+    key_store: &'a domain::MerchantKeyStore,
+    key_manager_state: &'a KeyManagerState,
+}
+
+impl<'a> PMAuthConfigValidation<'a> {
+    async fn validate_pm_auth(&self, val: &pii::SecretSerdeValue) -> RouterResponse<()> {
+        let config = serde_json::from_value::<api_models::pm_auth::PaymentMethodAuthConfig>(
+            val.clone().expose(),
+        )
+        .change_context(errors::ApiErrorResponse::InvalidRequestData {
+            message: "invalid data received for payment method auth config".to_string(),
+        })
+        .attach_printable("Failed to deserialize Payment Method Auth config")?;
+
+        let all_mcas = self
+            .db
+            .find_merchant_connector_account_by_merchant_id_and_disabled_list(
+                self.key_manager_state,
+                self.merchant_id,
+                true,
+                self.key_store,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                id: self.merchant_id.get_string_repr().to_owned(),
+            })?;
+
+        for conn_choice in config.enabled_payment_methods {
+            let pm_auth_mca = all_mcas
+                .clone()
+                .into_iter()
+                .find(|mca| mca.merchant_connector_id == conn_choice.mca_id)
+                .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
+                    message: "payment method auth connector account not found".to_string(),
+                })?;
+
+            if &pm_auth_mca.profile_id != self.profile_id {
+                return Err(errors::ApiErrorResponse::GenericNotFoundError {
+                    message: "payment method auth profile_id differs from connector profile_id"
+                        .to_string(),
+                }
+                .into());
+            }
+        }
+
+        Ok(services::ApplicationResponse::StatusOk)
+    }
+
+    async fn validate_pm_auth_config(&self) -> RouterResult<()> {
+        if self.connector_type != &api_enums::ConnectorType::PaymentMethodAuth {
+            if let Some(val) = self.pm_auth_config.clone() {
+                self.validate_pm_auth(&val).await?;
+            }
+        }
+        Ok(())
+    }
+}
+
+struct ConnectorTypeAndConnectorName<'a> {
+    connector_type: &'a api_enums::ConnectorType,
+    connector_name: &'a api_enums::Connector,
+}
+
+impl<'a> ConnectorTypeAndConnectorName<'a> {
+    fn get_routable_connector(&self) -> RouterResult<Option<api_enums::RoutableConnectors>> {
+        let mut routable_connector =
+            api_enums::RoutableConnectors::from_str(&self.connector_name.to_string()).ok();
+
+        let pm_auth_connector =
+            api_enums::convert_pm_auth_connector(self.connector_name.to_string().as_str());
+        let authentication_connector =
+            api_enums::convert_authentication_connector(self.connector_name.to_string().as_str());
+
+        if pm_auth_connector.is_some() {
+            if self.connector_type != &api_enums::ConnectorType::PaymentMethodAuth
+                && self.connector_type != &api_enums::ConnectorType::PaymentProcessor
+            {
+                return Err(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "Invalid connector type given".to_string(),
+                }
+                .into());
+            }
+        } else if authentication_connector.is_some() {
+            if self.connector_type != &api_enums::ConnectorType::AuthenticationProcessor {
+                return Err(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "Invalid connector type given".to_string(),
+                }
+                .into());
+            }
+        } else {
+            let routable_connector_option = self
+                .connector_name
+                .to_string()
+                .parse::<api_enums::RoutableConnectors>()
+                .change_context(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "Invalid connector name given".to_string(),
+                })?;
+            routable_connector = Some(routable_connector_option);
+        };
+        Ok(routable_connector)
+    }
+}
+
+struct MerchantDefaultConfigUpdate<'a> {
+    routable_connector: &'a Option<api_enums::RoutableConnectors>,
+    merchant_connector_id: &'a String,
+    store: &'a dyn StorageInterface,
+    merchant_id: &'a id_type::MerchantId,
+    default_routing_config: &'a Vec<api_models::routing::RoutableConnectorChoice>,
+    default_routing_config_for_profile: &'a Vec<api_models::routing::RoutableConnectorChoice>,
+    profile_id: &'a String,
+    transaction_type: &'a api_enums::TransactionType,
+}
+
+impl<'a> MerchantDefaultConfigUpdate<'a> {
+    async fn update_merchant_default_config(&self) -> RouterResult<()> {
+        let mut default_routing_config = self.default_routing_config.to_owned();
+        let mut default_routing_config_for_profile =
+            self.default_routing_config_for_profile.to_owned();
+        if let Some(routable_connector_val) = self.routable_connector {
+            let choice = routing_types::RoutableConnectorChoice {
+                choice_kind: routing_types::RoutableChoiceKind::FullStruct,
+                connector: *routable_connector_val,
+                merchant_connector_id: Some(self.merchant_connector_id.clone()),
+            };
+            if !default_routing_config.contains(&choice) {
+                default_routing_config.push(choice.clone());
+                routing_helpers::update_merchant_default_config(
+                    self.store,
+                    self.merchant_id.get_string_repr(),
+                    default_routing_config.clone(),
+                    self.transaction_type,
+                )
+                .await?;
+            }
+            if !default_routing_config_for_profile.contains(&choice.clone()) {
+                default_routing_config_for_profile.push(choice);
+                routing_helpers::update_merchant_default_config(
+                    self.store,
+                    self.profile_id,
+                    default_routing_config_for_profile.clone(),
+                    self.transaction_type,
+                )
+                .await?;
+            }
+        }
+        Ok(())
+    }
+}
+
+#[cfg(any(feature = "v1", feature = "v2", feature = "olap"))]
+#[async_trait::async_trait]
+trait MerchantConnectorAccountCreateBridge {
+    async fn create_domain_model_from_request(
+        self,
+        state: &SessionState,
+        key_store: domain::MerchantKeyStore,
+        business_profile: &storage::business_profile::BusinessProfile,
+        key_manager_state: &KeyManagerState,
+    ) -> RouterResult<domain::MerchantConnectorAccount>;
+
+    async fn validate_and_get_profile_id(
+        self,
+        merchant_account: &domain::MerchantAccount,
+        db: &dyn StorageInterface,
+        should_validate: bool,
+    ) -> RouterResult<String>;
+}
+
+#[cfg(all(
+    feature = "v2",
+    feature = "merchant_connector_account_v2",
+    feature = "olap"
+))]
+#[async_trait::async_trait]
+impl MerchantConnectorAccountCreateBridge for api::MerchantConnectorCreate {
+    async fn create_domain_model_from_request(
+        self,
+        state: &SessionState,
+        key_store: domain::MerchantKeyStore,
+        business_profile: &storage::business_profile::BusinessProfile,
+        key_manager_state: &KeyManagerState,
+    ) -> RouterResult<domain::MerchantConnectorAccount> {
+        // If connector label is not passed in the request, generate one
+        let connector_label = self.get_connector_label(business_profile.profile_name.clone());
+        let payment_methods_enabled = PaymentMethodsEnabled {
+            payment_methods_enabled: &self.payment_methods_enabled,
+        };
+        let payment_methods_enabled = payment_methods_enabled.get_payment_methods_enabled()?;
+        let frm_configs = self.get_frm_config_as_secret();
+        // Validate Merchant api details and return error if not in correct format
+        let auth = types::ConnectorAuthType::from_option_secret_value(
+            self.connector_account_details.clone(),
+        )
         .change_context(errors::ApiErrorResponse::InvalidDataFormat {
-            field_name: "metadata".to_string(),
-            expected_format: "connector metadata".to_string(),
+            field_name: "connector_account_details".to_string(),
+            expected_format: "auth_type and api_key".to_string(),
         })?;
 
-    parsed_connector_metadata
-        .apple_pay
-        .and_then(|applepay_metadata| {
-            applepay_metadata
-                .session_token_data
-                .map(|session_token_data| {
-                    let api_models::payments::SessionTokenInfo {
-                        certificate,
-                        certificate_keys,
-                        ..
-                    } = session_token_data;
-
-                    helpers::create_identity_from_certificate_and_key(certificate, certificate_keys)
-                        .change_context(errors::ApiErrorResponse::InvalidDataValue {
-                            field_name: "certificate/certificate key",
-                        })
-                        .map(|_identity_result| ())
-                })
+        let connector_auth_type_and_metadata_validation = ConnectorAuthTypeAndMetadataValidation {
+            connector_name: &self.connector_name,
+            auth_type: &auth,
+            connector_meta_data: &self.metadata,
+        };
+        connector_auth_type_and_metadata_validation.validate_auth_and_metadata_type()?;
+        let connector_status_and_disabled_validation = ConnectorStatusAndDisabledValidation {
+            status: &self.status,
+            disabled: &self.disabled,
+            auth: &auth,
+            current_status: &api_enums::ConnectorStatus::Active,
+        };
+        let (connector_status, disabled) =
+            connector_status_and_disabled_validation.validate_status_and_disabled()?;
+        let identifier = km_types::Identifier::Merchant(business_profile.merchant_id.clone());
+        let merchant_recipient_data = if let Some(data) = &self.additional_merchant_data {
+            Some(
+                process_open_banking_connectors(
+                    state,
+                    &business_profile.merchant_id,
+                    &auth,
+                    &self.connector_type,
+                    &self.connector_name,
+                    types::AdditionalMerchantData::foreign_from(data.clone()),
+                )
+                .await?,
+            )
+        } else {
+            None
+        }
+        .map(|data| {
+            serde_json::to_value(types::AdditionalMerchantData::OpenBankingRecipientData(
+                data,
+            ))
         })
-        .transpose()?;
+        .transpose()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get MerchantRecipientData")?;
+        Ok(domain::MerchantConnectorAccount {
+            merchant_id: business_profile.merchant_id.clone(),
+            connector_type: self.connector_type,
+            connector_name: self.connector_name.to_string(),
+            merchant_connector_id: utils::generate_id(consts::ID_LENGTH, "mca"),
+            connector_account_details: domain_types::encrypt(
+                key_manager_state,
+                self.connector_account_details.ok_or(
+                    errors::ApiErrorResponse::MissingRequiredField {
+                        field_name: "connector_account_details",
+                    },
+                )?,
+                identifier.clone(),
+                key_store.key.peek(),
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encrypt connector account details")?,
+            payment_methods_enabled,
+            disabled,
+            metadata: self.metadata.clone(),
+            frm_configs,
+            connector_label: Some(connector_label.clone()),
+            created_at: date_time::now(),
+            modified_at: date_time::now(),
+            connector_webhook_details: match self.connector_webhook_details {
+                Some(connector_webhook_details) => {
+                    connector_webhook_details.encode_to_value(
+                    )
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable(format!("Failed to serialize api_models::admin::MerchantConnectorWebhookDetails for Merchant: {}", business_profile.merchant_id))
+                    .map(Some)?
+                    .map(Secret::new)
+                }
+                None => None,
+            },
+            profile_id: Some(business_profile.profile_id.clone()),
+            applepay_verified_domains: None,
+            pm_auth_config: self.pm_auth_config.clone(),
+            status: connector_status,
+            connector_wallets_details: helpers::get_encrypted_apple_pay_connector_wallets_details(state, &key_store, &self.metadata).await?,
+            test_mode: None,
+            business_country: None,
+            business_label: None,
+            business_sub_label: None,
+            additional_merchant_data: if let Some(mcd) =  merchant_recipient_data {
+                Some(domain_types::encrypt(
+                    key_manager_state,
+                    Secret::new(mcd),
+                    identifier,
+                    key_store.key.peek(),
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Unable to encrypt additional_merchant_data")?)
+            } else {
+                None
+            },
+        })
+    }
 
-    Ok(())
+    /// If profile_id is not passed, use default profile if available
+    /// or return a `MissingRequiredField` error
+    async fn validate_and_get_profile_id(
+        self,
+        merchant_account: &domain::MerchantAccount,
+        db: &dyn StorageInterface,
+        should_validate: bool,
+    ) -> RouterResult<String> {
+        match self.profile_id {
+            Some(profile_id) => {
+                // Check whether this business profile belongs to the merchant
+                if should_validate {
+                    let _ = core_utils::validate_and_get_business_profile(
+                        db,
+                        Some(&profile_id),
+                        merchant_account.get_id(),
+                    )
+                    .await?;
+                }
+                Ok(profile_id.clone())
+            }
+            None => Err(report!(errors::ApiErrorResponse::MissingRequiredField {
+                field_name: "profile_id"
+            })),
+        }
+    }
+}
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2", feature = "olap"),
+    not(feature = "merchant_connector_account_v2")
+))]
+#[async_trait::async_trait]
+impl MerchantConnectorAccountCreateBridge for api::MerchantConnectorCreate {
+    async fn create_domain_model_from_request(
+        self,
+        state: &SessionState,
+        key_store: domain::MerchantKeyStore,
+        business_profile: &storage::business_profile::BusinessProfile,
+        key_manager_state: &KeyManagerState,
+    ) -> RouterResult<domain::MerchantConnectorAccount> {
+        // If connector label is not passed in the request, generate one
+        let connector_label = self
+            .connector_label
+            .clone()
+            .or(core_utils::get_connector_label(
+                self.business_country,
+                self.business_label.as_ref(),
+                self.business_sub_label.as_ref(),
+                &self.connector_name.to_string(),
+            ))
+            .unwrap_or(format!(
+                "{}_{}",
+                self.connector_name, business_profile.profile_name
+            ));
+        let payment_methods_enabled = PaymentMethodsEnabled {
+            payment_methods_enabled: &self.payment_methods_enabled,
+        };
+        let payment_methods_enabled = payment_methods_enabled.get_payment_methods_enabled()?;
+        let frm_configs = self.get_frm_config_as_secret();
+        // Validate Merchant api details and return error if not in correct format
+        let auth = types::ConnectorAuthType::from_option_secret_value(
+            self.connector_account_details.clone(),
+        )
+        .change_context(errors::ApiErrorResponse::InvalidDataFormat {
+            field_name: "connector_account_details".to_string(),
+            expected_format: "auth_type and api_key".to_string(),
+        })?;
+
+        let connector_auth_type_and_metadata_validation = ConnectorAuthTypeAndMetadataValidation {
+            connector_name: &self.connector_name,
+            auth_type: &auth,
+            connector_meta_data: &self.metadata,
+        };
+        connector_auth_type_and_metadata_validation.validate_auth_and_metadata_type()?;
+        let connector_status_and_disabled_validation = ConnectorStatusAndDisabledValidation {
+            status: &self.status,
+            disabled: &self.disabled,
+            auth: &auth,
+            current_status: &api_enums::ConnectorStatus::Active,
+        };
+        let (connector_status, disabled) =
+            connector_status_and_disabled_validation.validate_status_and_disabled()?;
+        let identifier = km_types::Identifier::Merchant(business_profile.merchant_id.clone());
+        let merchant_recipient_data = if let Some(data) = &self.additional_merchant_data {
+            Some(
+                process_open_banking_connectors(
+                    state,
+                    &business_profile.merchant_id,
+                    &auth,
+                    &self.connector_type,
+                    &self.connector_name,
+                    types::AdditionalMerchantData::foreign_from(data.clone()),
+                )
+                .await?,
+            )
+        } else {
+            None
+        }
+        .map(|data| {
+            serde_json::to_value(types::AdditionalMerchantData::OpenBankingRecipientData(
+                data,
+            ))
+        })
+        .transpose()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get MerchantRecipientData")?;
+        Ok(domain::MerchantConnectorAccount {
+            merchant_id: business_profile.merchant_id.clone(),
+            connector_type: self.connector_type,
+            connector_name: self.connector_name.to_string(),
+            merchant_connector_id: utils::generate_id(consts::ID_LENGTH, "mca"),
+            connector_account_details: domain_types::encrypt(
+                key_manager_state,
+                self.connector_account_details.ok_or(
+                    errors::ApiErrorResponse::MissingRequiredField {
+                        field_name: "connector_account_details",
+                    },
+                )?,
+                identifier.clone(),
+                key_store.key.peek(),
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encrypt connector account details")?,
+            payment_methods_enabled,
+            disabled,
+            metadata: self.metadata.clone(),
+            frm_configs,
+            connector_label: Some(connector_label.clone()),
+            created_at: date_time::now(),
+            modified_at: date_time::now(),
+            connector_webhook_details: match self.connector_webhook_details {
+                Some(connector_webhook_details) => {
+                    connector_webhook_details.encode_to_value(
+                    )
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable(format!("Failed to serialize api_models::admin::MerchantConnectorWebhookDetails for Merchant: {}", business_profile.merchant_id))
+                    .map(Some)?
+                    .map(Secret::new)
+                }
+                None => None,
+            },
+            profile_id: Some(business_profile.profile_id.clone()),
+            applepay_verified_domains: None,
+            pm_auth_config: self.pm_auth_config.clone(),
+            status: connector_status,
+            connector_wallets_details: helpers::get_encrypted_apple_pay_connector_wallets_details(state, &key_store, &self.metadata).await?,
+            test_mode: self.test_mode,
+            business_country: self.business_country,
+            business_label: self.business_label.clone(),
+            business_sub_label: self.business_sub_label.clone(),
+            additional_merchant_data: if let Some(mcd) =  merchant_recipient_data {
+                Some(domain_types::encrypt(
+                    key_manager_state,
+                    Secret::new(mcd),
+                    identifier,
+                    key_store.key.peek(),
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Unable to encrypt additional_merchant_data")?)
+            } else {
+                None
+            },
+        })
+    }
+
+    /// If profile_id is not passed, use default profile if available, or
+    /// If business_details (business_country and business_label) are passed, get the business_profile
+    /// or return a `MissingRequiredField` error
+    async fn validate_and_get_profile_id(
+        self,
+        merchant_account: &domain::MerchantAccount,
+        db: &dyn StorageInterface,
+        should_validate: bool,
+    ) -> RouterResult<String> {
+        match self.profile_id.or(merchant_account.default_profile.clone()) {
+            Some(profile_id) => {
+                // Check whether this business profile belongs to the merchant
+                if should_validate {
+                    let _ = core_utils::validate_and_get_business_profile(
+                        db,
+                        Some(&profile_id),
+                        merchant_account.get_id(),
+                    )
+                    .await?;
+                }
+                Ok(profile_id.clone())
+            }
+            None => match self.business_country.zip(self.business_label) {
+                Some((business_country, business_label)) => {
+                    let profile_name = format!("{business_country}_{business_label}");
+                    let business_profile = db
+                        .find_business_profile_by_profile_name_merchant_id(
+                            &profile_name,
+                            merchant_account.get_id(),
+                        )
+                        .await
+                        .to_not_found_response(
+                            errors::ApiErrorResponse::BusinessProfileNotFound { id: profile_name },
+                        )?;
+
+                    Ok(business_profile.profile_id)
+                }
+                _ => Err(report!(errors::ApiErrorResponse::MissingRequiredField {
+                    field_name: "profile_id or business_country, business_label"
+                })),
+            },
+        }
+    }
 }
 
 pub async fn create_payment_connector(
@@ -1225,7 +2265,13 @@ pub async fn create_payment_connector(
     let store = state.store.as_ref();
     let key_manager_state = &(&state).into();
     #[cfg(feature = "dummy_connector")]
-    validate_dummy_connector_enabled(&state, &req.connector_name).await?;
+    req.connector_name
+        .clone()
+        .validate_dummy_connector_enabled(state.conf.dummy_connector.enabled)
+        .change_context(errors::ApiErrorResponse::InvalidRequestData {
+            message: "Invalid connector name".to_string(),
+        })?;
+
     let key_store = store
         .get_merchant_key_store_by_merchant_id(
             key_manager_state,
@@ -1235,10 +2281,11 @@ pub async fn create_payment_connector(
         .await
         .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
 
-    req.metadata
-        .clone()
-        .map(validate_certificate_in_mca_metadata)
-        .transpose()?;
+    let connector_metadata = ConnectorMetadata {
+        connector_metadata: &req.metadata,
+    };
+
+    connector_metadata.validate_apple_pay_certificates_in_mca_metadata()?;
 
     let merchant_account = state
         .store
@@ -1246,25 +2293,31 @@ pub async fn create_payment_connector(
         .await
         .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "merchant_connector_account_v2")
+    ))]
     helpers::validate_business_details(
         req.business_country,
         req.business_label.as_ref(),
         &merchant_account,
     )?;
 
-    // Business label support will be deprecated soon
-    let profile_id = core_utils::get_profile_id_from_business_details(
-        req.business_country,
-        req.business_label.as_ref(),
-        &merchant_account,
-        req.profile_id.as_ref(),
-        store,
-        true,
-    )
-    .await?;
+    let profile_id = req
+        .clone()
+        .validate_and_get_profile_id(&merchant_account, store, true)
+        .await?;
 
-    let mut routable_connector =
-        api_enums::RoutableConnectors::from_str(&req.connector_name.to_string()).ok();
+    let pm_auth_config_validation = PMAuthConfigValidation {
+        connector_type: &req.connector_type,
+        pm_auth_config: &req.pm_auth_config,
+        db: store,
+        merchant_id,
+        profile_id: &Some(profile_id.clone()),
+        key_store: &key_store,
+        key_manager_state,
+    };
+    pm_auth_config_validation.validate_pm_auth_config().await?;
 
     let business_profile = state
         .store
@@ -1274,106 +2327,11 @@ pub async fn create_payment_connector(
             id: profile_id.to_owned(),
         })?;
 
-    let pm_auth_connector =
-        api_enums::convert_pm_auth_connector(req.connector_name.to_string().as_str());
-    let authentication_connector =
-        api_enums::convert_authentication_connector(req.connector_name.to_string().as_str());
-
-    if pm_auth_connector.is_some() {
-        if req.connector_type != api_enums::ConnectorType::PaymentMethodAuth
-            && req.connector_type != api_enums::ConnectorType::PaymentProcessor
-        {
-            return Err(errors::ApiErrorResponse::InvalidRequestData {
-                message: "Invalid connector type given".to_string(),
-            }
-            .into());
-        }
-    } else if authentication_connector.is_some() {
-        if req.connector_type != api_enums::ConnectorType::AuthenticationProcessor {
-            return Err(errors::ApiErrorResponse::InvalidRequestData {
-                message: "Invalid connector type given".to_string(),
-            }
-            .into());
-        }
-    } else {
-        let routable_connector_option = req
-            .connector_name
-            .to_string()
-            .parse::<api_enums::RoutableConnectors>()
-            .change_context(errors::ApiErrorResponse::InvalidRequestData {
-                message: "Invalid connector name given".to_string(),
-            })?;
-        routable_connector = Some(routable_connector_option);
+    let connector_type_and_connector_enum = ConnectorTypeAndConnectorName {
+        connector_type: &req.connector_type,
+        connector_name: &req.connector_name,
     };
-
-    // If connector label is not passed in the request, generate one
-    let connector_label = req
-        .connector_label
-        .or(core_utils::get_connector_label(
-            req.business_country,
-            req.business_label.as_ref(),
-            req.business_sub_label.as_ref(),
-            &req.connector_name.to_string(),
-        ))
-        .unwrap_or(format!(
-            "{}_{}",
-            req.connector_name, business_profile.profile_name
-        ));
-
-    let mut vec = Vec::new();
-    let payment_methods_enabled = match req.payment_methods_enabled {
-        Some(val) => {
-            for pm in val.into_iter() {
-                let pm_value = pm
-                    .encode_to_value()
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable(
-                        "Failed while encoding to serde_json::Value, PaymentMethod",
-                    )?;
-                vec.push(pm_value)
-            }
-            Some(vec)
-        }
-        None => None,
-    };
-
-    // Validate Merchant api details and return error if not in correct format
-    let auth: types::ConnectorAuthType = req
-        .connector_account_details
-        .clone()
-        .parse_value("ConnectorAuthType")
-        .change_context(errors::ApiErrorResponse::InvalidDataFormat {
-            field_name: "connector_account_details".to_string(),
-            expected_format: "auth_type and api_key".to_string(),
-        })?;
-
-    validate_auth_and_metadata_type(req.connector_name, &auth, &req.metadata)?;
-
-    let merchant_recipient_data = if let Some(data) = &req.additional_merchant_data {
-        Some(
-            process_open_banking_connectors(
-                &state,
-                merchant_id,
-                &auth,
-                &req.connector_type,
-                &req.connector_name,
-                types::AdditionalMerchantData::foreign_from(data.clone()),
-            )
-            .await?,
-        )
-    } else {
-        None
-    }
-    .map(|data| {
-        serde_json::to_value(types::AdditionalMerchantData::OpenBankingRecipientData(
-            data,
-        ))
-    })
-    .transpose()
-    .change_context(errors::ApiErrorResponse::InternalServerError)
-    .attach_printable("Failed to get MerchantRecipientData")?;
-
-    let frm_configs = get_frm_config_as_secret(req.frm_configs);
+    let routable_connector = connector_type_and_connector_enum.get_routable_connector()?;
 
     // The purpose of this merchant account update is just to update the
     // merchant account `modified_at` field for KGraph cache invalidation
@@ -1389,93 +2347,17 @@ pub async fn create_payment_connector(
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("error updating the merchant account when creating payment connector")?;
 
-    let (connector_status, disabled) = validate_status_and_disabled(
-        req.status,
-        req.disabled,
-        auth.clone(),
-        // The validate_status_and_disabled function will use this value only
-        // when the status can be active. So we are passing this as fallback.
-        api_enums::ConnectorStatus::Active,
-    )?;
-
-    if req.connector_type != api_enums::ConnectorType::PaymentMethodAuth {
-        if let Some(val) = req.pm_auth_config.clone() {
-            validate_pm_auth(
-                val,
-                &state,
-                merchant_id,
-                &key_store,
-                merchant_account,
-                &Some(profile_id.clone()),
-            )
-            .await?;
-        }
-    }
-
-    let connector_auth = serde_json::to_value(auth)
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("Failed while encoding ConnectorAuthType to serde_json::Value")?;
-    let conn_auth = Secret::new(connector_auth);
-    let identifier = km_types::Identifier::Merchant(key_store.merchant_id.clone());
-    let merchant_connector_account = domain::MerchantConnectorAccount {
-        merchant_id: merchant_id.to_owned(),
-        connector_type: req.connector_type,
-        connector_name: req.connector_name.to_string(),
-        merchant_connector_id: utils::generate_id(consts::ID_LENGTH, "mca"),
-        connector_account_details: domain_types::encrypt(key_manager_state,
-            conn_auth,
-            identifier.clone(),
-            key_store.key.peek(),
+    let merchant_connector_account = req
+        .clone()
+        .create_domain_model_from_request(
+            &state,
+            key_store.clone(),
+            &business_profile,
+            key_manager_state,
         )
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("Unable to encrypt connector account details")?,
-        payment_methods_enabled,
-        test_mode: req.test_mode,
-        disabled,
-        metadata: req.metadata.clone(),
-        frm_configs,
-        connector_label: Some(connector_label.clone()),
-        business_country: req.business_country,
-        business_label: req.business_label.clone(),
-        business_sub_label: req.business_sub_label.clone(),
-        created_at: date_time::now(),
-        modified_at: date_time::now(),
-        connector_webhook_details: match req.connector_webhook_details {
-            Some(connector_webhook_details) => {
-                connector_webhook_details.encode_to_value(
-                )
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable(format!("Failed to serialize api_models::admin::MerchantConnectorWebhookDetails for Merchant: {:?}", merchant_id))
-                .map(Some)?
-                .map(Secret::new)
-            }
-            None => None,
-        },
-        profile_id: Some(profile_id.clone()),
-        applepay_verified_domains: None,
-        pm_auth_config: req.pm_auth_config.clone(),
-        status: connector_status,
-        connector_wallets_details: helpers::get_encrypted_apple_pay_connector_wallets_details(&state, &key_store, &req.metadata).await?,
-        additional_merchant_data: if let Some(mcd) =  merchant_recipient_data {
-            Some(domain_types::encrypt(key_manager_state,
-                Secret::new(mcd),
-                identifier,
-                key_store.key.peek(),
-            )
-            .await
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Unable to encrypt additional_merchant_data")?)
-        } else {
-            None
-        },
-    };
+        .await?;
 
-    let transaction_type = match req.connector_type {
-        #[cfg(feature = "payouts")]
-        api_enums::ConnectorType::PayoutProcessor => api_enums::TransactionType::Payout,
-        _ => api_enums::TransactionType::Payment,
-    };
+    let transaction_type = req.get_transaction_type();
 
     let mut default_routing_config = routing_helpers::get_merchant_default_config(
         &*state.store,
@@ -1495,45 +2377,34 @@ pub async fn create_payment_connector(
         .store
         .insert_merchant_connector_account(
             key_manager_state,
-            merchant_connector_account,
+            merchant_connector_account.clone(),
             &key_store,
         )
         .await
         .to_duplicate_response(
             errors::ApiErrorResponse::DuplicateMerchantConnectorAccount {
                 profile_id: profile_id.clone(),
-                connector_label,
+                connector_label: merchant_connector_account
+                    .connector_label
+                    .unwrap_or_default(),
             },
         )?;
 
-    if let Some(routable_connector_val) = routable_connector {
-        let choice = routing_types::RoutableConnectorChoice {
-            choice_kind: routing_types::RoutableChoiceKind::FullStruct,
-            connector: routable_connector_val,
-            merchant_connector_id: Some(mca.merchant_connector_id.clone()),
-        };
+    //update merchant default config
+    let merchant_default_config_update = MerchantDefaultConfigUpdate {
+        routable_connector: &routable_connector,
+        merchant_connector_id: &mca.merchant_connector_id,
+        store,
+        merchant_id,
+        default_routing_config: &mut default_routing_config,
+        default_routing_config_for_profile: &mut default_routing_config_for_profile,
+        profile_id: &profile_id,
+        transaction_type: &transaction_type,
+    };
 
-        if !default_routing_config.contains(&choice) {
-            default_routing_config.push(choice.clone());
-            routing_helpers::update_merchant_default_config(
-                &*state.store,
-                merchant_id.get_string_repr(),
-                default_routing_config.clone(),
-                &transaction_type,
-            )
-            .await?;
-        }
-        if !default_routing_config_for_profile.contains(&choice.clone()) {
-            default_routing_config_for_profile.push(choice);
-            routing_helpers::update_merchant_default_config(
-                &*state.store,
-                &profile_id.clone(),
-                default_routing_config_for_profile.clone(),
-                &transaction_type,
-            )
-            .await?;
-        }
-    }
+    merchant_default_config_update
+        .update_merchant_default_config()
+        .await?;
 
     metrics::MCA_CREATE.add(
         &metrics::CONTEXT,
@@ -1549,18 +2420,19 @@ pub async fn create_payment_connector(
 }
 
 async fn validate_pm_auth(
-    val: serde_json::Value,
+    val: pii::SecretSerdeValue,
     state: &SessionState,
     merchant_id: &id_type::MerchantId,
     key_store: &domain::MerchantKeyStore,
     merchant_account: domain::MerchantAccount,
     profile_id: &Option<String>,
 ) -> RouterResponse<()> {
-    let config = serde_json::from_value::<api_models::pm_auth::PaymentMethodAuthConfig>(val)
-        .change_context(errors::ApiErrorResponse::InvalidRequestData {
-            message: "invalid data received for payment method auth config".to_string(),
-        })
-        .attach_printable("Failed to deserialize Payment Method Auth config")?;
+    let config =
+        serde_json::from_value::<api_models::pm_auth::PaymentMethodAuthConfig>(val.expose())
+            .change_context(errors::ApiErrorResponse::InvalidRequestData {
+                message: "invalid data received for payment method auth config".to_string(),
+            })
+            .attach_printable("Failed to deserialize Payment Method Auth config")?;
 
     let all_mcas = &*state
         .store
@@ -1731,10 +2603,20 @@ pub async fn update_payment_connector(
             field_name: "connector",
         })
         .attach_printable_lazy(|| format!("unable to parse connector name {connector_name:?}"))?;
-    validate_auth_and_metadata_type(connector_enum, &auth, &metadata)?;
-
+    let connector_auth_type_and_metadata_validation = ConnectorAuthTypeAndMetadataValidation {
+        connector_name: &connector_enum,
+        auth_type: &auth,
+        connector_meta_data: &metadata,
+    };
+    connector_auth_type_and_metadata_validation.validate_auth_and_metadata_type()?;
+    let connector_status_and_disabled_validation = ConnectorStatusAndDisabledValidation {
+        status: &req.status,
+        disabled: &req.disabled,
+        auth: &auth,
+        current_status: &mca.status,
+    };
     let (connector_status, disabled) =
-        validate_status_and_disabled(req.status, req.disabled, auth, mca.status)?;
+        connector_status_and_disabled_validation.validate_status_and_disabled()?;
 
     if req.connector_type != api_enums::ConnectorType::PaymentMethodAuth {
         if let Some(val) = req.pm_auth_config.clone() {
@@ -2429,409 +3311,6 @@ pub async fn connector_agnostic_mit_toggle(
     ))
 }
 
-pub fn validate_auth_and_metadata_type(
-    connector_name: api_models::enums::Connector,
-    auth_type: &types::ConnectorAuthType,
-    connector_meta_data: &Option<pii::SecretSerdeValue>,
-) -> Result<(), error_stack::Report<errors::ApiErrorResponse>> {
-    validate_connector_auth_type(auth_type)?;
-    validate_auth_and_metadata_type_with_connector(connector_name, auth_type, connector_meta_data)
-        .map_err(|err| match *err.current_context() {
-            errors::ConnectorError::InvalidConnectorName => {
-                err.change_context(errors::ApiErrorResponse::InvalidRequestData {
-                    message: "The connector name is invalid".to_string(),
-                })
-            }
-            errors::ConnectorError::InvalidConnectorConfig { config: field_name } => err
-                .change_context(errors::ApiErrorResponse::InvalidRequestData {
-                    message: format!("The {} is invalid", field_name),
-                }),
-            errors::ConnectorError::FailedToObtainAuthType => {
-                err.change_context(errors::ApiErrorResponse::InvalidRequestData {
-                    message: "The auth type is invalid for the connector".to_string(),
-                })
-            }
-            _ => err.change_context(errors::ApiErrorResponse::InvalidRequestData {
-                message: "The request body is invalid".to_string(),
-            }),
-        })
-}
-
-pub(crate) fn validate_auth_and_metadata_type_with_connector(
-    connector_name: api_models::enums::Connector,
-    val: &types::ConnectorAuthType,
-    connector_meta_data: &Option<pii::SecretSerdeValue>,
-) -> Result<(), error_stack::Report<errors::ConnectorError>> {
-    use crate::connector::*;
-
-    match connector_name {
-        api_enums::Connector::Adyenplatform => {
-            adyenplatform::transformers::AdyenplatformAuthType::try_from(val)?;
-            Ok(())
-        }
-        // api_enums::Connector::Payone => {payone::transformers::PayoneAuthType::try_from(val)?;Ok(())} Added as a template code for future usage
-        #[cfg(feature = "dummy_connector")]
-        api_enums::Connector::DummyConnector1
-        | api_enums::Connector::DummyConnector2
-        | api_enums::Connector::DummyConnector3
-        | api_enums::Connector::DummyConnector4
-        | api_enums::Connector::DummyConnector5
-        | api_enums::Connector::DummyConnector6
-        | api_enums::Connector::DummyConnector7 => {
-            dummyconnector::transformers::DummyConnectorAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Aci => {
-            aci::transformers::AciAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Adyen => {
-            adyen::transformers::AdyenAuthType::try_from(val)?;
-            adyen::transformers::AdyenConnectorMetadataObject::try_from(connector_meta_data)?;
-            Ok(())
-        }
-        api_enums::Connector::Airwallex => {
-            airwallex::transformers::AirwallexAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Authorizedotnet => {
-            authorizedotnet::transformers::AuthorizedotnetAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Bankofamerica => {
-            bankofamerica::transformers::BankOfAmericaAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Billwerk => {
-            billwerk::transformers::BillwerkAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Bitpay => {
-            bitpay::transformers::BitpayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Bambora => {
-            bambora::transformers::BamboraAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Bamboraapac => {
-            bamboraapac::transformers::BamboraapacAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Boku => {
-            boku::transformers::BokuAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Bluesnap => {
-            bluesnap::transformers::BluesnapAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Braintree => {
-            braintree::transformers::BraintreeAuthType::try_from(val)?;
-            braintree::braintree_graphql_transformers::BraintreeMeta::try_from(
-                connector_meta_data,
-            )?;
-            Ok(())
-        }
-        api_enums::Connector::Cashtocode => {
-            cashtocode::transformers::CashtocodeAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Checkout => {
-            checkout::transformers::CheckoutAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Coinbase => {
-            coinbase::transformers::CoinbaseAuthType::try_from(val)?;
-            coinbase::transformers::CoinbaseConnectorMeta::try_from(connector_meta_data)?;
-            Ok(())
-        }
-        api_enums::Connector::Cryptopay => {
-            cryptopay::transformers::CryptopayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Cybersource => {
-            cybersource::transformers::CybersourceAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Datatrans => {
-            datatrans::transformers::DatatransAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Dlocal => {
-            dlocal::transformers::DlocalAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Ebanx => {
-            ebanx::transformers::EbanxAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Fiserv => {
-            fiserv::transformers::FiservAuthType::try_from(val)?;
-            fiserv::transformers::FiservSessionObject::try_from(connector_meta_data)?;
-            Ok(())
-        }
-        api_enums::Connector::Forte => {
-            forte::transformers::ForteAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Globalpay => {
-            globalpay::transformers::GlobalpayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Globepay => {
-            globepay::transformers::GlobepayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Gocardless => {
-            gocardless::transformers::GocardlessAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Gpayments => {
-            gpayments::transformers::GpaymentsAuthType::try_from(val)?;
-            gpayments::transformers::GpaymentsMetaData::try_from(connector_meta_data)?;
-            Ok(())
-        }
-        api_enums::Connector::Helcim => {
-            helcim::transformers::HelcimAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Iatapay => {
-            iatapay::transformers::IatapayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Itaubank => {
-            itaubank::transformers::ItaubankAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Klarna => {
-            klarna::transformers::KlarnaAuthType::try_from(val)?;
-            klarna::transformers::KlarnaConnectorMetadataObject::try_from(connector_meta_data)?;
-            Ok(())
-        }
-        api_enums::Connector::Mifinity => {
-            mifinity::transformers::MifinityAuthType::try_from(val)?;
-            mifinity::transformers::MifinityConnectorMetadataObject::try_from(connector_meta_data)?;
-            Ok(())
-        }
-        api_enums::Connector::Mollie => {
-            mollie::transformers::MollieAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Multisafepay => {
-            multisafepay::transformers::MultisafepayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Netcetera => {
-            netcetera::transformers::NetceteraAuthType::try_from(val)?;
-            netcetera::transformers::NetceteraMetaData::try_from(connector_meta_data)?;
-            Ok(())
-        }
-        api_enums::Connector::Nexinets => {
-            nexinets::transformers::NexinetsAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Nmi => {
-            nmi::transformers::NmiAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Noon => {
-            noon::transformers::NoonAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Nuvei => {
-            nuvei::transformers::NuveiAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Opennode => {
-            opennode::transformers::OpennodeAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Payme => {
-            payme::transformers::PaymeAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Paypal => {
-            paypal::transformers::PaypalAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Payone => {
-            payone::transformers::PayoneAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Payu => {
-            payu::transformers::PayuAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Placetopay => {
-            placetopay::transformers::PlacetopayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Powertranz => {
-            powertranz::transformers::PowertranzAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Prophetpay => {
-            prophetpay::transformers::ProphetpayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Rapyd => {
-            rapyd::transformers::RapydAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Razorpay => {
-            razorpay::transformers::RazorpayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Shift4 => {
-            shift4::transformers::Shift4AuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Square => {
-            square::transformers::SquareAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Stax => {
-            stax::transformers::StaxAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Stripe => {
-            stripe::transformers::StripeAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Trustpay => {
-            trustpay::transformers::TrustpayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Tsys => {
-            tsys::transformers::TsysAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Volt => {
-            volt::transformers::VoltAuthType::try_from(val)?;
-            Ok(())
-        }
-        // api_enums::Connector::Wellsfargo => {
-        //     wellsfargo::transformers::WellsfargoAuthType::try_from(val)?;
-        //     Ok(())
-        // }
-        api_enums::Connector::Wise => {
-            wise::transformers::WiseAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Worldline => {
-            worldline::transformers::WorldlineAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Worldpay => {
-            worldpay::transformers::WorldpayAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Zen => {
-            zen::transformers::ZenAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Zsl => {
-            zsl::transformers::ZslAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Signifyd => {
-            signifyd::transformers::SignifydAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Riskified => {
-            riskified::transformers::RiskifiedAuthType::try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Plaid => {
-            PlaidAuthType::foreign_try_from(val)?;
-            Ok(())
-        }
-        api_enums::Connector::Threedsecureio => {
-            threedsecureio::transformers::ThreedsecureioAuthType::try_from(val)?;
-            Ok(())
-        }
-    }
-}
-
-pub(crate) fn validate_connector_auth_type(
-    auth_type: &types::ConnectorAuthType,
-) -> Result<(), error_stack::Report<errors::ApiErrorResponse>> {
-    let validate_non_empty_field = |field_value: &str, field_name: &str| {
-        if field_value.trim().is_empty() {
-            Err(errors::ApiErrorResponse::InvalidDataFormat {
-                field_name: format!("connector_account_details.{}", field_name),
-                expected_format: "a non empty String".to_string(),
-            }
-            .into())
-        } else {
-            Ok(())
-        }
-    };
-    match auth_type {
-        hyperswitch_domain_models::router_data::ConnectorAuthType::TemporaryAuth => Ok(()),
-        hyperswitch_domain_models::router_data::ConnectorAuthType::HeaderKey { api_key } => {
-            validate_non_empty_field(api_key.peek(), "api_key")
-        }
-        hyperswitch_domain_models::router_data::ConnectorAuthType::BodyKey { api_key, key1 } => {
-            validate_non_empty_field(api_key.peek(), "api_key")?;
-            validate_non_empty_field(key1.peek(), "key1")
-        }
-        hyperswitch_domain_models::router_data::ConnectorAuthType::SignatureKey {
-            api_key,
-            key1,
-            api_secret,
-        } => {
-            validate_non_empty_field(api_key.peek(), "api_key")?;
-            validate_non_empty_field(key1.peek(), "key1")?;
-            validate_non_empty_field(api_secret.peek(), "api_secret")
-        }
-        hyperswitch_domain_models::router_data::ConnectorAuthType::MultiAuthKey {
-            api_key,
-            key1,
-            api_secret,
-            key2,
-        } => {
-            validate_non_empty_field(api_key.peek(), "api_key")?;
-            validate_non_empty_field(key1.peek(), "key1")?;
-            validate_non_empty_field(api_secret.peek(), "api_secret")?;
-            validate_non_empty_field(key2.peek(), "key2")
-        }
-        hyperswitch_domain_models::router_data::ConnectorAuthType::CurrencyAuthKey {
-            auth_key_map,
-        } => {
-            if auth_key_map.is_empty() {
-                Err(errors::ApiErrorResponse::InvalidDataFormat {
-                    field_name: "connector_account_details.auth_key_map".to_string(),
-                    expected_format: "a non empty map".to_string(),
-                }
-                .into())
-            } else {
-                Ok(())
-            }
-        }
-        hyperswitch_domain_models::router_data::ConnectorAuthType::CertificateAuth {
-            certificate,
-            private_key,
-        } => {
-            helpers::create_identity_from_certificate_and_key(
-                certificate.to_owned(),
-                private_key.to_owned(),
-            )
-            .change_context(errors::ApiErrorResponse::InvalidDataFormat {
-                field_name:
-                    "connector_account_details.certificate or connector_account_details.private_key"
-                        .to_string(),
-                expected_format:
-                    "a valid base64 encoded string of PEM encoded Certificate and Private Key"
-                        .to_string(),
-            })?;
-            Ok(())
-        }
-        hyperswitch_domain_models::router_data::ConnectorAuthType::NoKey => Ok(()),
-    }
-}
-
 pub async fn transfer_key_store_to_key_manager(
     state: SessionState,
     req: admin_types::MerchantKeyTransferRequest,
@@ -2845,65 +3324,6 @@ pub async fn transfer_key_store_to_key_manager(
     ))
 }
 
-#[cfg(feature = "dummy_connector")]
-pub async fn validate_dummy_connector_enabled(
-    state: &SessionState,
-    connector_name: &api_enums::Connector,
-) -> Result<(), errors::ApiErrorResponse> {
-    if !state.conf.dummy_connector.enabled
-        && matches!(
-            connector_name,
-            api_enums::Connector::DummyConnector1
-                | api_enums::Connector::DummyConnector2
-                | api_enums::Connector::DummyConnector3
-                | api_enums::Connector::DummyConnector4
-                | api_enums::Connector::DummyConnector5
-                | api_enums::Connector::DummyConnector6
-                | api_enums::Connector::DummyConnector7
-        )
-    {
-        Err(errors::ApiErrorResponse::InvalidRequestData {
-            message: "Invalid connector name".to_string(),
-        })
-    } else {
-        Ok(())
-    }
-}
-
-pub fn validate_status_and_disabled(
-    status: Option<api_enums::ConnectorStatus>,
-    disabled: Option<bool>,
-    auth: types::ConnectorAuthType,
-    current_status: api_enums::ConnectorStatus,
-) -> RouterResult<(api_enums::ConnectorStatus, Option<bool>)> {
-    let connector_status = match (status, auth) {
-        (Some(common_enums::ConnectorStatus::Active), types::ConnectorAuthType::TemporaryAuth) => {
-            return Err(errors::ApiErrorResponse::InvalidRequestData {
-                message: "Connector status cannot be active when using TemporaryAuth".to_string(),
-            }
-            .into());
-        }
-        (Some(status), _) => status,
-        (None, types::ConnectorAuthType::TemporaryAuth) => common_enums::ConnectorStatus::Inactive,
-        (None, _) => current_status,
-    };
-
-    let disabled = match (disabled, connector_status) {
-        (Some(false), common_enums::ConnectorStatus::Inactive) => {
-            return Err(errors::ApiErrorResponse::InvalidRequestData {
-                message: "Connector cannot be enabled when connector_status is inactive or when using TemporaryAuth"
-                    .to_string(),
-            }
-            .into());
-        }
-        (Some(disabled), _) => Some(disabled),
-        (None, common_enums::ConnectorStatus::Inactive) => Some(true),
-        (None, _) => None,
-    };
-
-    Ok((connector_status, disabled))
-}
-
 async fn process_open_banking_connectors(
     state: &SessionState,
     merchant_id: &id_type::MerchantId,
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index 47f3ba0ab4..70a0a4e4cd 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -2656,7 +2656,7 @@ pub async fn list_payment_methods(
                     .pm_auth_config
                     .as_ref()
                     .map(|config| {
-                        serde_json::from_value::<PaymentMethodAuthConfig>(config.clone())
+                        serde_json::from_value::<PaymentMethodAuthConfig>(config.clone().expose())
                             .change_context(errors::StorageError::DeserializationFailed)
                             .attach_printable("Failed to deserialize Payment Method Auth config")
                     })
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index e1d98c5056..003c956109 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -3069,12 +3069,7 @@ pub async fn get_payment_filters(
                 .connector_label
                 .as_ref()
                 .map(|label| {
-                    let info = MerchantConnectorInfo {
-                        connector_label: label.clone(),
-                        merchant_connector_id: merchant_connector_account
-                            .merchant_connector_id
-                            .clone(),
-                    };
+                    let info = merchant_connector_account.to_merchant_connector_info(label);
                     (merchant_connector_account.connector_name.clone(), info)
                 })
         })
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index 5a8e12992c..f2c230f0b8 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -989,13 +989,15 @@ pub async fn get_filters_for_refunds(
     let connector_map = merchant_connector_accounts
         .into_iter()
         .filter_map(|merchant_connector_account| {
-            merchant_connector_account.connector_label.map(|label| {
-                let info = MerchantConnectorInfo {
-                    connector_label: label,
-                    merchant_connector_id: merchant_connector_account.merchant_connector_id,
-                };
-                (merchant_connector_account.connector_name, info)
-            })
+            merchant_connector_account
+                .connector_label
+                .clone()
+                .map(|label| {
+                    let info = merchant_connector_account
+                        .clone()
+                        .to_merchant_connector_info(&label.clone());
+                    (merchant_connector_account.connector_name, info)
+                })
         })
         .fold(
             HashMap::new(),
diff --git a/crates/router/src/routes/admin.rs b/crates/router/src/routes/admin.rs
index 05f65b8e47..5044070037 100644
--- a/crates/router/src/routes/admin.rs
+++ b/crates/router/src/routes/admin.rs
@@ -243,6 +243,10 @@ pub async fn delete_merchant_account(
 /// Merchant Connector - Create
 ///
 /// Create a new Merchant Connector for the merchant account. The connector could be a payment processor / facilitator / acquirer or specialized services like Fraud / Accounting etc."
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "merchant_connector_account_v2")
+))]
 #[utoipa::path(
     post,
     path = "/accounts/{account_id}/connectors",
@@ -263,12 +267,56 @@ pub async fn payment_connector_create(
     json_payload: web::Json<admin::MerchantConnectorCreate>,
 ) -> HttpResponse {
     let flow = Flow::MerchantConnectorsCreate;
+    let payload = json_payload.into_inner();
     let merchant_id = path.into_inner();
     Box::pin(api::server_wrap(
         flow,
         state,
         &req,
-        json_payload.into_inner(),
+        payload,
+        |state, _, req, _| create_payment_connector(state, req, &merchant_id),
+        auth::auth_type(
+            &auth::AdminApiAuth,
+            &auth::JWTAuthMerchantFromRoute {
+                merchant_id: merchant_id.clone(),
+                required_permission: Permission::MerchantConnectorAccountWrite,
+            },
+            req.headers(),
+        ),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
+/// Merchant Connector - Create
+///
+/// Create a new Merchant Connector for the merchant account. The connector could be a payment processor / facilitator / acquirer or specialized services like Fraud / Accounting etc."
+#[cfg(all(feature = "v2", feature = "merchant_connector_account_v2"))]
+#[utoipa::path(
+    post,
+    path = "/accounts/{account_id}/connectors",
+    request_body = MerchantConnectorCreate,
+    responses(
+        (status = 200, description = "Merchant Connector Created", body = MerchantConnectorResponse),
+        (status = 400, description = "Missing Mandatory fields"),
+    ),
+    tag = "Merchant Connector Account",
+    operation_id = "Create a Merchant Connector",
+    security(("admin_api_key" = []))
+)]
+#[instrument(skip_all, fields(flow = ?Flow::MerchantConnectorsCreate))]
+pub async fn payment_connector_create(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    json_payload: web::Json<admin::MerchantConnectorCreate>,
+) -> HttpResponse {
+    let flow = Flow::MerchantConnectorsCreate;
+    let payload = json_payload.into_inner();
+    let merchant_id = payload.merchant_id.clone();
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        payload,
         |state, _, req, _| create_payment_connector(state, req, &merchant_id),
         auth::auth_type(
             &auth::AdminApiAuth,
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 309a94c141..0e1be147fb 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -58,7 +58,7 @@ use crate::analytics::AnalyticsProvider;
 use crate::routes::fraud_check as frm_routes;
 #[cfg(all(feature = "recon", feature = "olap"))]
 use crate::routes::recon as recon_routes;
-#[cfg(feature = "olap")]
+#[cfg(all(feature = "olap", not(feature = "v2")))]
 use crate::routes::verify_connector::payment_connector_verify;
 pub use crate::{
     configs::settings,
@@ -1082,7 +1082,31 @@ impl MerchantAccount {
 
 pub struct MerchantConnectorAccount;
 
-#[cfg(any(feature = "olap", feature = "oltp"))]
+#[cfg(all(
+    any(feature = "olap", feature = "oltp"),
+    feature = "v2",
+    feature = "merchant_connector_account_v2"
+))]
+impl MerchantConnectorAccount {
+    pub fn server(state: AppState) -> Scope {
+        let mut route = web::scope("/connector_accounts").app_data(web::Data::new(state));
+
+        #[cfg(feature = "olap")]
+        {
+            use super::admin::*;
+
+            route =
+                route.service(web::resource("").route(web::post().to(payment_connector_create)));
+        }
+        route
+    }
+}
+
+#[cfg(all(
+    any(feature = "olap", feature = "oltp"),
+    any(feature = "v1", feature = "v2"),
+    not(feature = "merchant_connector_account_v2")
+))]
 impl MerchantConnectorAccount {
     pub fn server(state: AppState) -> Scope {
         let mut route = web::scope("/account").app_data(web::Data::new(state));
diff --git a/crates/router/src/types/domain/merchant_connector_account.rs b/crates/router/src/types/domain/merchant_connector_account.rs
index 170113c7a2..86d367c1f1 100644
--- a/crates/router/src/types/domain/merchant_connector_account.rs
+++ b/crates/router/src/types/domain/merchant_connector_account.rs
@@ -35,7 +35,7 @@ pub struct MerchantConnectorAccount {
     pub connector_webhook_details: Option<pii::SecretSerdeValue>,
     pub profile_id: Option<String>,
     pub applepay_verified_domains: Option<Vec<String>>,
-    pub pm_auth_config: Option<serde_json::Value>,
+    pub pm_auth_config: Option<pii::SecretSerdeValue>,
     pub status: enums::ConnectorStatus,
     pub connector_wallets_details: Option<Encryptable<Secret<serde_json::Value>>>,
     pub additional_merchant_data: Option<Encryptable<Secret<serde_json::Value>>>,
@@ -55,7 +55,7 @@ pub enum MerchantConnectorAccountUpdate {
         frm_configs: Option<Vec<Secret<serde_json::Value>>>,
         connector_webhook_details: Option<pii::SecretSerdeValue>,
         applepay_verified_domains: Option<Vec<String>>,
-        pm_auth_config: Option<serde_json::Value>,
+        pm_auth_config: Option<pii::SecretSerdeValue>,
         connector_label: Option<String>,
         status: Option<enums::ConnectorStatus>,
         connector_wallets_details: Option<Encryptable<Secret<serde_json::Value>>>,
diff --git a/crates/router/src/types/transformers.rs b/crates/router/src/types/transformers.rs
index c81e45d3e6..8a29ac7e79 100644
--- a/crates/router/src/types/transformers.rs
+++ b/crates/router/src/types/transformers.rs
@@ -949,7 +949,51 @@ impl TryFrom<domain::MerchantConnectorAccount> for api_models::admin::MerchantCo
             }
             None => None,
         };
-        Ok(Self {
+        #[cfg(all(feature = "v2", feature = "merchant_connector_account_v2"))]
+        let response = Self {
+            connector_type: item.connector_type,
+            connector_name: item.connector_name,
+            connector_label: item.connector_label,
+            connector_id: item.merchant_connector_id,
+            connector_account_details: item.connector_account_details.into_inner(),
+            disabled: item.disabled,
+            payment_methods_enabled,
+            metadata: item.metadata,
+            frm_configs,
+            connector_webhook_details: item
+                .connector_webhook_details
+                .map(|webhook_details| {
+                    serde_json::Value::parse_value(
+                        webhook_details.expose(),
+                        "MerchantConnectorWebhookDetails",
+                    )
+                    .attach_printable("Unable to deserialize connector_webhook_details")
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                })
+                .transpose()?,
+            profile_id: item.profile_id,
+            applepay_verified_domains: item.applepay_verified_domains,
+            pm_auth_config: item.pm_auth_config,
+            status: item.status,
+            additional_merchant_data: item
+                .additional_merchant_data
+                .map(|data| {
+                    let data = data.into_inner();
+                    serde_json::Value::parse_value::<router_types::AdditionalMerchantData>(
+                        data.expose(),
+                        "AdditionalMerchantData",
+                    )
+                    .attach_printable("Unable to deserialize additional_merchant_data")
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                })
+                .transpose()?
+                .map(api_models::admin::AdditionalMerchantData::foreign_from),
+        };
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "merchant_connector_account_v2")
+        ))]
+        let response = Self {
             connector_type: item.connector_type,
             connector_name: item.connector_name,
             connector_label: item.connector_label,
@@ -991,7 +1035,8 @@ impl TryFrom<domain::MerchantConnectorAccount> for api_models::admin::MerchantCo
                 })
                 .transpose()?
                 .map(api_models::admin::AdditionalMerchantData::foreign_from),
-        })
+        };
+        Ok(response)
     }
 }
 
diff --git a/scripts/ci-checks.sh b/scripts/ci-checks.sh
index 219bde3c68..6bc37451a1 100755
--- a/scripts/ci-checks.sh
+++ b/scripts/ci-checks.sh
@@ -69,7 +69,7 @@ crates_with_v1_feature="$(
     --null-input \
     '$crates_with_features[]
     | select( IN("v1"; .features[]))  # Select crates with `v1` feature
-    | { name, features: (.features - ["v1", "v2", "default", "payment_v2", "merchant_account_v2","customer_v2"]) }  # Remove specific features to generate feature combinations
+    | { name, features: (.features - ["v1", "v2", "default", "payment_v2", "merchant_account_v2","customer_v2", "merchant_connector_account_v2"]) }  # Remove specific features to generate feature combinations
     | { name, features: ( .features | map([., "v1"] | join(",")) ) }  # Add `v1` to remaining features and join them by comma
     | .name as $name | .features[] | { $name, features: . }  # Expand nested features object to have package - features combinations
     | "\(.name) \(.features)" # Print out package name and features separated by space'