diff --git a/CHANGELOG.md b/CHANGELOG.md
index a10827389..ffd72c9f7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -20,6 +20,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Added new `AzureSTTService`. This allows you to use Azure Speech-To-Text.
 
+### Performance
+
+- Convert `BaseOutputTransport` and `BaseOutputTransport` to fully use asyncio
+  and remove the use of threads.
+
 ### Other
 
 - Added `twilio-chatbot`. This is an example that shows how to integrate Twilio
diff --git a/dev-requirements.txt b/dev-requirements.txt
index 2d7da9ca6..dc9190db2 100644
--- a/dev-requirements.txt
+++ b/dev-requirements.txt
@@ -2,6 +2,7 @@ autopep8~=2.1.0
 build~=1.2.1
 grpcio-tools~=1.62.2
 pip-tools~=7.4.1
+pyright~=1.1.367
 pytest~=8.2.0
 setuptools~=69.5.1
 setuptools_scm~=8.1.0
diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index cf9a9404e..ac3e1f618 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -71,6 +71,8 @@ class PipelineTask:
         await self._source.process_frame(CancelFrame(), FrameDirection.DOWNSTREAM)
         self._process_down_task.cancel()
         self._process_up_task.cancel()
+        await self._process_down_task
+        await self._process_up_task
 
     async def run(self):
         self._process_up_task = asyncio.create_task(self._process_up_queue())
@@ -122,6 +124,7 @@ class PipelineTask:
             await self._pipeline.cleanup()
         # We just enqueue None to terminate the task gracefully.
         self._process_up_task.cancel()
+        await self._process_up_task
 
     async def _process_up_queue(self):
         while True:
diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py
index e9423d8f1..85ac49ae3 100644
--- a/src/pipecat/services/azure.py
+++ b/src/pipecat/services/azure.py
@@ -146,10 +146,11 @@ class AzureSTTService(AIService):
     async def cancel(self, frame: CancelFrame):
         self._speech_recognizer.stop_continuous_recognition_async()
         self._push_frame_task.cancel()
+        await self._push_frame_task
 
     def _create_push_task(self):
-        self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
         self._push_queue = asyncio.Queue()
+        self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
 
     async def _push_frame_task_handler(self):
         running = True
diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py
index 810d7518d..3a76574ad 100644
--- a/src/pipecat/services/deepgram.py
+++ b/src/pipecat/services/deepgram.py
@@ -134,10 +134,11 @@ class DeepgramSTTService(AIService):
     async def cancel(self, frame: CancelFrame):
         await self._connection.finish()
         self._push_frame_task.cancel()
+        await self._push_frame_task
 
     def _create_push_task(self):
-        self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
         self._push_queue = asyncio.Queue()
+        self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
 
     async def _push_frame_task_handler(self):
         running = True
diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py
index d7e3c1c9d..cb7be5d55 100644
--- a/src/pipecat/transports/base_input.py
+++ b/src/pipecat/transports/base_input.py
@@ -5,7 +5,6 @@
 #
 
 import asyncio
-import queue
 
 from concurrent.futures import ThreadPoolExecutor
 
@@ -33,8 +32,6 @@ class BaseInputTransport(FrameProcessor):
 
         self._params = params
 
-        self._running = False
-
         self._executor = ThreadPoolExecutor(max_workers=5)
 
         # Create push frame task. This is the task that will push frames in
@@ -42,34 +39,21 @@ class BaseInputTransport(FrameProcessor):
         self._create_push_task()
 
     async def start(self, frame: StartFrame):
-        if self._running:
-            return
-
-        self._running = True
-
-        # Create audio input queue and thread if needed.
+        # Create audio input queue and task if needed.
         if self._params.audio_in_enabled or self._params.vad_enabled:
-            self._audio_in_queue = queue.Queue()
-            self._audio_thread = self._loop.run_in_executor(
-                self._executor, self._audio_thread_handler)
+            self._audio_in_queue = asyncio.Queue()
+            self._audio_task = self.get_event_loop().create_task(self._audio_task_handler())
 
     async def stop(self):
-        if not self._running:
-            return
-
-        # This will exit all threads.
-        self._running = False
-
-        # Wait for the threads to finish.
+        # Wait for the task to finish.
         if self._params.audio_in_enabled or self._params.vad_enabled:
-            await self._audio_thread
-
-        self._push_frame_task.cancel()
+            self._audio_task.cancel()
+            await self._audio_task
 
     def vad_analyzer(self) -> VADAnalyzer | None:
         return self._params.vad_analyzer
 
-    def push_audio_frame(self, frame: AudioRawFrame):
+    async def push_audio_frame(self, frame: AudioRawFrame):
         if self._params.audio_in_enabled or self._params.vad_enabled:
             self._audio_in_queue.put_nowait(frame)
 
@@ -78,7 +62,8 @@ class BaseInputTransport(FrameProcessor):
     #
 
     async def cleanup(self):
-        pass
+        self._push_frame_task.cancel()
+        await self._push_frame_task
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
@@ -102,8 +87,8 @@ class BaseInputTransport(FrameProcessor):
 
     def _create_push_task(self):
         loop = self.get_event_loop()
-        self._push_frame_task = loop.create_task(self._push_frame_task_handler())
         self._push_queue = asyncio.Queue()
+        self._push_frame_task = loop.create_task(self._push_frame_task_handler())
 
     async def _internal_push_frame(
             self,
@@ -129,6 +114,7 @@ class BaseInputTransport(FrameProcessor):
             if isinstance(frame, UserStartedSpeakingFrame):
                 logger.debug("User started speaking")
                 self._push_frame_task.cancel()
+                await self._push_frame_task
                 self._create_push_task()
                 await self.push_frame(StartInterruptionFrame())
             elif isinstance(frame, UserStoppedSpeakingFrame):
@@ -140,15 +126,16 @@ class BaseInputTransport(FrameProcessor):
     # Audio input
     #
 
-    def _vad_analyze(self, audio_frames: bytes) -> VADState:
+    async def _vad_analyze(self, audio_frames: bytes) -> VADState:
         state = VADState.QUIET
         vad_analyzer = self.vad_analyzer()
         if vad_analyzer:
-            state = vad_analyzer.analyze_audio(audio_frames)
+            state = await self.get_event_loop().run_in_executor(
+                self._executor, vad_analyzer.analyze_audio, audio_frames)
         return state
 
-    def _handle_vad(self, audio_frames: bytes, vad_state: VADState):
-        new_vad_state = self._vad_analyze(audio_frames)
+    async def _handle_vad(self, audio_frames: bytes, vad_state: VADState):
+        new_vad_state = await self._vad_analyze(audio_frames)
         if new_vad_state != vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING:
             frame = None
             if new_vad_state == VADState.SPEAKING:
@@ -157,33 +144,29 @@ class BaseInputTransport(FrameProcessor):
                 frame = UserStoppedSpeakingFrame()
 
             if frame:
-                future = asyncio.run_coroutine_threadsafe(
-                    self._handle_interruptions(frame), self.get_event_loop())
-                future.result()
+                await self._handle_interruptions(frame)
 
             vad_state = new_vad_state
         return vad_state
 
-    def _audio_thread_handler(self):
+    async def _audio_task_handler(self):
         vad_state: VADState = VADState.QUIET
-        while self._running:
+        while True:
             try:
-                frame: AudioRawFrame = self._audio_in_queue.get(timeout=1)
+                frame: AudioRawFrame = await self._audio_in_queue.get()
 
                 audio_passthrough = True
 
                 # Check VAD and push event if necessary. We just care about
                 # changes from QUIET to SPEAKING and vice versa.
                 if self._params.vad_enabled:
-                    vad_state = self._handle_vad(frame.audio, vad_state)
+                    vad_state = await self._handle_vad(frame.audio, vad_state)
                     audio_passthrough = self._params.vad_audio_passthrough
 
                 # Push audio downstream if passthrough.
                 if audio_passthrough:
-                    future = asyncio.run_coroutine_threadsafe(
-                        self._internal_push_frame(frame), self._loop)
-                    future.result()
-            except queue.Empty:
-                pass
+                    await self._internal_push_frame(frame)
+            except asyncio.CancelledError:
+                break
             except BaseException as e:
                 logger.error(f"{self} error reading audio frames: {e}")
diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py
index 2a1461563..f3c298b4c 100644
--- a/src/pipecat/transports/base_output.py
+++ b/src/pipecat/transports/base_output.py
@@ -7,11 +7,6 @@
 
 import asyncio
 import itertools
-import queue
-import time
-import threading
-
-from concurrent.futures import ThreadPoolExecutor
 
 from PIL import Image
 from typing import List
@@ -42,67 +37,51 @@ class BaseOutputTransport(FrameProcessor):
 
         self._params = params
 
-        self._running = False
-
-        self._executor = ThreadPoolExecutor(max_workers=5)
-
         # These are the images that we should send to the camera at our desired
         # framerate.
         self._camera_images = None
 
-        # Create media threads queues.
-        if self._params.camera_out_enabled:
-            self._camera_out_queue = queue.Queue()
-        self._sink_queue = queue.Queue()
-        self._sink_thread = None
-
-        self._stopped_event = asyncio.Event()
-        self._is_interrupted = threading.Event()
-
         # We will write 20ms audio at a time. If we receive long audio frames we
         # will chunk them. This will help with interruption handling.
         audio_bytes_10ms = int(self._params.audio_out_sample_rate / 100) * \
             self._params.audio_out_channels * 2
         self._audio_chunk_size = audio_bytes_10ms * 2
 
+        self._stopped_event = asyncio.Event()
+
+        # Create sink frame task. This is the task that will actually write
+        # audio or video frames. We write audio/video in a task so we can keep
+        # generating frames upstream while, for example, the audio is playing.
+        self._create_sink_task()
+
         # Create push frame task. This is the task that will push frames in
         # order. We also guarantee that all frames are pushed in the same task.
         self._create_push_task()
 
     async def start(self, frame: StartFrame):
-        if self._running:
-            return
-
-        self._running = True
-
-        loop = self.get_event_loop()
-
-        # Create queues and threads.
+        # Create media threads queues.
         if self._params.camera_out_enabled:
-            self._camera_out_thread = loop.run_in_executor(
-                self._executor, self._camera_out_thread_handler)
-
-        self._sink_thread = loop.run_in_executor(self._executor, self._sink_thread_handler)
+            self._camera_out_queue = asyncio.Queue()
+            self._camera_out_task = self.get_event_loop().create_task(self._camera_out_task_handler())
 
     async def stop(self):
-        if not self._running:
-            return
-
-        # This will exit all threads.
-        self._running = False
+        # Wait on the threads to finish.
+        if self._params.camera_out_enabled:
+            self._camera_out_task.cancel()
+            await self._camera_out_task
 
         self._stopped_event.set()
 
-    def send_message(self, frame: TransportMessageFrame):
+    async def send_message(self, frame: TransportMessageFrame):
         pass
 
-    def send_metrics(self, frame: MetricsFrame):
+    async def send_metrics(self, frame: MetricsFrame):
         pass
 
-    def write_frame_to_camera(self, frame: ImageRawFrame):
+    async def write_frame_to_camera(self, frame: ImageRawFrame):
         pass
 
-    def write_raw_audio_frames(self, frames: bytes):
+    async def write_raw_audio_frames(self, frames: bytes):
         pass
 
     #
@@ -110,12 +89,12 @@ class BaseOutputTransport(FrameProcessor):
     #
 
     async def cleanup(self):
-        # Wait on the threads to finish.
-        if self._params.camera_out_enabled:
-            await self._camera_out_thread
+        if self._sink_task:
+            self._sink_task.cancel()
+            await self._sink_task
 
-        if self._sink_thread:
-            await self._sink_thread
+        self._push_frame_task.cancel()
+        await self._push_frame_task
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
@@ -128,7 +107,7 @@ class BaseOutputTransport(FrameProcessor):
         if isinstance(frame, StartFrame):
             await self.start(frame)
             await self.push_frame(frame, direction)
-        # EndFrame is managed in the queue handler.
+        # EndFrame is managed in the sink queue handler.
         elif isinstance(frame, CancelFrame):
             await self.stop()
             await self.push_frame(frame, direction)
@@ -136,14 +115,14 @@ class BaseOutputTransport(FrameProcessor):
             await self._handle_interruptions(frame)
             await self.push_frame(frame, direction)
         elif isinstance(frame, MetricsFrame):
-            self.send_metrics(frame)
+            await self.send_metrics(frame)
             await self.push_frame(frame, direction)
         elif isinstance(frame, SystemFrame):
             await self.push_frame(frame, direction)
         elif isinstance(frame, AudioRawFrame):
             await self._handle_audio(frame)
         else:
-            self._sink_queue.put_nowait(frame)
+            await self._sink_queue.put(frame)
 
         # If we are finishing, wait here until we have stopped, otherwise we might
         # close things too early upstream. We need this event because we don't
@@ -156,50 +135,51 @@ class BaseOutputTransport(FrameProcessor):
             return
 
         if isinstance(frame, StartInterruptionFrame):
-            self._is_interrupted.set()
+            # Stop sink task.
+            self._sink_task.cancel()
+            await self._sink_task
+            self._create_sink_task()
+            # Stop push task.
             self._push_frame_task.cancel()
+            await self._push_frame_task
             self._create_push_task()
-        elif isinstance(frame, StopInterruptionFrame):
-            self._is_interrupted.clear()
 
     async def _handle_audio(self, frame: AudioRawFrame):
         audio = frame.audio
         for i in range(0, len(audio), self._audio_chunk_size):
             chunk = AudioRawFrame(audio[i: i + self._audio_chunk_size],
                                   sample_rate=frame.sample_rate, num_channels=frame.num_channels)
-            self._sink_queue.put_nowait(chunk)
+            await self._sink_queue.put(chunk)
 
-    def _sink_thread_handler(self):
+    def _create_sink_task(self):
+        loop = self.get_event_loop()
+        self._sink_queue = asyncio.Queue()
+        self._sink_task = loop.create_task(self._sink_task_handler())
+
+    async def _sink_task_handler(self):
         # Audio accumlation buffer
         buffer = bytearray()
-        while self._running:
+        while True:
             try:
-                frame = self._sink_queue.get(timeout=1)
-                if not self._is_interrupted.is_set():
-                    if isinstance(frame, AudioRawFrame) and self._params.audio_out_enabled:
-                        buffer.extend(frame.audio)
-                        buffer = self._maybe_send_audio(buffer)
-                    elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled:
-                        self._set_camera_image(frame)
-                    elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled:
-                        self._set_camera_images(frame.images)
-                    elif isinstance(frame, TransportMessageFrame):
-                        self.send_message(frame)
-                    else:
-                        future = asyncio.run_coroutine_threadsafe(
-                            self._internal_push_frame(frame), self.get_event_loop())
-                        future.result()
+                frame = await self._sink_queue.get()
+                if isinstance(frame, AudioRawFrame) and self._params.audio_out_enabled:
+                    buffer.extend(frame.audio)
+                    buffer = await self._maybe_send_audio(buffer)
+                elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled:
+                    await self._set_camera_image(frame)
+                elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled:
+                    await self._set_camera_images(frame.images)
+                elif isinstance(frame, TransportMessageFrame):
+                    await self.send_message(frame)
                 else:
-                    # If we get interrupted just clear the output buffer.
-                    buffer = bytearray()
+                    await self._internal_push_frame(frame)
 
                 if isinstance(frame, EndFrame):
-                    future = asyncio.run_coroutine_threadsafe(self.stop(), self.get_event_loop())
-                    future.result()
+                    await self.stop()
 
                 self._sink_queue.task_done()
-            except queue.Empty:
-                pass
+            except asyncio.CancelledError:
+                break
             except BaseException as e:
                 logger.error(f"{self} error processing sink queue: {e}")
 
@@ -209,8 +189,8 @@ class BaseOutputTransport(FrameProcessor):
 
     def _create_push_task(self):
         loop = self.get_event_loop()
-        self._push_frame_task = loop.create_task(self._push_frame_task_handler())
         self._push_queue = asyncio.Queue()
+        self._push_frame_task = loop.create_task(self._push_frame_task_handler())
 
     async def _internal_push_frame(
             self,
@@ -233,7 +213,7 @@ class BaseOutputTransport(FrameProcessor):
     async def send_image(self, frame: ImageRawFrame | SpriteFrame):
         await self.process_frame(frame, FrameDirection.DOWNSTREAM)
 
-    def _draw_image(self, frame: ImageRawFrame):
+    async def _draw_image(self, frame: ImageRawFrame):
         desired_size = (self._params.camera_out_width, self._params.camera_out_height)
 
         if frame.size != desired_size:
@@ -243,32 +223,32 @@ class BaseOutputTransport(FrameProcessor):
                 f"{frame} does not have the expected size {desired_size}, resizing")
             frame = ImageRawFrame(resized_image.tobytes(), resized_image.size, resized_image.format)
 
-        self.write_frame_to_camera(frame)
+        await self.write_frame_to_camera(frame)
 
-    def _set_camera_image(self, image: ImageRawFrame):
+    async def _set_camera_image(self, image: ImageRawFrame):
         if self._params.camera_out_is_live:
-            self._camera_out_queue.put_nowait(image)
+            await self._camera_out_queue.put(image)
         else:
             self._camera_images = itertools.cycle([image])
 
-    def _set_camera_images(self, images: List[ImageRawFrame]):
+    async def _set_camera_images(self, images: List[ImageRawFrame]):
         self._camera_images = itertools.cycle(images)
 
-    def _camera_out_thread_handler(self):
-        while self._running:
+    async def _camera_out_task_handler(self):
+        while True:
             try:
                 if self._params.camera_out_is_live:
-                    image = self._camera_out_queue.get(timeout=1)
-                    self._draw_image(image)
+                    image = await self._camera_out_queue.get()
+                    await self._draw_image(image)
                     self._camera_out_queue.task_done()
                 elif self._camera_images:
                     image = next(self._camera_images)
-                    self._draw_image(image)
-                    time.sleep(1.0 / self._params.camera_out_framerate)
+                    await self._draw_image(image)
+                    await asyncio.sleep(1.0 / self._params.camera_out_framerate)
                 else:
-                    time.sleep(1.0 / self._params.camera_out_framerate)
-            except queue.Empty:
-                pass
+                    await asyncio.sleep(1.0 / self._params.camera_out_framerate)
+            except asyncio.CancelledError:
+                break
             except Exception as e:
                 logger.error(f"{self} error writing to camera: {e}")
 
@@ -279,12 +259,8 @@ class BaseOutputTransport(FrameProcessor):
     async def send_audio(self, frame: AudioRawFrame):
         await self.process_frame(frame, FrameDirection.DOWNSTREAM)
 
-    def _maybe_send_audio(self, buffer: bytearray) -> bytearray:
-        try:
-            if len(buffer) >= self._audio_chunk_size:
-                self.write_raw_audio_frames(bytes(buffer[:self._audio_chunk_size]))
-                buffer = buffer[self._audio_chunk_size:]
-            return buffer
-        except BaseException as e:
-            logger.error(f"{self} error writing audio frames: {e}")
-            return buffer
+    async def _maybe_send_audio(self, buffer: bytearray) -> bytearray:
+        if len(buffer) >= self._audio_chunk_size:
+            await self.write_raw_audio_frames(bytes(buffer[:self._audio_chunk_size]))
+            buffer = buffer[self._audio_chunk_size:]
+        return buffer
diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py
index 33d6062aa..cd05550a9 100644
--- a/src/pipecat/transports/local/audio.py
+++ b/src/pipecat/transports/local/audio.py
@@ -6,6 +6,8 @@
 
 import asyncio
 
+from concurrent.futures import ThreadPoolExecutor
+
 from pipecat.frames.frames import AudioRawFrame, StartFrame
 from pipecat.processors.frame_processor import FrameProcessor
 from pipecat.transports.base_input import BaseInputTransport
@@ -43,26 +45,20 @@ class LocalAudioInputTransport(BaseInputTransport):
         await super().start(frame)
         self._in_stream.start_stream()
 
-    async def stop(self):
-        await super().stop()
-        self._in_stream.stop_stream()
-
     async def cleanup(self):
+        await super().cleanup()
+        self._in_stream.stop_stream()
         # This is not very pretty (taken from PyAudio docs).
         while self._in_stream.is_active():
             await asyncio.sleep(0.1)
         self._in_stream.close()
 
-        await super().cleanup()
-
     def _audio_in_callback(self, in_data, frame_count, time_info, status):
-        if not self._running:
-            return (None, pyaudio.paAbort)
-
         frame = AudioRawFrame(audio=in_data,
                               sample_rate=self._params.audio_in_sample_rate,
                               num_channels=self._params.audio_in_channels)
-        self.push_audio_frame(frame)
+
+        asyncio.run_coroutine_threadsafe(self.push_audio_frame(frame), self.get_event_loop())
 
         return (None, pyaudio.paContinue)
 
@@ -72,19 +68,29 @@ class LocalAudioOutputTransport(BaseOutputTransport):
     def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
         super().__init__(params)
 
+        self._executor = ThreadPoolExecutor(max_workers=5)
+
         self._out_stream = py_audio.open(
             format=py_audio.get_format_from_width(2),
             channels=params.audio_out_channels,
             rate=params.audio_out_sample_rate,
             output=True)
 
-    def write_raw_audio_frames(self, frames: bytes):
-        self._out_stream.write(frames)
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._out_stream.start_stream()
 
     async def cleanup(self):
         await super().cleanup()
+        self._out_stream.stop_stream()
+        # This is not very pretty (taken from PyAudio docs).
+        while self._out_stream.is_active():
+            await asyncio.sleep(0.1)
         self._out_stream.close()
 
+    async def write_raw_audio_frames(self, frames: bytes):
+        await self.get_event_loop().run_in_executor(self._executor, self._out_stream.write, frames)
+
 
 class LocalAudioTransport(BaseTransport):
 
diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py
index 6370e8d39..0e2a404df 100644
--- a/src/pipecat/transports/local/tk.py
+++ b/src/pipecat/transports/local/tk.py
@@ -6,6 +6,8 @@
 
 import asyncio
 
+from concurrent.futures import ThreadPoolExecutor
+
 import numpy as np
 import tkinter as tk
 
@@ -53,25 +55,20 @@ class TkInputTransport(BaseInputTransport):
         await super().start(frame)
         self._in_stream.start_stream()
 
-    async def stop(self):
-        await super().stop()
-        self._in_stream.stop_stream()
-
     async def cleanup(self):
         await super().cleanup()
+        self._in_stream.stop_stream()
         # This is not very pretty (taken from PyAudio docs).
         while self._in_stream.is_active():
             await asyncio.sleep(0.1)
         self._in_stream.close()
 
     def _audio_in_callback(self, in_data, frame_count, time_info, status):
-        if not self._running:
-            return (None, pyaudio.paAbort)
-
         frame = AudioRawFrame(audio=in_data,
                               sample_rate=self._params.audio_in_sample_rate,
                               num_channels=self._params.audio_in_channels)
-        self.push_audio_frame(frame)
+
+        asyncio.run_coroutine_threadsafe(self.push_audio_frame(frame), self.get_event_loop())
 
         return (None, pyaudio.paContinue)
 
@@ -81,6 +78,8 @@ class TkOutputTransport(BaseOutputTransport):
     def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams):
         super().__init__(params)
 
+        self._executor = ThreadPoolExecutor(max_workers=5)
+
         self._out_stream = py_audio.open(
             format=py_audio.get_format_from_width(2),
             channels=params.audio_out_channels,
@@ -94,16 +93,24 @@ class TkOutputTransport(BaseOutputTransport):
         self._image_label = tk.Label(tk_root, image=photo)
         self._image_label.pack()
 
-    def write_raw_audio_frames(self, frames: bytes):
-        self._out_stream.write(frames)
-
-    def write_frame_to_camera(self, frame: ImageRawFrame):
-        self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._out_stream.start_stream()
 
     async def cleanup(self):
         await super().cleanup()
+        self._out_stream.stop_stream()
+        # This is not very pretty (taken from PyAudio docs).
+        while self._out_stream.is_active():
+            await asyncio.sleep(0.1)
         self._out_stream.close()
 
+    async def write_raw_audio_frames(self, frames: bytes):
+        await self.get_event_loop().run_in_executor(self._executor, self._out_stream.write, frames)
+
+    async def write_frame_to_camera(self, frame: ImageRawFrame):
+        self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
+
     def _write_frame_to_tk(self, frame: ImageRawFrame):
         width = frame.size[0]
         height = frame.size[1]
diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py
index 6e5307780..0b3eb906f 100644
--- a/src/pipecat/transports/network/fastapi_websocket.py
+++ b/src/pipecat/transports/network/fastapi_websocket.py
@@ -75,7 +75,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
                 continue
 
             if isinstance(frame, AudioRawFrame):
-                self.push_audio_frame(frame)
+                await self.push_audio_frame(frame)
 
         await self._callbacks.on_client_disconnected(self._websocket)
 
@@ -89,7 +89,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
         self._params = params
         self._audio_buffer = bytes()
 
-    def write_raw_audio_frames(self, frames: bytes):
+    async def write_raw_audio_frames(self, frames: bytes):
         self._audio_buffer += frames
         while len(self._audio_buffer) >= self._params.audio_frame_size:
             frame = AudioRawFrame(
@@ -115,9 +115,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
 
             payload = self._params.serializer.serialize(frame)
             if payload:
-                future = asyncio.run_coroutine_threadsafe(
-                    self._websocket.send_text(payload), self.get_event_loop())
-                future.result()
+                await self._websocket.send_text(payload)
 
             self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]
 
diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py
index 63fd2a541..d17174eba 100644
--- a/src/pipecat/transports/network/websocket_server.py
+++ b/src/pipecat/transports/network/websocket_server.py
@@ -93,7 +93,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
                 continue
 
             if isinstance(frame, AudioRawFrame):
-                self.push_audio_frame(frame)
+                await self.push_audio_frame(frame)
             else:
                 await self._internal_push_frame(frame)
 
@@ -123,7 +123,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
             logger.warning("Only one client allowed, using new connection")
         self._websocket = websocket
 
-    def write_raw_audio_frames(self, frames: bytes):
+    async def write_raw_audio_frames(self, frames: bytes):
         self._audio_buffer += frames
         while len(self._audio_buffer) >= self._params.audio_frame_size:
             frame = AudioRawFrame(
@@ -149,9 +149,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
 
             proto = self._params.serializer.serialize(frame)
 
-            future = asyncio.run_coroutine_threadsafe(
-                self._websocket.send(proto), self.get_event_loop())
-            future.result()
+            await self._websocket.send(proto)
 
             self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]
 
diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py
index 2476fc8e6..9fdf9dab1 100644
--- a/src/pipecat/transports/services/daily.py
+++ b/src/pipecat/transports/services/daily.py
@@ -6,11 +6,10 @@
 
 import aiohttp
 import asyncio
-import queue
 import time
 
 from dataclasses import dataclass
-from typing import Any, Callable, Mapping
+from typing import Any, Awaitable, Callable, Mapping
 from concurrent.futures import ThreadPoolExecutor
 
 from daily import (
@@ -108,20 +107,27 @@ class DailyParams(TransportParams):
 
 
 class DailyCallbacks(BaseModel):
-    on_joined: Callable[[Mapping[str, Any]], None]
-    on_left: Callable[[], None]
-    on_error: Callable[[str], None]
-    on_app_message: Callable[[Any, str], None]
-    on_call_state_updated: Callable[[str], None]
-    on_dialin_ready: Callable[[str], None]
-    on_dialout_answered: Callable[[Any], None]
-    on_dialout_connected: Callable[[Any], None]
-    on_dialout_stopped: Callable[[Any], None]
-    on_dialout_error: Callable[[Any], None]
-    on_dialout_warning: Callable[[Any], None]
-    on_first_participant_joined: Callable[[Mapping[str, Any]], None]
-    on_participant_joined: Callable[[Mapping[str, Any]], None]
-    on_participant_left: Callable[[Mapping[str, Any], str], None]
+    on_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
+    on_left: Callable[[], Awaitable[None]]
+    on_error: Callable[[str], Awaitable[None]]
+    on_app_message: Callable[[Any, str], Awaitable[None]]
+    on_call_state_updated: Callable[[str], Awaitable[None]]
+    on_dialin_ready: Callable[[str], Awaitable[None]]
+    on_dialout_answered: Callable[[Any], Awaitable[None]]
+    on_dialout_connected: Callable[[Any], Awaitable[None]]
+    on_dialout_stopped: Callable[[Any], Awaitable[None]]
+    on_dialout_error: Callable[[Any], Awaitable[None]]
+    on_dialout_warning: Callable[[Any], Awaitable[None]]
+    on_first_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
+    on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
+    on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]]
+
+
+def completion_callback(future):
+    def _callback(*args):
+        if not future.cancelled():
+            future.get_loop().call_soon_threadsafe(future.set_result, *args)
+    return _callback
 
 
 class DailyTransportClient(EventHandler):
