diff --git a/backend/adapter_processor_v2/__init__.py b/backend/adapter_processor_v2/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/backend/adapter_processor_v2/adapter_processor.py b/backend/adapter_processor_v2/adapter_processor.py
new file mode 100644
index 0000000000..dd9d44ff3c
--- /dev/null
+++ b/backend/adapter_processor_v2/adapter_processor.py
@@ -0,0 +1,257 @@
+import json
+import logging
+from typing import Any, Optional
+
+from account_v2.models import User
+from adapter_processor_v2.constants import AdapterKeys
+from adapter_processor_v2.exceptions import (
+    InternalServiceError,
+    InValidAdapterId,
+    TestAdapterError,
+)
+from django.conf import settings
+from django.core.exceptions import ObjectDoesNotExist
+from platform_settings_v2.platform_auth_service import PlatformAuthenticationService
+from unstract.adapters.adapterkit import Adapterkit
+from unstract.adapters.base import Adapter
+from unstract.adapters.enums import AdapterTypes
+from unstract.adapters.exceptions import AdapterError
+from unstract.adapters.x2text.constants import X2TextConstants
+
+from .models import AdapterInstance, UserDefaultAdapter
+
+logger = logging.getLogger(__name__)
+
+
+class AdapterProcessor:
+    @staticmethod
+    def get_json_schema(adapter_id: str) -> dict[str, Any]:
+        """Function to return JSON Schema for Adapters."""
+        schema_details: dict[str, Any] = {}
+        updated_adapters = AdapterProcessor.__fetch_adapters_by_key_value(
+            AdapterKeys.ID, adapter_id
+        )
+        if len(updated_adapters) != 0:
+            try:
+                schema_details[AdapterKeys.JSON_SCHEMA] = json.loads(
+                    updated_adapters[0].get(AdapterKeys.JSON_SCHEMA)
+                )
+            except Exception as exc:
+                logger.error(f"Error occured while parsing JSON Schema : {exc}")
+                raise InternalServiceError()
+        else:
+            logger.error(
+                f"Invalid adapter Id : {adapter_id} while fetching JSON Schema"
+            )
+            raise InValidAdapterId()
+        return schema_details
+
+    @staticmethod
+    def get_all_supported_adapters(type: str) -> list[dict[Any, Any]]:
+        """Function to return list of all supported adapters."""
+        supported_adapters = []
+        updated_adapters = []
+        updated_adapters = AdapterProcessor.__fetch_adapters_by_key_value(
+            AdapterKeys.ADAPTER_TYPE, type
+        )
+        for each_adapter in updated_adapters:
+            supported_adapters.append(
+                {
+                    AdapterKeys.ID: each_adapter.get(AdapterKeys.ID),
+                    AdapterKeys.NAME: each_adapter.get(AdapterKeys.NAME),
+                    AdapterKeys.DESCRIPTION: each_adapter.get(AdapterKeys.DESCRIPTION),
+                    AdapterKeys.ICON: each_adapter.get(AdapterKeys.ICON),
+                    AdapterKeys.ADAPTER_TYPE: each_adapter.get(
+                        AdapterKeys.ADAPTER_TYPE
+                    ),
+                }
+            )
+        return supported_adapters
+
+    @staticmethod
+    def get_adapter_data_with_key(adapter_id: str, key_value: str) -> Any:
+        """Generic Function to get adapter data with provided key."""
+        updated_adapters = AdapterProcessor.__fetch_adapters_by_key_value(
+            "id", adapter_id
+        )
+        if len(updated_adapters) == 0:
+            logger.error(f"Invalid adapter ID {adapter_id} while invoking utility")
+            raise InValidAdapterId()
+        return updated_adapters[0].get(key_value)
+
+    @staticmethod
+    def test_adapter(adapter_id: str, adapter_metadata: dict[str, Any]) -> bool:
+        logger.info(f"Testing adapter: {adapter_id}")
+        try:
+            adapter_class = Adapterkit().get_adapter_class_by_adapter_id(adapter_id)
+
+            if adapter_metadata.pop(AdapterKeys.ADAPTER_TYPE) == AdapterKeys.X2TEXT:
+                adapter_metadata[X2TextConstants.X2TEXT_HOST] = settings.X2TEXT_HOST
+                adapter_metadata[X2TextConstants.X2TEXT_PORT] = settings.X2TEXT_PORT
+                platform_key = PlatformAuthenticationService.get_active_platform_key()
+                adapter_metadata[X2TextConstants.PLATFORM_SERVICE_API_KEY] = str(
+                    platform_key.key
+                )
+
+            adapter_instance = adapter_class(adapter_metadata)
+            test_result: bool = adapter_instance.test_connection()
+            logger.info(f"{adapter_id} test result: {test_result}")
+            return test_result
+        except AdapterError as e:
+            raise TestAdapterError(str(e))
+
+    @staticmethod
+    def __fetch_adapters_by_key_value(key: str, value: Any) -> Adapter:
+        """Fetches a list of adapters that have an attribute matching key and
+        value."""
+        logger.info(f"Fetching adapter list for {key} with {value}")
+        adapter_kit = Adapterkit()
+        adapters = adapter_kit.get_adapters_list()
+        return [iterate for iterate in adapters if iterate[key] == value]
+
+    @staticmethod
+    def set_default_triad(default_triad: dict[str, str], user: User) -> None:
+        try:
+            (
+                user_default_adapter,
+                created,
+            ) = UserDefaultAdapter.objects.get_or_create(user=user)
+
+            if default_triad.get(AdapterKeys.LLM_DEFAULT, None):
+                user_default_adapter.default_llm_adapter = AdapterInstance.objects.get(
+                    pk=default_triad[AdapterKeys.LLM_DEFAULT]
+                )
+            if default_triad.get(AdapterKeys.EMBEDDING_DEFAULT, None):
+                user_default_adapter.default_embedding_adapter = (
+                    AdapterInstance.objects.get(
+                        pk=default_triad[AdapterKeys.EMBEDDING_DEFAULT]
+                    )
+                )
+
+            if default_triad.get(AdapterKeys.VECTOR_DB_DEFAULT, None):
+                user_default_adapter.default_vector_db_adapter = (
+                    AdapterInstance.objects.get(
+                        pk=default_triad[AdapterKeys.VECTOR_DB_DEFAULT]
+                    )
+                )
+
+            if default_triad.get(AdapterKeys.X2TEXT_DEFAULT, None):
+                user_default_adapter.default_x2text_adapter = (
+                    AdapterInstance.objects.get(
+                        pk=default_triad[AdapterKeys.X2TEXT_DEFAULT]
+                    )
+                )
+
+            user_default_adapter.save()
+
+            logger.info("Changed defaults successfully")
+        except Exception as e:
+            logger.error(f"Unable to save defaults because: {e}")
+            if isinstance(e, InValidAdapterId):
+                raise e
+            else:
+                raise InternalServiceError()
+
+    @staticmethod
+    def get_adapter_instance_by_id(adapter_instance_id: str) -> Adapter:
+        """Get the adapter instance by its ID.
+
+        Parameters:
+        - adapter_instance_id (str): The ID of the adapter instance.
+
+        Returns:
+        - Adapter: The adapter instance with the specified ID.
+
+        Raises:
+        - Exception: If there is an error while fetching the adapter instance.
+        """
+        try:
+            adapter = AdapterInstance.objects.get(id=adapter_instance_id)
+        except Exception as e:
+            logger.error(f"Unable to fetch adapter: {e}")
+        if not adapter:
+            logger.error("Unable to fetch adapter")
+        return adapter.adapter_name
+
+    @staticmethod
+    def get_adapters_by_type(
+        adapter_type: AdapterTypes, user: User
+    ) -> list[AdapterInstance]:
+        """Get a list of adapters by their type.
+
+        Parameters:
+        - adapter_type (AdapterTypes): The type of adapters to retrieve.
+        - user: Logged in User
+
+        Returns:
+        - list[AdapterInstance]: A list of AdapterInstance objects that match
+            the specified adapter type.
+        """
+
+        adapters: list[AdapterInstance] = AdapterInstance.objects.for_user(user).filter(
+            adapter_type=adapter_type.value,
+        )
+        return adapters
+
+    @staticmethod
+    def get_adapter_by_name_and_type(
+        adapter_type: AdapterTypes,
+        adapter_name: Optional[str] = None,
+    ) -> Optional[AdapterInstance]:
+        """Get the adapter instance by its name and type.
+
+        Parameters:
+        - adapter_name (str): The name of the adapter instance.
+        - adapter_type (AdapterTypes): The type of the adapter instance.
+
+        Returns:
+        - AdapterInstance: The adapter with the specified name and type.
+        """
+        if adapter_name:
+            adapter: AdapterInstance = AdapterInstance.objects.get(
+                adapter_name=adapter_name, adapter_type=adapter_type.value
+            )
+        else:
+            try:
+                adapter = AdapterInstance.objects.get(
+                    adapter_type=adapter_type.value, is_default=True
+                )
+            except AdapterInstance.DoesNotExist:
+                return None
+        return adapter
+
+    @staticmethod
+    def get_default_adapters(user: User) -> list[AdapterInstance]:
+        """Retrieve a list of default adapter instances. This method queries
+        the database to fetch all adapter instances marked as default.
+
+        Raises:
+            InternalServiceError: If an unexpected error occurs during
+            the database query.
+
+        Returns:
+            list[AdapterInstance]: A list of AdapterInstance objects that are
+            marked as default.
+        """
+        try:
+            adapters: list[AdapterInstance] = []
+            default_adapter = UserDefaultAdapter.objects.get(user=user)
+
+            if default_adapter.default_embedding_adapter:
+                adapters.append(default_adapter.default_embedding_adapter)
+            if default_adapter.default_llm_adapter:
+                adapters.append(default_adapter.default_llm_adapter)
+            if default_adapter.default_vector_db_adapter:
+                adapters.append(default_adapter.default_vector_db_adapter)
+            if default_adapter.default_x2text_adapter:
+                adapters.append(default_adapter.default_x2text_adapter)
+
+            return adapters
+        except ObjectDoesNotExist as e:
+            logger.error(f"No default adapters found: {e}")
+            raise InternalServiceError(
+                "No default adapters found, " "configure them through Platform Settings"
+            )
+        except Exception as e:
+            logger.error(f"Error occurred while fetching default adapters: {e}")
+            raise InternalServiceError("Error fetching default adapters")
diff --git a/backend/adapter_processor_v2/constants.py b/backend/adapter_processor_v2/constants.py
new file mode 100644
index 0000000000..e35f2d0acc
--- /dev/null
+++ b/backend/adapter_processor_v2/constants.py
@@ -0,0 +1,28 @@
+class AdapterKeys:
+    JSON_SCHEMA = "json_schema"
+    ADAPTER_TYPE = "adapter_type"
+    IS_DEFAULT = "is_default"
+    LLM = "LLM"
+    X2TEXT = "X2TEXT"
+    OCR = "OCR"
+    VECTOR_DB = "VECTOR_DB"
+    EMBEDDING = "EMBEDDING"
+    NAME = "name"
+    DESCRIPTION = "description"
+    ICON = "icon"
+    ADAPTER_ID = "adapter_id"
+    ADAPTER_METADATA = "adapter_metadata"
+    ADAPTER_METADATA_B = "adapter_metadata_b"
+    ID = "id"
+    IS_VALID = "is_valid"
+    LLM_DEFAULT = "llm_default"
+    VECTOR_DB_DEFAULT = "vector_db_default"
+    EMBEDDING_DEFAULT = "embedding_default"
+    X2TEXT_DEFAULT = "x2text_default"
+    SHARED_USERS = "shared_users"
+    ADAPTER_NAME_EXISTS = (
+        "Configuration with this Name already exists. "
+        "Please try with a different Name"
+    )
+    ADAPTER_CREATED_BY = "created_by_email"
+    ADAPTER_CONTEXT_WINDOW_SIZE = "context_window_size"
diff --git a/backend/adapter_processor_v2/exceptions.py b/backend/adapter_processor_v2/exceptions.py
new file mode 100644
index 0000000000..5bddbb3c21
--- /dev/null
+++ b/backend/adapter_processor_v2/exceptions.py
@@ -0,0 +1,68 @@
+from typing import Optional
+
+from rest_framework.exceptions import APIException
+
+from backend.exceptions import UnstractBaseException
+
+
+class IdIsMandatory(APIException):
+    status_code = 400
+    default_detail = "ID is Mandatory."
+
+
+class InValidType(APIException):
+    status_code = 400
+    default_detail = "Type is not Valid."
+
+
+class InValidAdapterId(APIException):
+    status_code = 400
+    default_detail = "Adapter ID is not Valid."
+
+
+class InvalidEncryptionKey(APIException):
+    status_code = 403
+    default_detail = (
+        "Platform encryption key for storing adapter credentials has changed! "
+        "Please inform the organization admin to contact support."
+    )
+
+
+class InternalServiceError(APIException):
+    status_code = 500
+    default_detail = "Internal Service error"
+
+
+class CannotDeleteDefaultAdapter(APIException):
+    status_code = 500
+    default_detail = (
+        "This is configured as default and cannot be deleted. "
+        "Please configure a different default before you try again!"
+    )
+
+
+class UniqueConstraintViolation(APIException):
+    status_code = 400
+    default_detail = "Unique constraint violated"
+
+
+class TestAdapterError(UnstractBaseException):
+    status_code = 500
+    default_detail = "Error while testing adapter"
+
+
+class DeleteAdapterInUseError(APIException):
+    status_code = 409
+
+    def __init__(
+        self,
+        detail: Optional[str] = None,
+        code: Optional[str] = None,
+        adapter_name: str = "adapter",
+    ):
+        if detail is None:
+            detail = (
+                f"Cannot delete {adapter_name}. "
+                "It is used in a workflow or a prompt studio project"
+            )
+        super().__init__(detail, code)
diff --git a/backend/adapter_processor_v2/models.py b/backend/adapter_processor_v2/models.py
new file mode 100644
index 0000000000..5e63056c55
--- /dev/null
+++ b/backend/adapter_processor_v2/models.py
@@ -0,0 +1,208 @@
+import json
+import logging
+import uuid
+from typing import Any
+
+from account_v2.models import User
+from cryptography.fernet import Fernet
+from django.conf import settings
+from django.db import models
+from django.db.models import QuerySet
+from tenant_account_v2.models import OrganizationMember
+from unstract.adapters.adapterkit import Adapterkit
+from unstract.adapters.enums import AdapterTypes
+from unstract.adapters.exceptions import AdapterError
+from utils.models.base_model import BaseModel
+from utils.models.organization_mixin import (
+    DefaultOrganizationManagerMixin,
+    DefaultOrganizationMixin,
+)
+
+logger = logging.getLogger(__name__)
+
+ADAPTER_NAME_SIZE = 128
+VERSION_NAME_SIZE = 64
+ADAPTER_ID_LENGTH = 128
+
+logger = logging.getLogger(__name__)
+
+
+class AdapterInstanceModelManager(DefaultOrganizationManagerMixin, models.Manager):
+    def get_queryset(self) -> QuerySet[Any]:
+        return super().get_queryset()
+
+    def for_user(self, user: User) -> QuerySet[Any]:
+        return (
+            self.get_queryset()
+            .filter(
+                models.Q(created_by=user)
+                | models.Q(shared_users=user)
+                | models.Q(shared_to_org=True)
+                | models.Q(is_friction_less=True)
+            )
+            .distinct("id")
+        )
+
+
+class AdapterInstance(DefaultOrganizationMixin, BaseModel):
+    id = models.UUIDField(
+        primary_key=True,
+        default=uuid.uuid4,
+        editable=False,
+        db_comment="Unique identifier for the Adapter Instance",
+    )
+    adapter_name = models.TextField(
+        max_length=ADAPTER_NAME_SIZE,
+        null=False,
+        blank=False,
+        db_comment="Name of the Adapter Instance",
+    )
+    adapter_id = models.CharField(
+        max_length=ADAPTER_ID_LENGTH,
+        default="",
+        db_comment="Unique identifier of the Adapter",
+    )
+
+    # TODO to be removed once the migration for encryption
+    adapter_metadata = models.JSONField(
+        db_column="adapter_metadata",
+        null=False,
+        blank=False,
+        default=dict,
+        db_comment="JSON adapter metadata submitted by the user",
+    )
+    adapter_metadata_b = models.BinaryField(null=True)
+    adapter_type = models.CharField(
+        choices=[(tag.value, tag.name) for tag in AdapterTypes],
+        db_comment="Type of adapter LLM/EMBEDDING/VECTOR_DB",
+    )
+    created_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        related_name="adapters_created",
+        null=True,
+        blank=True,
+    )
+    modified_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        related_name="adapters_modified",
+        null=True,
+        blank=True,
+    )
+
+    is_active = models.BooleanField(
+        default=False,
+        db_comment="Is the adapter instance currently being used",
+    )
+    shared_to_org = models.BooleanField(
+        default=False,
+        db_comment="Is the adapter shared to entire org",
+    )
+
+    is_friction_less = models.BooleanField(
+        default=False,
+        db_comment="Was the adapter created through frictionless onboarding",
+    )
+
+    # Can be used if the adapter usage gets exhausted
+    # Can also be used in other possible scenarios in feature
+    is_usable = models.BooleanField(
+        default=True,
+        db_comment="Is the Adpater Usable",
+    )
+
+    # Introduced field to establish M2M relation between users and adapters.
+    # This will introduce intermediary table which relates both the models.
+    shared_users = models.ManyToManyField(User, related_name="shared_adapters_instance")
+    description = models.TextField(blank=True, null=True, default=None)
+
+    objects = AdapterInstanceModelManager()
+
+    class Meta:
+        verbose_name = "adapter instance"
+        verbose_name_plural = "adapter instances"
+        db_table = "adapter_instance_v2"
+        constraints = [
+            models.UniqueConstraint(
+                fields=["adapter_name", "adapter_type", "organization"],
+                name="unique_organization_adapter",
+            ),
+        ]
+
+    def create_adapter(self) -> None:
+
+        encryption_secret: str = settings.ENCRYPTION_KEY
+        f: Fernet = Fernet(encryption_secret.encode("utf-8"))
+
+        self.adapter_metadata_b = f.encrypt(
+            json.dumps(self.adapter_metadata).encode("utf-8")
+        )
+        self.adapter_metadata = {}
+
+        self.save()
+
+    def get_adapter_meta_data(self) -> Any:
+        encryption_secret: str = settings.ENCRYPTION_KEY
+        f: Fernet = Fernet(encryption_secret.encode("utf-8"))
+
+        adapter_metadata = json.loads(
+            f.decrypt(bytes(self.adapter_metadata_b).decode("utf-8"))
+        )
+        return adapter_metadata
+
+    def get_context_window_size(self) -> int:
+
+        adapter_metadata = self.get_adapter_meta_data()
+        # Get the adapter_instance
+        adapter_class = Adapterkit().get_adapter_class_by_adapter_id(self.adapter_id)
+        try:
+            adapter_instance = adapter_class(adapter_metadata)
+            return adapter_instance.get_context_window_size()
+        except AdapterError as e:
+            logger.warning(f"Unable to retrieve context window size - {e}")
+        return 0
+
+
+class UserDefaultAdapter(BaseModel):
+    user = models.OneToOneField(
+        User, on_delete=models.CASCADE, related_name="organization_default_adapters"
+    )
+    organization_member = models.OneToOneField(
+        OrganizationMember,
+        on_delete=models.CASCADE,
+        default=None,
+        null=True,
+        db_comment="Foreign key reference to the OrganizationMember model.",
+        related_name="default_adapters",
+    )
+    default_llm_adapter = models.ForeignKey(
+        AdapterInstance,
+        on_delete=models.SET_NULL,
+        null=True,
+        related_name="user_default_llm_adapter",
+    )
+    default_embedding_adapter = models.ForeignKey(
+        AdapterInstance,
+        on_delete=models.SET_NULL,
+        null=True,
+        related_name="user_default_embedding_adapter",
+    )
+    default_vector_db_adapter = models.ForeignKey(
+        AdapterInstance,
+        on_delete=models.SET_NULL,
+        null=True,
+        related_name="user_default_vector_db_adapter",
+    )
+
+    default_x2text_adapter = models.ForeignKey(
+        AdapterInstance,
+        on_delete=models.SET_NULL,
+        null=True,
+        related_name="user_default_x2text_adapter",
+    )
+
+    class Meta:
+        verbose_name = "Default Adapter for Organization User"
+        verbose_name_plural = "Default Adapters for Organization Users"
+        db_table = "default_organization_user_adapter_v2"
diff --git a/backend/adapter_processor_v2/serializers.py b/backend/adapter_processor_v2/serializers.py
new file mode 100644
index 0000000000..927fe7e912
--- /dev/null
+++ b/backend/adapter_processor_v2/serializers.py
@@ -0,0 +1,164 @@
+import json
+from typing import Any
+
+from account_v2.serializer import UserSerializer
+from adapter_processor_v2.adapter_processor import AdapterProcessor
+from adapter_processor_v2.constants import AdapterKeys
+from adapter_processor_v2.exceptions import InvalidEncryptionKey
+from cryptography.fernet import Fernet, InvalidToken
+from django.conf import settings
+from rest_framework import serializers
+from rest_framework.serializers import ModelSerializer
+from unstract.adapters.constants import Common as common
+from unstract.adapters.enums import AdapterTypes
+
+from backend.constants import FieldLengthConstants as FLC
+from backend.serializers import AuditSerializer
+
+from .models import AdapterInstance, UserDefaultAdapter
+
+
+class TestAdapterSerializer(serializers.Serializer):
+    adapter_id = serializers.CharField(max_length=FLC.ADAPTER_ID_LENGTH)
+    adapter_metadata = serializers.JSONField()
+    adapter_type = serializers.JSONField()
+
+
+class BaseAdapterSerializer(AuditSerializer):
+    class Meta:
+        model = AdapterInstance
+        fields = "__all__"
+
+
+class DefaultAdapterSerializer(serializers.Serializer):
+    llm_default = serializers.CharField(max_length=FLC.UUID_LENGTH, required=False)
+    embedding_default = serializers.CharField(
+        max_length=FLC.UUID_LENGTH, required=False
+    )
+    vector_db_default = serializers.CharField(
+        max_length=FLC.UUID_LENGTH, required=False
+    )
+
+
+class AdapterInstanceSerializer(BaseAdapterSerializer):
+    """Inherits BaseAdapterSerializer.
+
+    Used for CRUD other than listing
+    """
+
+    def to_internal_value(self, data: dict[str, Any]) -> dict[str, Any]:
+        if data.get(AdapterKeys.ADAPTER_METADATA, None):
+            encryption_secret: str = settings.ENCRYPTION_KEY
+            f: Fernet = Fernet(encryption_secret.encode("utf-8"))
+            json_string: str = json.dumps(data.pop(AdapterKeys.ADAPTER_METADATA))
+
+            data[AdapterKeys.ADAPTER_METADATA_B] = f.encrypt(
+                json_string.encode("utf-8")
+            )
+
+        return data
+
+    def to_representation(self, instance: AdapterInstance) -> dict[str, str]:
+        rep: dict[str, str] = super().to_representation(instance)
+
+        rep.pop(AdapterKeys.ADAPTER_METADATA_B)
+
+        try:
+            adapter_metadata = instance.get_adapter_meta_data()
+        except InvalidToken:
+            raise InvalidEncryptionKey
+        rep[AdapterKeys.ADAPTER_METADATA] = adapter_metadata
+        # Retrieve context window if adapter is a LLM
+        # For other adapter types, context_window is not relevant.
+        if instance.adapter_type == AdapterTypes.LLM.value:
+            adapter_metadata[AdapterKeys.ADAPTER_CONTEXT_WINDOW_SIZE] = (
+                instance.get_context_window_size()
+            )
+
+        rep[common.ICON] = AdapterProcessor.get_adapter_data_with_key(
+            instance.adapter_id, common.ICON
+        )
+        rep[AdapterKeys.ADAPTER_CREATED_BY] = instance.created_by.email
+
+        return rep
+
+
+class AdapterInfoSerializer(BaseAdapterSerializer):
+
+    context_window_size = serializers.SerializerMethodField()
+
+    class Meta(BaseAdapterSerializer.Meta):
+        model = AdapterInstance
+        fields = (
+            "id",
+            "adapter_id",
+            "adapter_name",
+            "adapter_type",
+            "created_by",
+            "context_window_size",
+        )  # type: ignore
+
+    def get_context_window_size(self, obj: AdapterInstance) -> int:
+        return obj.get_context_window_size()
+
+
+class AdapterListSerializer(BaseAdapterSerializer):
+    """Inherits BaseAdapterSerializer.
+
+    Used for listing adapters
+    """
+
+    class Meta(BaseAdapterSerializer.Meta):
+        model = AdapterInstance
+        fields = (
+            "id",
+            "adapter_id",
+            "adapter_name",
+            "adapter_type",
+            "created_by",
+            "description",
+        )  # type: ignore
+
+    def to_representation(self, instance: AdapterInstance) -> dict[str, str]:
+        rep: dict[str, str] = super().to_representation(instance)
+        rep[common.ICON] = AdapterProcessor.get_adapter_data_with_key(
+            instance.adapter_id, common.ICON
+        )
+        adapter_metadata = instance.get_adapter_meta_data()
+        model = adapter_metadata.get("model")
+        if model:
+            rep["model"] = model
+
+        if instance.is_friction_less:
+            rep["created_by_email"] = "Unstract"
+        else:
+            rep["created_by_email"] = instance.created_by.email
+
+        return rep
+
+
+class SharedUserListSerializer(BaseAdapterSerializer):
+    """Inherits BaseAdapterSerializer.
+
+    Used for listing adapter users
+    """
+
+    shared_users = UserSerializer(many=True)
+    created_by = UserSerializer()
+
+    class Meta(BaseAdapterSerializer.Meta):
+        model = AdapterInstance
+        fields = (
+            "id",
+            "adapter_id",
+            "adapter_name",
+            "adapter_type",
+            "created_by",
+            "shared_users",
+        )  # type: ignore
+
+
+class UserDefaultAdapterSerializer(ModelSerializer):
+    class Meta:
+        model = UserDefaultAdapter
+        fields = "__all__"
diff --git a/backend/adapter_processor_v2/urls.py b/backend/adapter_processor_v2/urls.py
new file mode 100644
index 0000000000..741d3c3e18
--- /dev/null
+++ b/backend/adapter_processor_v2/urls.py
@@ -0,0 +1,42 @@
+from adapter_processor_v2.views import (
+    AdapterInstanceViewSet,
+    AdapterViewSet,
+    DefaultAdapterViewSet,
+)
+from django.urls import path
+from rest_framework.urlpatterns import format_suffix_patterns
+
+default_triad = DefaultAdapterViewSet.as_view(
+    {"post": "configure_default_triad", "get": "get_default_triad"}
+)
+adapter = AdapterViewSet.as_view({"get": "list"})
+adapter_schema = AdapterViewSet.as_view({"get": "get_adapter_schema"})
+adapter_test = AdapterViewSet.as_view({"post": "test"})
+adapter_list = AdapterInstanceViewSet.as_view({"post": "create", "get": "list"})
+adapter_detail = AdapterInstanceViewSet.as_view(
+    {
+        "get": "retrieve",
+        "put": "update",
+        "patch": "partial_update",
+        "delete": "destroy",
+    }
+)
+
+adapter_users = AdapterInstanceViewSet.as_view({"get": "list_of_shared_users"})
+adapter_info = AdapterInstanceViewSet.as_view({"get": "adapter_info"})
+urlpatterns = format_suffix_patterns(
+    [
+        path("adapter_schema/", adapter_schema, name="get_adapter_schema"),
+        path("supported_adapters/", adapter, name="adapter-list"),
+        path("adapter/", adapter_list, name="adapter-list"),
+        path("adapter/default_triad/", default_triad, name="default_triad"),
+        path("adapter/<uuid:pk>/", adapter_detail, name="adapter_detail"),
+        path("adapter/info/<uuid:pk>/", adapter_info, name="adapter_info"),
+        path("test_adapters/", adapter_test, name="adapter-test"),
+        path(
+            "adapter/users/<uuid:pk>/",
+            adapter_users,
+            name="adapter-users",
+        ),
+    ]
+)
diff --git a/backend/adapter_processor_v2/views.py b/backend/adapter_processor_v2/views.py
new file mode 100644
index 0000000000..93cddf5aa3
--- /dev/null
+++ b/backend/adapter_processor_v2/views.py
@@ -0,0 +1,321 @@
+import logging
+import uuid
+from typing import Any, Optional
+
+from adapter_processor_v2.adapter_processor import AdapterProcessor
+from adapter_processor_v2.constants import AdapterKeys
+from adapter_processor_v2.exceptions import (
+    CannotDeleteDefaultAdapter,
+    DeleteAdapterInUseError,
+    IdIsMandatory,
+    InValidType,
+    UniqueConstraintViolation,
+)
+from adapter_processor_v2.serializers import (
+    AdapterInfoSerializer,
+    AdapterInstanceSerializer,
+    AdapterListSerializer,
+    DefaultAdapterSerializer,
+    SharedUserListSerializer,
+    TestAdapterSerializer,
+    UserDefaultAdapterSerializer,
+)
+from django.db import IntegrityError
+from django.db.models import ProtectedError, QuerySet
+from django.http import HttpRequest
+from django.http.response import HttpResponse
+from permissions.permission import (
+    IsFrictionLessAdapter,
+    IsFrictionLessAdapterDelete,
+    IsOwner,
+    IsOwnerOrSharedUserOrSharedToOrg,
+)
+from rest_framework import status
+from rest_framework.decorators import action
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.serializers import ModelSerializer
+from rest_framework.versioning import URLPathVersioning
+from rest_framework.viewsets import GenericViewSet, ModelViewSet
+from tenant_account_v2.organization_member_service import OrganizationMemberService
+from utils.filtering import FilterHelper
+
+from .constants import AdapterKeys as constant
+from .exceptions import InternalServiceError
+from .models import AdapterInstance, UserDefaultAdapter
+
+logger = logging.getLogger(__name__)
+
+
+class DefaultAdapterViewSet(ModelViewSet):
+    versioning_class = URLPathVersioning
+    serializer_class = DefaultAdapterSerializer
+
+    def configure_default_triad(
+        self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
+    ) -> HttpResponse:
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        # Convert request data to json
+        default_triad = request.data
+        AdapterProcessor.set_default_triad(default_triad, request.user)
+        return Response(status=status.HTTP_200_OK)
+
+    def get_default_triad(
+        self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
+    ) -> HttpResponse:
+        try:
+            user_default_adapter = UserDefaultAdapter.objects.get(user=request.user)
+            serializer = UserDefaultAdapterSerializer(user_default_adapter).data
+            return Response(serializer)
+
+        except UserDefaultAdapter.DoesNotExist:
+            # Handle the case when no records are found
+            return Response(status=status.HTTP_200_OK, data={})
+
+
+class AdapterViewSet(GenericViewSet):
+    versioning_class = URLPathVersioning
+    serializer_class = TestAdapterSerializer
+
+    def list(
+        self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
+    ) -> HttpResponse:
+        if request.method == "GET":
+            adapter_type = request.GET.get(AdapterKeys.ADAPTER_TYPE)
+            if (
+                adapter_type == AdapterKeys.LLM
+                or adapter_type == AdapterKeys.EMBEDDING
+                or adapter_type == AdapterKeys.VECTOR_DB
+                or adapter_type == AdapterKeys.X2TEXT
+                or adapter_type == AdapterKeys.OCR
+            ):
+                json_schema = AdapterProcessor.get_all_supported_adapters(
+                    type=adapter_type
+                )
+                return Response(json_schema, status=status.HTTP_200_OK)
+            else:
+                raise InValidType
+
+    def get_adapter_schema(
+        self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
+    ) -> HttpResponse:
+        if request.method == "GET":
+            adapter_name = request.GET.get(AdapterKeys.ID)
+            if adapter_name is None or adapter_name == "":
+                raise IdIsMandatory()
+            json_schema = AdapterProcessor.get_json_schema(adapter_id=adapter_name)
+            return Response(data=json_schema, status=status.HTTP_200_OK)
+
+    def test(self, request: Request) -> Response:
+        """Tests the connector against the credentials passed."""
+        serializer: AdapterInstanceSerializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        adapter_id = serializer.validated_data.get(AdapterKeys.ADAPTER_ID)
+        adapter_metadata = serializer.validated_data.get(AdapterKeys.ADAPTER_METADATA)
+        adapter_metadata[AdapterKeys.ADAPTER_TYPE] = serializer.validated_data.get(
+            AdapterKeys.ADAPTER_TYPE
+        )
+        try:
+            test_result = AdapterProcessor.test_adapter(
+                adapter_id=adapter_id, adapter_metadata=adapter_metadata
+            )
+            return Response(
+                {AdapterKeys.IS_VALID: test_result},
+                status=status.HTTP_200_OK,
+            )
+        except Exception as e:
+            logger.error(f"Error testing adapter : {str(e)}")
+            raise e
+
+
+class AdapterInstanceViewSet(ModelViewSet):
+
+    serializer_class = AdapterInstanceSerializer
+
+    def get_permissions(self) -> list[Any]:
+
+        if self.action in ["update", "retrieve"]:
+            return [IsFrictionLessAdapter()]
+
+        elif self.action == "destroy":
+            return [IsFrictionLessAdapterDelete()]
+
+        elif self.action in ["list_of_shared_users", "adapter_info"]:
+            return [IsOwnerOrSharedUserOrSharedToOrg()]
+
+        # Hack for friction-less onboarding
+        # User cant view/update metadata but can delete/share etc
+        return [IsOwner()]
+
+    def get_queryset(self) -> Optional[QuerySet]:
+        if filter_args := FilterHelper.build_filter_args(
+            self.request,
+            constant.ADAPTER_TYPE,
+        ):
+            queryset = AdapterInstance.objects.for_user(self.request.user).filter(
+                **filter_args
+            )
+        else:
+            queryset = AdapterInstance.objects.for_user(self.request.user)
+        return queryset
+
+    def get_serializer_class(
+        self,
+    ) -> ModelSerializer:
+        if self.action == "list":
+            return AdapterListSerializer
+        return AdapterInstanceSerializer
+
+    def create(self, request: Any) -> Response:
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        try:
+            instance = serializer.save()
+
+            # Check to see if there is a default configured
+            # for this adapter_type and for the current user
+            (
+                user_default_adapter,
+                created,
+            ) = UserDefaultAdapter.objects.get_or_create(user=request.user)
+
+            adapter_type = serializer.validated_data.get(AdapterKeys.ADAPTER_TYPE)
+            if (adapter_type == AdapterKeys.LLM) and (
+                not user_default_adapter.default_llm_adapter
+            ):
+                user_default_adapter.default_llm_adapter = instance
+
+            elif (adapter_type == AdapterKeys.EMBEDDING) and (
+                not user_default_adapter.default_embedding_adapter
+            ):
+                user_default_adapter.default_embedding_adapter = instance
+            elif (adapter_type == AdapterKeys.VECTOR_DB) and (
+                not user_default_adapter.default_vector_db_adapter
+            ):
+                user_default_adapter.default_vector_db_adapter = instance
+            elif (adapter_type == AdapterKeys.X2TEXT) and (
+                not user_default_adapter.default_x2text_adapter
+            ):
+                user_default_adapter.default_x2text_adapter = instance
+
+            organization_member = OrganizationMemberService.get_user_by_id(
+                request.user.id
+            )
+            user_default_adapter.organization_member = organization_member
+
+            user_default_adapter.save()
+
+        except IntegrityError:
+            raise UniqueConstraintViolation(f"{AdapterKeys.ADAPTER_NAME_EXISTS}")
+        except Exception as e:
+            logger.error(f"Error saving adapter to DB: {e}")
+            raise InternalServiceError
+        headers = self.get_success_headers(serializer.data)
+        return Response(
+            serializer.data, status=status.HTTP_201_CREATED, headers=headers
+        )
+
+    def destroy(
+        self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
+    ) -> Response:
+        adapter_instance: AdapterInstance = self.get_object()
+        adapter_type = adapter_instance.adapter_type
+        try:
+            user_default_adapter: UserDefaultAdapter = UserDefaultAdapter.objects.get(
+                user=request.user
+            )
+
+            if (
+                (
+                    adapter_type == AdapterKeys.LLM
+                    and adapter_instance == user_default_adapter.default_llm_adapter
+                )
+                or (
+                    adapter_type == AdapterKeys.EMBEDDING
+                    and adapter_instance
+                    == user_default_adapter.default_embedding_adapter
+                )
+                or (
+                    adapter_type == AdapterKeys.VECTOR_DB
+                    and adapter_instance
+                    == user_default_adapter.default_vector_db_adapter
+                )
+                or (
+                    adapter_type == AdapterKeys.X2TEXT
+                    and adapter_instance == user_default_adapter.default_x2text_adapter
+                )
+            ):
+                logger.error("Cannot delete a default adapter")
+                raise CannotDeleteDefaultAdapter()
+        except UserDefaultAdapter.DoesNotExist:
+            # We can go head and remove adapter here
+            logger.info("User default adpater doesnt not exist")
+
+        try:
+            super().perform_destroy(adapter_instance)
+        except ProtectedError:
+            logger.error(
+                f"Failed to delete adapter: {adapter_instance.adapter_id}"
+                f" named {adapter_instance.adapter_name}"
+            )
+            # TODO: Provide details of adpter usage with exception object
+            raise DeleteAdapterInUseError(adapter_name=adapter_instance.adapter_name)
+        return Response(status=status.HTTP_204_NO_CONTENT)
+
+    def partial_update(
+        self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
+    ) -> Response:
+        if AdapterKeys.SHARED_USERS in request.data:
+            # find the deleted users
+            adapter = self.get_object()
+            shared_users = {
+                int(user_id) for user_id in request.data.get("shared_users", {})
+            }
+            current_users = {user.id for user in adapter.shared_users.all()}
+            removed_users = current_users.difference(shared_users)
+
+            # if removed user use this adapter as default
+            # Remove the same from his default
+            for user_id in removed_users:
+                try:
+                    user_default_adapter = UserDefaultAdapter.objects.get(
+                        user_id=user_id
+                    )
+
+                    if user_default_adapter.default_llm_adapter == adapter:
+                        user_default_adapter.default_llm_adapter = None
+                    elif user_default_adapter.default_embedding_adapter == adapter:
+                        user_default_adapter.default_embedding_adapter = None
+                    elif user_default_adapter.default_vector_db_adapter == adapter:
+                        user_default_adapter.default_vector_db_adapter = None
+                    elif user_default_adapter.default_x2text_adapter == adapter:
+                        user_default_adapter.default_x2text_adapter = None
+
+                    user_default_adapter.save()
+                except UserDefaultAdapter.DoesNotExist:
+                    logger.debug(
+                        "User id : %s doesnt have default adapters configured",
+                        user_id,
+                    )
+                    continue
+
+        return super().partial_update(request, *args, **kwargs)
+
+    @action(detail=True, methods=["get"])
+    def list_of_shared_users(self, request: HttpRequest, pk: Any = None) -> Response:
+
+        adapter = self.get_object()
+
+        serialized_instances = SharedUserListSerializer(adapter).data
+
+        return Response(serialized_instances)
+
+    @action(detail=True, methods=["get"])
+    def adapter_info(self, request: HttpRequest, pk: uuid) -> Response:
+
+        adapter = self.get_object()
+
+        serialized_instances = AdapterInfoSerializer(adapter).data
+
+        return Response(serialized_instances)
diff --git a/backend/connector_processor/connector_processor.py b/backend/connector_processor/connector_processor.py
index e014d69790..770d03c490 100644
--- a/backend/connector_processor/connector_processor.py
+++ b/backend/connector_processor/connector_processor.py
@@ -3,9 +3,6 @@
 import logging
 from typing import Any, Optional
 
-from connector.constants import ConnectorInstanceKey as CIKey
-from connector_auth.constants import ConnectorAuthKey
-from connector_auth.pipeline.common import ConnectorAuthHelper
 from connector_processor.constants import ConnectorKeys
 from connector_processor.exceptions import (
     InValidConnectorId,
@@ -14,11 +11,22 @@
     TestConnectorInputError,
 )
 
+from backend.constants import FeatureFlag
 from unstract.connectors.base import UnstractConnector
 from unstract.connectors.connectorkit import Connectorkit
 from unstract.connectors.enums import ConnectorMode
 from unstract.connectors.exceptions import ConnectorError
 from unstract.connectors.filesystems.ucs import UnstractCloudStorage
+from unstract.flags.feature_flag import check_feature_flag_status
+
+if check_feature_flag_status(FeatureFlag.MULTI_TENANCY_V2):
+    from connector_auth_v2.constants import ConnectorAuthKey
+    from connector_auth_v2.pipeline.common import ConnectorAuthHelper
+    from connector_v2.constants import ConnectorInstanceKey as CIKey
+else:
+    from connector.constants import ConnectorInstanceKey as CIKey
+    from connector_auth.constants import ConnectorAuthKey
+    from connector_auth.pipeline.common import ConnectorAuthHelper
 
 logger = logging.getLogger(__name__)
 
diff --git a/backend/connector_processor/views.py b/backend/connector_processor/views.py
index edca86ba15..20c9277a76 100644
--- a/backend/connector_processor/views.py
+++ b/backend/connector_processor/views.py
@@ -1,4 +1,3 @@
-from connector.constants import ConnectorInstanceKey as CIKey
 from connector_processor.connector_processor import ConnectorProcessor
 from connector_processor.constants import ConnectorKeys
 from connector_processor.exceptions import IdIsMandatory, InValidType
@@ -13,6 +12,14 @@
 from rest_framework.versioning import URLPathVersioning
 from rest_framework.viewsets import GenericViewSet
 
+from backend.constants import FeatureFlag
+from unstract.flags.feature_flag import check_feature_flag_status
+
+if check_feature_flag_status(FeatureFlag.MULTI_TENANCY_V2):
+    from connector_v2.constants import ConnectorInstanceKey as CIKey
+else:
+    from connector.constants import ConnectorInstanceKey as CIKey
+
 
 @api_view(("GET",))
 def get_connector_schema(request: HttpRequest) -> HttpResponse:
