Merge pull request #662 from pipecat-ai/aleix/daily-transport-async-functions

transports(daily): make functions async

CHANGELOG.md

@@ -18,6 +18,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- The following `DailyTransport` functions are now `async` which means they need
+  to be awaited: `start_dialout`, `stop_dialout`, `start_recording`,
+  `stop_recording`, `capture_participant_transcription` and
+  `capture_participant_video`.
+
 - Changed default output sample rate to 24000. This changes all TTS service to
   output to 24000 and also the default output transport sample rate. This
   improves audio quality at the cost of some extra bandwidth.

examples/canonical-metrics/bot.py

@@ -124,7 +124,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             await task.queue_frames([LLMMessagesFrame(messages)])
 
         @transport.event_handler("on_participant_left")

examples/chatbot-audio-recording/bot.py

@@ -123,7 +123,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             await task.queue_frames([LLMMessagesFrame(messages)])
 
         @transport.event_handler("on_participant_left")

examples/deployment/flyio-example/bot.py

@@ -75,7 +75,7 @@ async def main(room_url: str, token: str):
 
     @transport.event_handler("on_first_participant_joined")
     async def on_first_participant_joined(transport, participant):
-        transport.capture_participant_transcription(participant["id"])
+        await transport.capture_participant_transcription(participant["id"])
         await task.queue_frames([LLMMessagesFrame(messages)])
 
     @transport.event_handler("on_participant_left")

examples/dialin-chatbot/bot_daily.py

@@ -81,7 +81,7 @@ async def main(room_url: str, token: str, callId: str, callDomain: str):
 
     @transport.event_handler("on_first_participant_joined")
     async def on_first_participant_joined(transport, participant):
-        transport.capture_participant_transcription(participant["id"])
+        await transport.capture_participant_transcription(participant["id"])
         await task.queue_frames([LLMMessagesFrame(messages)])
 
     @transport.event_handler("on_participant_left")

examples/dialin-chatbot/bot_twilio.py

@@ -84,7 +84,7 @@ async def main(room_url: str, token: str, callId: str, sipUri: str):
 
     @transport.event_handler("on_first_participant_joined")
     async def on_first_participant_joined(transport, participant):
-        transport.capture_participant_transcription(participant["id"])
+        await transport.capture_participant_transcription(participant["id"])
         await task.queue_frames([LLMMessagesFrame(messages)])
 
     @transport.event_handler("on_participant_left")

examples/foundational/06-listen-and-respond.py

@@ -110,7 +110,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/06a-image-sync.py

@@ -127,7 +127,7 @@ async def main():
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             participant_name = participant.get("info", {}).get("userName", "")
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             await task.queue_frames([TextFrame(f"Hi there {participant_name}!")])
 
         runner = PipelineRunner()

examples/foundational/07-interruptible-vad.py

@@ -89,7 +89,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07-interruptible.py

@@ -87,7 +87,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07a-interruptible-anthropic.py

@@ -82,7 +82,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([LLMMessagesFrame(messages)])
 

examples/foundational/07b-interruptible-langchain.py

@@ -109,7 +109,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             lc.set_participant_id(participant["id"])
             # Kick off the conversation.
             # the `LLMMessagesFrame` will be picked up by the LangchainProcessor using

examples/foundational/07d-interruptible-elevenlabs.py

@@ -85,7 +85,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07e-interruptible-playht.py

@@ -88,7 +88,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07f-interruptible-azure.py

@@ -89,7 +89,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07g-interruptible-openai-tts.py

@@ -74,7 +74,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07h-interruptible-openpipe.py

@@ -86,7 +86,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07i-interruptible-xtts.py

@@ -81,7 +81,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07j-interruptible-gladia.py

@@ -83,7 +83,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07k-interruptible-lmnt.py

@@ -77,7 +77,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07l-interruptible-together.py

@@ -96,7 +96,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([LLMMessagesFrame(messages)])
 

examples/foundational/07m-interruptible-aws.py

@@ -84,7 +84,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07n-interruptible-google.py

@@ -82,7 +82,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/07o-interruptible-assemblyai.py

@@ -83,7 +83,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/09-mirror.py

@@ -74,7 +74,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_video(participant["id"])
+            await transport.capture_participant_video(participant["id"])
 
         pipeline = Pipeline([transport.input(), MirrorProcessor(), transport.output()])
 