@@ -161,7 +167,6 @@ class DailyTransportClient(EventHandler):
         self._joined = False
         self._joining = False
         self._leaving = False
-        self._sync_response = {k: queue.Queue() for k in ["join", "leave"]}
 
         self._executor = ThreadPoolExecutor(max_workers=5)
 
@@ -174,10 +179,16 @@ class DailyTransportClient(EventHandler):
             color_format=self._params.camera_out_color_format)
 
         self._mic: VirtualMicrophoneDevice = Daily.create_microphone_device(
-            "mic", sample_rate=self._params.audio_out_sample_rate, channels=self._params.audio_out_channels)
+            "mic",
+            sample_rate=self._params.audio_out_sample_rate,
+            channels=self._params.audio_out_channels,
+            non_blocking=True)
 
         self._speaker: VirtualSpeakerDevice = Daily.create_speaker_device(
-            "speaker", sample_rate=self._params.audio_in_sample_rate, channels=self._params.audio_in_channels)
+            "speaker",
+            sample_rate=self._params.audio_in_sample_rate,
+            channels=self._params.audio_in_channels,
+            non_blocking=True)
         Daily.select_speaker_device("speaker")
 
     @property
@@ -187,30 +198,39 @@ class DailyTransportClient(EventHandler):
     def set_callbacks(self, callbacks: DailyCallbacks):
         self._callbacks = callbacks
 
