diff --git a/libraries/botbuilder-core/botbuilder/core/adapters/test_adapter.py b/libraries/botbuilder-core/botbuilder/core/adapters/test_adapter.py
index 18c0e2962..56f122344 100644
--- a/libraries/botbuilder-core/botbuilder/core/adapters/test_adapter.py
+++ b/libraries/botbuilder-core/botbuilder/core/adapters/test_adapter.py
@@ -217,7 +217,7 @@ async def continue_conversation(
             reference, callback, bot_id, claims_identity, audience
         )
 
-    async def create_conversation(
+    async def create_conversation(  # pylint: disable=arguments-differ
         self, channel_id: str, callback: Callable  # pylint: disable=unused-argument
     ):
         self.activity_buffer.clear()
diff --git a/libraries/botbuilder-core/botbuilder/core/bot_adapter.py b/libraries/botbuilder-core/botbuilder/core/bot_adapter.py
index cb073bc51..5ab04eafb 100644
--- a/libraries/botbuilder-core/botbuilder/core/bot_adapter.py
+++ b/libraries/botbuilder-core/botbuilder/core/bot_adapter.py
@@ -3,8 +3,13 @@
 
 from abc import ABC, abstractmethod
 from typing import List, Callable, Awaitable
-from botbuilder.schema import Activity, ConversationReference, ResourceResponse
-from botframework.connector.auth import ClaimsIdentity
+from botbuilder.schema import (
+    Activity,
+    ConversationReference,
+    ConversationParameters,
+    ResourceResponse,
+)
+from botframework.connector.auth import AppCredentials, ClaimsIdentity
 
 from . import conversation_reference_extension
 from .bot_assert import BotAssert
@@ -108,6 +113,47 @@ async def continue_conversation(
         )
         return await self.run_pipeline(context, callback)
 
+    async def create_conversation(
+        self,
+        reference: ConversationReference,
+        logic: Callable[[TurnContext], Awaitable] = None,
+        conversation_parameters: ConversationParameters = None,
+        channel_id: str = None,
+        service_url: str = None,
+        credentials: AppCredentials = None,
+    ):
+        """
+        Starts a new conversation with a user. Used to direct message to a member of a group.
+
+        :param reference: The conversation reference that contains the tenant
+        :type reference: :class:`botbuilder.schema.ConversationReference`
+        :param logic: The logic to use for the creation of the conversation
+        :type logic: :class:`typing.Callable`
+        :param conversation_parameters: The information to use to create the conversation
+        :type conversation_parameters:
+        :param channel_id: The ID for the channel.
+        :type channel_id: :class:`typing.str`
+        :param service_url: The channel's service URL endpoint.
+        :type service_url: :class:`typing.str`
+        :param credentials: The application credentials for the bot.
+        :type credentials: :class:`botframework.connector.auth.AppCredentials`
+
+        :raises: It raises a generic exception error.
+
+        :return: A task representing the work queued to execute.
+
+        .. remarks::
+            To start a conversation, your bot must know its account information and the user's
+            account information on that channel.
+            Most channels only support initiating a direct message (non-group) conversation.
+            The adapter attempts to create a new conversation on the channel, and
+            then sends a conversation update activity through its middleware pipeline
+            to the the callback method.
+            If the conversation is established with the specified users, the ID of the activity
+            will contain the ID of the new conversation.
+        """
+        raise Exception("Not Implemented")
+
     async def run_pipeline(
         self, context: TurnContext, callback: Callable[[TurnContext], Awaitable] = None
     ):
diff --git a/libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py b/libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py
index 15e23e8f0..41ce8ff72 100644
--- a/libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py
+++ b/libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py
@@ -363,7 +363,11 @@ async def create_conversation(
             )
 
             # Mix in the tenant ID if specified. This is required for MS Teams.
-            if reference.conversation and reference.conversation.tenant_id:
+            if (
+                reference
+                and reference.conversation
+                and reference.conversation.tenant_id
+            ):
                 # Putting tenant_id in channel_data is a temporary while we wait for the Teams API to be updated
                 if parameters.channel_data is None:
                     parameters.channel_data = {}
diff --git a/libraries/botbuilder-core/botbuilder/core/cloud_adapter_base.py b/libraries/botbuilder-core/botbuilder/core/cloud_adapter_base.py
index 6b1301f1e..56976ac94 100644
--- a/libraries/botbuilder-core/botbuilder/core/cloud_adapter_base.py
+++ b/libraries/botbuilder-core/botbuilder/core/cloud_adapter_base.py
@@ -6,13 +6,18 @@
 from copy import Error
 from http import HTTPStatus
 from typing import Awaitable, Callable, List, Union
+from uuid import uuid4
 
 from botbuilder.core.invoke_response import InvokeResponse
 
 from botbuilder.schema import (
     Activity,
+    ActivityEventNames,
     ActivityTypes,
+    ConversationAccount,
     ConversationReference,
+    ConversationResourceResponse,
+    ConversationParameters,
     DeliveryModes,
     ExpectedReplies,
     ResourceResponse,
@@ -175,6 +180,71 @@ async def continue_conversation_with_claims(
             claims_identity, get_continuation_activity(reference), audience, logic
         )
 
+    async def create_conversation(  # pylint: disable=arguments-differ
+        self,
+        bot_app_id: ConversationReference,
+        callback: Callable[[TurnContext], Awaitable] = None,
+        conversation_parameters: ConversationParameters = None,
+        channel_id: str = None,
+        service_url: str = None,
+        audience: str = None,
+    ):
+        if not service_url:
+            raise TypeError(
+                "CloudAdapter.create_conversation(): service_url is required."
+            )
+        if not conversation_parameters:
+            raise TypeError(
+                "CloudAdapter.create_conversation(): conversation_parameters is required."
+            )
+        if not callback:
+            raise TypeError("CloudAdapter.create_conversation(): callback is required.")
+
+        # Create a ClaimsIdentity, to create the connector and for adding to the turn context.
+        claims_identity = self.create_claims_identity(bot_app_id)
+        claims_identity.claims[AuthenticationConstants.SERVICE_URL_CLAIM] = service_url
+
+        # create the connectror factory
+        connector_factory = self.bot_framework_authentication.create_connector_factory(
+            claims_identity
+        )
+
+        # Create the connector client to use for outbound requests.
+        connector_client = await connector_factory.create(service_url, audience)
+
+        # Make the actual create conversation call using the connector.
+        create_conversation_result = (
+            await connector_client.conversations.create_conversation(
+                conversation_parameters
+            )
+        )
+
+        # Create the create activity to communicate the results to the application.
+        create_activity = self._create_create_activity(
+            create_conversation_result, channel_id, service_url, conversation_parameters
+        )
+
+        # Create a UserTokenClient instance for the application to use. (For example, in the OAuthPrompt.)
+        user_token_client = (
+            await self.bot_framework_authentication.create_user_token_client(
+                claims_identity
+            )
+        )
+
+        # Create a turn context and run the pipeline.
+        context = self._create_turn_context(
+            create_activity,
+            claims_identity,
+            None,
+            connector_client,
+            user_token_client,
+            callback,
+            connector_factory,
+        )
+
+        # Run the pipeline
+        await self.run_pipeline(context, callback)
+
     async def process_proactive(
         self,
         claims_identity: ClaimsIdentity,
@@ -301,6 +371,28 @@ def create_claims_identity(self, bot_app_id: str = "") -> ClaimsIdentity:
             True,
         )
 
+    def _create_create_activity(
+        self,
+        create_conversation_result: ConversationResourceResponse,
+        channel_id: str,
+        service_url: str,
+        conversation_parameters: ConversationParameters,
+    ) -> Activity:
+        # Create a conversation update activity to represent the result.
+        activity = Activity.create_event_activity()
+        activity.name = ActivityEventNames.create_conversation
+        activity.channel_id = channel_id
+        activity.service_url = service_url
+        activity.id = create_conversation_result.activity_id or str(uuid4())
+        activity.conversation = ConversationAccount(
+            id=create_conversation_result.id,
+            tenant_id=conversation_parameters.tenant_id,
+        )
+        activity.channel_data = conversation_parameters.channel_data
+        activity.recipient = conversation_parameters.bot
+
+        return activity
+
     def _create_turn_context(
         self,
         activity: Activity,
diff --git a/libraries/botbuilder-core/botbuilder/core/teams/teams_info.py b/libraries/botbuilder-core/botbuilder/core/teams/teams_info.py
index 2cd9ee0c7..3226cb053 100644
--- a/libraries/botbuilder-core/botbuilder/core/teams/teams_info.py
+++ b/libraries/botbuilder-core/botbuilder/core/teams/teams_info.py
@@ -3,14 +3,15 @@
 
 from typing import List, Tuple
 
+from botframework.connector import Channels
 from botframework.connector.aio import ConnectorClient
-from botframework.connector.teams.teams_connector_client import TeamsConnectorClient
-from botbuilder.schema import ConversationParameters, ConversationReference
+from botframework.connector.teams import TeamsConnectorClient
 from botbuilder.core.teams.teams_activity_extensions import (
     teams_get_meeting_info,
     teams_get_channel_data,
 )
-from botbuilder.core.turn_context import Activity, TurnContext
+from botbuilder.core import CloudAdapterBase, BotFrameworkAdapter, TurnContext
+from botbuilder.schema import Activity, ConversationParameters, ConversationReference
 from botbuilder.schema.teams import (
     ChannelInfo,
     MeetingInfo,
@@ -25,15 +26,61 @@
 class TeamsInfo:
     @staticmethod
     async def send_message_to_teams_channel(
-        turn_context: TurnContext, activity: Activity, teams_channel_id: str
+        turn_context: TurnContext,
+        activity: Activity,
+        teams_channel_id: str,
+        *,
+        bot_app_id: str = None,
     ) -> Tuple[ConversationReference, str]:
         if not turn_context:
             raise ValueError("The turn_context cannot be None")
+        if not turn_context.activity:
+            raise ValueError("The activity inside turn context cannot be None")
         if not activity:
             raise ValueError("The activity cannot be None")
         if not teams_channel_id:
             raise ValueError("The teams_channel_id cannot be None or empty")
 
+        if not bot_app_id:
+            return await TeamsInfo._legacy_send_message_to_teams_channel(
+                turn_context, activity, teams_channel_id
+            )
+
+        conversation_reference: ConversationReference = None
+        new_activity_id = ""
+        service_url = turn_context.activity.service_url
+        conversation_parameters = ConversationParameters(
+            is_group=True,
+            channel_data=TeamsChannelData(channel=ChannelInfo(id=teams_channel_id)),
+            activity=activity,
+        )
+
+        async def aux_callback(
+            new_turn_context,
+        ):
+            nonlocal new_activity_id
+            nonlocal conversation_reference
+            new_activity_id = new_turn_context.activity.id
+            conversation_reference = TurnContext.get_conversation_reference(
+                new_turn_context.activity
+            )
+
+        adapter: CloudAdapterBase = turn_context.adapter
+        await adapter.create_conversation(
+            bot_app_id,
+            aux_callback,
+            conversation_parameters,
+            Channels.ms_teams,
+            service_url,
+            None,
+        )
+
+        return (conversation_reference, new_activity_id)
+
+    @staticmethod
+    async def _legacy_send_message_to_teams_channel(
+        turn_context: TurnContext, activity: Activity, teams_channel_id: str
+    ) -> Tuple[ConversationReference, str]:
         old_ref = TurnContext.get_conversation_reference(turn_context.activity)
         conversation_parameters = ConversationParameters(
             is_group=True,
@@ -41,7 +88,9 @@ async def send_message_to_teams_channel(
             activity=activity,
         )
 
-        result = await turn_context.adapter.create_conversation(
+        # if this version of the method is called the adapter probably wont be CloudAdapter
+        adapter: BotFrameworkAdapter = turn_context.adapter
+        result = await adapter.create_conversation(
             old_ref, TeamsInfo._create_conversation_callback, conversation_parameters
         )
         return (result[0], result[1])
diff --git a/libraries/botbuilder-core/tests/simple_adapter.py b/libraries/botbuilder-core/tests/simple_adapter.py
index b8dd3c404..ae68dc323 100644
--- a/libraries/botbuilder-core/tests/simple_adapter.py
+++ b/libraries/botbuilder-core/tests/simple_adapter.py
@@ -59,7 +59,7 @@ async def send_activities(
 
         return responses
 
-    async def create_conversation(
+    async def create_conversation(  # pylint: disable=arguments-differ
         self,
         reference: ConversationReference,
         logic: Callable[[TurnContext], Awaitable] = None,
diff --git a/libraries/botbuilder-core/tests/teams/simple_adapter_with_create_conversation.py b/libraries/botbuilder-core/tests/teams/simple_adapter_with_create_conversation.py
index 477aa3b28..d1801d978 100644
--- a/libraries/botbuilder-core/tests/teams/simple_adapter_with_create_conversation.py
+++ b/libraries/botbuilder-core/tests/teams/simple_adapter_with_create_conversation.py
@@ -58,7 +58,7 @@ async def send_activities(
 
         return responses
 
-    async def create_conversation(
+    async def create_conversation(  # pylint: disable=arguments-differ
         self,
         reference: ConversationReference,
         logic: Callable[[TurnContext], Awaitable] = None,