Issue:
Bug: Create Subscription with autocollection off



Request to Subscription Povider to create subscription
curl --location 'https://site/api/v2/customers/AzyUM7UusAXA7mMT/subscription_for_items' \
--header 'Content-Type: application/x-www-form-urlencoded' \
--header 'Authorization: Basic =' \
--data-urlencode 'billing_address%5Bline1%5D=PO Box 9999' \
--data-urlencode 'billing_address%5Bcity%5D=Walnut' \
--data-urlencode 'billing_address%5Bzip%5D=91789' \
--data-urlencode 'billing_address%5Bcountry%5D=US' \
--data-urlencode 'subscription_items%5Bitem_price_id%5D%5B0%5D=cbdemo_enterprise-suite-monthly' \
--data-urlencode 'auto_collection=off'

Response from Subscription Provider
```
{
    "subscription": {
        "id": "AzyvCyUut31wPEoY",
        "billing_period": 1,
        "billing_period_unit": "month",
        "auto_collection": "off",
        "customer_id": "AzyUM7UusAXA7mMT",
        "status": "active",
        "current_term_start": 1756123927,
        "current_term_end": 1758802327,
        "next_billing_at": 1758802327,
        "created_at": 1756123927,
        "started_at": 1756123927,
        "activated_at": 1756123927,
        "updated_at": 1756123927,
        "has_scheduled_changes": false,
        "channel": "web",
        "resource_version": 1756123927399,
        "deleted": false,
        "object": "subscription",
        "currency_code": "INR",
        "subscription_items": [
            {
                "item_price_id": "cbdemo_enterprise-suite-monthly",
                "item_type": "plan",
                "quantity": 1,
                "unit_price": 14100,
                "amount": 14100,
                "free_quantity": 0,
                "object": "subscription_item"
            }
        ],
        "due_invoices_count": 1,
        "due_since": 1756123927,
        "total_dues": 14100,
        "mrr": 0,
        "has_scheduled_advance_invoices": false
    },
    "customer": {
        "id": "AzyUM7UusAXA7mMT",
        "first_name": "John",
        "last_name": "Doe",
        "email": "john@test.com",
        "auto_collection": "on",
        "net_term_days": 0,
        "allow_direct_debit": false,
        "created_at": 1756110939,
        "taxability": "taxable",
        "updated_at": 1756110939,
        "locale": "fr-CA",
        "pii_cleared": "active",
        "channel": "web",
        "resource_version": 1756110939077,
        "deleted": false,
        "object": "customer",
        "billing_address": {
            "first_name": "John",
            "last_name": "Doe",
            "line1": "PO Box 9999",
            "city": "Walnut",
            "state_code": "CA",
            "state": "California",
            "country": "US",
            "zip": "91789",
            "validation_status": "not_validated",
            "object": "billing_address"
        },
        "card_status": "no_card",
        "promotional_credits": 0,
        "refundable_credits": 0,
        "excess_payments": 0,
        "unbilled_charges": 0,
        "preferred_currency_code": "INR",
        "mrr": 0
    },
    "invoice": {
        "id": "48",
        "customer_id": "AzyUM7UusAXA7mMT",
        "subscription_id": "AzyvCyUut31wPEoY",
        "recurring": true,
        "status": "payment_due",
        "date": 1756123927,
        "due_date": 1756123927,
        "net_term_days": 0,
        "price_type": "tax_exclusive",
        "exchange_rate": 1.0,
        "total": 14100,
        "amount_due": 14100,
        "amount_adjusted": 0,
        "amount_paid": 0,
        "write_off_amount": 0,
        "credits_applied": 0,
        "updated_at": 1756123927,
        "resource_version": 1756123927364,
        "deleted": false,
        "object": "invoice",
        "first_invoice": true,
        "amount_to_collect": 14100,
        "round_off_amount": 0,
        "new_sales_amount": 14100,
        "has_advance_charges": false,
        "currency_code": "INR",
        "base_currency_code": "INR",
        "generated_at": 1756123927,
        "is_gifted": false,
        "term_finalized": true,
        "channel": "web",
        "tax": 0,
        "line_items": [
            {
                "id": "li_AzyvCyUut31yqEoa",
                "date_from": 1756123927,
                "date_to": 1758802327,
                "unit_amount": 14100,
                "quantity": 1,
                "amount": 14100,
                "pricing_model": "flat_fee",
                "is_taxed": false,
                "tax_amount": 0,
                "object": "line_item",
                "subscription_id": "AzyvCyUut31wPEoY",
                "customer_id": "AzyUM7UusAXA7mMT",
                "description": "Enterprise Suite Monthly",
                "entity_type": "plan_item_price",
                "entity_id": "cbdemo_enterprise-suite-monthly",
                "tax_exempt_reason": "tax_not_configured",
                "discount_amount": 0,
                "item_level_discount_amount": 0
            }
        ],
        "sub_total": 14100,
        "linked_payments": [],
        "applied_credits": [],
        "adjustment_credit_notes": [],
        "issued_credit_notes": [],
        "linked_orders": [],
        "dunning_attempts": [],
        "billing_address": {
            "first_name": "John",
            "last_name": "Doe",
            "line1": "PO Box 9999",
            "city": "Walnut",
            "state_code": "CA",
            "state": "California",
            "country": "US",
            "zip": "91789",
            "validation_status": "not_validated",
            "object": "billing_address"
        },
        "site_details_at_creation": {
            "timezone": "Asia/Calcutta"
        }
    }
}
```

Common Status for Chargebee Recurly and Stripe Billing
```
Common Status	    Chargebee	    Recurly	        Stripe Billing
Pending	            Future	        -	            incomplete, 
Trial	            InTrial	        -	            trialing
Active	            Active	        ACTIVE	        active
Paused	            Paused	        PAUSED	        paused
unpaid  	        -	            -	            unpaid
Onetime             NonRenewing     -               -
Canceled	        Cancelled,      CANCELLED, 	    canceled,
                    Transferred     EXPIRED         	
Failed	            -	            FAILED	        incomplete_expired
```

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
- Added Chargebee subscription creation request/response structures.
- Implemented conversion traits for Chargebee subscription data.
- Enhanced Recurly connector to support subscription creation flows.
- Introduced new router flow types and request/response types for subscriptions.
- Updated default implementations to include subscription-related traits and integrations.
- Added new traits for subscription management in the hyperswitch interfaces.
- Implemented conversion logic for subscription data between old and new router data formats.
- Updated API service to handle subscription connector integration.

Common Status for Chargebee, Recurly and Stripe Billing
```
Common Status	    Chargebee	    Recurly	        Stripe Billing
Pending	            Future	        -	            incomplete, 
Trial	            InTrial	        -	            trialing
Active	            Active	        ACTIVE	        active
Paused	            Paused	        PAUSED	        paused
unpaid  	        -	            -	            unpaid
Onetime             NonRenewing     -               -
Cancelled	        Cancelled,      CANCELLED, 	    canceled,
                    Transferred     EXPIRED         	
Failed	            -	            FAILED	        incomplete_expired
```

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


closes [9056](https://github.com/juspay/hyperswitch/issues/9056)

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
Since the API handler is being implemented in a separate PR, this connector integration code cannot be called directly and was tested only by successful compilation of code
<img width="996" height="805" alt="image" src="https://github.com/user-attachments/assets/2849b14f-c922-4f63-8ba0-68f5b3e21d29" />

<img width="981" height="623" alt="image" src="https://github.com/user-attachments/assets/b5f2524b-8e46-4c9a-be91-3730d9335865" />


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

Since the API handler is being implemented in a separate PR, this connector integration code cannot be called directly and was tested only by successful compilation of code
<img width="996" height="805" alt="image" src="https://github.com/user-attachments/assets/2849b14f-c922-4f63-8ba0-68f5b3e21d29" />

<img width="981" height="623" alt="image" src="https://github.com/user-attachments/assets/b5f2524b-8e46-4c9a-be91-3730d9335865" />

Patch:
diff --git a/crates/hyperswitch_connectors/src/connectors/chargebee.rs b/crates/hyperswitch_connectors/src/connectors/chargebee.rs
index b9006f9e82e..bf6f3fe637d 100644
--- a/crates/hyperswitch_connectors/src/connectors/chargebee.rs
+++ b/crates/hyperswitch_connectors/src/connectors/chargebee.rs
@@ -16,30 +16,38 @@ use error_stack::ResultExt;
 use hyperswitch_domain_models::{revenue_recovery, router_data_v2::RouterDataV2};
 use hyperswitch_domain_models::{
     router_data::{AccessToken, ConnectorAuthType, ErrorResponse, RouterData},
+    router_data_v2::flow_common_types::SubscriptionCreateData,
     router_flow_types::{
         access_token_auth::AccessTokenAuth,
         payments::{Authorize, Capture, PSync, PaymentMethodToken, Session, SetupMandate, Void},
         refunds::{Execute, RSync},
         revenue_recovery::InvoiceRecordBack,
-        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans},
+        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans, SubscriptionCreate},
         CreateConnectorCustomer,
     },
     router_request_types::{
         revenue_recovery::InvoiceRecordBackRequest,
-        subscriptions::{GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest},
+        subscriptions::{
+            GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest,
+            SubscriptionCreateRequest,
+        },
         AccessTokenRequestData, ConnectorCustomerData, PaymentMethodTokenizationData,
         PaymentsAuthorizeData, PaymentsCancelData, PaymentsCaptureData, PaymentsSessionData,
         PaymentsSyncData, RefundsData, SetupMandateRequestData,
     },
     router_response_types::{
         revenue_recovery::InvoiceRecordBackResponse,
-        subscriptions::{GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse},
+        subscriptions::{
+            GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse,
+            SubscriptionCreateResponse,
+        },
         ConnectorInfo, PaymentsResponseData, RefundsResponseData,
     },
     types::{
         ConnectorCustomerRouterData, GetSubscriptionPlanPricesRouterData,
         GetSubscriptionPlansRouterData, InvoiceRecordBackRouterData, PaymentsAuthorizeRouterData,
         PaymentsCaptureRouterData, PaymentsSyncRouterData, RefundSyncRouterData, RefundsRouterData,
+        SubscriptionCreateRouterData,
     },
 };
 use hyperswitch_interfaces::{
@@ -94,10 +102,117 @@ impl api::Refund for Chargebee {}
 impl api::RefundExecute for Chargebee {}
 impl api::RefundSync for Chargebee {}
 impl api::PaymentToken for Chargebee {}
+impl api::subscriptions::Subscriptions for Chargebee {}
 
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 impl api::revenue_recovery::RevenueRecoveryRecordBack for Chargebee {}
 
+impl api::subscriptions::SubscriptionCreate for Chargebee {}
+
+impl ConnectorIntegration<SubscriptionCreate, SubscriptionCreateRequest, SubscriptionCreateResponse>
+    for Chargebee
+{
+    fn get_headers(
+        &self,
+        req: &SubscriptionCreateRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<Vec<(String, masking::Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_url(
+        &self,
+        req: &SubscriptionCreateRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        let metadata: chargebee::ChargebeeMetadata =
+            utils::to_connector_meta_from_secret(req.connector_meta_data.clone())?;
+        let url = self
+            .base_url(connectors)
+            .to_string()
+            .replace("{{merchant_endpoint_prefix}}", metadata.site.peek());
+        let customer_id = &req.request.customer_id.get_string_repr().to_string();
+        Ok(format!(
+            "{url}v2/customers/{customer_id}/subscription_for_items"
+        ))
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_request_body(
+        &self,
+        req: &SubscriptionCreateRouterData,
+        _connectors: &Connectors,
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
+        let connector_router_data = chargebee::ChargebeeRouterData::from((MinorUnit::new(0), req));
+        let connector_req =
+            chargebee::ChargebeeSubscriptionCreateRequest::try_from(&connector_router_data)?;
+        Ok(RequestContent::FormUrlEncoded(Box::new(connector_req)))
+    }
+
+    fn build_request(
+        &self,
+        req: &SubscriptionCreateRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<Option<Request>, errors::ConnectorError> {
+        Ok(Some(
+            RequestBuilder::new()
+                .method(Method::Post)
+                .url(&types::SubscriptionCreateType::get_url(
+                    self, req, connectors,
+                )?)
+                .attach_default_headers()
+                .headers(types::SubscriptionCreateType::get_headers(
+                    self, req, connectors,
+                )?)
+                .set_body(types::SubscriptionCreateType::get_request_body(
+                    self, req, connectors,
+                )?)
+                .build(),
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        data: &SubscriptionCreateRouterData,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: Response,
+    ) -> CustomResult<SubscriptionCreateRouterData, errors::ConnectorError> {
+        let response: chargebee::ChargebeeSubscriptionCreateResponse = res
+            .response
+            .parse_struct("chargebee ChargebeeSubscriptionCreateResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_response_body(&response));
+        router_env::logger::info!(connector_response=?response);
+        RouterData::try_from(ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }
+}
+
+impl
+    ConnectorIntegrationV2<
+        SubscriptionCreate,
+        SubscriptionCreateData,
+        SubscriptionCreateRequest,
+        SubscriptionCreateResponse,
+    > for Chargebee
+{
+    // Not Implemented (R)
+}
+
 impl ConnectorIntegration<PaymentMethodToken, PaymentMethodTokenizationData, PaymentsResponseData>
     for Chargebee
 {
@@ -689,7 +804,6 @@ fn get_chargebee_plans_query_params(
     Ok(param)
 }
 
-impl api::subscriptions::Subscriptions for Chargebee {}
 impl api::subscriptions::GetSubscriptionPlansFlow for Chargebee {}
 
 impl
diff --git a/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs b/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
index 7d76b869912..5f16d054890 100644
--- a/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
+++ b/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
@@ -5,7 +5,7 @@ use common_enums::enums;
 use common_utils::{
     errors::CustomResult,
     ext_traits::ByteSliceExt,
-    id_type::CustomerId,
+    id_type::{CustomerId, SubscriptionId},
     pii::{self, Email},
     types::MinorUnit,
 };
@@ -17,14 +17,20 @@ use hyperswitch_domain_models::{
     router_data::{ConnectorAuthType, RouterData},
     router_flow_types::{
         refunds::{Execute, RSync},
+        subscriptions::SubscriptionCreate,
         CreateConnectorCustomer, InvoiceRecordBack,
     },
     router_request_types::{
-        revenue_recovery::InvoiceRecordBackRequest, ConnectorCustomerData, ResponseId,
+        revenue_recovery::InvoiceRecordBackRequest,
+        subscriptions::{SubscriptionAutoCollection, SubscriptionCreateRequest},
+        ConnectorCustomerData, ResponseId,
     },
     router_response_types::{
         revenue_recovery::InvoiceRecordBackResponse,
-        subscriptions::{self, GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse},
+        subscriptions::{
+            self, GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse,
+            SubscriptionCreateResponse, SubscriptionStatus,
+        },
         ConnectorCustomerResponseData, PaymentsResponseData, RefundsResponseData,
     },
     types::{InvoiceRecordBackRouterData, PaymentsAuthorizeRouterData, RefundsRouterData},
@@ -36,9 +42,158 @@ use time::PrimitiveDateTime;
 
 use crate::{
     types::{RefundsResponseRouterData, ResponseRouterData},
-    utils::{self, PaymentsAuthorizeRequestData},
+    utils::{self, PaymentsAuthorizeRequestData, RouterData as OtherRouterData},
 };
 
+// SubscriptionCreate structures
+#[derive(Debug, Serialize)]
+pub struct ChargebeeSubscriptionCreateRequest {
+    #[serde(rename = "id")]
+    pub subscription_id: SubscriptionId,
+    #[serde(rename = "subscription_items[item_price_id][0]")]
+    pub item_price_id: String,
+    #[serde(rename = "subscription_items[quantity][0]")]
+    pub quantity: Option<u32>,
+    #[serde(rename = "billing_address[line1]")]
+    pub billing_address_line1: Option<Secret<String>>,
+    #[serde(rename = "billing_address[city]")]
+    pub billing_address_city: Option<String>,
+    #[serde(rename = "billing_address[state]")]
+    pub billing_address_state: Option<Secret<String>>,
+    #[serde(rename = "billing_address[zip]")]
+    pub billing_address_zip: Option<Secret<String>>,
+    #[serde(rename = "billing_address[country]")]
+    pub billing_address_country: Option<common_enums::CountryAlpha2>,
+    pub auto_collection: ChargebeeAutoCollection,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+#[serde(rename_all = "snake_case")]
+pub enum ChargebeeAutoCollection {
+    On,
+    Off,
+}
+
+impl From<SubscriptionAutoCollection> for ChargebeeAutoCollection {
+    fn from(auto_collection: SubscriptionAutoCollection) -> Self {
+        match auto_collection {
+            SubscriptionAutoCollection::On => Self::On,
+            SubscriptionAutoCollection::Off => Self::Off,
+        }
+    }
+}
+
+impl TryFrom<&ChargebeeRouterData<&hyperswitch_domain_models::types::SubscriptionCreateRouterData>>
+    for ChargebeeSubscriptionCreateRequest
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: &ChargebeeRouterData<&hyperswitch_domain_models::types::SubscriptionCreateRouterData>,
+    ) -> Result<Self, Self::Error> {
+        let req = &item.router_data.request;
+
+        let first_item =
+            req.subscription_items
+                .first()
+                .ok_or(errors::ConnectorError::MissingRequiredField {
+                    field_name: "subscription_items",
+                })?;
+
+        Ok(Self {
+            subscription_id: req.subscription_id.clone(),
+            item_price_id: first_item.item_price_id.clone(),
+            quantity: first_item.quantity,
+            billing_address_line1: item.router_data.get_optional_billing_line1(),
+            billing_address_city: item.router_data.get_optional_billing_city(),
+            billing_address_state: item.router_data.get_optional_billing_state(),
+            billing_address_zip: item.router_data.get_optional_billing_zip(),
+            billing_address_country: item.router_data.get_optional_billing_country(),
+            auto_collection: req.auto_collection.clone().into(),
+        })
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ChargebeeSubscriptionCreateResponse {
+    pub subscription: ChargebeeSubscriptionDetails,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ChargebeeSubscriptionDetails {
+    pub id: SubscriptionId,
+    pub status: ChargebeeSubscriptionStatus,
+    pub customer_id: CustomerId,
+    pub currency_code: enums::Currency,
+    pub total_dues: Option<MinorUnit>,
+    #[serde(default, with = "common_utils::custom_serde::timestamp::option")]
+    pub next_billing_at: Option<PrimitiveDateTime>,
+    #[serde(default, with = "common_utils::custom_serde::timestamp::option")]
+    pub created_at: Option<PrimitiveDateTime>,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+#[serde(rename_all = "snake_case")]
+pub enum ChargebeeSubscriptionStatus {
+    Future,
+    #[serde(rename = "in_trial")]
+    InTrial,
+    Active,
+    #[serde(rename = "non_renewing")]
+    NonRenewing,
+    Paused,
+    Cancelled,
+    Transferred,
+}
+
+impl From<ChargebeeSubscriptionStatus> for SubscriptionStatus {
+    fn from(status: ChargebeeSubscriptionStatus) -> Self {
+        match status {
+            ChargebeeSubscriptionStatus::Future => Self::Pending,
+            ChargebeeSubscriptionStatus::InTrial => Self::Trial,
+            ChargebeeSubscriptionStatus::Active => Self::Active,
+            ChargebeeSubscriptionStatus::NonRenewing => Self::Onetime,
+            ChargebeeSubscriptionStatus::Paused => Self::Paused,
+            ChargebeeSubscriptionStatus::Cancelled => Self::Cancelled,
+            ChargebeeSubscriptionStatus::Transferred => Self::Cancelled,
+        }
+    }
+}
+
+impl
+    TryFrom<
+        ResponseRouterData<
+            SubscriptionCreate,
+            ChargebeeSubscriptionCreateResponse,
+            SubscriptionCreateRequest,
+            SubscriptionCreateResponse,
+        >,
+    > for hyperswitch_domain_models::types::SubscriptionCreateRouterData
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: ResponseRouterData<
+            SubscriptionCreate,
+            ChargebeeSubscriptionCreateResponse,
+            SubscriptionCreateRequest,
+            SubscriptionCreateResponse,
+        >,
+    ) -> Result<Self, Self::Error> {
+        let subscription = &item.response.subscription;
+        Ok(Self {
+            response: Ok(SubscriptionCreateResponse {
+                subscription_id: subscription.id.clone(),
+                status: subscription.status.clone().into(),
+                customer_id: subscription.customer_id.clone(),
+                currency_code: subscription.currency_code,
+                total_amount: subscription.total_dues.unwrap_or(MinorUnit::new(0)),
+                next_billing_at: subscription.next_billing_at,
+                created_at: subscription.created_at,
+            }),
+            ..item.data
+        })
+    }
+}
+
 //TODO: Fill the struct with respective fields
 pub struct ChargebeeRouterData<T> {
     pub amount: MinorUnit, // The type of amount that a connector accepts, for example, String, i64, f64, etc.
diff --git a/crates/hyperswitch_connectors/src/connectors/recurly.rs b/crates/hyperswitch_connectors/src/connectors/recurly.rs
index abadb7cb199..5bac349e886 100644
--- a/crates/hyperswitch_connectors/src/connectors/recurly.rs
+++ b/crates/hyperswitch_connectors/src/connectors/recurly.rs
@@ -10,17 +10,22 @@ use error_stack::ResultExt;
 use hyperswitch_domain_models::{
     router_data::{ConnectorAuthType, ErrorResponse},
     router_data_v2::{
-        flow_common_types::{GetSubscriptionPlanPricesData, GetSubscriptionPlansData},
+        flow_common_types::{
+            GetSubscriptionPlanPricesData, GetSubscriptionPlansData, SubscriptionCreateData,
+        },
         UasFlowData,
     },
     router_flow_types::{
-        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans},
+        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans, SubscriptionCreate},
         unified_authentication_service::{
             Authenticate, AuthenticationConfirmation, PostAuthenticate, PreAuthenticate,
         },
     },
     router_request_types::{
-        subscriptions::{GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest},
+        subscriptions::{
+            GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest,
+            SubscriptionCreateRequest,
+        },
         unified_authentication_service::{
             UasAuthenticationRequestData, UasAuthenticationResponseData,
             UasConfirmationRequestData, UasPostAuthenticationRequestData,
@@ -28,7 +33,7 @@ use hyperswitch_domain_models::{
         },
     },
     router_response_types::subscriptions::{
-        GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse,
+        GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse, SubscriptionCreateResponse,
     },
 };
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
@@ -168,6 +173,16 @@ impl
     > for Recurly
 {
 }
+impl api::subscriptions_v2::SubscriptionsCreateV2 for Recurly {}
+impl
+    ConnectorIntegrationV2<
+        SubscriptionCreate,
+        SubscriptionCreateData,
+        SubscriptionCreateRequest,
+        SubscriptionCreateResponse,
+    > for Recurly
+{
+}
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 impl api::revenue_recovery_v2::RevenueRecoveryRecordBackV2 for Recurly {}
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
diff --git a/crates/hyperswitch_connectors/src/default_implementations.rs b/crates/hyperswitch_connectors/src/default_implementations.rs
index 96fbfb6b311..f109ea3fb2f 100644
--- a/crates/hyperswitch_connectors/src/default_implementations.rs
+++ b/crates/hyperswitch_connectors/src/default_implementations.rs
@@ -44,10 +44,14 @@ use hyperswitch_domain_models::{
         AccessTokenAuthentication, Authenticate, AuthenticationConfirmation,
         ExternalVaultCreateFlow, ExternalVaultDeleteFlow, ExternalVaultInsertFlow,
         ExternalVaultProxy, ExternalVaultRetrieveFlow, PostAuthenticate, PreAuthenticate,
+        SubscriptionCreate as SubscriptionCreateFlow,
     },
     router_request_types::{
         authentication,
-        subscriptions::{GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest},
+        subscriptions::{
+            GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest,
+            SubscriptionCreateRequest,
+        },
         unified_authentication_service::{
             UasAuthenticationRequestData, UasAuthenticationResponseData,
             UasConfirmationRequestData, UasPostAuthenticationRequestData,
@@ -66,7 +70,10 @@ use hyperswitch_domain_models::{
         VerifyWebhookSourceRequestData,
     },
     router_response_types::{
-        subscriptions::{GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse},
+        subscriptions::{
+            GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse,
+            SubscriptionCreateResponse,
+        },
         AcceptDisputeResponse, AuthenticationResponseData, DefendDisputeResponse,
         DisputeSyncResponse, FetchDisputesResponse, GiftCardBalanceCheckResponseData,
         MandateRevokeResponseData, PaymentsResponseData, RetrieveFileResponse,
@@ -130,7 +137,10 @@ use hyperswitch_interfaces::{
             PaymentsPreAuthenticate, PaymentsPreProcessing, TaxCalculation,
         },
         revenue_recovery::RevenueRecovery,
-        subscriptions::{GetSubscriptionPlanPricesFlow, GetSubscriptionPlansFlow, Subscriptions},
+        subscriptions::{
+            GetSubscriptionPlanPricesFlow, GetSubscriptionPlansFlow, SubscriptionCreate,
+            Subscriptions,
+        },
         vault::{
             ExternalVault, ExternalVaultCreate, ExternalVaultDelete, ExternalVaultInsert,
             ExternalVaultRetrieve,
@@ -6941,6 +6951,7 @@ macro_rules! default_imp_for_subscriptions {
     ($($path:ident::$connector:ident),*) => {
         $(  impl Subscriptions for $path::$connector {}
             impl GetSubscriptionPlansFlow for $path::$connector {}
+            impl SubscriptionCreate for $path::$connector {}
             impl
             ConnectorIntegration<
             GetSubscriptionPlans,
@@ -6956,6 +6967,12 @@ macro_rules! default_imp_for_subscriptions {
             GetSubscriptionPlanPricesResponse
             > for $path::$connector
             {}
+            impl
+            ConnectorIntegration<
+            SubscriptionCreateFlow,
+            SubscriptionCreateRequest,
+            SubscriptionCreateResponse,
+            > for $path::$connector {}
         )*
     };
 }
@@ -9253,3 +9270,16 @@ impl<const T: u8>
     > for connectors::DummyConnector<T>
 {
 }
+
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8> SubscriptionCreate for connectors::DummyConnector<T> {}
+
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8>
+    ConnectorIntegration<
+        SubscriptionCreateFlow,
+        SubscriptionCreateRequest,
+        SubscriptionCreateResponse,
+    > for connectors::DummyConnector<T>
+{
+}
diff --git a/crates/hyperswitch_domain_models/src/router_data_v2/flow_common_types.rs b/crates/hyperswitch_domain_models/src/router_data_v2/flow_common_types.rs
index 759e2f9943b..0d89445b6a3 100644
--- a/crates/hyperswitch_domain_models/src/router_data_v2/flow_common_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_data_v2/flow_common_types.rs
@@ -150,6 +150,9 @@ pub struct FilesFlowData {
 #[derive(Debug, Clone)]
 pub struct InvoiceRecordBackData;
 
+#[derive(Debug, Clone)]
+pub struct SubscriptionCreateData;
+
 #[derive(Debug, Clone)]
 pub struct GetSubscriptionPlansData;
 
diff --git a/crates/hyperswitch_domain_models/src/router_flow_types.rs b/crates/hyperswitch_domain_models/src/router_flow_types.rs
index dc5418ac8cf..18cf7a53fa6 100644
--- a/crates/hyperswitch_domain_models/src/router_flow_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_flow_types.rs
@@ -20,6 +20,7 @@ pub use payments::*;
 pub use payouts::*;
 pub use refunds::*;
 pub use revenue_recovery::*;
+pub use subscriptions::*;
 pub use unified_authentication_service::*;
 pub use vault::*;
 pub use webhooks::*;
diff --git a/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs
index 4f277c07678..28c78e94393 100644
--- a/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs
+++ b/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs
@@ -1,4 +1,6 @@
 #[derive(Debug, Clone)]
+pub struct SubscriptionCreate;
+#[derive(Debug, Clone)]
 pub struct GetSubscriptionPlans;
 
 #[derive(Debug, Clone)]
diff --git a/crates/hyperswitch_domain_models/src/router_request_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_request_types/subscriptions.rs
index bc9049d10f9..832140e1690 100644
--- a/crates/hyperswitch_domain_models/src/router_request_types/subscriptions.rs
+++ b/crates/hyperswitch_domain_models/src/router_request_types/subscriptions.rs
@@ -1,3 +1,28 @@
+use api_models::payments::Address;
+use common_utils::id_type;
+
+use crate::connector_endpoints;
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionItem {
+    pub item_price_id: String,
+    pub quantity: Option<u32>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionCreateRequest {
+    pub customer_id: id_type::CustomerId,
+    pub subscription_id: id_type::SubscriptionId,
+    pub subscription_items: Vec<SubscriptionItem>,
+    pub billing_address: Address,
+    pub auto_collection: SubscriptionAutoCollection,
+    pub connector_params: connector_endpoints::ConnectorParams,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SubscriptionAutoCollection {
+    On,
+    Off,
+}
 #[derive(Debug, Clone)]
 pub struct GetSubscriptionPlansRequest {
     pub limit: Option<u32>,
diff --git a/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
index 0b06a6aa269..c61f9fd7ff4 100644
--- a/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
+++ b/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
@@ -1,4 +1,29 @@
 use common_enums::Currency;
+use common_utils::{id_type, types::MinorUnit};
+use time::PrimitiveDateTime;
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionCreateResponse {
+    pub subscription_id: id_type::SubscriptionId,
+    pub status: SubscriptionStatus,
+    pub customer_id: id_type::CustomerId,
+    pub currency_code: Currency,
+    pub total_amount: MinorUnit,
+    pub next_billing_at: Option<PrimitiveDateTime>,
+    pub created_at: Option<PrimitiveDateTime>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SubscriptionStatus {
+    Pending,
+    Trial,
+    Active,
+    Paused,
+    Unpaid,
+    Onetime,
+    Cancelled,
+    Failed,
+}
 
 #[derive(Debug, Clone)]
 pub struct GetSubscriptionPlansResponse {
@@ -21,7 +46,7 @@ pub struct GetSubscriptionPlanPricesResponse {
 pub struct SubscriptionPlanPrices {
     pub price_id: String,
     pub plan_id: Option<String>,
-    pub amount: common_utils::types::MinorUnit,
+    pub amount: MinorUnit,
     pub currency: Currency,
     pub interval: PeriodUnit,
     pub interval_count: i64,
diff --git a/crates/hyperswitch_domain_models/src/types.rs b/crates/hyperswitch_domain_models/src/types.rs
index cc57f7f82f0..06fbeb267d3 100644
--- a/crates/hyperswitch_domain_models/src/types.rs
+++ b/crates/hyperswitch_domain_models/src/types.rs
@@ -6,7 +6,7 @@ use crate::{
     router_flow_types::{
         mandate_revoke::MandateRevoke,
         revenue_recovery::InvoiceRecordBack,
-        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans},
+        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans, SubscriptionCreate},
         AccessTokenAuth, AccessTokenAuthentication, Authenticate, AuthenticationConfirmation,
         Authorize, AuthorizeSessionToken, BillingConnectorInvoiceSync,
         BillingConnectorPaymentsSync, CalculateTax, Capture, CompleteAuthorize,
@@ -20,7 +20,10 @@ use crate::{
             BillingConnectorInvoiceSyncRequest, BillingConnectorPaymentsSyncRequest,
             InvoiceRecordBackRequest,
         },
-        subscriptions::{GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest},
+        subscriptions::{
+            GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest,
+            SubscriptionCreateRequest,
+        },
         unified_authentication_service::{
             UasAuthenticationRequestData, UasAuthenticationResponseData,
             UasConfirmationRequestData, UasPostAuthenticationRequestData,
@@ -42,7 +45,10 @@ use crate::{
             BillingConnectorInvoiceSyncResponse, BillingConnectorPaymentsSyncResponse,
             InvoiceRecordBackResponse,
         },
-        subscriptions::{GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse},
+        subscriptions::{
+            GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse,
+            SubscriptionCreateResponse,
+        },
         GiftCardBalanceCheckResponseData, MandateRevokeResponseData, PaymentsResponseData,
         RefundsResponseData, TaxCalculationResponseData, VaultResponseData,
         VerifyWebhookSourceResponseData,
@@ -195,3 +201,6 @@ pub type ExternalVaultProxyPaymentsRouterDataV2 = RouterDataV2<
     ExternalVaultProxyPaymentsData,
     PaymentsResponseData,
 >;
+
+pub type SubscriptionCreateRouterData =
+    RouterData<SubscriptionCreate, SubscriptionCreateRequest, SubscriptionCreateResponse>;
diff --git a/crates/hyperswitch_interfaces/src/api/subscriptions.rs b/crates/hyperswitch_interfaces/src/api/subscriptions.rs
index 7e710c63d13..1d15a7f2b64 100644
--- a/crates/hyperswitch_interfaces/src/api/subscriptions.rs
+++ b/crates/hyperswitch_interfaces/src/api/subscriptions.rs
@@ -1,12 +1,13 @@
 //! Subscriptions Interface for V1
 #[cfg(feature = "v1")]
 use hyperswitch_domain_models::{
+    router_flow_types::subscriptions::SubscriptionCreate as SubscriptionCreateFlow,
     router_flow_types::subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans},
     router_request_types::subscriptions::{
-        GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest,
+        GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest, SubscriptionCreateRequest,
     },
     router_response_types::subscriptions::{
-        GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse,
+        GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse, SubscriptionCreateResponse,
     },
 };
 
@@ -37,12 +38,20 @@ pub trait GetSubscriptionPlanPricesFlow:
 {
 }
 
+#[cfg(feature = "v1")]
+/// trait SubscriptionCreate
+pub trait SubscriptionCreate:
+    ConnectorIntegration<SubscriptionCreateFlow, SubscriptionCreateRequest, SubscriptionCreateResponse>
+{
+}
+
 /// trait Subscriptions
 #[cfg(feature = "v1")]
 pub trait Subscriptions:
     ConnectorCommon
     + GetSubscriptionPlansFlow
     + GetSubscriptionPlanPricesFlow
+    + SubscriptionCreate
     + PaymentsConnectorCustomer
 {
 }
@@ -62,3 +71,7 @@ pub trait GetSubscriptionPlanPricesFlow {}
 #[cfg(not(feature = "v1"))]
 /// trait CreateCustomer (disabled when not V1)
 pub trait ConnectorCustomer {}
+
+/// trait SubscriptionCreate
+#[cfg(not(feature = "v1"))]
+pub trait SubscriptionCreate {}
diff --git a/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs b/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs
index 609ab2dbc3f..f14d8439e2c 100644
--- a/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs
+++ b/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs
@@ -1,12 +1,16 @@
 //! SubscriptionsV2
 use hyperswitch_domain_models::{
-    router_data_v2::flow_common_types::{GetSubscriptionPlanPricesData, GetSubscriptionPlansData},
-    router_flow_types::subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans},
+    router_data_v2::flow_common_types::{
+        GetSubscriptionPlanPricesData, GetSubscriptionPlansData, SubscriptionCreateData,
+    },
+    router_flow_types::subscriptions::{
+        GetSubscriptionPlanPrices, GetSubscriptionPlans, SubscriptionCreate,
+    },
     router_request_types::subscriptions::{
-        GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest,
+        GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest, SubscriptionCreateRequest,
     },
     router_response_types::subscriptions::{
-        GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse,
+        GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse, SubscriptionCreateResponse,
     },
 };
 
@@ -15,7 +19,7 @@ use crate::connector_integration_v2::ConnectorIntegrationV2;
 
 /// trait SubscriptionsV2
 pub trait SubscriptionsV2:
-    GetSubscriptionPlansV2 + ConnectorCustomerV2 + GetSubscriptionPlanPricesV2
+    GetSubscriptionPlansV2 + SubscriptionsCreateV2 + ConnectorCustomerV2 + GetSubscriptionPlanPricesV2
 {
 }
 
@@ -40,3 +44,14 @@ pub trait GetSubscriptionPlanPricesV2:
 >
 {
 }
+
+/// trait SubscriptionsCreateV2
+pub trait SubscriptionsCreateV2:
+    ConnectorIntegrationV2<
+    SubscriptionCreate,
+    SubscriptionCreateData,
+    SubscriptionCreateRequest,
+    SubscriptionCreateResponse,
+>
+{
+}
diff --git a/crates/hyperswitch_interfaces/src/conversion_impls.rs b/crates/hyperswitch_interfaces/src/conversion_impls.rs
index d119cf332e1..a9d4a1bcd62 100644
--- a/crates/hyperswitch_interfaces/src/conversion_impls.rs
+++ b/crates/hyperswitch_interfaces/src/conversion_impls.rs
@@ -13,8 +13,8 @@ use hyperswitch_domain_models::{
             BillingConnectorPaymentsSyncFlowData, DisputesFlowData, ExternalAuthenticationFlowData,
             ExternalVaultProxyFlowData, FilesFlowData, GetSubscriptionPlanPricesData,
             GetSubscriptionPlansData, GiftCardBalanceCheckFlowData, InvoiceRecordBackData,
-            MandateRevokeFlowData, PaymentFlowData, RefundFlowData, UasFlowData,
-            VaultConnectorFlowData, WebhookSourceVerifyData,
+            MandateRevokeFlowData, PaymentFlowData, RefundFlowData, SubscriptionCreateData,
+            UasFlowData, VaultConnectorFlowData, WebhookSourceVerifyData,
         },
         RouterDataV2,
     },
@@ -878,6 +878,7 @@ macro_rules! default_router_data_conversion {
 }
 default_router_data_conversion!(GetSubscriptionPlansData);
 default_router_data_conversion!(GetSubscriptionPlanPricesData);
+default_router_data_conversion!(SubscriptionCreateData);
 
 impl<T, Req: Clone, Resp: Clone> RouterDataConversion<T, Req, Resp> for UasFlowData {
     fn from_old_router_data(
diff --git a/crates/hyperswitch_interfaces/src/types.rs b/crates/hyperswitch_interfaces/src/types.rs
index dbb9ddf8869..d9975a8c058 100644
--- a/crates/hyperswitch_interfaces/src/types.rs
+++ b/crates/hyperswitch_interfaces/src/types.rs
@@ -16,7 +16,7 @@ use hyperswitch_domain_models::{
         },
         refunds::{Execute, RSync},
         revenue_recovery::{BillingConnectorPaymentsSync, InvoiceRecordBack},
-        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans},
+        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans, SubscriptionCreate},
         unified_authentication_service::{
             Authenticate, AuthenticationConfirmation, PostAuthenticate, PreAuthenticate,
         },
@@ -32,7 +32,10 @@ use hyperswitch_domain_models::{
             BillingConnectorInvoiceSyncRequest, BillingConnectorPaymentsSyncRequest,
             InvoiceRecordBackRequest,
         },
-        subscriptions::{GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest},
+        subscriptions::{
+            GetSubscriptionPlanPricesRequest, GetSubscriptionPlansRequest,
+            SubscriptionCreateRequest,
+        },
         unified_authentication_service::{
             UasAuthenticationRequestData, UasAuthenticationResponseData,
             UasConfirmationRequestData, UasPostAuthenticationRequestData,
@@ -57,7 +60,10 @@ use hyperswitch_domain_models::{
             BillingConnectorInvoiceSyncResponse, BillingConnectorPaymentsSyncResponse,
             InvoiceRecordBackResponse,
         },
-        subscriptions::{GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse},
+        subscriptions::{
+            GetSubscriptionPlanPricesResponse, GetSubscriptionPlansResponse,
+            SubscriptionCreateResponse,
+        },
         AcceptDisputeResponse, DefendDisputeResponse, DisputeSyncResponse, FetchDisputesResponse,
         GiftCardBalanceCheckResponseData, MandateRevokeResponseData, PaymentsResponseData,
         RefundsResponseData, RetrieveFileResponse, SubmitEvidenceResponse,
@@ -303,6 +309,13 @@ pub type InvoiceRecordBackType = dyn ConnectorIntegration<
     InvoiceRecordBackResponse,
 >;
 
+/// Type alias for `ConnectorIntegration<SubscriptionCreate, SubscriptionCreateRequest, SubscriptionCreateResponse>`
+pub type SubscriptionCreateType = dyn ConnectorIntegration<
+    SubscriptionCreate,
+    SubscriptionCreateRequest,
+    SubscriptionCreateResponse,
+>;
+
 /// Type alias for `ConnectorIntegration<BillingConnectorPaymentsSync, BillingConnectorPaymentsSyncRequest, BillingConnectorPaymentsSyncResponse>`
 pub type BillingConnectorPaymentsSyncType = dyn ConnectorIntegration<
     BillingConnectorPaymentsSync,
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 4e8970d9309..0fd012e9f7a 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -211,6 +211,9 @@ where
     Ok(())
 }
 
+pub type BoxedSubscriptionConnectorIntegrationInterface<T, Req, Res> =
+    BoxedConnectorIntegrationInterface<T, common_types::SubscriptionCreateData, Req, Res>;
+
 /// Handle the flow by interacting with connector module
 /// `connector_request` is applicable only in case if the `CallConnectorAction` is `Trigger`
 /// In other cases, It will be created if required, even if it is not passed