-    def send_message(self, frame: DailyTransportMessageFrame):
-        self._client.send_app_message(frame.message, frame.participant_id)
+    async def send_message(self, frame: DailyTransportMessageFrame):
+        future = self._loop.create_future()
+        self._client.send_app_message(
+            frame.message,
+            frame.participant_id,
+            completion=completion_callback(future))
+        await future
 
-    def read_next_audio_frame(self) -> AudioRawFrame | None:
+    async def read_next_audio_frame(self) -> AudioRawFrame | None:
         sample_rate = self._params.audio_in_sample_rate
         num_channels = self._params.audio_in_channels
 
         if self._other_participant_has_joined:
             num_frames = int(sample_rate / 100) * 2  # 20ms of audio
 
-            audio = self._speaker.read_frames(num_frames)
+            future = self._loop.create_future()
+            self._speaker.read_frames(num_frames, completion=completion_callback(future))
+            audio = await future
 
             return AudioRawFrame(audio=audio, sample_rate=sample_rate, num_channels=num_channels)
         else:
             # If no one has ever joined the meeting `read_frames()` would block,
             # instead we just wait a bit. daily-python should probably return
             # silence instead.
-            time.sleep(0.01)
+            await asyncio.sleep(0.01)
             return None
 
-    def write_raw_audio_frames(self, frames: bytes):
-        self._mic.write_frames(frames)
+    async def write_raw_audio_frames(self, frames: bytes):
+        future = self._loop.create_future()
+        self._mic.write_frames(frames, completion=completion_callback(future))
+        await future
 
-    def write_frame_to_camera(self, frame: ImageRawFrame):
+    async def write_frame_to_camera(self, frame: ImageRawFrame):
         self._camera.write_frame(frame.image)
 
     async def join(self):
@@ -218,13 +238,10 @@ class DailyTransportClient(EventHandler):
         if self._joined or self._joining:
             return
 
-        self._joining = True
-
-        await self._loop.run_in_executor(self._executor, self._join)
-
-    def _join(self):
         logger.info(f"Joining {self._room_url}")
 
+        self._joining = True
+
         # For performance reasons, never subscribe to video streams (unless a
         # video renderer is registered).
         self._client.update_subscription_profiles({
@@ -236,10 +253,42 @@ class DailyTransportClient(EventHandler):
 
         self._client.set_user_name(self._bot_name)
 
+        try:
+            (data, error) = await self._join()
+
+            if not error:
+                self._joined = True
+                self._joining = False
+
+                logger.info(f"Joined {self._room_url}")
+
+                if self._token and self._params.transcription_enabled:
+                    logger.info(
+                        f"Enabling transcription with settings {self._params.transcription_settings}")
+                    self._client.start_transcription(
+                        self._params.transcription_settings.model_dump())
+
+                await self._callbacks.on_joined(data["participants"]["local"])
+            else:
+                error_msg = f"Error joining {self._room_url}: {error}"
+                logger.error(error_msg)
+                await self._callbacks.on_error(error_msg)
+        except asyncio.TimeoutError:
+            error_msg = f"Time out joining {self._room_url}"
+            logger.error(error_msg)
+            await self._callbacks.on_error(error_msg)
+
+    async def _join(self):
+        future = self._loop.create_future()
+
+        def handle_join_response(data, error):
+            if not future.cancelled():
+                future.get_loop().call_soon_threadsafe(future.set_result, (data, error))
+
         self._client.join(
             self._room_url,
             self._token,
-            completion=self._call_joined,
+            completion=handle_join_response,
             client_settings={
                 "inputs": {
                     "camera": {
@@ -275,33 +324,7 @@ class DailyTransportClient(EventHandler):
                 },
             })
 
-        self._handle_join_response()
-
-    def _handle_join_response(self):
-        try:
-            (data, error) = self._sync_response["join"].get(timeout=10)
-            if not error:
-                self._joined = True
-                self._joining = False
-
-                logger.info(f"Joined {self._room_url}")
-
-                if self._token and self._params.transcription_enabled:
-                    logger.info(
-                        f"Enabling transcription with settings {self._params.transcription_settings}")
-                    self._client.start_transcription(
-                        self._params.transcription_settings.model_dump())
-
-                self._callbacks.on_joined(data["participants"]["local"])
-            else:
-                error_msg = f"Error joining {self._room_url}: {error}"
-                logger.error(error_msg)
-                self._callbacks.on_error(error_msg)
-            self._sync_response["join"].task_done()
-        except queue.Empty:
-            error_msg = f"Time out joining {self._room_url}"
-            logger.error(error_msg)
-            self._callbacks.on_error(error_msg)
+        return await asyncio.wait_for(future, timeout=10)
 
     async def leave(self):
         # Transport not joined, ignore.
@@ -311,34 +334,36 @@ class DailyTransportClient(EventHandler):
         self._joined = False
         self._leaving = True
 
-        await self._loop.run_in_executor(self._executor, self._leave)
-
-    def _leave(self):
         logger.info(f"Leaving {self._room_url}")
 
         if self._params.transcription_enabled:
             self._client.stop_transcription()
 
-        self._client.leave(completion=self._call_left)
-
-        self._handle_leave_response()
-
-    def _handle_leave_response(self):
         try:
-            error = self._sync_response["leave"].get(timeout=10)
+            error = await self._leave()
             if not error:
                 self._leaving = False
                 logger.info(f"Left {self._room_url}")
-                self._callbacks.on_left()
+                await self._callbacks.on_left()
             else:
                 error_msg = f"Error leaving {self._room_url}: {error}"
                 logger.error(error_msg)
-                self._callbacks.on_error(error_msg)
-            self._sync_response["leave"].task_done()
-        except queue.Empty:
+                await self._callbacks.on_error(error_msg)
+        except asyncio.TimeoutError:
             error_msg = f"Time out leaving {self._room_url}"
             logger.error(error_msg)
-            self._callbacks.on_error(error_msg)
+            await self._callbacks.on_error(error_msg)
+
+    async def _leave(self):
+        future = self._loop.create_future()
+
+        def handle_leave_response(error):
+            if not future.cancelled():
+                future.get_loop().call_soon_threadsafe(future.set_result, error)
+
+        self._client.leave(completion=handle_leave_response)
+
+        return await asyncio.wait_for(future, timeout=10)
 
     async def cleanup(self):
         await self._loop.run_in_executor(self._executor, self._cleanup)
@@ -400,28 +425,28 @@ class DailyTransportClient(EventHandler):
     #
 
     def on_app_message(self, message: Any, sender: str):
-        self._callbacks.on_app_message(message, sender)
+        self._call_async_callback(self._callbacks.on_app_message, message, sender)
 
     def on_call_state_updated(self, state: str):
-        self._callbacks.on_call_state_updated(state)
+        self._call_async_callback(self._callbacks.on_call_state_updated, state)
 
     def on_dialin_ready(self, sip_endpoint: str):
-        self._callbacks.on_dialin_ready(sip_endpoint)
+        self._call_async_callback(self._callbacks.on_dialin_ready, sip_endpoint)
 
     def on_dialout_answered(self, data: Any):
-        self._callbacks.on_dialout_answered(data)
+        self._call_async_callback(self._callbacks.on_dialout_answered, data)
 
     def on_dialout_connected(self, data: Any):
-        self._callbacks.on_dialout_connected(data)
+        self._call_async_callback(self._callbacks.on_dialout_connected, data)
 
     def on_dialout_stopped(self, data: Any):
-        self._callbacks.on_dialout_stopped(data)
+        self._call_async_callback(self._callbacks.on_dialout_stopped, data)
 
     def on_dialout_error(self, data: Any):
-        self._callbacks.on_dialout_error(data)
+        self._call_async_callback(self._callbacks.on_dialout_error, data)
 
     def on_dialout_warning(self, data: Any):
-        self._callbacks.on_dialout_warning(data)
+        self._call_async_callback(self._callbacks.on_dialout_warning, data)
 
     def on_participant_joined(self, participant):
         id = participant["id"]
@@ -429,15 +454,15 @@ class DailyTransportClient(EventHandler):
 
         if not self._other_participant_has_joined:
             self._other_participant_has_joined = True
-            self._callbacks.on_first_participant_joined(participant)
+            self._call_async_callback(self._callbacks.on_first_participant_joined, participant)
 
-        self._callbacks.on_participant_joined(participant)
+        self._call_async_callback(self._callbacks.on_participant_joined, participant)
 
     def on_participant_left(self, participant, reason):
         id = participant["id"]
         logger.info(f"Participant left {id}")
 
-        self._callbacks.on_participant_left(participant, reason)
+        self._call_async_callback(self._callbacks.on_participant_left, participant, reason)
 
     def on_transcription_message(self, message: Mapping[str, Any]):
         participant_id = ""
@@ -446,7 +471,7 @@ class DailyTransportClient(EventHandler):
 
         if participant_id in self._transcription_renderers:
             callback = self._transcription_renderers[participant_id]
-            callback(participant_id, message)
+            self._call_async_callback(callback, participant_id, message)
 
     def on_transcription_error(self, message):
         logger.error(f"Transcription error: {message}")
@@ -461,18 +486,19 @@ class DailyTransportClient(EventHandler):
     # Daily (CallClient callbacks)
     #
 
-    def _call_joined(self, data, error):
-        self._sync_response["join"].put((data, error))
-
-    def _call_left(self, error):
-        self._sync_response["leave"].put(error)
-
     def _video_frame_received(self, participant_id, video_frame):
         callback = self._video_renderers[participant_id]
-        callback(participant_id,
-                 video_frame.buffer,
-                 (video_frame.width, video_frame.height),
-                 video_frame.color_format)
+        self._call_async_callback(
+            callback,
+            participant_id,
+            video_frame.buffer,
+            (video_frame.width,
+             video_frame.height),
+            video_frame.color_format)
+
+    def _call_async_callback(self, callback, *args):
+        future = asyncio.run_coroutine_threadsafe(callback(*args), self._loop)
+        future.result()
 
 
 class DailyInputTransport(BaseInputTransport):
@@ -491,8 +517,6 @@ class DailyInputTransport(BaseInputTransport):
                 num_channels=self._params.audio_in_channels)
 
     async def start(self, frame: StartFrame):
-        if self._running:
-            return
         # Parent start.
         await super().start(frame)
         # Join the room.
@@ -500,19 +524,17 @@ class DailyInputTransport(BaseInputTransport):
         # Create audio task. It reads audio frames from Daily and push them
         # internally for VAD processing.
         if self._params.audio_in_enabled or self._params.vad_enabled:
-            self._audio_in_thread = self._loop.run_in_executor(
-                self._executor, self._audio_in_thread_handler)
+            self._audio_in_task = self.get_event_loop().create_task(self._audio_in_task_handler())
 
     async def stop(self):
-        if not self._running:
-            return
-        # Parent stop. This will set _running to False.
+        # Parent stop.
         await super().stop()
         # Leave the room.
         await self._client.leave()
         # Stop audio thread.
         if self._params.audio_in_enabled or self._params.vad_enabled:
-            await self._audio_in_thread
+            self._audio_in_task.cancel()
+            await self._audio_in_task
 
     async def cleanup(self):
         await super().cleanup()
@@ -535,26 +557,25 @@ class DailyInputTransport(BaseInputTransport):
     # Frames
     #
 
-    def push_transcription_frame(self, frame: TranscriptionFrame | InterimTranscriptionFrame):
-        future = asyncio.run_coroutine_threadsafe(
-            self._internal_push_frame(frame), self.get_event_loop())
-        future.result()
+    async def push_transcription_frame(self, frame: TranscriptionFrame | InterimTranscriptionFrame):
+        await self._internal_push_frame(frame)
 
-    def push_app_message(self, message: Any, sender: str):
+    async def push_app_message(self, message: Any, sender: str):
         frame = DailyTransportMessageFrame(message=message, participant_id=sender)
-        future = asyncio.run_coroutine_threadsafe(
-            self._internal_push_frame(frame), self.get_event_loop())
-        future.result()
+        await self._internal_push_frame(frame)
 
     #
     # Audio in
     #
 
-    def _audio_in_thread_handler(self):
-        while self._running:
-            frame = self._client.read_next_audio_frame()
-            if frame:
-                self.push_audio_frame(frame)
+    async def _audio_in_task_handler(self):
+        while True:
+            try:
+                frame = await self._client.read_next_audio_frame()
+                if frame:
+                    await self.push_audio_frame(frame)
+            except asyncio.CancelledError:
+                break
 
     #
     # Camera in
@@ -584,7 +605,7 @@ class DailyInputTransport(BaseInputTransport):
         if participant_id in self._video_renderers:
             self._video_renderers[participant_id]["render_next_frame"] = True
 
-    def _on_participant_video_frame(self, participant_id: str, buffer, size, format):
+    async def _on_participant_video_frame(self, participant_id: str, buffer, size, format):
         render_frame = False
 
         curr_time = time.time()
@@ -604,9 +625,7 @@ class DailyInputTransport(BaseInputTransport):
                 image=buffer,
                 size=size,
                 format=format)
-            future = asyncio.run_coroutine_threadsafe(
-                self._internal_push_frame(frame), self.get_event_loop())
-            future.result()
+            await self._internal_push_frame(frame)
 
         self._video_renderers[participant_id]["timestamp"] = curr_time
 
@@ -619,17 +638,13 @@ class DailyOutputTransport(BaseOutputTransport):
         self._client = client
 
     async def start(self, frame: StartFrame):
-        if self._running:
-            return
         # Parent start.
         await super().start(frame)
         # Join the room.
         await self._client.join()
 
     async def stop(self):
-        if not self._running:
-            return
-        # Parent stop. This will set _running to False.
+        # Parent stop.
         await super().stop()
         # Leave the room.
         await self._client.leave()
@@ -638,10 +653,10 @@ class DailyOutputTransport(BaseOutputTransport):
         await super().cleanup()
         await self._client.cleanup()
 
-    def send_message(self, frame: DailyTransportMessageFrame):
-        self._client.send_message(frame)
+    async def send_message(self, frame: DailyTransportMessageFrame):
+        await self._client.send_message(frame)
 
-    def send_metrics(self, frame: MetricsFrame):
+    async def send_metrics(self, frame: MetricsFrame):
         ttfb = [{"name": n, "time": t} for n, t in frame.ttfb.items()]
         message = DailyTransportMessageFrame(message={
             "type": "pipecat-metrics",
@@ -649,13 +664,13 @@ class DailyOutputTransport(BaseOutputTransport):
                 "ttfb": ttfb
             },
         })
