diff --git a/examples/canonical-metrics/.gitignore b/examples/canonical-metrics/.gitignore
new file mode 100644
index 000000000..50d9d205e
--- /dev/null
+++ b/examples/canonical-metrics/.gitignore
@@ -0,0 +1,161 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+recordings/
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+runpod.toml
diff --git a/examples/canonical-metrics/Dockerfile b/examples/canonical-metrics/Dockerfile
new file mode 100644
index 000000000..704080eec
--- /dev/null
+++ b/examples/canonical-metrics/Dockerfile
@@ -0,0 +1,16 @@
+FROM python:3.10-bullseye
+
+RUN mkdir /app
+RUN mkdir /app/assets
+RUN mkdir /app/utils
+COPY *.py /app/
+COPY requirements.txt /app/
+copy assets/* /app/assets/
+copy utils/* /app/utils/
+
+WORKDIR /app
+RUN pip3 install -r requirements.txt
+
+EXPOSE 7860
+
+CMD ["python3", "server.py"]
\ No newline at end of file
diff --git a/examples/canonical-metrics/README.md b/examples/canonical-metrics/README.md
new file mode 100644
index 000000000..13c0b31e0
--- /dev/null
+++ b/examples/canonical-metrics/README.md
@@ -0,0 +1,37 @@
+# Simple Chatbot
+
+<img src="image.png" width="420px">
+
+This app connects you to a chatbot powered by GPT-4, complete with animations generated by Stable Video Diffusion.
+
+See a video of it in action: https://x.com/kwindla/status/1778628911817183509
+
+And a quick video walkthrough of the code: https://www.loom.com/share/13df1967161f4d24ade054e7f8753416
+
+ℹ️ The first time, things might take extra time to get started since VAD (Voice Activity Detection) model needs to be downloaded.
+
+## Get started
+
+```python
+python3 -m venv venv
+source venv/bin/activate
+pip install -r requirements.txt
+
+cp env.example .env # and add your credentials
+
+```
+
+## Run the server
+
+```bash
+python server.py
+```
+
+Then, visit `http://localhost:7860/start` in your browser to start a chatbot session.
+
+## Build and test the Docker image
+
+```
+docker build -t chatbot .
+docker run --env-file .env -p 7860:7860 chatbot
+```
diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py
new file mode 100644
index 000000000..b6da3fbd4
--- /dev/null
+++ b/examples/canonical-metrics/bot.py
@@ -0,0 +1,146 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+import uuid
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.frames.frames import EndFrame, LLMMessagesFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
+from pipecat.processors.canonical_metrics_processor import CanonicalMetrics
+from pipecat.processors.user_marker_processor import UserMarkerProcessor
+from pipecat.services.elevenlabs import ElevenLabsTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Chatbot",
+            DailyParams(
+                audio_out_enabled=True,
+                audio_in_enabled=True,
+                camera_out_enabled=False,
+                vad_enabled=True,
+                vad_audio_passthrough=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                transcription_enabled=True,
+                #
+                # Spanish
+                #
+                # transcription_settings=DailyTranscriptionSettings(
+                #     language="es",
+                #     tier="nova",
+                #     model="2-general"
+                # )
+            )
+        )
+
+        tts = ElevenLabsTTSService(
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+
+            #
+            # English
+            #
+            voice_id="cgSgspJ2msm6clMCkdW9",
+            aiohttp_session=session,
+
+            #
+            # Spanish
+            #
+            # model="eleven_multilingual_v2",
+            # voice_id="gD1IexrzCvsXPHUuT0s3",
+        )
+
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                #
+                # English
+                #
+                "content": "You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by introducing yourself. Keep all your responses to 12 words or fewer.",
+
+                #
+                # Spanish
+                #
+                # "content": "Eres Chatbot, un amigable y útil robot. Tu objetivo es demostrar tus capacidades de una manera breve. Tus respuestas se convertiran a audio así que nunca no debes incluir caracteres especiales. Contesta a lo que el usuario pregunte de una manera creativa, útil y breve. Empieza por presentarte a ti mismo.",
+            },
+        ]
+
+        user_response = LLMUserResponseAggregator()
+        assistant_response = LLMAssistantResponseAggregator()
+
+        """
+        CanonicalMetrics uses AudioBufferProcessor under the hood to buffer the audio. On
+        call completion, CanonicalMetrics will send the audio buffer to Canonical for
+        analysis. Visit https://voice.canonical.chat to learn more.
+        """
+        canonical = CanonicalMetrics(
+            call_id=str(uuid.uuid4()),
+            assistant="pipecat-chatbot",
+            assistant_speaks_first=True,
+        )
+        usermarker = UserMarkerProcessor()
+        pipeline = Pipeline([
+            transport.input(),  # microphone
+            usermarker,  # used to mark the user's audio in the pipeline
+            user_response,
+            llm,
+            tts,
+            canonical,  # captures audio and uploads to Canonical AI for metrics
+            transport.output(),
+            assistant_response,
+        ])
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            transport.capture_participant_transcription(participant["id"])
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            print(f"Participant left: {participant}")
+            await task.queue_frame(EndFrame())
+
+        @transport.event_handler("on_call_state_updated")
+        async def on_call_state_updated(transport, state):
+            if state == "left":
+                await task.queue_frame(EndFrame())
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/examples/canonical-metrics/env.example b/examples/canonical-metrics/env.example
new file mode 100644
index 000000000..39f209372
--- /dev/null
+++ b/examples/canonical-metrics/env.example
@@ -0,0 +1,5 @@
+DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
+DAILY_API_KEY=7df...
+OPENAI_API_KEY=sk-PL...
+ELEVENLABS_API_KEY=aeb...
+CANONICAL_API_KEY=can...
\ No newline at end of file
diff --git a/examples/canonical-metrics/requirements.txt b/examples/canonical-metrics/requirements.txt
new file mode 100644
index 000000000..7e53edc6b
--- /dev/null
+++ b/examples/canonical-metrics/requirements.txt
@@ -0,0 +1,5 @@
+python-dotenv
+fastapi[all]
+uvicorn
+pipecat-ai[daily,openai,silero,elevenlabs,canonical]
+
diff --git a/examples/canonical-metrics/runner.py b/examples/canonical-metrics/runner.py
new file mode 100644
index 000000000..7507d28d6
--- /dev/null
+++ b/examples/canonical-metrics/runner.py
@@ -0,0 +1,54 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import aiohttp
+import argparse
+import os
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+
+
+async def configure(aiohttp_session: aiohttp.ClientSession):
+    parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u",
+        "--url",
+        type=str,
+        required=False,
+        help="URL of the Daily room to join")
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL.")
+
+    if not key:
+        raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
+
+    daily_rest_helper = DailyRESTHelper(
+        daily_api_key=key,
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session
+    )
+
+    # Create a meeting token for the given room with an expiration 1 hour in
+    # the future.
+    expiry_time: float = 60 * 60
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token)
diff --git a/examples/canonical-metrics/server.py b/examples/canonical-metrics/server.py
new file mode 100644
index 000000000..2c717ffa2
--- /dev/null
+++ b/examples/canonical-metrics/server.py
@@ -0,0 +1,147 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+import subprocess
+from contextlib import asynccontextmanager
+
+import aiohttp
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, RedirectResponse
+
+from pipecat.transports.services.helpers.daily_rest import (DailyRESTHelper,
+                                                            DailyRoomParams)
+
+MAX_BOTS_PER_ROOM = 1
+
+# Bot sub-process dict for status reporting and concurrency control
+bot_procs = {}
+
+daily_helpers = {}
+
+load_dotenv(override=True)
+
+
+def cleanup():
+    # Clean up function, just to be extra safe
+    for entry in bot_procs.values():
+        proc = entry[0]
+        proc.terminate()
+        proc.wait()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    aiohttp_session = aiohttp.ClientSession()
+    daily_helpers["rest"] = DailyRESTHelper(
+        daily_api_key=os.getenv("DAILY_API_KEY", ""),
+        daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
+        aiohttp_session=aiohttp_session
+    )
+    yield
+    await aiohttp_session.close()
+    cleanup()
+
+app = FastAPI(lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/start")
+async def start_agent(request: Request):
+    print(f"!!! Creating room")
+    room = await daily_helpers["rest"].create_room(DailyRoomParams())
+    print(f"!!! Room URL: {room.url}")
+    # Ensure the room property is present
+    if not room.url:
+        raise HTTPException(
+            status_code=500,
+            detail="Missing 'room' property in request data. Cannot start agent without a target room!")
+
+    # Check if there is already an existing process running in this room
+    num_bots_in_room = sum(
+        1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
+    if num_bots_in_room >= MAX_BOTS_PER_ROOM:
+        raise HTTPException(
+            status_code=500, detail=f"Max bot limited reach for room: {room.url}")
+
+    # Get the token for the room
+    token = await daily_helpers["rest"].get_token(room.url)
+
+    if not token:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to get token for room: {room.url}")
+
+    # Spawn a new agent, and join the user session
+    # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
+    try:
+        proc = subprocess.Popen(
+            [
+                f"python3 -m bot -u {room.url} -t {token}"
+            ],
+            shell=True,
+            bufsize=1,
+            cwd=os.path.dirname(os.path.abspath(__file__))
+        )
+        bot_procs[proc.pid] = (proc, room.url)
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to start subprocess: {e}")
+
+    return RedirectResponse(room.url)
+
+
+@app.get("/status/{pid}")
+def get_status(pid: int):
+    # Look up the subprocess
+    proc = bot_procs.get(pid)
+
+    # If the subprocess doesn't exist, return an error
+    if not proc:
+        raise HTTPException(
+            status_code=404, detail=f"Bot with process id: {pid} not found")
+
+    # Check the status of the subprocess
+    if proc[0].poll() is None:
+        status = "running"
+    else:
+        status = "finished"
+
+    return JSONResponse({"bot_id": pid, "status": status})
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    default_host = os.getenv("HOST", "0.0.0.0")
+    default_port = int(os.getenv("FAST_API_PORT", "7860"))
+
+    parser = argparse.ArgumentParser(
+        description="Daily Storyteller FastAPI server")
+    parser.add_argument("--host", type=str,
+                        default=default_host, help="Host address")
+    parser.add_argument("--port", type=int,
+                        default=default_port, help="Port number")
+    parser.add_argument("--reload", action="store_true",
+                        help="Reload code on change")
+
+    config = parser.parse_args()
+
+    uvicorn.run(
+        "server:app",
+        host=config.host,
+        port=config.port,
+        reload=config.reload,
+    )
diff --git a/examples/chatbot-audio-recording/.gitignore b/examples/chatbot-audio-recording/.gitignore
new file mode 100644
index 000000000..2bc1403d1
--- /dev/null
+++ b/examples/chatbot-audio-recording/.gitignore
@@ -0,0 +1,161 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+runpod.toml
diff --git a/examples/chatbot-audio-recording/Dockerfile b/examples/chatbot-audio-recording/Dockerfile
new file mode 100644
index 000000000..704080eec
--- /dev/null
+++ b/examples/chatbot-audio-recording/Dockerfile
@@ -0,0 +1,16 @@
+FROM python:3.10-bullseye
+
+RUN mkdir /app
+RUN mkdir /app/assets
+RUN mkdir /app/utils
+COPY *.py /app/
+COPY requirements.txt /app/
+copy assets/* /app/assets/
+copy utils/* /app/utils/
+
+WORKDIR /app
+RUN pip3 install -r requirements.txt
+
+EXPOSE 7860
+
+CMD ["python3", "server.py"]
\ No newline at end of file
diff --git a/examples/chatbot-audio-recording/README.md b/examples/chatbot-audio-recording/README.md
new file mode 100644
index 000000000..13c0b31e0
--- /dev/null
+++ b/examples/chatbot-audio-recording/README.md
@@ -0,0 +1,37 @@
+# Simple Chatbot
+
+<img src="image.png" width="420px">
+
+This app connects you to a chatbot powered by GPT-4, complete with animations generated by Stable Video Diffusion.
+
+See a video of it in action: https://x.com/kwindla/status/1778628911817183509
+
+And a quick video walkthrough of the code: https://www.loom.com/share/13df1967161f4d24ade054e7f8753416
+
+ℹ️ The first time, things might take extra time to get started since VAD (Voice Activity Detection) model needs to be downloaded.
+
+## Get started
+
+```python
+python3 -m venv venv
+source venv/bin/activate
+pip install -r requirements.txt
+
+cp env.example .env # and add your credentials
+
+```
+
+## Run the server
+
+```bash
+python server.py
+```
+
+Then, visit `http://localhost:7860/start` in your browser to start a chatbot session.
+
+## Build and test the Docker image
+
+```
+docker build -t chatbot .
+docker run --env-file .env -p 7860:7860 chatbot
+```
diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py
new file mode 100644
index 000000000..7297de215
--- /dev/null
+++ b/examples/chatbot-audio-recording/bot.py
@@ -0,0 +1,136 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.frames.frames import EndFrame, LLMMessagesFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
+from pipecat.processors.audio_buffer_processor import AudioBufferProcessor
+from pipecat.processors.user_marker_processor import UserMarkerProcessor
+from pipecat.services.elevenlabs import ElevenLabsTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Chatbot",
+            DailyParams(
+                audio_out_enabled=True,
+                audio_in_enabled=True,
+                camera_out_enabled=False,
+                vad_enabled=True,
+                vad_audio_passthrough=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                transcription_enabled=True,
+                #
+                # Spanish
+                #
+                # transcription_settings=DailyTranscriptionSettings(
+                #     language="es",
+                #     tier="nova",
+                #     model="2-general"
+                # )
+            )
+        )
+
+        tts = ElevenLabsTTSService(
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+
+            #
+            # English
+            #
+            voice_id="cgSgspJ2msm6clMCkdW9",
+            aiohttp_session=session,
+
+            #
+            # Spanish
+            #
+            # model="eleven_multilingual_v2",
+            # voice_id="gD1IexrzCvsXPHUuT0s3",
+        )
+
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                #
+                # English
+                #
+                "content": "You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by introducing yourself. Keep all your response to 12 words or fewer.",
+
+                #
+                # Spanish
+                #
+                # "content": "Eres Chatbot, un amigable y útil robot. Tu objetivo es demostrar tus capacidades de una manera breve. Tus respuestas se convertiran a audio así que nunca no debes incluir caracteres especiales. Contesta a lo que el usuario pregunte de una manera creativa, útil y breve. Empieza por presentarte a ti mismo.",
+            },
+        ]
+
+        user_response = LLMUserResponseAggregator()
+        assistant_response = LLMAssistantResponseAggregator()
+
+        audiobuffer = AudioBufferProcessor()
+        usermarker = UserMarkerProcessor()
+        pipeline = Pipeline([
+            transport.input(),  # microphone
+            usermarker,  # used to mark the user's audio in the pipeline
+            user_response,
+            llm,
+            tts,
+            audiobuffer,  # used to buffer the audio in the pipeline
+            transport.output(),
+            assistant_response,
+        ])
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            transport.capture_participant_transcription(participant["id"])
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            print(f"Participant left: {participant}")
+            await task.queue_frame(EndFrame())
+
+        @transport.event_handler("on_call_state_updated")
+        async def on_call_state_updated(transport, state):
+            if state == "left":
+                await task.queue_frame(EndFrame())
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/examples/chatbot-audio-recording/env.example b/examples/chatbot-audio-recording/env.example
new file mode 100644
index 000000000..d368ae510
--- /dev/null
+++ b/examples/chatbot-audio-recording/env.example
@@ -0,0 +1,4 @@
+DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
+DAILY_API_KEY=7df...
+OPENAI_API_KEY=sk-PL...
+ELEVENLABS_API_KEY=aeb...
\ No newline at end of file
diff --git a/examples/chatbot-audio-recording/requirements.txt b/examples/chatbot-audio-recording/requirements.txt
new file mode 100644
index 000000000..9786b52de
--- /dev/null
+++ b/examples/chatbot-audio-recording/requirements.txt
@@ -0,0 +1,4 @@
+python-dotenv
+fastapi[all]
+uvicorn
+pipecat-ai[daily,openai,silero,elevenlabs]
diff --git a/examples/chatbot-audio-recording/runner.py b/examples/chatbot-audio-recording/runner.py
new file mode 100644
index 000000000..7507d28d6
--- /dev/null
+++ b/examples/chatbot-audio-recording/runner.py
@@ -0,0 +1,54 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import aiohttp
+import argparse
+import os
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+
+
+async def configure(aiohttp_session: aiohttp.ClientSession):
+    parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u",
+        "--url",
+        type=str,
+        required=False,
+        help="URL of the Daily room to join")
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL.")
+
+    if not key:
+        raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
+
+    daily_rest_helper = DailyRESTHelper(
+        daily_api_key=key,
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session
+    )
+
+    # Create a meeting token for the given room with an expiration 1 hour in
+    # the future.
+    expiry_time: float = 60 * 60
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token)
diff --git a/examples/chatbot-audio-recording/server.py b/examples/chatbot-audio-recording/server.py
new file mode 100644
index 000000000..2c717ffa2
--- /dev/null
+++ b/examples/chatbot-audio-recording/server.py
@@ -0,0 +1,147 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+import subprocess
+from contextlib import asynccontextmanager
+
+import aiohttp
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, RedirectResponse
+
+from pipecat.transports.services.helpers.daily_rest import (DailyRESTHelper,
+                                                            DailyRoomParams)
+
+MAX_BOTS_PER_ROOM = 1
+
+# Bot sub-process dict for status reporting and concurrency control
+bot_procs = {}
+
+daily_helpers = {}
+
+load_dotenv(override=True)
+
+
+def cleanup():
+    # Clean up function, just to be extra safe
+    for entry in bot_procs.values():
+        proc = entry[0]
+        proc.terminate()
+        proc.wait()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    aiohttp_session = aiohttp.ClientSession()
+    daily_helpers["rest"] = DailyRESTHelper(
+        daily_api_key=os.getenv("DAILY_API_KEY", ""),
+        daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
+        aiohttp_session=aiohttp_session
+    )
+    yield
+    await aiohttp_session.close()
+    cleanup()
+
+app = FastAPI(lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/start")
+async def start_agent(request: Request):
+    print(f"!!! Creating room")
+    room = await daily_helpers["rest"].create_room(DailyRoomParams())
+    print(f"!!! Room URL: {room.url}")
+    # Ensure the room property is present
+    if not room.url:
+        raise HTTPException(
+            status_code=500,
+            detail="Missing 'room' property in request data. Cannot start agent without a target room!")
+
+    # Check if there is already an existing process running in this room
+    num_bots_in_room = sum(
+        1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
+    if num_bots_in_room >= MAX_BOTS_PER_ROOM:
+        raise HTTPException(
+            status_code=500, detail=f"Max bot limited reach for room: {room.url}")
+
+    # Get the token for the room
+    token = await daily_helpers["rest"].get_token(room.url)
+
+    if not token:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to get token for room: {room.url}")
+
+    # Spawn a new agent, and join the user session
+    # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
+    try:
+        proc = subprocess.Popen(
+            [
+                f"python3 -m bot -u {room.url} -t {token}"
+            ],
+            shell=True,
+            bufsize=1,
+            cwd=os.path.dirname(os.path.abspath(__file__))
+        )
+        bot_procs[proc.pid] = (proc, room.url)
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to start subprocess: {e}")
+
+    return RedirectResponse(room.url)
+
+
+@app.get("/status/{pid}")
+def get_status(pid: int):
+    # Look up the subprocess
+    proc = bot_procs.get(pid)
+
+    # If the subprocess doesn't exist, return an error
+    if not proc:
+        raise HTTPException(
+            status_code=404, detail=f"Bot with process id: {pid} not found")
+
+    # Check the status of the subprocess
+    if proc[0].poll() is None:
+        status = "running"
+    else:
+        status = "finished"
+
+    return JSONResponse({"bot_id": pid, "status": status})
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    default_host = os.getenv("HOST", "0.0.0.0")
+    default_port = int(os.getenv("FAST_API_PORT", "7860"))
+
+    parser = argparse.ArgumentParser(
+        description="Daily Storyteller FastAPI server")
+    parser.add_argument("--host", type=str,
+                        default=default_host, help="Host address")
+    parser.add_argument("--port", type=int,
+                        default=default_port, help="Port number")
+    parser.add_argument("--reload", action="store_true",
+                        help="Reload code on change")
+
+    config = parser.parse_args()
+
+    uvicorn.run(
+        "server:app",
+        host=config.host,
+        port=config.port,
+        reload=config.reload,
+    )
diff --git a/examples/simple-chatbot/requirements.txt b/examples/simple-chatbot/requirements.txt
index a7a8729df..9786b52de 100644
--- a/examples/simple-chatbot/requirements.txt
+++ b/examples/simple-chatbot/requirements.txt
@@ -1,4 +1,4 @@
 python-dotenv
 fastapi[all]
 uvicorn
-pipecat-ai[daily,openai,silero]
+pipecat-ai[daily,openai,silero,elevenlabs]
diff --git a/pyproject.toml b/pyproject.toml
index 73c643ddc..41e1af149 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -36,6 +36,7 @@ Website = "https://pipecat.ai"
 [project.optional-dependencies]
 anthropic = [ "anthropic~=0.34.0" ]
 azure = [ "azure-cognitiveservices-speech~=1.40.0" ]
+canonical = [ "aiofiles~=24.1.0" ]
 cartesia = [ "websockets~=12.0" ]
 daily = [ "daily-python~=0.10.1" ]
 deepgram = [ "deepgram-sdk~=3.5.0" ]
diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py
index 13c2f53f1..4eee87cc9 100644
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -4,9 +4,9 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from typing import Any, List, Mapping, Optional, Tuple
-
+import time
 from dataclasses import dataclass, field
+from typing import Any, List, Mapping, Optional, Tuple
 
 from pipecat.transcriptions.language import Language
 from pipecat.utils.utils import obj_count, obj_id
@@ -223,6 +223,16 @@ class TransportMessageFrame(DataFrame):
 class AppFrame(Frame):
     pass
 
+
+@dataclass
+class UserAudioFrame(AudioRawFrame):
+    """
+    Indicates user audio in the pipeline.
+    """
+
+    def __init__(self, frame: AudioRawFrame):
+        super().__init__(frame.audio, frame.sample_rate, frame.num_channels)
+
 #
 # System frames
 #
diff --git a/src/pipecat/processors/audio_buffer_processor.py b/src/pipecat/processors/audio_buffer_processor.py
new file mode 100644
index 000000000..f724e7c5e
--- /dev/null
+++ b/src/pipecat/processors/audio_buffer_processor.py
@@ -0,0 +1,75 @@
+from pipecat.frames.frames import (AudioRawFrame, BotStartedSpeakingFrame,
+                                   BotStoppedSpeakingFrame, Frame,
+                                   UserAudioFrame, UserStoppedSpeakingFrame)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+class AudioBufferProcessor(FrameProcessor):
+
+    def __init__(self):
+        """
+        Initialize the AudioBufferProcessor.
+
+        This constructor sets up the initial state for audio processing:
+        - audio_buffer: A bytearray to store incoming audio data.
+        - num_channels: The number of audio channels (initialized as None).
+        - sample_rate: The sample rate of the audio (initialized as None).
+        - assistant_audio: A boolean flag to indicate if assistant audio is being processed.
+        - user_audio: A boolean flag to indicate if user audio is being processed.
+
+        The num_channels and sample_rate are set to None initially and will be
+        populated when the first audio frame is processed.
+        """
+        super().__init__()
+        self.audio_buffer = bytearray()
+        self.num_channels = None
+        self.sample_rate = None
+        self.assistant_audio = False
+        self.user_audio = False
+
+    def has_audio(self):
+        return (
+            self.audio_buffer and
+            len(self.audio_buffer) > 0 and
+            self.num_channels and
+            self.sample_rate
+        )
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, AudioRawFrame) or isinstance(frame, UserAudioFrame):
+            if self.num_channels is None:
+                self.num_channels = frame.num_channels
+            if self.sample_rate is None:
+                self.sample_rate = frame.sample_rate
+
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            self.user_audio = False
+
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self.assistant_audio = True
+            self.user_audio = False  # do not capture user audio if assistant is speaking
+        if isinstance(frame, BotStoppedSpeakingFrame):
+            self.assistant_audio = False
+            # Capture user audio if assistant is not speaking, even if it's silence, the point
+            # here is to capture so that the conversation is as close to reality as possible.
+            # This is important for evaluation and metrics capture.
+            self.user_audio = True
+
+        # only include audio from the user if the user is speaking, this is because audio from the user's
+        # mic is always coming in. if we include all the user's audio there will be a long latency before
+        # the user starts speaking because all of the user's silence during the assistant's speech will have been
+        # added to the buffer.
+        if isinstance(frame, UserAudioFrame) and self.user_audio:
+            self.audio_buffer.extend(frame.audio)
+
+        # include all audio from the assistant
+        if (
+            isinstance(frame, AudioRawFrame)
+            and not isinstance(frame, UserAudioFrame)
+        ):
+            self.audio_buffer.extend(frame.audio)
+
+        # do not push the user's audio frame, doing so will result in echo
+        if not isinstance(frame, UserAudioFrame):
+            await self.push_frame(frame, direction)
diff --git a/src/pipecat/processors/canonical_metrics_processor.py b/src/pipecat/processors/canonical_metrics_processor.py
new file mode 100644
index 000000000..e7c0afb61
--- /dev/null
+++ b/src/pipecat/processors/canonical_metrics_processor.py
@@ -0,0 +1,202 @@
+import os
+import uuid
+import wave
+from datetime import datetime
+from io import BytesIO
+from typing import Dict, List, Tuple
+
+import aiohttp
+from loguru import logger
+
+try:
+    import aiofiles
+    import aiofiles.os
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use Canonical Metrics, you need to `pip install pipecat-ai[canonical]`. " +
+        "Also, set the `CANONICAL_API_KEY` environment variable.")
+    raise Exception(f"Missing module: {e}")
+
+
+from pipecat.frames.frames import CancelFrame, EndFrame, Frame
+from pipecat.processors.audio_buffer_processor import AudioBufferProcessor
+from pipecat.processors.frame_processor import FrameDirection
+
+"""
+This class extends AudioBufferProcessor to handle audio processing and uploading
+for the Canonical Voice API.
+"""
+
+
+class CanonicalMetrics(AudioBufferProcessor):
+    """
+    Initialize a CanonicalAudioProcessor instance.
+
+    This class extends AudioBufferProcessor to handle audio processing and uploading
+    for the Canonical Voice API.
+
+    Args:
+        call_id (str): Your unique identifier for the call. This is used to match the call in the Canonical Voice system to the call in your system.
+        assistant (str): Identifier for the AI assistant. This can be whatever you want, it's intended for you convenience so you can distinguish
+        between different assistants and a grouping mechanism for calls.
+        assistant_speaks_first (bool, optional): Indicates if the assistant speaks first in the conversation. Defaults to True.
+        output_dir (str, optional): Directory to save temporary audio files. Defaults to "recordings".
+        default_part_size (int, optional): Default size for multipart upload parts in bytes. Defaults to 1MB (1024 * 1024 * 1).
+
+    Attributes:
+        call_id (str): Stores the unique call identifier.
+        assistant (str): Stores the assistant identifier.
+        assistant_speaks_first (bool): Indicates whether the assistant speaks first.
+        output_dir (str): Directory path for saving temporary audio files.
+        partsize (int): Size of each part for multipart uploads.
+
+    The constructor also ensures that the output directory exists.
+    This class requires a Canonical API key to be set in the CANONICAL_API_KEY environment variable.
+    """
+
+    def __init__(
+            self,
+            call_id: str,
+            assistant: str,
+            assistant_speaks_first: bool = True,
+            output_dir: str = "recordings",
+            default_part_size: int = 1024 * 1024 * 1):
+        super().__init__()
+        if not os.environ.get("CANONICAL_API_KEY"):
+            raise ValueError(
+                "CANONICAL_API_KEY is not set, a Canonical API key is required to use this class")
+        self.call_id = call_id
+        self.assistant = assistant
+        self.assistant_speaks_first = assistant_speaks_first
+        self.output_dir = output_dir
+        self.partsize = default_part_size
+        self.end_of_call = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if self.end_of_call:
+            return
+
+        if (isinstance(frame, EndFrame) or isinstance(frame, CancelFrame)):
+            self.end_of_call = True
+            if self.has_audio():
+                os.makedirs(self.output_dir, exist_ok=True)
+                filename = self.get_output_filename()
+                with BytesIO() as buffer:
+                    with wave.open(buffer, 'wb') as wf:
+                        wf.setnchannels(self.num_channels)
+                        wf.setsampwidth(self.sample_rate // 8000)
+                        wf.setframerate(self.sample_rate)
+                        wf.writeframes(self.audio_buffer)
+                    wave_data = buffer.getvalue()
+
+                async with aiofiles.open(filename, 'wb') as file:
+                    await file.write(wave_data)
+
+                try:
+                    await self.multipart_upload(filename)
+                    await aiofiles.os.remove(filename)
+                except FileNotFoundError:
+                    pass
+                except Exception as e:
+                    raise e
+                self.audio_buffer = bytearray()
+
+    def get_output_filename(self):
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        return f"{self.output_dir}/{timestamp}-{uuid.uuid4().hex}.wav"
+
+    def canonical_api_url(self):
+        return os.environ.get("CANONICAL_API_URL", "https://voiceapp.canonical.chat/api/v1")
+
+    def request_headers(self):
+        return {
+            "Content-Type": "application/json",
+            "X-Canonical-Api-Key": os.environ.get("CANONICAL_API_KEY")
+        }
+
+    async def multipart_upload(self, file_path: str):
+        upload_request, upload_response = await self.request_upload(file_path)
+        parts = await self.upload_parts(file_path, upload_request, upload_response)
+        await self.upload_complete(parts, upload_request, upload_response)
+
+    async def request_upload(self, file_path: str) -> Tuple[Dict, Dict]:
+        filename = os.path.basename(file_path)
+        filename = f"{str(uuid.uuid4())}-{filename}"
+        filesize = os.path.getsize(file_path)
+        numparts = int((filesize + self.partsize - 1) / self.partsize)
+
+        params = {
+            'filename': filename,
+            'parts': numparts,
+            'assistant': self.assistant,
+            'assistantSpeaksFirst': self.assistant_speaks_first
+        }
+        print(f"Requesting presigned URLs for {numparts} parts")
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                f"{self.canonical_api_url()}/recording/uploadRequest",
+                headers=self.request_headers(),
+                json=params
+            ) as response:
+                if not response.ok:
+                    raise Exception(f"Failed to get presigned URLs: {await response.text()}")
+                response_json = await response.json()
+        return params, response_json
+
+    async def upload_parts(
+            self,
+            file_path: str,
+            upload_request: Dict,
+            upload_response: Dict) -> List[Dict]:
+
+        urls = upload_response['urls']
+        parts = []
+        try:
+            async with aiofiles.open(file_path, 'rb') as file:
+                async with aiohttp.ClientSession() as session:
+                    for partnum, upload_url in enumerate(urls, start=1):
+                        data = await file.read(self.partsize)
+                        if not data:
+                            break
+
+                        async with session.put(upload_url, data=data) as response:
+                            if not response.ok:
+                                logger.error(f"Failed to upload part {partnum}: {await response.text()}")
+                                raise Exception(f"Failed to upload part {partnum}: {await response.text()}")
+
+                            etag = response.headers['ETag']
+                            parts.append({'partnum': str(partnum), 'etag': etag})
+
+        except Exception as e:
+            logger.error(f"Multipart upload aborted, an error occurred: {str(e)}")
+        return parts
+
+    async def upload_complete(
+            self,
+            parts: List[Dict],
+            upload_request: Dict,
+            upload_response: Dict):
+
+        params = {
+            'filename': upload_request['filename'],
+            'parts': parts,
+            'slug': upload_response['slug'],
+            'callId': self.call_id,
+            'assistant': {
+                'id': self.assistant,
+                'speaksFirst': self.assistant_speaks_first
+            }
+        }
+        print(f"Completing upload for {params['filename']}")
+        print(f"Slug: {params['slug']}")
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                f"{self.canonical_api_url()}/recording/uploadComplete",
+                headers=self.request_headers(),
+                json=params
+            ) as response:
+                if not response.ok:
+                    logger.error(f"Failed to complete upload: {await response.text()}")
+                    raise Exception(f"Failed to complete upload: {await response.text()}")
diff --git a/src/pipecat/processors/user_marker_processor.py b/src/pipecat/processors/user_marker_processor.py
new file mode 100644
index 000000000..cadbfffeb
--- /dev/null
+++ b/src/pipecat/processors/user_marker_processor.py
@@ -0,0 +1,19 @@
+from pipecat.frames.frames import AudioRawFrame, Frame, UserAudioFrame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+class UserMarkerProcessor(FrameProcessor):
+    """
+    This class extends FrameProcessor, used to mark the user's audio in the pipeline.
+    This FrameProcessor must be inserted after transport.input() so that the only
+    AudioRaw it receives are from the user.
+    """
+
+    def __init__(self):
+        super().__init__()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, AudioRawFrame):
+            frame = UserAudioFrame(frame)
+        await self.push_frame(frame, direction)