Merge pull request #4324 from pipecat-ai/mb/pyright-initial

Add pyright type checking: step 1

.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

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.

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.

pyrightconfig.json

@@ -0,0 +1,21 @@
+{
+  "typeCheckingMode": "basic",
+  "pythonVersion": "3.11",
+  "pythonPlatform": "All",
+  "include": [
+    "src/pipecat/clocks",
+    "src/pipecat/metrics",
+    "src/pipecat/transcriptions",
+    "src/pipecat/frames",
+    "src/pipecat/observers",
+    "src/pipecat/extensions",
+    "src/pipecat/turns",
+    "src/pipecat/pipeline",
+    "src/pipecat/runner"
+  ],
+  "exclude": [
+    "**/*_pb2.py",
+    "**/__pycache__"
+  ],
+  "reportMissingImports": false
+}

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

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()

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,
             },
         ]

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
 
 

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).
+        # Captured once when the first StartFrame arrives.
-        self._start_frame_id: str | None = None
+        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:
+        if self._start_frame is None:
-            self._start_frame_id = data.frame.id
+            self._start_frame = _StartFrameInfo(
-            self._start_frame_arrival_ns = data.timestamp
+                frame_id=data.frame.id,
-            self._start_wall_clock = time.time()
+                arrival_ns=data.timestamp,
-        elif data.frame.id != self._start_frame_id:
+                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,
         )

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 (
@@ -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]:
@@ -47,7 +48,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 +57,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.

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()
 

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.

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)
 

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,

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,27 @@ 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")
+    required_vars = [
-    WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID")
+        "WHATSAPP_APP_SECRET",
-    WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN")
+        "WHATSAPP_PHONE_NUMBER_ID",
-    WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN")
+        "WHATSAPP_TOKEN",
-
+        "WHATSAPP_WEBHOOK_VERIFICATION_TOKEN",
-    if not all(
+    ]
-        [
+    missing = [v for v in required_vars if not os.getenv(v)]
-            WHATSAPP_APP_SECRET,
+    if missing:
-            WHATSAPP_PHONE_NUMBER_ID,
+        missing_list = "\n    ".join(missing)
-            WHATSAPP_TOKEN,
-            WHATSAPP_WEBHOOK_VERIFICATION_TOKEN,
-        ]
-    ):
         logger.error(
-            """Missing required environment variables for WhatsApp transport:
+            f"""Missing required environment variables for WhatsApp transport:
-    WHATSAPP_APP_SECRET
+    {missing_list}
-    WHATSAPP_PHONE_NUMBER_ID
-    WHATSAPP_TOKEN
-    WHATSAPP_WEBHOOK_VERIFICATION_TOKEN
             """
         )
         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

src/pipecat/runner/types.py

@@ -122,4 +122,4 @@ class LiveKitRunnerArguments(RunnerArguments):
 
     room_name: str
     url: str
-    token: str | None = None
+    token: str

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,
+    params: Any,
-    transport_type: str = None,
+    transport_type: str,
-    call_data: dict = None,
+    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
 

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
+            A ProcessFrameResult indicating the outcome, or None (treated as
-            None are treated as CONTINUE for backward compatibility.
+            CONTINUE for backward compatibility).
         """
         pass
 

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
+            A ProcessFrameResult indicating the outcome, or None (treated as
-            None are treated as CONTINUE for backward compatibility.
+            CONTINUE for backward compatibility).
         """
         pass
 

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.
@@ -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())