-        self._client.send_message(message)
+        await self._client.send_message(message)
 
-    def write_raw_audio_frames(self, frames: bytes):
-        self._client.write_raw_audio_frames(frames)
+    async def write_raw_audio_frames(self, frames: bytes):
+        await self._client.write_raw_audio_frames(frames)
 
-    def write_frame_to_camera(self, frame: ImageRawFrame):
-        self._client.write_frame_to_camera(frame)
+    async def write_frame_to_camera(self, frame: ImageRawFrame):
+        await self._client.write_frame_to_camera(frame)
 
 
 class DailyTransport(BaseTransport):
@@ -774,24 +789,24 @@ class DailyTransport(BaseTransport):
             self._input.capture_participant_video(
                 participant_id, framerate, video_source, color_format)
 
-    def _on_joined(self, participant):
-        self._call_async_event_handler("on_joined", participant)
+    async def _on_joined(self, participant):
+        await self._call_event_handler("on_joined", participant)
 
-    def _on_left(self):
-        self._call_async_event_handler("on_left")
+    async def _on_left(self):
+        await self._call_event_handler("on_left")
 
-    def _on_error(self, error):
+    async def _on_error(self, error):
         # TODO(aleix): Report error to input/output transports. The one managing
         # the client should report the error.
         pass
 
