From 9bd51cd88c7182246261486bb48087cbadf0a5b7 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 18:04:42 -0400 Subject: [PATCH 1/9] Add incremental pyright type checking with CI enforcement Add pyrightconfig.json with basic type checking for zero-error modules (clocks, metrics, transcriptions, frames) and enforce via CI. The include list will expand as modules are fixed. --- .github/workflows/format.yaml | 4 ++++ pyrightconfig.json | 16 ++++++++++++++++ 2 files changed, 20 insertions(+) create mode 100644 pyrightconfig.json diff --git a/.github/workflows/format.yaml b/.github/workflows/format.yaml index 8284be043..c02245708 100644 --- a/.github/workflows/format.yaml +++ b/.github/workflows/format.yaml @@ -41,3 +41,7 @@ jobs: - name: Ruff linter (all rules) id: ruff-check run: uv run ruff check + + - name: Type check (pyright) + id: pyright + run: uv run pyright diff --git a/pyrightconfig.json b/pyrightconfig.json new file mode 100644 index 000000000..fbca2b790 --- /dev/null +++ b/pyrightconfig.json @@ -0,0 +1,16 @@ +{ + "typeCheckingMode": "basic", + "pythonVersion": "3.11", + "pythonPlatform": "All", + "include": [ + "src/pipecat/clocks", + "src/pipecat/metrics", + "src/pipecat/transcriptions", + "src/pipecat/frames" + ], + "exclude": [ + "**/*_pb2.py", + "**/__pycache__" + ], + "reportMissingImports": false +} From aa355e3d32e79a98ef8427ebacc505f8895a7a98 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 18:25:10 -0400 Subject: [PATCH 2/9] Fix type errors in observers and add to pyright checked set Group three co-assigned fields (_start_frame_id, _start_frame_arrival_ns, _start_wall_clock) into a single _StartFrameInfo dataclass. This makes the "always set together" invariant structural rather than implicit, and fixes the incorrect str | None annotation on _start_frame_id (Frame.id is int). --- pyrightconfig.json | 3 +- .../observers/startup_timing_observer.py | 49 ++++++++++--------- 2 files changed, 29 insertions(+), 23 deletions(-) diff --git a/pyrightconfig.json b/pyrightconfig.json index fbca2b790..69d2ba8db 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -6,7 +6,8 @@ "src/pipecat/clocks", "src/pipecat/metrics", "src/pipecat/transcriptions", - "src/pipecat/frames" + "src/pipecat/frames", + "src/pipecat/observers" ], "exclude": [ "**/*_pb2.py", diff --git a/src/pipecat/observers/startup_timing_observer.py b/src/pipecat/observers/startup_timing_observer.py index 6c6eb8204..ac63ab10a 100644 --- a/src/pipecat/observers/startup_timing_observer.py +++ b/src/pipecat/observers/startup_timing_observer.py @@ -49,6 +49,15 @@ from pipecat.processors.frame_processor import FrameProcessor _INTERNAL_TYPES = (PipelineSource, BasePipeline) +@dataclass +class _StartFrameInfo: + """Captured once when the first StartFrame arrives at a processor.""" + + frame_id: int + arrival_ns: int + wall_clock: float + + @dataclass class _ArrivalInfo: """Internal record of when a StartFrame arrived at a processor.""" @@ -175,8 +184,8 @@ class StartupTimingObserver(BaseObserver): # Collected timings in pipeline order. self._timings: list[ProcessorStartupTiming] = [] - # Lock onto the first StartFrame we see (by frame ID). - self._start_frame_id: str | None = None + # Captured once when the first StartFrame arrives. + self._start_frame: _StartFrameInfo | None = None # Whether we've already emitted the startup timing report. self._startup_timing_reported = False @@ -184,15 +193,9 @@ class StartupTimingObserver(BaseObserver): # Whether we've already measured transport timing. self._transport_timing_reported = False - # Timestamp (ns) when we first see a StartFrame arrive at a processor. - self._start_frame_arrival_ns: int | None = None - # Bot connected timing (stored for inclusion in the transport report). self._bot_connected_secs: float | None = None - # Wall clock time when the StartFrame was first seen. - self._start_wall_clock: float | None = None - self._register_event_handler("on_startup_timing_report") self._register_event_handler("on_transport_timing_report") @@ -233,11 +236,13 @@ class StartupTimingObserver(BaseObserver): return # Lock onto the first StartFrame. - if self._start_frame_id is None: - self._start_frame_id = data.frame.id - self._start_frame_arrival_ns = data.timestamp - self._start_wall_clock = time.time() - elif data.frame.id != self._start_frame_id: + if self._start_frame is None: + self._start_frame = _StartFrameInfo( + frame_id=data.frame.id, + arrival_ns=data.timestamp, + wall_clock=time.time(), + ) + elif data.frame.id != self._start_frame.frame_id: return if self._should_track(data.processor): @@ -268,16 +273,16 @@ class StartupTimingObserver(BaseObserver): if not isinstance(data.frame, StartFrame): return - if self._start_frame_id is not None and data.frame.id != self._start_frame_id: + if self._start_frame is not None and data.frame.id != self._start_frame.frame_id: return arrival = self._arrivals.pop(data.source.id, None) - if arrival is None: + if arrival is None or self._start_frame is None: return duration_ns = data.timestamp - arrival.arrival_ts_ns duration_secs = duration_ns / 1e9 - start_offset_secs = (arrival.arrival_ts_ns - self._start_frame_arrival_ns) / 1e9 + start_offset_secs = (arrival.arrival_ts_ns - self._start_frame.arrival_ns) / 1e9 self._timings.append( ProcessorStartupTiming( @@ -289,22 +294,22 @@ class StartupTimingObserver(BaseObserver): def _handle_bot_connected(self, data: FramePushed): """Record bot connected timing on first BotConnectedFrame.""" - if self._bot_connected_secs is not None or self._start_frame_arrival_ns is None: + if self._bot_connected_secs is not None or self._start_frame is None: return - delta_ns = data.timestamp - self._start_frame_arrival_ns + delta_ns = data.timestamp - self._start_frame.arrival_ns self._bot_connected_secs = delta_ns / 1e9 async def _handle_client_connected(self, data: FramePushed): """Emit transport timing report on first ClientConnectedFrame.""" - if self._transport_timing_reported or self._start_frame_arrival_ns is None: + if self._transport_timing_reported or self._start_frame is None: return self._transport_timing_reported = True - delta_ns = data.timestamp - self._start_frame_arrival_ns + delta_ns = data.timestamp - self._start_frame.arrival_ns client_connected_secs = delta_ns / 1e9 report = TransportTimingReport( - start_time=self._start_wall_clock or 0.0, + start_time=self._start_frame.wall_clock, bot_connected_secs=self._bot_connected_secs, client_connected_secs=client_connected_secs, ) @@ -319,7 +324,7 @@ class StartupTimingObserver(BaseObserver): total = sum(t.duration_secs for t in self._timings) report = StartupTimingReport( - start_time=self._start_wall_clock or 0.0, + start_time=self._start_frame.wall_clock if self._start_frame else 0.0, total_duration_secs=total, processor_timings=self._timings, ) From c6a18378441eacddd18393abd9b60940617bf1b1 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 21:22:46 -0400 Subject: [PATCH 3/9] Fix type errors in extensions and add to pyright checked set Tighten LLMMessagesAppendFrame and LLMMessagesUpdateFrame message fields from list[dict] to list[LLMContextMessage] to match actual usage. Add type annotations on inline message lists in IVR navigator and voicemail detector. --- pyrightconfig.json | 3 ++- src/pipecat/extensions/ivr/ivr_navigator.py | 15 +++++++++------ .../extensions/voicemail/voicemail_detector.py | 6 +++--- src/pipecat/frames/frames.py | 9 ++++----- 4 files changed, 18 insertions(+), 15 deletions(-) diff --git a/pyrightconfig.json b/pyrightconfig.json index 69d2ba8db..ea03dc37b 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -7,7 +7,8 @@ "src/pipecat/metrics", "src/pipecat/transcriptions", "src/pipecat/frames", - "src/pipecat/observers" + "src/pipecat/observers", + "src/pipecat/extensions" ], "exclude": [ "**/*_pb2.py", diff --git a/src/pipecat/extensions/ivr/ivr_navigator.py b/src/pipecat/extensions/ivr/ivr_navigator.py index f6b36655b..37b2469bb 100644 --- a/src/pipecat/extensions/ivr/ivr_navigator.py +++ b/src/pipecat/extensions/ivr/ivr_navigator.py @@ -30,6 +30,7 @@ from pipecat.frames.frames import ( VADParamsUpdateFrame, ) from pipecat.pipeline.pipeline import Pipeline +from pipecat.processors.aggregators.llm_context import LLMContextMessage from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.llm_service import LLMService from pipecat.utils.text.pattern_pair_aggregator import ( @@ -87,7 +88,7 @@ class IVRProcessor(FrameProcessor): self._classifier_prompt = classifier_prompt # Store saved context messages - self._saved_messages: list[dict] = [] + self._saved_messages: list[LLMContextMessage] = [] # XML pattern aggregation self._aggregator = PatternPairAggregator() @@ -97,18 +98,18 @@ class IVRProcessor(FrameProcessor): self._register_event_handler("on_conversation_detected") self._register_event_handler("on_ivr_status_changed") - def update_saved_messages(self, messages: list[dict]) -> None: + def update_saved_messages(self, messages: list[LLMContextMessage]) -> None: """Update the saved context messages. Sets the messages that are saved when switching between conversation and IVR navigation modes. Args: - messages: List of message dictionaries to save. + messages: List of context messages to save. """ self._saved_messages = messages - def _get_conversation_history(self) -> list[dict]: + def _get_conversation_history(self) -> list[LLMContextMessage]: """Get saved context messages without the system message. Returns: @@ -144,7 +145,9 @@ class IVRProcessor(FrameProcessor): await self.push_frame(frame, direction) # Set the classifier prompt and push it upstream - messages = [{"role": "system", "content": self._classifier_prompt}] + messages: list[LLMContextMessage] = [ + {"role": "developer", "content": self._classifier_prompt} + ] llm_update_frame = LLMMessagesUpdateFrame(messages=messages) await self.push_frame(llm_update_frame, FrameDirection.UPSTREAM) @@ -261,7 +264,7 @@ class IVRProcessor(FrameProcessor): logger.debug("IVR detected - switching to IVR navigation mode") # Create new context with IVR system prompt and saved messages - messages = [{"role": "system", "content": self._ivr_prompt}] + messages: list[LLMContextMessage] = [{"role": "developer", "content": self._ivr_prompt}] # Add saved conversation history if available conversation_history = self._get_conversation_history() diff --git a/src/pipecat/extensions/voicemail/voicemail_detector.py b/src/pipecat/extensions/voicemail/voicemail_detector.py index 3ab7f2f7d..d7e9d559b 100644 --- a/src/pipecat/extensions/voicemail/voicemail_detector.py +++ b/src/pipecat/extensions/voicemail/voicemail_detector.py @@ -35,7 +35,7 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.pipeline.parallel_pipeline import ParallelPipeline -from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage from pipecat.processors.aggregators.llm_response_universal import ( LLMContextAggregatorPair, LLMUserAggregatorParams, @@ -617,9 +617,9 @@ VOICEMAIL SYSTEM (respond "VOICEMAIL"): self._validate_prompt(custom_system_prompt) # Set up the LLM context with the classification prompt - self._messages = [ + self._messages: list[LLMContextMessage] = [ { - "role": "system", + "role": "developer", "content": self._prompt, }, ] diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 7fd215caf..24006428d 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -20,7 +20,6 @@ from typing import ( TYPE_CHECKING, Any, Literal, - Optional, ) from pipecat.adapters.schemas.tools_schema import ToolsSchema @@ -559,11 +558,11 @@ class LLMMessagesAppendFrame(DataFrame): current context. Parameters: - messages: List of message dictionaries to append. + messages: List of context messages to append. run_llm: Whether the context update should be sent to the LLM. """ - messages: list[dict] + messages: list[LLMContextMessage] run_llm: bool | None = None @@ -575,11 +574,11 @@ class LLMMessagesUpdateFrame(DataFrame): context LLM messages. Parameters: - messages: List of message dictionaries to replace current context. + messages: List of context messages to replace current context. run_llm: Whether the context update should be sent to the LLM. """ - messages: list[dict] + messages: list[LLMContextMessage] run_llm: bool | None = None From 36319ecbf082e84a89fac28fba70cc4305511890 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 21:26:08 -0400 Subject: [PATCH 4/9] Replace system role message In UserTurnCompletionMixin, use a developer role message for LLM messages following an incomplete turn --- src/pipecat/turns/user_turn_completion_mixin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/turns/user_turn_completion_mixin.py b/src/pipecat/turns/user_turn_completion_mixin.py index 1d8da1e07..ab5397c06 100644 --- a/src/pipecat/turns/user_turn_completion_mixin.py +++ b/src/pipecat/turns/user_turn_completion_mixin.py @@ -292,7 +292,7 @@ class UserTurnCompletionLLMServiceMixin: # Push through pipeline to trigger LLM response await self.push_frame( - LLMMessagesAppendFrame(messages=[{"role": "system", "content": prompt}]) + LLMMessagesAppendFrame(messages=[{"role": "developer", "content": prompt}]) ) await self.push_frame(LLMRunFrame()) From 3127cc6161b4841c42edac1cd992d6d26cbcb6b8 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 21:33:43 -0400 Subject: [PATCH 5/9] Fix type errors in turns and add to pyright checked set Widen base strategy process_frame return types to ProcessFrameResult | None to match actual behavior (None treated as CONTINUE). Give UserTurnCompletionLLMServiceMixin a FrameProcessor base class so pyright can see create_task, cancel_task, process_frame, and push_frame. --- pyrightconfig.json | 3 ++- .../turns/user_start/base_user_turn_start_strategy.py | 6 +++--- src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py | 6 +++--- src/pipecat/turns/user_turn_completion_mixin.py | 4 ++-- 4 files changed, 10 insertions(+), 9 deletions(-) diff --git a/pyrightconfig.json b/pyrightconfig.json index ea03dc37b..4016fd6e9 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -8,7 +8,8 @@ "src/pipecat/transcriptions", "src/pipecat/frames", "src/pipecat/observers", - "src/pipecat/extensions" + "src/pipecat/extensions", + "src/pipecat/turns" ], "exclude": [ "**/*_pb2.py", diff --git a/src/pipecat/turns/user_start/base_user_turn_start_strategy.py b/src/pipecat/turns/user_start/base_user_turn_start_strategy.py index 401178cc7..b424d7a9e 100644 --- a/src/pipecat/turns/user_start/base_user_turn_start_strategy.py +++ b/src/pipecat/turns/user_start/base_user_turn_start_strategy.py @@ -101,7 +101,7 @@ class BaseUserTurnStartStrategy(BaseObject): """Reset the strategy to its initial state.""" pass - async def process_frame(self, frame: Frame) -> ProcessFrameResult: + async def process_frame(self, frame: Frame) -> ProcessFrameResult | None: """Process an incoming frame. Subclasses should override this to implement logic that decides whether @@ -111,8 +111,8 @@ class BaseUserTurnStartStrategy(BaseObject): frame: The frame to be processed. Returns: - A ProcessFrameResult indicating the outcome. Subclasses that return - None are treated as CONTINUE for backward compatibility. + A ProcessFrameResult indicating the outcome, or None (treated as + CONTINUE for backward compatibility). """ pass diff --git a/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py index 1f8497359..e5dc1bd66 100644 --- a/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py +++ b/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py @@ -89,7 +89,7 @@ class BaseUserTurnStopStrategy(BaseObject): """Reset the strategy to its initial state.""" pass - async def process_frame(self, frame: Frame) -> ProcessFrameResult: + async def process_frame(self, frame: Frame) -> ProcessFrameResult | None: """Process an incoming frame to decide whether the user stopped speaking. Subclasses should override this to implement logic that decides whether @@ -99,8 +99,8 @@ class BaseUserTurnStopStrategy(BaseObject): frame: The frame to be analyzed. Returns: - A ProcessFrameResult indicating the outcome. Subclasses that return - None are treated as CONTINUE for backward compatibility. + A ProcessFrameResult indicating the outcome, or None (treated as + CONTINUE for backward compatibility). """ pass diff --git a/src/pipecat/turns/user_turn_completion_mixin.py b/src/pipecat/turns/user_turn_completion_mixin.py index ab5397c06..51d6c7828 100644 --- a/src/pipecat/turns/user_turn_completion_mixin.py +++ b/src/pipecat/turns/user_turn_completion_mixin.py @@ -26,7 +26,7 @@ from pipecat.frames.frames import ( LLMRunFrame, LLMTextFrame, ) -from pipecat.processors.frame_processor import FrameDirection +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor # Turn completion markers USER_TURN_COMPLETE_MARKER = "✓" @@ -178,7 +178,7 @@ class UserTurnCompletionConfig: return self.incomplete_long_prompt or DEFAULT_INCOMPLETE_LONG_PROMPT -class UserTurnCompletionLLMServiceMixin: +class UserTurnCompletionLLMServiceMixin(FrameProcessor): """Mixin that adds turn completion detection to LLM services. This mixin provides methods to push LLM text with turn completion detection. From ab91047300161fabbcca96d96bb8ffee65f3ac8e Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 21:47:11 -0400 Subject: [PATCH 6/9] Fix type errors in pipeline and add to pyright checked set Use Sequence[FrameProcessor] instead of list[FrameProcessor] in Pipeline, ServiceSwitcher, and ServiceSwitcherStrategy parameters to accept subtype lists. Add cast() in LLMSwitcher for narrowed return types. Guard against None in task_observer._send_to_proxy and replace hasattr with truthiness check in task._cleanup. --- pyrightconfig.json | 3 ++- src/pipecat/pipeline/llm_switcher.py | 6 +++--- src/pipecat/pipeline/pipeline.py | 8 ++++---- src/pipecat/pipeline/service_switcher.py | 11 ++++++----- src/pipecat/pipeline/task.py | 2 +- src/pipecat/pipeline/task_observer.py | 2 ++ 6 files changed, 18 insertions(+), 14 deletions(-) diff --git a/pyrightconfig.json b/pyrightconfig.json index 4016fd6e9..7176ab774 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -9,7 +9,8 @@ "src/pipecat/frames", "src/pipecat/observers", "src/pipecat/extensions", - "src/pipecat/turns" + "src/pipecat/turns", + "src/pipecat/pipeline" ], "exclude": [ "**/*_pb2.py", diff --git a/src/pipecat/pipeline/llm_switcher.py b/src/pipecat/pipeline/llm_switcher.py index 71a7c7974..031f9e875 100644 --- a/src/pipecat/pipeline/llm_switcher.py +++ b/src/pipecat/pipeline/llm_switcher.py @@ -6,7 +6,7 @@ """LLM switcher for switching between different LLMs at runtime, with different switching strategies.""" -from typing import Any +from typing import Any, cast from pipecat.adapters.schemas.direct_function import DirectFunction from pipecat.pipeline.service_switcher import ( @@ -47,7 +47,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]): Returns: List of LLM services managed by this switcher. """ - return self.services + return cast(list[LLMService], self.services) @property def active_llm(self) -> LLMService: @@ -56,7 +56,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]): Returns: The currently active LLM service, or None if no LLM is active. """ - return self.strategy.active_service + return cast(LLMService, self.strategy.active_service) async def run_inference(self, context: LLMContext, **kwargs) -> str | None: """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context, using the currently active LLM. diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index 325cedb82..d7999b4fe 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -11,7 +11,7 @@ in sequence and manages frame flow between them, along with helper classes for pipeline source and sink operations. """ -from collections.abc import Callable, Coroutine +from collections.abc import Callable, Coroutine, Sequence from pipecat.frames.frames import Frame from pipecat.pipeline.base_pipeline import BasePipeline @@ -98,7 +98,7 @@ class Pipeline(BasePipeline): def __init__( self, - processors: list[FrameProcessor], + processors: Sequence[FrameProcessor], *, source: FrameProcessor | None = None, sink: FrameProcessor | None = None, @@ -106,7 +106,7 @@ class Pipeline(BasePipeline): """Initialize the pipeline with a list of processors. Args: - processors: List of frame processors to connect in sequence. + processors: Sequence of frame processors to connect in sequence. source: An optional pipeline source processor. sink: An optional pipeline sink processor. """ @@ -116,7 +116,7 @@ class Pipeline(BasePipeline): # downstream outside of the pipeline. self._source = source or PipelineSource(self.push_frame, name=f"{self}::Source") self._sink = sink or PipelineSink(self.push_frame, name=f"{self}::Sink") - self._processors: list[FrameProcessor] = [self._source] + processors + [self._sink] + self._processors: list[FrameProcessor] = [self._source, *processors, self._sink] self._link_processors() diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py index a7f98f9b1..dd5e330e9 100644 --- a/src/pipecat/pipeline/service_switcher.py +++ b/src/pipecat/pipeline/service_switcher.py @@ -6,6 +6,7 @@ """Service switcher for switching between different services at runtime, with different switching strategies.""" +from collections.abc import Sequence from typing import Any, Generic, TypeVar from loguru import logger @@ -42,7 +43,7 @@ class ServiceSwitcherStrategy(BaseObject): ... """ - def __init__(self, services: list[FrameProcessor]): + def __init__(self, services: Sequence[FrameProcessor]): """Initialize the service switcher strategy with a list of services. Note: @@ -56,7 +57,7 @@ class ServiceSwitcherStrategy(BaseObject): if len(services) == 0: raise Exception(f"ServiceSwitcherStrategy needs at least one service") - self._services = services + self._services = list(services) self._active_service = services[0] self._register_event_handler("on_service_switched") @@ -223,7 +224,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): def __init__( self, - services: list[FrameProcessor], + services: Sequence[FrameProcessor], strategy_type: type[StrategyType] = ServiceSwitcherStrategyManual, ): """Initialize the service switcher with a list of services and a switching strategy. @@ -235,7 +236,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): """ _strategy = strategy_type(services) super().__init__(*self._make_pipeline_definitions(services, _strategy)) - self._services = services + self._services = list(services) self._strategy = _strategy @property @@ -250,7 +251,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): @staticmethod def _make_pipeline_definitions( - services: list[FrameProcessor], strategy: ServiceSwitcherStrategy + services: Sequence[FrameProcessor], strategy: ServiceSwitcherStrategy ) -> list[Any]: pipelines = [] for service in services: diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 394b3d2e2..08e08ea00 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -742,7 +742,7 @@ class PipelineTask(BasePipelineTask): await self._observer.cleanup() # End conversation tracing if it's active - this will also close any active turn span - if self._enable_tracing and hasattr(self, "_turn_trace_observer"): + if self._enable_tracing and self._turn_trace_observer: self._turn_trace_observer.end_conversation_tracing() # Cleanup pipeline processors. diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index c6603c1d8..5b29850a6 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -173,6 +173,8 @@ class TaskObserver(BaseObserver): return proxies async def _send_to_proxy(self, data: Any): + if not self._proxies: + return for proxy in self._proxies.values(): await proxy.queue.put(data) From cb1463f9f125ae01fbdaa619fd38ca8765b65770 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 22:05:01 -0400 Subject: [PATCH 7/9] Fix type errors in runner and add to pyright checked set Make required parameters non-optional: LiveKitRunnerArguments.token, _create_telephony_transport args. Use os.environ[] instead of os.getenv() for required WhatsApp env vars. Guard spec/loader None in module loading. Tighten sip_caller_phone guard in daily.py. --- pyrightconfig.json | 3 ++- src/pipecat/runner/daily.py | 2 +- src/pipecat/runner/run.py | 35 ++++++++++++++++------------------- src/pipecat/runner/types.py | 2 +- src/pipecat/runner/utils.py | 12 +++--------- 5 files changed, 23 insertions(+), 31 deletions(-) diff --git a/pyrightconfig.json b/pyrightconfig.json index 7176ab774..17359ed4a 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -10,7 +10,8 @@ "src/pipecat/observers", "src/pipecat/extensions", "src/pipecat/turns", - "src/pipecat/pipeline" + "src/pipecat/pipeline", + "src/pipecat/runner" ], "exclude": [ "**/*_pb2.py", diff --git a/src/pipecat/runner/daily.py b/src/pipecat/runner/daily.py index bc80f0641..168b2353d 100644 --- a/src/pipecat/runner/daily.py +++ b/src/pipecat/runner/daily.py @@ -228,7 +228,7 @@ async def configure( room_properties.enable_dialout = True # Add SIP configuration if enabled - if sip_enabled: + if sip_enabled and sip_caller_phone: sip_params = DailyRoomSipParams( display_name=sip_caller_phone, video=sip_enable_video, diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index c6d43fbbd..ca296bd78 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -156,6 +156,8 @@ def _get_bot_module(): spec = importlib.util.spec_from_file_location( module_name, os.path.join(cwd, filename) ) + if spec is None or spec.loader is None: + continue module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) @@ -386,29 +388,24 @@ def _add_lifespan_to_app(app: FastAPI, new_lifespan): def _setup_whatsapp_routes(app: FastAPI, args: argparse.Namespace): """Set up WhatsApp-specific routes.""" - WHATSAPP_APP_SECRET = os.getenv("WHATSAPP_APP_SECRET") - WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") - WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN") - - if not all( - [ - WHATSAPP_APP_SECRET, - WHATSAPP_PHONE_NUMBER_ID, - WHATSAPP_TOKEN, - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN, - ] - ): + required_vars = [ + "WHATSAPP_APP_SECRET", + "WHATSAPP_PHONE_NUMBER_ID", + "WHATSAPP_TOKEN", + "WHATSAPP_WEBHOOK_VERIFICATION_TOKEN", + ] + missing = [v for v in required_vars if not os.getenv(v)] + if missing: logger.error( - """Missing required environment variables for WhatsApp transport: - WHATSAPP_APP_SECRET - WHATSAPP_PHONE_NUMBER_ID - WHATSAPP_TOKEN - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN - """ + f"Missing required environment variables for WhatsApp transport: {', '.join(missing)}" ) return + WHATSAPP_APP_SECRET = os.environ["WHATSAPP_APP_SECRET"] + WHATSAPP_PHONE_NUMBER_ID = os.environ["WHATSAPP_PHONE_NUMBER_ID"] + WHATSAPP_TOKEN = os.environ["WHATSAPP_TOKEN"] + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.environ["WHATSAPP_WEBHOOK_VERIFICATION_TOKEN"] + try: from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest diff --git a/src/pipecat/runner/types.py b/src/pipecat/runner/types.py index 055824a22..b009b3350 100644 --- a/src/pipecat/runner/types.py +++ b/src/pipecat/runner/types.py @@ -122,4 +122,4 @@ class LiveKitRunnerArguments(RunnerArguments): room_name: str url: str - token: str | None = None + token: str diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 7a4b3034c..84316a743 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -416,9 +416,9 @@ def _get_transport_params(transport_key: str, transport_params: dict[str, Callab async def _create_telephony_transport( websocket: WebSocket, - params: Any | None = None, - transport_type: str = None, - call_data: dict = None, + params: Any, + transport_type: str, + call_data: dict, ) -> BaseTransport: """Create a telephony transport with pre-parsed WebSocket data. @@ -433,12 +433,6 @@ async def _create_telephony_transport( """ from pipecat.transports.websocket.fastapi import FastAPIWebsocketTransport - if params is None: - raise ValueError( - "FastAPIWebsocketParams must be provided. " - "The serializer and add_wav_header will be set automatically." - ) - # Always set add_wav_header to False for telephony params.add_wav_header = False From f5f92dea6309ca20591e9dfb115ec50602d5fc38 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 22:55:50 -0400 Subject: [PATCH 8/9] Add changelog entries and restore multi-line WhatsApp error log Add changelog entries for the pyright introduction and the LiveKitRunnerArguments.token signature tightening. Restore the indented multi-line format for the WhatsApp missing-env error, now listing only the vars that are actually missing. --- changelog/4324.added.md | 1 + changelog/4324.changed.md | 1 + src/pipecat/runner/run.py | 5 ++++- 3 files changed, 6 insertions(+), 1 deletion(-) create mode 100644 changelog/4324.added.md create mode 100644 changelog/4324.changed.md diff --git a/changelog/4324.added.md b/changelog/4324.added.md new file mode 100644 index 000000000..d38887585 --- /dev/null +++ b/changelog/4324.added.md @@ -0,0 +1 @@ +- Added incremental `pyright` type checking. A `pyrightconfig.json` at the repo root uses `typeCheckingMode: "basic"` with an explicit `include` list of modules that pass cleanly (`clocks`, `metrics`, `transcriptions`, `frames`, `observers`, `extensions`, `turns`, `pipeline`, `runner`). Remaining modules will be added in subsequent PRs. CI enforces the checked set via `uv run pyright` in the format workflow. diff --git a/changelog/4324.changed.md b/changelog/4324.changed.md new file mode 100644 index 000000000..672c60b62 --- /dev/null +++ b/changelog/4324.changed.md @@ -0,0 +1 @@ +- `LiveKitRunnerArguments.token` is now a required `str` (previously `str | None` with a default of `None`). LiveKit requires a token to join a room, so the type now reflects reality. This only affects custom runners that construct `LiveKitRunnerArguments` directly; code consuming the argument from the standard runner is unaffected. diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index ca296bd78..a53dea519 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -396,8 +396,11 @@ def _setup_whatsapp_routes(app: FastAPI, args: argparse.Namespace): ] missing = [v for v in required_vars if not os.getenv(v)] if missing: + missing_list = "\n ".join(missing) logger.error( - f"Missing required environment variables for WhatsApp transport: {', '.join(missing)}" + f"""Missing required environment variables for WhatsApp transport: + {missing_list} + """ ) return From 0340e25e9feef6d295162320b8fb9a97885b9008 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Fri, 17 Apr 2026 12:44:57 -0300 Subject: [PATCH 9/9] Fixing typecheck for service switcher. --- scripts/{fix-ruff.sh => fix-ruff-and-typecheck.sh} | 4 ++++ src/pipecat/pipeline/llm_switcher.py | 3 ++- src/pipecat/pipeline/service_switcher.py | 11 +++++------ 3 files changed, 11 insertions(+), 7 deletions(-) rename scripts/{fix-ruff.sh => fix-ruff-and-typecheck.sh} (83%) diff --git a/scripts/fix-ruff.sh b/scripts/fix-ruff-and-typecheck.sh similarity index 83% rename from scripts/fix-ruff.sh rename to scripts/fix-ruff-and-typecheck.sh index 96c6a278b..31cce6dc3 100755 --- a/scripts/fix-ruff.sh +++ b/scripts/fix-ruff-and-typecheck.sh @@ -6,5 +6,9 @@ PROJECT_ROOT="$(dirname "$SCRIPT_DIR")" echo "Running ruff format..." uv run ruff format "$PROJECT_ROOT" + echo "Running ruff check..." uv run ruff check --fix "$PROJECT_ROOT" + +echo "Running pyright check..." +uv run pyright diff --git a/src/pipecat/pipeline/llm_switcher.py b/src/pipecat/pipeline/llm_switcher.py index 031f9e875..9c89641e3 100644 --- a/src/pipecat/pipeline/llm_switcher.py +++ b/src/pipecat/pipeline/llm_switcher.py @@ -15,6 +15,7 @@ from pipecat.pipeline.service_switcher import ( StrategyType, ) from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameProcessor from pipecat.services.llm_service import LLMService @@ -38,7 +39,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]): strategy_type: The strategy class to use for switching between LLMs. Defaults to ``ServiceSwitcherStrategyManual``. """ - super().__init__(llms, strategy_type) + super().__init__(cast(list[FrameProcessor], llms), strategy_type) @property def llms(self) -> list[LLMService]: diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py index dd5e330e9..a7f98f9b1 100644 --- a/src/pipecat/pipeline/service_switcher.py +++ b/src/pipecat/pipeline/service_switcher.py @@ -6,7 +6,6 @@ """Service switcher for switching between different services at runtime, with different switching strategies.""" -from collections.abc import Sequence from typing import Any, Generic, TypeVar from loguru import logger @@ -43,7 +42,7 @@ class ServiceSwitcherStrategy(BaseObject): ... """ - def __init__(self, services: Sequence[FrameProcessor]): + def __init__(self, services: list[FrameProcessor]): """Initialize the service switcher strategy with a list of services. Note: @@ -57,7 +56,7 @@ class ServiceSwitcherStrategy(BaseObject): if len(services) == 0: raise Exception(f"ServiceSwitcherStrategy needs at least one service") - self._services = list(services) + self._services = services self._active_service = services[0] self._register_event_handler("on_service_switched") @@ -224,7 +223,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): def __init__( self, - services: Sequence[FrameProcessor], + services: list[FrameProcessor], strategy_type: type[StrategyType] = ServiceSwitcherStrategyManual, ): """Initialize the service switcher with a list of services and a switching strategy. @@ -236,7 +235,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): """ _strategy = strategy_type(services) super().__init__(*self._make_pipeline_definitions(services, _strategy)) - self._services = list(services) + self._services = services self._strategy = _strategy @property @@ -251,7 +250,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): @staticmethod def _make_pipeline_definitions( - services: Sequence[FrameProcessor], strategy: ServiceSwitcherStrategy + services: list[FrameProcessor], strategy: ServiceSwitcherStrategy ) -> list[Any]: pipelines = [] for service in services: