diff --git a/libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py b/libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py
index 8300bfefe..ffe81ec7d 100644
--- a/libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py
+++ b/libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py
@@ -39,6 +39,7 @@
     ConversationReference,
     TokenResponse,
     ResourceResponse,
+    DeliveryModes,
 )
 
 from . import __version__
@@ -456,6 +457,15 @@ async def process_activity_with_identity(
                 return InvokeResponse(status=501)
             return invoke_response.value
 
+        # Return the buffered activities in the response.  In this case, the invoker
+        # should deserialize accordingly:
+        #    activities = [Activity().deserialize(activity) for activity in response.body]
+        if context.activity.delivery_mode == DeliveryModes.buffered_replies:
+            serialized_activities = [
+                activity.serialize() for activity in context.buffered_replies
+            ]
+            return InvokeResponse(status=200, body=serialized_activities)
+
         return None
 
     async def _authenticate_request(
diff --git a/libraries/botbuilder-core/botbuilder/core/bot_framework_http_client.py b/libraries/botbuilder-core/botbuilder/core/bot_framework_http_client.py
index a72e3a8f5..963cf8bd4 100644
--- a/libraries/botbuilder-core/botbuilder/core/bot_framework_http_client.py
+++ b/libraries/botbuilder-core/botbuilder/core/bot_framework_http_client.py
@@ -91,7 +91,7 @@ async def post_activity(
             content = json.loads(data) if data else None
 
             if content:
-                return InvokeResponse(status=resp.status_code, body=content)
+                return InvokeResponse(status=resp.status, body=content)
 
         finally:
             # Restore activity properties.
@@ -99,6 +99,26 @@ async def post_activity(
             activity.service_url = original_service_url
             activity.caller_id = original_caller_id
 
+    async def post_buffered_activity(
+        self,
+        from_bot_id: str,
+        to_bot_id: str,
+        to_url: str,
+        service_url: str,
+        conversation_id: str,
+        activity: Activity,
+    ) -> [Activity]:
+        """
+        Helper method to return a list of activities when an Activity is being
+        sent with DeliveryMode == bufferedReplies.
+        """
+        response = await self.post_activity(
+            from_bot_id, to_bot_id, to_url, service_url, conversation_id, activity
+        )
+        if not response or (response.status / 100) != 2:
+            return []
+        return [Activity().deserialize(activity) for activity in response.body]
+
     async def _get_app_credentials(
         self, app_id: str, oauth_scope: str
     ) -> MicrosoftAppCredentials:
diff --git a/libraries/botbuilder-core/botbuilder/core/invoke_response.py b/libraries/botbuilder-core/botbuilder/core/invoke_response.py
index 408662707..7d258559e 100644
--- a/libraries/botbuilder-core/botbuilder/core/invoke_response.py
+++ b/libraries/botbuilder-core/botbuilder/core/invoke_response.py
@@ -1,20 +1,26 @@
-# Copyright (c) Microsoft Corporation. All rights reserved.
-# Licensed under the MIT License.
-
-
-class InvokeResponse:
-    """
-    Tuple class containing an HTTP Status Code and a JSON Serializable
-    object. The HTTP Status code is, in the invoke activity scenario, what will
-    be set in the resulting POST. The Body of the resulting POST will be
-    the JSON Serialized content from the Body property.
-    """
-
-    def __init__(self, status: int = None, body: object = None):
-        """
-        Gets or sets the HTTP status and/or body code for the response
-        :param status: The HTTP status code.
-        :param body: The body content for the response.
-        """
-        self.status = status
-        self.body = body
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+
+class InvokeResponse:
+    """
+    Tuple class containing an HTTP Status Code and a JSON serializable
+    object. The HTTP Status code is, in the invoke activity scenario, what will
+    be set in the resulting POST. The Body of the resulting POST will be
+    JSON serialized content.
+
+    The body content is defined by the producer.  The caller must know what
+    the content is and deserialize as needed.
+    """
+
+    def __init__(self, status: int = None, body: object = None):
+        """
+        Gets or sets the HTTP status and/or body code for the response
+        :param status: The HTTP status code.
+        :param body: The JSON serializable body content for the response.  This object
+        must be serializable by the core Python json routines.  The caller is responsible
+        for serializing more complex/nested objects into native classes (lists and
+        dictionaries of strings are acceptable).
+        """
+        self.status = status
+        self.body = body
diff --git a/libraries/botbuilder-core/botbuilder/core/turn_context.py b/libraries/botbuilder-core/botbuilder/core/turn_context.py
index 614a43ce1..b3ec326c8 100644
--- a/libraries/botbuilder-core/botbuilder/core/turn_context.py
+++ b/libraries/botbuilder-core/botbuilder/core/turn_context.py
@@ -12,6 +12,7 @@
     InputHints,
     Mention,
     ResourceResponse,
+    DeliveryModes,
 )
 from .re_escape import escape
 
@@ -50,6 +51,9 @@ def __init__(self, adapter_or_context, request: Activity = None):
 
         self._turn_state = {}
 
+        # A list of activities to send when `context.Activity.DeliveryMode == 'bufferedReplies'`
+        self.buffered_replies = []
+
     @property
     def turn_state(self) -> Dict[str, object]:
         return self._turn_state
@@ -190,7 +194,21 @@ def activity_validator(activity: Activity) -> Activity:
             for act in activities
         ]
 
+        # send activities through adapter
         async def logic():
+            nonlocal sent_non_trace_activity
+
+            if self.activity.delivery_mode == DeliveryModes.buffered_replies:
+                responses = []
+                for activity in output:
+                    self.buffered_replies.append(activity)
+                    responses.append(ResourceResponse())
+
+                if sent_non_trace_activity:
+                    self.responded = True
+
+                return responses
+
             responses = await self.adapter.send_activities(self, output)
             if sent_non_trace_activity:
                 self.responded = True
diff --git a/libraries/botbuilder-core/tests/test_bot_framework_adapter.py b/libraries/botbuilder-core/tests/test_bot_framework_adapter.py
index e53148d94..871e616a1 100644
--- a/libraries/botbuilder-core/tests/test_bot_framework_adapter.py
+++ b/libraries/botbuilder-core/tests/test_bot_framework_adapter.py
@@ -19,6 +19,7 @@
     ConversationReference,
     ConversationResourceResponse,
     ChannelAccount,
+    DeliveryModes,
 )
 from botframework.connector.aio import ConnectorClient
 from botframework.connector.auth import (
@@ -58,6 +59,7 @@ def __init__(self, settings=None):
         self.fail_operation = False
         self.expect_auth_header = ""
         self.new_service_url = None
+        self.connector_client_mock = None
 
     def aux_test_authenticate_request(self, request: Activity, auth_header: str):
         return super()._authenticate_request(request, auth_header)
@@ -102,7 +104,10 @@ def _get_or_create_connector_client(
         self.tester.assertIsNotNone(
             service_url, "create_connector_client() not passed service_url."
         )
-        connector_client_mock = Mock()
+
+        if self.connector_client_mock:
+            return self.connector_client_mock
+        self.connector_client_mock = Mock()
 
         async def mock_reply_to_activity(conversation_id, activity_id, activity):
             nonlocal self
@@ -160,23 +165,23 @@ async def mock_create_conversation(parameters):
             )
             return response
 
-        connector_client_mock.conversations.reply_to_activity.side_effect = (
+        self.connector_client_mock.conversations.reply_to_activity.side_effect = (
             mock_reply_to_activity
         )
-        connector_client_mock.conversations.send_to_conversation.side_effect = (
+        self.connector_client_mock.conversations.send_to_conversation.side_effect = (
             mock_send_to_conversation
         )
-        connector_client_mock.conversations.update_activity.side_effect = (
+        self.connector_client_mock.conversations.update_activity.side_effect = (
             mock_update_activity
         )
-        connector_client_mock.conversations.delete_activity.side_effect = (
+        self.connector_client_mock.conversations.delete_activity.side_effect = (
             mock_delete_activity
         )
-        connector_client_mock.conversations.create_conversation.side_effect = (
+        self.connector_client_mock.conversations.create_conversation.side_effect = (
             mock_create_conversation
         )
 
-        return connector_client_mock
+        return self.connector_client_mock
 
 
 async def process_activity(
@@ -572,3 +577,88 @@ async def callback(context: TurnContext):
         await adapter.continue_conversation(
             refs, callback, claims_identity=skills_identity, audience=skill_2_app_id
         )
+
+    async def test_delivery_mode_buffered_replies(self):
+        mock_credential_provider = unittest.mock.create_autospec(CredentialProvider)
+
+        settings = BotFrameworkAdapterSettings(
+            app_id="bot_id", credential_provider=mock_credential_provider
+        )
+        adapter = AdapterUnderTest(settings)
+
+        async def callback(context: TurnContext):
+            await context.send_activity("activity 1")
+            await context.send_activity("activity 2")
+            await context.send_activity("activity 3")
+
+        inbound_activity = Activity(
+            type=ActivityTypes.message,
+            channel_id="emulator",
+            service_url="http://tempuri.org/whatever",
+            delivery_mode=DeliveryModes.buffered_replies,
+            text="hello world",
+        )
+
+        identity = ClaimsIdentity(
+            claims={
+                AuthenticationConstants.AUDIENCE_CLAIM: "bot_id",
+                AuthenticationConstants.APP_ID_CLAIM: "bot_id",
+                AuthenticationConstants.VERSION_CLAIM: "1.0",
+            },
+            is_authenticated=True,
+        )
+
+        invoke_response = await adapter.process_activity_with_identity(
+            inbound_activity, identity, callback
+        )
+        assert invoke_response
+        assert invoke_response.status == 200
+        activities = invoke_response.body
+        assert len(activities) == 3
+        assert activities[0]["text"] == "activity 1"
+        assert activities[1]["text"] == "activity 2"
+        assert activities[2]["text"] == "activity 3"
+        assert (
+            adapter.connector_client_mock.conversations.send_to_conversation.call_count
+            == 0
+        )
+
+    async def test_delivery_mode_normal(self):
+        mock_credential_provider = unittest.mock.create_autospec(CredentialProvider)
+
+        settings = BotFrameworkAdapterSettings(
+            app_id="bot_id", credential_provider=mock_credential_provider
+        )
+        adapter = AdapterUnderTest(settings)
+
+        async def callback(context: TurnContext):
+            await context.send_activity("activity 1")
+            await context.send_activity("activity 2")
+            await context.send_activity("activity 3")
+
+        inbound_activity = Activity(
+            type=ActivityTypes.message,
+            channel_id="emulator",
+            service_url="http://tempuri.org/whatever",
+            delivery_mode=DeliveryModes.normal,
+            text="hello world",
+            conversation=ConversationAccount(id="conversationId"),
+        )
+
+        identity = ClaimsIdentity(
+            claims={
+                AuthenticationConstants.AUDIENCE_CLAIM: "bot_id",
+                AuthenticationConstants.APP_ID_CLAIM: "bot_id",
+                AuthenticationConstants.VERSION_CLAIM: "1.0",
+            },
+            is_authenticated=True,
+        )
+
+        invoke_response = await adapter.process_activity_with_identity(
+            inbound_activity, identity, callback
+        )
+        assert not invoke_response
+        assert (
+            adapter.connector_client_mock.conversations.send_to_conversation.call_count
+            == 3
+        )
diff --git a/libraries/botbuilder-schema/botbuilder/schema/_connector_client_enums.py b/libraries/botbuilder-schema/botbuilder/schema/_connector_client_enums.py
index a725f880b..68aab5ecf 100644
--- a/libraries/botbuilder-schema/botbuilder/schema/_connector_client_enums.py
+++ b/libraries/botbuilder-schema/botbuilder/schema/_connector_client_enums.py
@@ -99,6 +99,7 @@ class DeliveryModes(str, Enum):
 
     normal = "normal"
     notification = "notification"
+    buffered_replies = "bufferedReplies"
 
 
 class ContactRelationUpdateActionTypes(str, Enum):
diff --git a/samples/experimental/skills-buffered/child/app.py b/samples/experimental/skills-buffered/child/app.py
new file mode 100644
index 000000000..27351c36d
--- /dev/null
+++ b/samples/experimental/skills-buffered/child/app.py
@@ -0,0 +1,78 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import sys
+import traceback
+
+from aiohttp import web
+from aiohttp.web import Request, Response
+from aiohttp.web_response import json_response
+from botbuilder.core import (
+    BotFrameworkAdapterSettings,
+    TurnContext,
+    BotFrameworkAdapter,
+)
+from botbuilder.schema import Activity
+
+from bots import ChildBot
+from config import DefaultConfig
+
+CONFIG = DefaultConfig()
+
+# Create adapter.
+# See https://aka.ms/about-bot-adapter to learn more about how bots work.
+SETTINGS = BotFrameworkAdapterSettings(
+    app_id=CONFIG.APP_ID, app_password=CONFIG.APP_PASSWORD,
+)
+ADAPTER = BotFrameworkAdapter(SETTINGS)
+
+
+# Catch-all for errors.
+async def on_error(context: TurnContext, error: Exception):
+    # This check writes out errors to console log .vs. app insights.
+    # NOTE: In production environment, you should consider logging this to Azure
+    #       application insights.
+    print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
+    traceback.print_exc()
+
+    # Send a message to the user
+    await context.send_activity("The bot encountered an error or bug.")
+    await context.send_activity(
+        "To continue to run this bot, please fix the bot source code."
+    )
+
+
+ADAPTER.on_turn_error = on_error
+
+# Create the Bot
+BOT = ChildBot()
+
+
+# Listen for incoming requests on /api/messages
+async def messages(req: Request) -> Response:
+    # Main bot message handler.
+    if "application/json" in req.headers["Content-Type"]:
+        body = await req.json()
+    else:
+        return Response(status=415)
+
+    activity = Activity().deserialize(body)
+    auth_header = req.headers["Authorization"] if "Authorization" in req.headers else ""
+
+    try:
+        response = await ADAPTER.process_activity(activity, auth_header, BOT.on_turn)
+        if response:
+            return json_response(data=response.body, status=response.status)
+        return Response(status=201)
+    except Exception as exception:
+        raise exception
+
+
+APP = web.Application()
+APP.router.add_post("/api/messages", messages)
+
+if __name__ == "__main__":
+    try:
+        web.run_app(APP, host="localhost", port=CONFIG.PORT)
+    except Exception as error:
+        raise error
diff --git a/samples/experimental/skills-buffered/child/bots/__init__.py b/samples/experimental/skills-buffered/child/bots/__init__.py
new file mode 100644
index 000000000..a1643fbf8
--- /dev/null
+++ b/samples/experimental/skills-buffered/child/bots/__init__.py
@@ -0,0 +1,6 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from .child_bot import ChildBot
+
+__all__ = ["ChildBot"]
diff --git a/samples/experimental/skills-buffered/child/bots/child_bot.py b/samples/experimental/skills-buffered/child/bots/child_bot.py
new file mode 100644
index 000000000..ad6a37839
--- /dev/null
+++ b/samples/experimental/skills-buffered/child/bots/child_bot.py
@@ -0,0 +1,12 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from botbuilder.core import ActivityHandler, TurnContext
+
+
+class ChildBot(ActivityHandler):
+    async def on_message_activity(self, turn_context: TurnContext):
+        await turn_context.send_activity("child: activity (1)")
+        await turn_context.send_activity("child: activity (2)")
+        await turn_context.send_activity("child: activity (3)")
+        await turn_context.send_activity(f"child: {turn_context.activity.text}")
diff --git a/samples/experimental/skills-buffered/child/config.py b/samples/experimental/skills-buffered/child/config.py
new file mode 100644
index 000000000..f21c1df0e
--- /dev/null
+++ b/samples/experimental/skills-buffered/child/config.py
@@ -0,0 +1,13 @@
+#!/usr/bin/env python3
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import os
+
+
+class DefaultConfig:
+    """ Bot Configuration """
+
+    PORT = 3979
+    APP_ID = os.environ.get("MicrosoftAppId", "")
+    APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "")
diff --git a/samples/experimental/skills-buffered/child/requirements.txt b/samples/experimental/skills-buffered/child/requirements.txt
new file mode 100644
index 000000000..20f8f8fe5
--- /dev/null
+++ b/samples/experimental/skills-buffered/child/requirements.txt
@@ -0,0 +1,2 @@
+botbuilder-core>=4.7.1
+aiohttp
diff --git a/samples/experimental/skills-buffered/parent/app.py b/samples/experimental/skills-buffered/parent/app.py
new file mode 100644
index 000000000..585a6873f
--- /dev/null
+++ b/samples/experimental/skills-buffered/parent/app.py
@@ -0,0 +1,100 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import sys
+import traceback
+
+from aiohttp import web
+from aiohttp.web import Request, Response
+from aiohttp.web_response import json_response
+from botbuilder.core import (
+    BotFrameworkAdapterSettings,
+    MemoryStorage,
+    TurnContext,
+    BotFrameworkAdapter,
+    BotFrameworkHttpClient,
+)
+from botbuilder.core.integration import (
+    aiohttp_channel_service_routes,
+    aiohttp_error_middleware,
+)
+from botbuilder.core.skills import SkillHandler
+from botbuilder.schema import Activity
+from botframework.connector.auth import (
+    AuthenticationConfiguration,
+    SimpleCredentialProvider,
+)
+
+from bots.parent_bot import ParentBot
+from skill_conversation_id_factory import SkillConversationIdFactory
+from config import DefaultConfig
+
+CONFIG = DefaultConfig()
+
+# Create adapter.
+# See https://aka.ms/about-bot-adapter to learn more about how bots work.
+SETTINGS = BotFrameworkAdapterSettings(
+    app_id=CONFIG.APP_ID, app_password=CONFIG.APP_PASSWORD,
+)
+ADAPTER = BotFrameworkAdapter(SETTINGS)
+
+CREDENTIAL_PROVIDER = SimpleCredentialProvider(CONFIG.APP_ID, CONFIG.APP_PASSWORD)
+CLIENT = BotFrameworkHttpClient(CREDENTIAL_PROVIDER)
+
+
+# Catch-all for errors.
+async def on_error(context: TurnContext, error: Exception):
+    # This check writes out errors to console log .vs. app insights.
+    # NOTE: In production environment, you should consider logging this to Azure
+    #       application insights.
+    print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
+    traceback.print_exc()
+
+    # Send a message to the user
+    await context.send_activity("The bot encountered an error or bug.")
+    await context.send_activity(
+        "To continue to run this bot, please fix the bot source code."
+    )
+
+
+ADAPTER.on_turn_error = on_error
+
+# Create the Bot
+BOT = ParentBot(CLIENT)
+
+STORAGE = MemoryStorage()
+ID_FACTORY = SkillConversationIdFactory(STORAGE)
+SKILL_HANDLER = SkillHandler(
+    ADAPTER, BOT, ID_FACTORY, CREDENTIAL_PROVIDER, AuthenticationConfiguration()
+)
+
+
+# Listen for incoming requests on /api/messages
+async def messages(req: Request) -> Response:
+    # Main bot message handler.
+    if "application/json" in req.headers["Content-Type"]:
+        body = await req.json()
+    else:
+        return Response(status=415)
+
+    activity = Activity().deserialize(body)
+    auth_header = req.headers["Authorization"] if "Authorization" in req.headers else ""
+
+    try:
+        response = await ADAPTER.process_activity(activity, auth_header, BOT.on_turn)
+        if response:
+            return json_response(data=response.body, status=response.status)
+        return Response(status=201)
+    except Exception as exception:
+        raise exception
+
+
+APP = web.Application(middlewares=[aiohttp_error_middleware])
+APP.router.add_post("/api/messages", messages)
+APP.router.add_routes(aiohttp_channel_service_routes(SKILL_HANDLER, "/api/skills"))
+
+if __name__ == "__main__":
+    try:
+        web.run_app(APP, host="localhost", port=CONFIG.PORT)
+    except Exception as error:
+        raise error
diff --git a/samples/experimental/skills-buffered/parent/bots/__init__.py b/samples/experimental/skills-buffered/parent/bots/__init__.py
new file mode 100644
index 000000000..01c37eaea
--- /dev/null
+++ b/samples/experimental/skills-buffered/parent/bots/__init__.py
@@ -0,0 +1,4 @@
+from .parent_bot import ParentBot
+
+
+__all__ = ["ParentBot"]
diff --git a/samples/experimental/skills-buffered/parent/bots/parent_bot.py b/samples/experimental/skills-buffered/parent/bots/parent_bot.py
new file mode 100644
index 000000000..91b85b654
--- /dev/null
+++ b/samples/experimental/skills-buffered/parent/bots/parent_bot.py
@@ -0,0 +1,43 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import uuid
+
+from botbuilder.core import (
+    ActivityHandler,
+    TurnContext,
+    BotFrameworkHttpClient,
+    MessageFactory,
+)
+
+from botbuilder.schema import DeliveryModes
+
+
+class ParentBot(ActivityHandler):
+    def __init__(
+        self, skill_client: BotFrameworkHttpClient,
+    ):
+        self.client = skill_client
+
+    async def on_message_activity(self, turn_context: TurnContext):
+        await turn_context.send_activity("parent: before child")
+
+        activity = MessageFactory.text("parent to child")
+        TurnContext.apply_conversation_reference(
+            activity, TurnContext.get_conversation_reference(turn_context.activity)
+        )
+        activity.delivery_mode = DeliveryModes.buffered_replies
+
+        activities = await self.client.post_buffered_activity(
+            None,
+            "toBotId",
+            "http://localhost:3979/api/messages",
+            "http://tempuri.org/whatever",
+            str(uuid.uuid4()),
+            activity,
+        )
+
+        if activities:
+            await turn_context.send_activities(activities)
+
+        await turn_context.send_activity("parent: after child")
diff --git a/samples/experimental/skills-buffered/parent/config.py b/samples/experimental/skills-buffered/parent/config.py
new file mode 100644
index 000000000..d66581d4c
--- /dev/null
+++ b/samples/experimental/skills-buffered/parent/config.py
@@ -0,0 +1,13 @@
+#!/usr/bin/env python3
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import os
+
+
+class DefaultConfig:
+    """ Bot Configuration """
+
+    PORT = 3978
+    APP_ID = os.environ.get("MicrosoftAppId", "")
+    APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "")
diff --git a/samples/experimental/skills-buffered/parent/requirements.txt b/samples/experimental/skills-buffered/parent/requirements.txt
new file mode 100644
index 000000000..20f8f8fe5
--- /dev/null
+++ b/samples/experimental/skills-buffered/parent/requirements.txt
@@ -0,0 +1,2 @@
+botbuilder-core>=4.7.1
+aiohttp
diff --git a/samples/experimental/skills-buffered/parent/skill_conversation_id_factory.py b/samples/experimental/skills-buffered/parent/skill_conversation_id_factory.py
new file mode 100644
index 000000000..8faaae025
--- /dev/null
+++ b/samples/experimental/skills-buffered/parent/skill_conversation_id_factory.py
@@ -0,0 +1,47 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from botbuilder.core import Storage
+from botbuilder.core.skills import ConversationIdFactoryBase
+from botbuilder.schema import ConversationReference
+
+
+class SkillConversationIdFactory(ConversationIdFactoryBase):
+    def __init__(self, storage: Storage):
+        if not storage:
+            raise TypeError("storage can't be None")
+
+        self._storage = storage
+
+    async def create_skill_conversation_id(
+        self, conversation_reference: ConversationReference
+    ) -> str:
+        if not conversation_reference:
+            raise TypeError("conversation_reference can't be None")
+
+        if not conversation_reference.conversation.id:
+            raise TypeError("conversation id in conversation reference can't be None")
+
+        if not conversation_reference.channel_id:
+            raise TypeError("channel id in conversation reference can't be None")
+
+        storage_key = f"{conversation_reference.channel_id}:{conversation_reference.conversation.id}"
+
+        skill_conversation_info = {storage_key: conversation_reference}
+
+        await self._storage.write(skill_conversation_info)
+
+        return storage_key
+
+    async def get_conversation_reference(
+        self, skill_conversation_id: str
+    ) -> ConversationReference:
+        if not skill_conversation_id:
+            raise TypeError("skill_conversation_id can't be None")
+
+        skill_conversation_info = await self._storage.read([skill_conversation_id])
+
+        return skill_conversation_info.get(skill_conversation_id)
+
+    async def delete_conversation_reference(self, skill_conversation_id: str):
+        await self._storage.delete([skill_conversation_id])