From 678dd22b8e682d918ee774ca8503ce94220d19c3 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 23 Sep 2025 15:29:20 -0400 Subject: [PATCH 1/6] Add missing `sender` argument to a few "on_app_message" handlers in examples --- examples/foundational/22b-natural-conversation-proposal.py | 2 +- examples/foundational/22c-natural-conversation-mixed-llms.py | 2 +- examples/foundational/22d-natural-conversation-gemini-audio.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 417aeca76..290f6624c 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -369,7 +369,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_app_message") - async def on_app_message(transport, message): + async def on_app_message(transport, message, sender): logger.debug(f"Received app message: {message}") if "message" not in message: return diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index e4c554b26..83c153f80 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -577,7 +577,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_app_message") - async def on_app_message(transport, message): + async def on_app_message(transport, message, sender): logger.debug(f"Received app message: {message}") if "message" not in message: return diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index d7ecf2ba7..2dbdac377 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -765,7 +765,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_app_message") - async def on_app_message(transport, message): + async def on_app_message(transport, message, sender): logger.debug(f"Received app message: {message}") if "message" not in message: return From 677f69971cd4d72656834e40d45f4a460113ef9b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 23 Sep 2025 16:14:36 -0400 Subject: [PATCH 2/6] Add filters in 22b and 22c examples to prevent function call results triggering the "statement judge" LLM from running unnecessarily, and with the wrong system prompt, which would result in garbled output statements comprised of both LLMs outputs combined --- examples/foundational/22b-natural-conversation-proposal.py | 4 ++++ examples/foundational/22c-natural-conversation-mixed-llms.py | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 290f6624c..a520d17be 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -237,6 +237,9 @@ class TurnDetectionLLM(Pipeline): or isinstance(frame, FunctionCallResultFrame) ) + async def filter_all(frame): + return False + super().__init__( [ ParallelPipeline( @@ -247,6 +250,7 @@ class TurnDetectionLLM(Pipeline): statement_judge_context_filter, statement_llm, completeness_check, + FunctionFilter(filter=filter_all, direction=FrameDirection.UPSTREAM), ], [ # Block everything except frames that trigger LLM inference. diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index 83c153f80..80811c2e7 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -431,6 +431,9 @@ class TurnDetectionLLM(Pipeline): or isinstance(frame, FunctionCallResultFrame) ) + async def filter_all(frame): + return False + super().__init__( [ ParallelPipeline( @@ -446,6 +449,7 @@ class TurnDetectionLLM(Pipeline): statement_judge_context_filter, statement_llm, completeness_check, + FunctionFilter(filter=filter_all, direction=FrameDirection.UPSTREAM), ], [ # Block everything except frames that trigger LLM inference. From 6ccbfd9b57be1ef7f7e7222b0fc8f874bc4673f5 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 23 Sep 2025 15:37:58 -0400 Subject: [PATCH 3/6] Update "natural conversation" examples to use universal `LLMContext` --- .../22b-natural-conversation-proposal.py | 65 +++++++++---------- .../22c-natural-conversation-mixed-llms.py | 65 +++++++++---------- 2 files changed, 58 insertions(+), 72 deletions(-) diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index a520d17be..9b3ff1d7a 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -9,8 +9,9 @@ import os from dotenv import load_dotenv from loguru import logger -from openai.types.chat import ChatCompletionToolParam +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( CancelFrame, @@ -19,6 +20,7 @@ from pipecat.frames.frames import ( FunctionCallInProgressFrame, FunctionCallResultFrame, InterruptionFrame, + LLMContextFrame, LLMRunFrame, StartFrame, SystemFrame, @@ -32,10 +34,8 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.user_idle_processor import UserIdleProcessor @@ -66,13 +66,13 @@ class StatementJudgeContextFilter(FrameProcessor): await self.push_frame(frame, direction) return - # We only want to handle OpenAILLMContextFrames, and only want to push through a simplified + # We only want to handle LLMContextFrames, and only want to push through a simplified # context frame that contains a system prompt and the most recent user messages, # concatenated. - if isinstance(frame, OpenAILLMContextFrame): + if isinstance(frame, LLMContextFrame): logger.debug(f"Context Frame: {frame}") # Take text content from the most recent user messages. - messages = frame.context.messages + messages = frame.context.get_messages() user_text_messages = [] last_assistant_message = None for message in reversed(messages): @@ -100,7 +100,7 @@ class StatementJudgeContextFilter(FrameProcessor): if last_assistant_message: messages.append(last_assistant_message) messages.append({"role": "user", "content": user_message}) - await self.push_frame(OpenAILLMContextFrame(OpenAILLMContext(messages))) + await self.push_frame(LLMContextFrame(LLMContext(messages))) class CompletenessCheck(FrameProcessor): @@ -231,7 +231,7 @@ class TurnDetectionLLM(Pipeline): async def pass_only_llm_trigger_frames(frame): return ( - isinstance(frame, OpenAILLMContextFrame) + isinstance(frame, LLMContextFrame) or isinstance(frame, InterruptionFrame) or isinstance(frame, FunctionCallInProgressFrame) or isinstance(frame, FunctionCallResultFrame) @@ -244,7 +244,7 @@ class TurnDetectionLLM(Pipeline): [ ParallelPipeline( [ - # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed + # Ignore everything except an LLMContextFrame. Pass a specially constructed # simplified context frame to the statement classifier LLM. The only frame this # sub-pipeline will output is a UserStoppedSpeakingFrame. statement_judge_context_filter, @@ -306,30 +306,23 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): async def on_function_calls_started(service, function_calls): await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) - tools = [ - ChatCompletionToolParam( - type="function", - function={ - "name": "get_current_weather", - "description": "Get the current weather", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The city and state, e.g. San Francisco, CA", - }, - "format": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "The temperature unit to use. Infer this from the users location.", - }, - }, - "required": ["location", "format"], - }, + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", }, - ) - ] + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + required=["location", "format"], + ) + tools = ToolsSchema(standard_tools=[weather_function]) messages = [ { @@ -338,8 +331,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ] - context = OpenAILLMContext(messages, tools) - context_aggregator = llm_main.create_context_aggregator(context) + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) # LLM + turn detection (with an extra LLM as a judge) llm = TurnDetectionLLM(llm_main) diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index 80811c2e7..461fab08d 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -9,8 +9,9 @@ import os from dotenv import load_dotenv from loguru import logger -from openai.types.chat import ChatCompletionToolParam +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( CancelFrame, @@ -19,6 +20,7 @@ from pipecat.frames.frames import ( FunctionCallInProgressFrame, FunctionCallResultFrame, InterruptionFrame, + LLMContextFrame, LLMRunFrame, StartFrame, SystemFrame, @@ -32,10 +34,8 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.user_idle_processor import UserIdleProcessor @@ -272,11 +272,11 @@ class StatementJudgeContextFilter(FrameProcessor): await self.push_frame(frame, direction) return - # We only want to handle OpenAILLMContextFrames, and only want to push through a simplified + # We only want to handle LLMContextFrames, and only want to push through a simplified # context frame that contains a system prompt and the most recent user messages, - if isinstance(frame, OpenAILLMContextFrame): + if isinstance(frame, LLMContextFrame): # Take text content from the most recent user messages. - messages = frame.context.messages + messages = frame.context.get_messages() user_text_messages = [] last_assistant_message = None for message in reversed(messages): @@ -303,7 +303,7 @@ class StatementJudgeContextFilter(FrameProcessor): if last_assistant_message: messages.append(last_assistant_message) messages.append({"role": "user", "content": user_message}) - await self.push_frame(OpenAILLMContextFrame(OpenAILLMContext(messages))) + await self.push_frame(LLMContextFrame(LLMContext(messages))) class CompletenessCheck(FrameProcessor): @@ -425,7 +425,7 @@ class TurnDetectionLLM(Pipeline): async def pass_only_llm_trigger_frames(frame): return ( - isinstance(frame, OpenAILLMContextFrame) + isinstance(frame, LLMContextFrame) or isinstance(frame, InterruptionFrame) or isinstance(frame, FunctionCallInProgressFrame) or isinstance(frame, FunctionCallResultFrame) @@ -443,7 +443,7 @@ class TurnDetectionLLM(Pipeline): FunctionFilter(filter=block_user_stopped_speaking), ], [ - # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed + # Ignore everything except an LLMContextFrame. Pass a specially constructed # simplified context frame to the statement classifier LLM. The only frame this # sub-pipeline will output is a UserStoppedSpeakingFrame. statement_judge_context_filter, @@ -509,30 +509,23 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): async def on_function_calls_started(service, function_calls): await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) - tools = [ - ChatCompletionToolParam( - type="function", - function={ - "name": "get_current_weather", - "description": "Get the current weather", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The city and state, e.g. San Francisco, CA", - }, - "format": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "The temperature unit to use. Infer this from the users location.", - }, - }, - "required": ["location", "format"], - }, + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", }, - ) - ] + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + required=["location", "format"], + ) + tools = ToolsSchema(standard_tools=[weather_function]) messages = [ { @@ -541,8 +534,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ] - context = OpenAILLMContext(messages, tools) - context_aggregator = llm_main.create_context_aggregator(context) + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) # LLM + turn detection (with an extra LLM as a judge) llm = TurnDetectionLLM(llm_main) From 66b7977a625aee7e58465de85019c9f8b90b6576 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 24 Sep 2025 09:34:28 -0400 Subject: [PATCH 4/6] Make `SmallWebRTCTransport` adhere to the expected "on_app_message" event signature --- src/pipecat/transports/smallwebrtc/transport.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 4a9ba9341..79f50cdb2 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -939,7 +939,7 @@ class SmallWebRTCTransport(BaseTransport): """Handle incoming application messages.""" if self._input: await self._input.push_app_message(message) - await self._call_event_handler("on_app_message", message) + await self._call_event_handler("on_app_message", message, "") async def _on_client_connected(self, webrtc_connection): """Handle client connection events.""" From 8896179b00f8241abcfcc7de4ad2d145b24e2600 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 24 Sep 2025 09:55:11 -0400 Subject: [PATCH 5/6] Update another "natural conversation" example to use universal `LLMContext`. Note that this one had to also be fixed in various ways, as it wasn't working. --- .../22d-natural-conversation-gemini-audio.py | 61 +++++++++---------- 1 file changed, 30 insertions(+), 31 deletions(-) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 2dbdac377..5819c9cb9 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -9,7 +9,6 @@ import os import time from dotenv import load_dotenv -from google.genai.types import Content, Part from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer @@ -21,6 +20,7 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, InputAudioRawFrame, InterruptionFrame, + LLMContextFrame, LLMFullResponseStartFrame, LLMRunFrame, StartFrame, @@ -34,20 +34,18 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMAssistantResponseAggregator, ) -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService -from pipecat.services.google.llm import GoogleLLMContext, GoogleLLMService +from pipecat.services.google.llm import GoogleLLMService from pipecat.services.llm_service import LLMService from pipecat.sync.base_notifier import BaseNotifier from pipecat.sync.event_notifier import EventNotifier @@ -375,7 +373,7 @@ class AudioAccumulator(FrameProcessor): await super().process_frame(frame, direction) # ignore context frame - if isinstance(frame, OpenAILLMContextFrame): + if isinstance(frame, LLMContextFrame): return if isinstance(frame, TranscriptionFrame): @@ -392,9 +390,9 @@ class AudioAccumulator(FrameProcessor): f"Processing audio buffer seconds: ({len(self._audio_frames)}) ({len(data)}) {len(data) / 2 / 16000}" ) self._user_speaking = False - context = GoogleLLMContext() + context = LLMContext() context.add_audio_frames_message(audio_frames=self._audio_frames) - await self.push_frame(OpenAILLMContextFrame(context=context)) + await self.push_frame(LLMContextFrame(context=context)) elif isinstance(frame, InputAudioRawFrame): # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest # frames as necessary. @@ -513,7 +511,7 @@ class LLMAggregatorBuffer(LLMAssistantResponseAggregator): class ConversationAudioContextAssembler(FrameProcessor): """Takes the single-message context generated by the AudioAccumulator and adds it to the conversation LLM's context.""" - def __init__(self, context: OpenAILLMContext, **kwargs): + def __init__(self, context: LLMContext, **kwargs): super().__init__(**kwargs) self._context = context @@ -525,11 +523,10 @@ class ConversationAudioContextAssembler(FrameProcessor): await self.push_frame(frame, direction) return - if isinstance(frame, OpenAILLMContextFrame): - GoogleLLMContext.upgrade_to_google(self._context) - last_message = frame.context.messages[-1] + if isinstance(frame, LLMContextFrame): + last_message = frame.context.get_messages()[-1] self._context._messages.append(last_message) - await self.push_frame(OpenAILLMContextFrame(context=self._context)) + await self.push_frame(LLMContextFrame(context=self._context)) class OutputGate(FrameProcessor): @@ -543,7 +540,7 @@ class OutputGate(FrameProcessor): def __init__( self, notifier: BaseNotifier, - context: OpenAILLMContext, + context: LLMContext, llm_transcription_buffer: LLMAggregatorBuffer, **kwargs, ): @@ -610,19 +607,23 @@ class OutputGate(FrameProcessor): self._gate_task = None async def _gate_task_handler(self): - await self._notifier.wait() + while True: + try: + await self._notifier.wait() - transcription = await self._transcription_buffer.wait_for_transcription() or "-" - self._context.add_message(Content(role="user", parts=[Part(text=transcription)])) + transcription = await self._transcription_buffer.wait_for_transcription() or "-" + self._context.add_message({"role": "user", "content": transcription}) - self.open_gate() - for frame, direction in self._frames_buffer: - await self.push_frame(frame, direction) - self._frames_buffer = [] + self.open_gate() + for frame, direction in self._frames_buffer: + await self.push_frame(frame, direction) + self._frames_buffer = [] + except asyncio.CancelledError: + break class TurnDetectionLLM(Pipeline): - def __init__(self, llm: LLMService, context: OpenAILLMContext): + def __init__(self, llm: LLMService, context: LLMContext): # This is the LLM that will transcribe user speech. tx_llm = GoogleLLMService( name="Transcriber", @@ -648,10 +649,10 @@ class TurnDetectionLLM(Pipeline): # as complete or incomplete. # statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier) - audio_accumulater = AudioAccumulator() + audio_accumulator = AudioAccumulator() # This sends a UserStoppedSpeakingFrame and triggers the notifier event completeness_check = CompletenessCheck( - notifier=notifier, audio_accumulator=audio_accumulater + notifier=notifier, audio_accumulator=audio_accumulator ) async def block_user_stopped_speaking(frame): @@ -667,7 +668,7 @@ class TurnDetectionLLM(Pipeline): super().__init__( [ - audio_accumulater, + audio_accumulator, ParallelPipeline( [ # Pass everything except UserStoppedSpeaking to the elements after @@ -734,8 +735,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): system_instruction=conversation_system_instruction, ) - context = OpenAILLMContext() - context_aggregator = conversation_llm.create_context_aggregator(context) + context = LLMContext() + context_aggregator = LLMContextAggregatorPair(context) llm = TurnDetectionLLM(conversation_llm, context) @@ -761,12 +762,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") - # Kick off the conversation. - await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_app_message") async def on_app_message(transport, message, sender): - logger.debug(f"Received app message: {message}") + logger.debug(f"Received app message: {message}, sender: {sender}") # TODO: revert if "message" not in message: return From 817c77f3fe1a4eaf79f19ba493ca2fbfedc33e8b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 24 Sep 2025 10:24:59 -0400 Subject: [PATCH 6/6] Update `SmallWebRTCTransport` to pass a sender ID in the "on_app_message" event --- src/pipecat/transports/smallwebrtc/transport.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 79f50cdb2..4b2437be1 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -66,7 +66,7 @@ class SmallWebRTCCallbacks(BaseModel): on_client_disconnected: Called when a client disconnects. """ - on_app_message: Callable[[Any], Awaitable[None]] + on_app_message: Callable[[Any, str], Awaitable[None]] on_client_connected: Callable[[SmallWebRTCConnection], Awaitable[None]] on_client_disconnected: Callable[[SmallWebRTCConnection], Awaitable[None]] @@ -254,7 +254,7 @@ class SmallWebRTCClient: @self._webrtc_connection.event_handler("app-message") async def on_app_message(connection: SmallWebRTCConnection, message: Any): - await self._handle_app_message(message) + await self._handle_app_message(message, connection.pc_id) def _convert_frame(self, frame_array: np.ndarray, format_name: str) -> np.ndarray: """Convert a video frame to RGB format based on the input format. @@ -512,9 +512,9 @@ class SmallWebRTCClient: if not self._closing: await self._callbacks.on_client_disconnected(self._webrtc_connection) - async def _handle_app_message(self, message: Any): + async def _handle_app_message(self, message: Any, sender: str): """Handle incoming application messages.""" - await self._callbacks.on_app_message(message) + await self._callbacks.on_app_message(message, sender) def _can_send(self): """Check if the connection is ready for sending data.""" @@ -935,11 +935,11 @@ class SmallWebRTCTransport(BaseTransport): if self._output: await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) - async def _on_app_message(self, message: Any): + async def _on_app_message(self, message: Any, sender: str): """Handle incoming application messages.""" if self._input: await self._input.push_app_message(message) - await self._call_event_handler("on_app_message", message, "") + await self._call_event_handler("on_app_message", message, sender) async def _on_client_connected(self, webrtc_connection): """Handle client connection events."""