examples/foundational/09a-local-mirror.py

@@ -81,7 +81,7 @@ async def main():
 
         @daily_transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_video(participant["id"])
+            await transport.capture_participant_video(participant["id"])
 
         pipeline = Pipeline([daily_transport.input(), MirrorProcessor(), tk_transport.output()])
 

examples/foundational/10-wake-phrase.py

@@ -82,7 +82,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             await tts.say("Hi! If you want to talk to me, just say 'Hey Robot'.")
 
         runner = PipelineRunner()

examples/foundational/11-sound-effects.py

@@ -134,7 +134,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             await tts.say("Hi, I'm listening!")
             await transport.send_audio(sounds["ding1.wav"])
 

examples/foundational/12-describe-video.py

@@ -84,8 +84,8 @@ async def main():
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await tts.say("Hi there! Feel free to ask me what I see.")
-            transport.capture_participant_video(participant["id"], framerate=0)
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_video(participant["id"], framerate=0)
+            await transport.capture_participant_transcription(participant["id"])
             image_requester.set_participant_id(participant["id"])
 
         pipeline = Pipeline(

examples/foundational/12a-describe-video-gemini-flash.py

@@ -86,8 +86,8 @@ async def main():
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await tts.say("Hi there! Feel free to ask me what I see.")
-            transport.capture_participant_video(participant["id"], framerate=0)
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_video(participant["id"], framerate=0)
+            await transport.capture_participant_transcription(participant["id"])
             image_requester.set_participant_id(participant["id"])
 
         pipeline = Pipeline(

examples/foundational/12b-describe-video-gpt-4o.py

@@ -83,8 +83,8 @@ async def main():
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await tts.say("Hi there! Feel free to ask me what I see.")
-            transport.capture_participant_video(participant["id"], framerate=0)
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_video(participant["id"], framerate=0)
+            await transport.capture_participant_transcription(participant["id"])
             image_requester.set_participant_id(participant["id"])
 
         pipeline = Pipeline(

examples/foundational/12c-describe-video-anthropic.py

@@ -83,8 +83,8 @@ async def main():
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await tts.say("Hi there! Feel free to ask me what I see.")
-            transport.capture_participant_video(participant["id"], framerate=0)
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_video(participant["id"], framerate=0)
+            await transport.capture_participant_transcription(participant["id"])
             image_requester.set_participant_id(participant["id"])
 
         pipeline = Pipeline(

examples/foundational/14-function-calling.py

@@ -127,7 +127,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/14a-function-calling-anthropic.py

@@ -105,7 +105,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/14b-function-calling-anthropic-video.py

@@ -160,8 +160,8 @@ If you need to use a tool, simply use the tool. Do not tell the user the tool yo
         async def on_first_participant_joined(transport, participant):
             global video_participant_id
             video_participant_id = participant["id"]
-            transport.capture_participant_transcription(video_participant_id)
-            transport.capture_participant_video(video_participant_id, framerate=0)
+            await transport.capture_participant_transcription(video_participant_id)
+            await transport.capture_participant_video(video_participant_id, framerate=0)
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/14c-function-calling-together.py

@@ -123,7 +123,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             # await tts.say("Hi! Ask me about the weather in San Francisco.")
 

examples/foundational/14d-function-calling-video.py

@@ -153,8 +153,8 @@ indicate you should use the get_image tool are:
         async def on_first_participant_joined(transport, participant):
             global video_participant_id
             video_participant_id = participant["id"]
-            transport.capture_participant_transcription(participant["id"])
-            transport.capture_participant_video(video_participant_id, framerate=0)
+            await transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_video(video_participant_id, framerate=0)
             # Kick off the conversation.
             await tts.say("Hi! Ask me about the weather in San Francisco.")
 

examples/foundational/14e-function-calling-gemini.py

@@ -159,8 +159,8 @@ indicate you should use the get_image tool are:
         async def on_first_participant_joined(transport, participant):
             global video_participant_id
             video_participant_id = participant["id"]
-            transport.capture_participant_transcription(participant["id"])
-            transport.capture_participant_video(video_participant_id, framerate=0)
+            await transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_video(video_participant_id, framerate=0)
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/15-switch-voices.py

@@ -141,7 +141,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append(
                 {

examples/foundational/15a-switch-languages.py

@@ -128,7 +128,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append(
                 {

examples/foundational/16-gpu-container-local-bot.py

@@ -92,7 +92,7 @@ async def main():
         # bot can "hear" and respond to them.
         @transport.event_handler("on_participant_joined")
         async def on_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
 
         # When the first participant joins, the bot should introduce itself.
         @transport.event_handler("on_first_participant_joined")

examples/foundational/17-detect-user-idle.py

@@ -99,7 +99,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

examples/foundational/19-openai-realtime-beta.py

@@ -166,7 +166,7 @@ Remember, your responses should be short. Just one or two sentences, usually."""
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/20a-persistent-context-openai.py

@@ -223,7 +223,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/20b-persistent-context-openai-realtime.py

@@ -249,7 +249,7 @@ Remember, your responses should be short. Just one or two sentences, usually."""
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/20c-persistent-context-anthropic.py

@@ -219,7 +219,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/20d-persistent-context-gemini.py

@@ -111,8 +111,8 @@ async def load_conversation(function_name, tool_call_id, args, llm, context, res
 messages = [
     {
         "role": "system",
-        "content": """You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your 
-capabilities in a succinct way. Your output will be converted to audio so don't include special 
+        "content": """You are a helpful LLM in a WebRTC call. 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.
 
 You have several tools you can use to help you.
@@ -276,8 +276,8 @@ async def main():
         async def on_first_participant_joined(transport, participant):
             global video_participant_id
             video_participant_id = participant["id"]
-            transport.capture_participant_transcription(participant["id"])
-            transport.capture_participant_video(video_participant_id, framerate=0)
+            await transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_video(video_participant_id, framerate=0)
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/moondream-chatbot/bot.py

@@ -203,8 +203,8 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            transport.capture_participant_video(participant["id"], framerate=0)
+            await transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_video(participant["id"], framerate=0)
             ir.set_participant_id(participant["id"])
             await task.queue_frames([LLMMessagesFrame(messages)])
 

examples/patient-intake/bot.py

@@ -352,7 +352,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             print(f"Context is: {context}")
             await task.queue_frames([OpenAILLMContextFrame(context)])
 

examples/storytelling-chatbot/src/bot.py

@@ -102,7 +102,7 @@ async def main(room_url, token=None):
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             logger.debug("Participant joined, storytime commence!")
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             await intro_task.queue_frames(
                 [
                     images["book1"],

examples/studypal/studypal.py

@@ -165,7 +165,7 @@ Your task is to help the user understand and learn from this article in 2 senten
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
             messages.append(
                 {
                     "role": "system",

examples/translation-chatbot/bot.py

@@ -121,7 +121,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
 
         runner = PipelineRunner()
 

src/pipecat/transports/services/daily.py

@@ -454,27 +454,37 @@ class DailyTransportClient(EventHandler):
     def participant_counts(self):
         return self._client.participant_counts()
 
-    def start_dialout(self, settings):
-        self._client.start_dialout(settings)
+    async def start_dialout(self, settings):
+        future = self._loop.create_future()
+        self._client.start_dialout(settings, completion=completion_callback(future))
+        await future
 
-    def stop_dialout(self, participant_id):
-        self._client.stop_dialout(participant_id)
+    async def stop_dialout(self, participant_id):
+        future = self._loop.create_future()
+        self._client.stop_dialout(participant_id, completion=completion_callback(future))
+        await future
 
-    def start_recording(self, streaming_settings, stream_id, force_new):
-        self._client.start_recording(streaming_settings, stream_id, force_new)
+    async def start_recording(self, streaming_settings, stream_id, force_new):
+        future = self._loop.create_future()
+        self._client.start_recording(
+            streaming_settings, stream_id, force_new, completion=completion_callback(future)
+        )
+        await future
 
-    def stop_recording(self, stream_id):
-        self._client.stop_recording(stream_id)
+    async def stop_recording(self, stream_id):
+        future = self._loop.create_future()
+        self._client.stop_recording(stream_id, completion=completion_callback(future))
+        await future
 
-    def capture_participant_transcription(self, participant_id: str):
+    async def capture_participant_transcription(self, participant_id: str):
         if not self._params.transcription_enabled:
             return
 
         self._transcription_ids.append(participant_id)
         if self._joined and self._transcription_status:
-            self.update_transcription(self._transcription_ids)
+            await self.update_transcription(self._transcription_ids)
 
-    def capture_participant_video(
+    async def capture_participant_video(
         self,
         participant_id: str,
         callback: Callable,
@@ -483,7 +493,7 @@ class DailyTransportClient(EventHandler):
         color_format: str = "RGB",
     ):
         # Only enable camera subscription on this participant
-        self._client.update_subscriptions(
+        await self.update_subscriptions(
             participant_settings={participant_id: {"media": "subscribed"}}
         )
 
@@ -496,8 +506,12 @@ class DailyTransportClient(EventHandler):
             color_format=color_format,
         )
 
-    def update_transcription(self, participants=None, instance_id=None):
-        self._client.update_transcription(participants, instance_id)
+    async def update_transcription(self, participants=None, instance_id=None):
+        future = self._loop.create_future()
+        self._client.update_transcription(
+            participants, instance_id, completion=completion_callback(future)
+        )
+        await future
 
     async def update_subscriptions(self, participant_settings=None, profile_settings=None):
         future = self._loop.create_future()
@@ -559,7 +573,7 @@ class DailyTransportClient(EventHandler):
     def on_transcription_started(self, status):
         logger.debug(f"Transcription started: {status}")
         self._transcription_status = status
-        self.update_transcription(self._transcription_ids)
+        self._call_async_callback(self.update_transcription, self._transcription_ids)
 
     def on_transcription_stopped(self, stopped_by, stopped_by_error):
         logger.debug("Transcription stopped")
@@ -662,7 +676,7 @@ class DailyInputTransport(BaseInputTransport):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, UserImageRequestFrame):
-            self.request_participant_image(frame.user_id)
+            await self.request_participant_image(frame.user_id)
 
     #
     # Frames
@@ -692,7 +706,7 @@ class DailyInputTransport(BaseInputTransport):
     # Camera in
     #
 
-    def capture_participant_video(
+    async def capture_participant_video(
         self,
         participant_id: str,
         framerate: int = 30,
@@ -705,11 +719,11 @@ class DailyInputTransport(BaseInputTransport):
             "render_next_frame": False,
         }
 
-        self._client.capture_participant_video(
+        await self._client.capture_participant_video(
             participant_id, self._on_participant_video_frame, framerate, video_source, color_format
         )
 
-    def request_participant_image(self, participant_id: str):
+    async def request_participant_image(self, participant_id: str):
         if participant_id in self._video_renderers:
             self._video_renderers[participant_id]["render_next_frame"] = True
 
@@ -866,22 +880,22 @@ class DailyTransport(BaseTransport):
     def participant_counts(self):
         return self._client.participant_counts()
 
-    def start_dialout(self, settings=None):
-        self._client.start_dialout(settings)
+    async def start_dialout(self, settings=None):
+        await self._client.start_dialout(settings)
 
-    def stop_dialout(self, participant_id):
-        self._client.stop_dialout(participant_id)
+    async def stop_dialout(self, participant_id):
+        await self._client.stop_dialout(participant_id)
 
-    def start_recording(self, streaming_settings=None, stream_id=None, force_new=None):
-        self._client.start_recording(streaming_settings, stream_id, force_new)
+    async def start_recording(self, streaming_settings=None, stream_id=None, force_new=None):
+        await self._client.start_recording(streaming_settings, stream_id, force_new)
 
-    def stop_recording(self, stream_id=None):
-        self._client.stop_recording(stream_id)
+    async def stop_recording(self, stream_id=None):
+        await self._client.stop_recording(stream_id)
 
-    def capture_participant_transcription(self, participant_id: str):
-        self._client.capture_participant_transcription(participant_id)
+    async def capture_participant_transcription(self, participant_id: str):
+        await self._client.capture_participant_transcription(participant_id)
 
-    def capture_participant_video(
+    async def capture_participant_video(
         self,
         participant_id: str,
         framerate: int = 30,
@@ -889,7 +903,7 @@ class DailyTransport(BaseTransport):
         color_format: str = "RGB",
     ):
         if self._input:
-            self._input.capture_participant_video(
+            await self._input.capture_participant_video(
                 participant_id, framerate, video_source, color_format
             )