-    def _on_app_message(self, message: Any, sender: str):
+    async def _on_app_message(self, message: Any, sender: str):
         if self._input:
-            self._input.push_app_message(message, sender)
-        self._call_async_event_handler("on_app_message", message, sender)
+            await self._input.push_app_message(message, sender)
+        await self._call_event_handler("on_app_message", message, sender)
 
-    def _on_call_state_updated(self, state: str):
-        self._call_async_event_handler("on_call_state_updated", state)
+    async def _on_call_state_updated(self, state: str):
+        await self._call_event_handler("on_call_state_updated", state)
 
     async def _handle_dialin_ready(self, sip_endpoint: str):
         if not self._params.dialin_settings:
@@ -824,36 +839,36 @@ class DailyTransport(BaseTransport):
             except BaseException as e:
                 logger.error(f"Error handling dialin-ready event ({url}): {e}")
 
-    def _on_dialin_ready(self, sip_endpoint):
+    async def _on_dialin_ready(self, sip_endpoint):
         if self._params.dialin_settings:
-            asyncio.run_coroutine_threadsafe(self._handle_dialin_ready(sip_endpoint), self._loop)
-        self._call_async_event_handler("on_dialin_ready", sip_endpoint)
+            await self._handle_dialin_ready(sip_endpoint)
+        await self._call_event_handler("on_dialin_ready", sip_endpoint)
 
-    def _on_dialout_answered(self, data):
-        self._call_async_event_handler("on_dialout_answered", data)
+    async def _on_dialout_answered(self, data):
+        await self._call_event_handler("on_dialout_answered", data)
 
-    def _on_dialout_connected(self, data):
-        self._call_async_event_handler("on_dialout_connected", data)
+    async def _on_dialout_connected(self, data):
+        await self._call_event_handler("on_dialout_connected", data)
 
-    def _on_dialout_stopped(self, data):
-        self._call_async_event_handler("on_dialout_stopped", data)
+    async def _on_dialout_stopped(self, data):
+        await self._call_event_handler("on_dialout_stopped", data)
 
-    def _on_dialout_error(self, data):
-        self._call_async_event_handler("on_dialout_error", data)
+    async def _on_dialout_error(self, data):
+        await self._call_event_handler("on_dialout_error", data)
 
-    def _on_dialout_warning(self, data):
-        self._call_async_event_handler("on_dialout_warning", data)
+    async def _on_dialout_warning(self, data):
+        await self._call_event_handler("on_dialout_warning", data)
 
-    def _on_participant_joined(self, participant):
-        self._call_async_event_handler("on_participant_joined", participant)
+    async def _on_participant_joined(self, participant):
+        await self._call_event_handler("on_participant_joined", participant)
 
-    def _on_participant_left(self, participant, reason):
-        self._call_async_event_handler("on_participant_left", participant, reason)
+    async def _on_participant_left(self, participant, reason):
+        await self._call_event_handler("on_participant_left", participant, reason)
 
-    def _on_first_participant_joined(self, participant):
-        self._call_async_event_handler("on_first_participant_joined", participant)
+    async def _on_first_participant_joined(self, participant):
+        await self._call_event_handler("on_first_participant_joined", participant)
 
-    def _on_transcription_message(self, participant_id, message):
+    async def _on_transcription_message(self, participant_id, message):
         text = message["text"]
         timestamp = message["timestamp"]
         is_final = message["rawResponse"]["is_final"]
@@ -864,9 +879,4 @@ class DailyTransport(BaseTransport):
             frame = InterimTranscriptionFrame(text, participant_id, timestamp)
 
         if self._input:
-            self._input.push_transcription_frame(frame)
-
-    def _call_async_event_handler(self, event_name: str, *args, **kwargs):
-        future = asyncio.run_coroutine_threadsafe(
-            self._call_event_handler(event_name, *args, **kwargs), self._loop)
-        future.result()
+            await self._input.push_transcription_frame(frame)