Remove typed-settings migration scaffolding and rename _update_settings_from_typed to _update_settings.

Now that all services use typed `ServiceSettings` objects, this removes the interim scaffolding that supported both dict-based and typed settings paths in parallel. Specifically: removes old dict-based `_update_settings(settings: Mapping)` methods from base classes, removes `isinstance(self._settings, ServiceSettings)` guards, simplifies `process_frame` branching, and renames `_update_settings_from_typed` to `_update_settings` across all ~30 service implementations. Also renames the no-arg `_update_settings()` helper on realtime services to `_send_session_update()` to avoid collision, adds `from_mapping` overrides on `GoogleLLMSettings` and `AnthropicLLMSettings` for ThinkingConfig dict-to-object conversion, and replaces a broken no-arg `_update_settings()` call in Gemini Live with a TODO.

src/pipecat/frames/frames.py

@@ -2113,13 +2113,13 @@ class TTSStoppedFrame(ControlFrame):
 class ServiceUpdateSettingsFrame(ControlFrame):
     """Base frame for updating service settings.
 
-    Supports both the legacy ``settings`` dict and the new typed ``update``
+    Supports both a ``settings`` dict (for backward compatibility) and an
-    object.  When both are provided, ``update`` takes precedence.
+    ``update`` object.  When both are provided, ``update`` takes precedence.
 
     Parameters:
         settings: Dictionary of setting name to value mappings (legacy).
-        update: Typed :class:`~pipecat.services.settings.ServiceSettings`
+        update: :class:`~pipecat.services.settings.ServiceSettings` object
-            object describing the delta to apply.
+            describing the delta to apply.
     """
 
     settings: Mapping[str, Any] = field(default_factory=dict)

src/pipecat/services/ai_service.py

@@ -10,7 +10,7 @@ Provides the foundation for all AI services in the Pipecat framework, including
 model management, settings handling, and frame processing lifecycle methods.
 """
 
-from typing import Any, AsyncGenerator, Dict, Mapping, Set
+from typing import Any, AsyncGenerator, Dict, Set
 
 from loguru import logger
 
@@ -43,7 +43,7 @@ class AIService(FrameProcessor):
         """
         super().__init__(**kwargs)
         self._model_name: str = ""
-        self._settings: Dict[str, Any] | ServiceSettings = {}
+        self._settings: ServiceSettings = ServiceSettings()
         self._session_properties: Dict[str, Any] = {}
 
     @property
@@ -97,71 +97,22 @@ class AIService(FrameProcessor):
         """
         pass
 
-    async def _update_settings(self, settings: Mapping[str, Any]):
+    async def _update_settings(self, update: ServiceSettings) -> Set[str]:
-        from pipecat.services.openai.realtime.events import SessionProperties
+        """Apply a settings update and return the set of changed field names.
 
-        for key, value in settings.items():
+        The update is applied to ``_settings`` and the changed-field set is
-            logger.debug("Update request for:", key, value)
+        returned.  The ``model`` field is handled specially: when it changes,
+        ``set_model_name`` is called.
 
-            if key in self._settings:
+        Concrete services should override this method (calling ``super()``)
-                logger.info(f"Updating LLM setting {key} to: [{value}]")
+        to react to specific changed fields (e.g. reconnect on voice change).
-                self._settings[key] = value
-            elif key in SessionProperties.model_fields:
-                logger.debug("Attempting to update", key, value)
-
-                try:
-                    from pipecat.services.openai.realtime.events import TurnDetection
-
-                    if isinstance(self._session_properties, SessionProperties):
-                        current_properties = self._session_properties
-                    else:
-                        current_properties = SessionProperties(**self._session_properties)
-
-                    if key == "turn_detection" and isinstance(value, dict):
-                        turn_detection = TurnDetection(**value)
-                        setattr(current_properties, key, turn_detection)
-                    else:
-                        setattr(current_properties, key, value)
-
-                    validated_properties = SessionProperties.model_validate(
-                        current_properties.model_dump()
-                    )
-                    logger.info(f"Updating LLM setting {key} to: [{value}]")
-                    self._session_properties = validated_properties.model_dump()
-                except Exception as e:
-                    logger.warning(f"Unexpected error updating session property {key}: {e}")
-            elif key == "model":
-                logger.info(f"Updating LLM setting {key} to: [{value}]")
-                self.set_model_name(value)
-            else:
-                logger.warning(f"Unknown setting for {self.name} service: {key}")
-
-    async def _update_settings_from_typed(self, update: ServiceSettings) -> Set[str]:
-        """Apply a typed settings update and return the set of changed field names.
-
-        If ``_settings`` is a :class:`ServiceSettings` object, the update is
-        applied to it and the changed-field set is returned.  The ``model``
-        field is handled specially: when it changes, ``set_model_name`` is
-        called.
-
-        Services that have been migrated to typed settings should override
-        this method (calling ``super()``) to react to specific changed fields
-        (e.g. reconnect on voice change).
 
         Args:
-            update: A typed settings delta.
+            update: A settings delta.
 
         Returns:
             Set of field names whose values actually changed.
         """
-        if not isinstance(self._settings, ServiceSettings):
-            logger.warning(
-                f"{self.name}: received typed settings update but _settings "
-                f"is not a ServiceSettings — falling back to dict-based update"
-            )
-            await self._update_settings(update.to_dict())
-            return set()
-
         changed = self._settings.apply_update(update)
 
         if "model" in changed:

src/pipecat/services/anthropic/llm.py

@@ -59,7 +59,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
 from pipecat.services.settings import NOT_GIVEN as _NOT_GIVEN
-from pipecat.services.settings import LLMSettings
+from pipecat.services.settings import LLMSettings, is_given
 from pipecat.utils.tracing.service_decorators import traced_llm
 
 try:
@@ -72,7 +72,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class AnthropicLLMSettings(LLMSettings):
-    """Typed settings for Anthropic LLM services.
+    """Settings for Anthropic LLM services.
 
     Parameters:
         enable_prompt_caching: Whether to enable prompt caching.
@@ -82,6 +82,18 @@ class AnthropicLLMSettings(LLMSettings):
     enable_prompt_caching: Any = field(default_factory=lambda: _NOT_GIVEN)
     thinking: Any = field(default_factory=lambda: _NOT_GIVEN)
 
+    @classmethod
+    def from_mapping(cls, settings):
+        """Convert a plain dict to settings, coercing thinking dicts.
+
+        For backward compatibility, a ``thinking`` value that is a plain dict
+        is converted to a :class:`AnthropicLLMService.ThinkingConfig`.
+        """
+        instance = super().from_mapping(settings)
+        if is_given(instance.thinking) and isinstance(instance.thinking, dict):
+            instance.thinking = AnthropicLLMService.ThinkingConfig(**instance.thinking)
+        return instance
+
 
 @dataclass
 class AnthropicContextAggregatorPair:

src/pipecat/services/assemblyai/stt.py

@@ -56,7 +56,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class AssemblyAISTTSettings(STTSettings):
-    """Typed settings for the AssemblyAI STT service.
+    """Settings for the AssemblyAI STT service.
 
     See :class:`AssemblyAIConnectionParams` for detailed parameter descriptions.
 
@@ -184,8 +184,8 @@ class AssemblyAISTTService(WebsocketSTTService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update and reconnect if anything changed.
+        """Apply a settings update and reconnect if anything changed.
 
         Any change triggers a WebSocket reconnect since all connection
         parameters are encoded in the WebSocket URL.
@@ -196,7 +196,7 @@ class AssemblyAISTTService(WebsocketSTTService):
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/asyncai/tts.py

@@ -76,7 +76,7 @@ def language_to_async_language(language: Language) -> Optional[str]:
 
 @dataclass
 class AsyncAITTSSettings(TTSSettings):
-    """Typed settings for Async AI TTS services.
+    """Settings for Async AI TTS services.
 
     Parameters:
         output_container: Audio container format (e.g. "raw").

src/pipecat/services/aws/llm.py

@@ -73,7 +73,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class AWSBedrockLLMSettings(LLMSettings):
-    """Typed settings for AWS Bedrock LLM services.
+    """Settings for AWS Bedrock LLM services.
 
     Parameters:
         latency: Performance mode - "standard" or "optimized".

src/pipecat/services/aws/stt.py

@@ -47,7 +47,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class AWSTranscribeSTTSettings(STTSettings):
-    """Typed settings for the AWS Transcribe STT service.
+    """Settings for the AWS Transcribe STT service.
 
     Parameters:
         sample_rate: Audio sample rate in Hz (8000 or 16000).
@@ -140,13 +140,13 @@ class AWSTranscribeSTTService(WebsocketSTTService):
         }
         return encoding_map.get(encoding, encoding)
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update, reconnecting if needed.
+        """Apply a settings update, reconnecting if needed.
 
         Any change to connection-relevant settings (model, language, etc.)
         triggers a WebSocket reconnect so the new configuration takes effect.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if changed and self._websocket:
             await self._disconnect()

src/pipecat/services/aws/tts.py

@@ -125,7 +125,7 @@ def language_to_aws_language(language: Language) -> Optional[str]:
 
 @dataclass
 class AWSPollyTTSSettings(TTSSettings):
-    """Typed settings for AWS Polly TTS service.
+    """Settings for AWS Polly TTS service.
 
     Parameters:
         engine: TTS engine to use ('standard', 'neural', etc.).

src/pipecat/services/azure/stt.py

@@ -52,7 +52,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class AzureSTTSettings(STTSettings):
-    """Typed settings for the Azure STT service.
+    """Settings for the Azure STT service.
 
     Parameters:
         region: Azure region for the Speech service.
@@ -123,13 +123,13 @@ class AzureSTTService(STTService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update, reconfiguring the recognizer if needed.
+        """Apply a settings update, reconfiguring the recognizer if needed.
 
         When ``language`` changes the ``SpeechConfig`` is updated and the
         speech recognizer is restarted so that the new language takes effect.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if "language" in changed:
             # Convert Language enum to Azure language code if needed.

src/pipecat/services/azure/tts.py

@@ -69,7 +69,7 @@ def sample_rate_to_output_format(sample_rate: int) -> SpeechSynthesisOutputForma
 
 @dataclass
 class AzureTTSSettings(TTSSettings):
-    """Typed settings for Azure TTS services.
+    """Settings for Azure TTS services.
 
     Parameters:
         emphasis: Emphasis level for speech ("strong", "moderate", "reduced").

src/pipecat/services/camb/tts.py

@@ -137,7 +137,7 @@ def _get_aligned_audio(buffer: bytes) -> tuple[bytes, bytes]:
 
 @dataclass
 class CambTTSSettings(TTSSettings):
-    """Typed settings for Camb.ai TTS service.
+    """Settings for Camb.ai TTS service.
 
     Parameters:
         user_instructions: Custom instructions for mars-instruct model only.

src/pipecat/services/cartesia/stt.py

@@ -46,7 +46,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class CartesiaSTTSettings(STTSettings):
-    """Typed settings for the Cartesia STT service.
+    """Settings for the Cartesia STT service.
 
     Parameters:
         encoding: Audio encoding format (e.g. ``"pcm_s16le"``).
@@ -294,8 +294,8 @@ class CartesiaSTTService(WebsocketSTTService):
 
         await self._disconnect_websocket()
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update and reconnect if anything changed.
+        """Apply a settings update and reconnect if anything changed.
 
         Args:
             update: A :class:`STTSettings` (or ``CartesiaSTTSettings``) delta.
@@ -303,7 +303,7 @@ class CartesiaSTTService(WebsocketSTTService):
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if changed:
             await self._disconnect()
             await self._connect()

src/pipecat/services/cartesia/tts.py

@@ -195,7 +195,7 @@ class CartesiaEmotion(str, Enum):
 
 @dataclass
 class CartesiaTTSSettings(TTSSettings):
-    """Typed settings for Cartesia TTS services.
+    """Settings for Cartesia TTS services.
 
     Parameters:
         output_container: Audio container format (e.g. "raw").

src/pipecat/services/deepgram/stt.py

@@ -49,7 +49,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class DeepgramSTTSettings(STTSettings):
-    """Typed settings for the Deepgram STT service.
+    """Settings for the Deepgram STT service.
 
     Parameters:
         live_options: Deepgram ``LiveOptions`` for detailed configuration.
@@ -195,8 +195,8 @@ class DeepgramSTTService(STTService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update, keeping ``live_options`` in sync.
+        """Apply a settings update, keeping ``live_options`` in sync.
 
         Top-level ``model`` and ``language`` are the source of truth.  When
         they are given in *update* their values are propagated into
@@ -213,7 +213,7 @@ class DeepgramSTTService(STTService):
             getattr(update, "language", NOT_GIVEN)
         )
 
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/deepgram/stt_sagemaker.py

@@ -51,7 +51,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class DeepgramSageMakerSTTSettings(STTSettings):
-    """Typed settings for the Deepgram SageMaker STT service.
+    """Settings for the Deepgram SageMaker STT service.
 
     Parameters:
         live_options: Deepgram ``LiveOptions`` for detailed configuration.
@@ -163,8 +163,8 @@ class DeepgramSageMakerSTTService(STTService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update, keeping ``live_options`` in sync.
+        """Apply a settings update, keeping ``live_options`` in sync.
 
         Top-level ``model`` and ``language`` are the source of truth.  When
         they change their values are propagated into ``live_options``.
@@ -178,7 +178,7 @@ class DeepgramSageMakerSTTService(STTService):
             getattr(update, "language", NOT_GIVEN)
         )
 
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/deepgram/tts.py

@@ -47,7 +47,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class DeepgramTTSSettings(TTSSettings):
-    """Typed settings for Deepgram TTS service.
+    """Settings for Deepgram TTS service.
 
     Parameters:
         encoding: Audio encoding format (linear16, mulaw, alaw).

src/pipecat/services/elevenlabs/stt.py

@@ -171,7 +171,7 @@ def language_to_elevenlabs_language(language: Language) -> Optional[str]:
 
 @dataclass
 class ElevenLabsSTTSettings(STTSettings):
-    """Typed settings for the ElevenLabs file-based STT service.
+    """Settings for the ElevenLabs file-based STT service.
 
     Parameters:
         tag_audio_events: Whether to include audio event tags in transcription.
@@ -182,7 +182,7 @@ class ElevenLabsSTTSettings(STTSettings):
 
 @dataclass
 class ElevenLabsRealtimeSTTSettings(STTSettings):
-    """Typed settings for the ElevenLabs Realtime STT service.
+    """Settings for the ElevenLabs Realtime STT service.
 
     See ``ElevenLabsRealtimeSTTService.InputParams`` for detailed descriptions.
 
@@ -294,8 +294,8 @@ class ElevenLabsSTTService(SegmentedSTTService):
         """
         return language_to_elevenlabs_language(language)
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update.
+        """Apply a settings update.
 
         Converts language to ElevenLabs format before applying and keeps
         ``_model_id`` in sync with the model setting.
@@ -312,7 +312,7 @@ class ElevenLabsSTTService(SegmentedSTTService):
             if converted is not None:
                 update.language = converted
 
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if "model" in changed:
             self._model_id = self._settings.model
@@ -543,8 +543,8 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update and reconnect if anything changed.
+        """Apply a settings update and reconnect if anything changed.
 
         Converts language to ElevenLabs format before applying and keeps
         ``_model_id`` in sync.
@@ -561,7 +561,7 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
             if converted is not None:
                 update.language = converted
 
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/elevenlabs/tts.py

@@ -154,7 +154,7 @@ def build_elevenlabs_voice_settings(
     """Build voice settings dictionary for ElevenLabs based on provided settings.
 
     Args:
-        settings: Dictionary or typed settings containing voice settings parameters.
+        settings: Dictionary or settings containing voice settings parameters.
 
     Returns:
         Dictionary of voice settings or None if no valid settings are provided.
@@ -186,7 +186,7 @@ class PronunciationDictionaryLocator(BaseModel):
 
 @dataclass
 class ElevenLabsTTSSettings(TTSSettings):
-    """Typed settings for the ElevenLabs WebSocket TTS service.
+    """Settings for the ElevenLabs WebSocket TTS service.
 
     Fields that appear in the WebSocket URL (``voice``, ``model``,
     ``language``) require a full reconnect when changed.  Fields that
@@ -230,7 +230,7 @@ class ElevenLabsTTSSettings(TTSSettings):
 
 @dataclass
 class ElevenLabsHttpTTSSettings(TTSSettings):
-    """Typed settings for the ElevenLabs HTTP TTS service.
+    """Settings for the ElevenLabs HTTP TTS service.
 
     Parameters:
         optimize_streaming_latency: Latency optimization level (0-4).
@@ -471,8 +471,8 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 voice_settings[key] = val
         return voice_settings or None
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update, reconnecting as needed.
+        """Apply a settings update, reconnecting as needed.
 
         Uses the declarative ``URL_FIELDS`` and ``VOICE_SETTINGS_FIELDS``
         sets on :class:`ElevenLabsTTSSettings` to decide whether to
@@ -484,7 +484,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed
@@ -958,8 +958,8 @@ class ElevenLabsHttpTTSService(WordTTSService):
     def _set_voice_settings(self):
         return build_elevenlabs_voice_settings(self._settings)
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update and rebuild voice settings.
+        """Apply a settings update and rebuild voice settings.
 
         Args:
             update: A :class:`TTSSettings` (or ``ElevenLabsHttpTTSSettings``) delta.
@@ -967,7 +967,7 @@ class ElevenLabsHttpTTSService(WordTTSService):
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if changed:
             self._voice_settings = self._set_voice_settings()
         return changed

src/pipecat/services/fal/stt.py

@@ -150,7 +150,7 @@ def language_to_fal_language(language: Language) -> Optional[str]:
 
 @dataclass
 class FalSTTSettings(STTSettings):
-    """Typed settings for the Fal Wizper STT service.
+    """Settings for the Fal Wizper STT service.
 
     Parameters:
         task: Task to perform ('transcribe' or 'translate'). Defaults to
@@ -251,9 +251,9 @@ class FalSTTService(SegmentedSTTService):
         """
         return language_to_fal_language(language)
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update, converting language if changed."""
+        """Apply a settings update, converting language if changed."""
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if "language" in changed:
             # Convert the Language enum to a Fal language code.

src/pipecat/services/fish/tts.py

@@ -49,7 +49,7 @@ FishAudioOutputFormat = Literal["opus", "mp3", "pcm", "wav"]
 
 @dataclass
 class FishAudioTTSSettings(TTSSettings):
-    """Typed settings for Fish Audio TTS service.
+    """Settings for Fish Audio TTS service.
 
     Parameters:
         fish_sample_rate: Audio sample rate sent to the API.
@@ -184,8 +184,8 @@ class FishAudioTTSService(InterruptibleTTSService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update and reconnect if needed.
+        """Apply a settings update and reconnect if needed.
 
         Any change to voice or model triggers a WebSocket reconnect.
 
@@ -195,7 +195,7 @@ class FishAudioTTSService(InterruptibleTTSService):
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if changed:
             await self._disconnect()
             await self._connect()

src/pipecat/services/gladia/stt.py

@@ -182,7 +182,7 @@ class _InputParamsDescriptor:
 
 @dataclass
 class GladiaSTTSettings(STTSettings):
-    """Typed settings for Gladia STT service.
+    """Settings for Gladia STT service.
 
     Parameters:
         input_params: Gladia ``GladiaInputParams`` for detailed configuration.
@@ -379,19 +379,19 @@ class GladiaSTTService(WebsocketSTTService):
         await super().start(frame)
         await self._connect()
 
-    async def _update_settings_from_typed(self, update: GladiaSTTSettings) -> set[str]:
+    async def _update_settings(self, update: GladiaSTTSettings) -> set[str]:
-        """Apply typed settings update.
+        """Apply settings update.
 
         Gladia sessions are fixed at creation time, so any change requires
         a full session teardown and reconnect.
 
         Args:
-            update: A typed settings delta.
+            update: A settings delta.
 
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/google/gemini_live/llm.py

@@ -604,7 +604,7 @@ class InputParams(BaseModel):
 
 @dataclass
 class GeminiLiveLLMSettings(LLMSettings):
-    """Typed settings for Gemini Live LLM services.
+    """Settings for Gemini Live LLM services.
 
     Parameters:
         modalities: Response modalities.
@@ -976,7 +976,8 @@ class GeminiLiveLLMService(LLMService):
             # (we have an example that does just that, actually).
             await self._create_single_response(frame.messages)
         elif isinstance(frame, LLMSetToolsFrame):
-            await self._update_settings()
+            # TODO: implement runtime tool updates for Gemini Live.
+            pass
         else:
             await self.push_frame(frame, direction)
 

src/pipecat/services/google/llm.py

@@ -58,7 +58,7 @@ from pipecat.services.openai.llm import (
     OpenAIAssistantContextAggregator,
     OpenAIUserContextAggregator,
 )
-from pipecat.services.settings import NOT_GIVEN, LLMSettings
+from pipecat.services.settings import NOT_GIVEN, LLMSettings, is_given
 from pipecat.utils.tracing.service_decorators import traced_llm
 
 # Suppress gRPC fork warnings
@@ -675,7 +675,7 @@ class GoogleLLMContext(OpenAILLMContext):
 
 @dataclass
 class GoogleLLMSettings(LLMSettings):
-    """Typed settings for Google LLM services.
+    """Settings for Google LLM services.
 
     Parameters:
         thinking: Thinking configuration.
@@ -683,6 +683,18 @@ class GoogleLLMSettings(LLMSettings):
 
     thinking: Any = field(default_factory=lambda: NOT_GIVEN)
 
+    @classmethod
+    def from_mapping(cls, settings):
+        """Convert a plain dict to settings, coercing thinking dicts.
+
+        For backward compatibility, a ``thinking`` value that is a plain dict
+        is converted to a :class:`GoogleLLMService.ThinkingConfig`.
+        """
+        instance = super().from_mapping(settings)
+        if is_given(instance.thinking) and isinstance(instance.thinking, dict):
+            instance.thinking = GoogleLLMService.ThinkingConfig(**instance.thinking)
+        return instance
+
 
 class GoogleLLMService(LLMService):
     """Google AI (Gemini) LLM service implementation.
@@ -1227,14 +1239,6 @@ class GoogleLLMService(LLMService):
             # Do nothing - we're shutting down anyway
             pass
 
-    async def _update_settings(self, settings):
-        """Override to handle ThinkingConfig validation."""
-        # Convert thinking dict to ThinkingConfig if needed
-        if "thinking" in settings and isinstance(settings["thinking"], dict):
-            settings = dict(settings)  # Make a copy to avoid modifying the original
-            settings["thinking"] = self.ThinkingConfig(**settings["thinking"])
-        await super()._update_settings(settings)
-
     def create_context_aggregator(
         self,
         context: OpenAILLMContext,

src/pipecat/services/google/stt.py

@@ -360,7 +360,7 @@ def language_to_google_stt_language(language: Language) -> Optional[str]:
 
 @dataclass
 class GoogleSTTSettings(STTSettings):
-    """Typed settings for Google Cloud Speech-to-Text V2.
+    """Settings for Google Cloud Speech-to-Text V2.
 
     Parameters:
         languages: List of ``Language`` enums for recognition
@@ -628,10 +628,10 @@ class GoogleSTTService(STTService):
                 DeprecationWarning,
             )
         logger.debug(f"Switching STT languages to: {languages}")
-        await self._update_settings_from_typed(GoogleSTTSettings(languages=list(languages)))
+        await self._update_settings(GoogleSTTSettings(languages=list(languages)))
 
-    async def _update_settings_from_typed(self, update: GoogleSTTSettings) -> set[str]:
+    async def _update_settings(self, update: GoogleSTTSettings) -> set[str]:
-        """Apply typed settings update and reconnect if anything changed.
+        """Apply settings update and reconnect if anything changed.
 
         Handles ``language`` from base ``set_language`` by converting it to
         ``languages``. Emits a deprecation warning if ``language_codes`` is
@@ -639,7 +639,7 @@ class GoogleSTTService(STTService):
         Reconnects the stream on any change.
 
         Args:
-            update: A typed settings delta.
+            update: A settings delta.
 
         Returns:
             Set of field names whose values actually changed.
@@ -663,7 +663,7 @@ class GoogleSTTService(STTService):
                     stacklevel=2,
                 )
 
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if changed:
             await self._reconnect_if_needed()
@@ -742,7 +742,7 @@ class GoogleSTTService(STTService):
                 "GoogleSTTSettings(...) instead.",
                 DeprecationWarning,
             )
-        # Build a typed settings delta from the provided options
+        # Build a settings delta from the provided options
         update = GoogleSTTSettings()
 
         if languages is not None:
@@ -770,7 +770,7 @@ class GoogleSTTService(STTService):
             logger.debug(f"Updating location to: {location}")
             self._location = location
 
-        await self._update_settings_from_typed(update)
+        await self._update_settings(update)
 
     async def _connect(self):
         """Initialize streaming recognition config and stream."""

src/pipecat/services/google/tts.py

@@ -478,7 +478,7 @@ def language_to_gemini_tts_language(language: Language) -> Optional[str]:
 
 @dataclass
 class GoogleHttpTTSSettings(TTSSettings):
-    """Typed settings for Google HTTP TTS service.
+    """Settings for Google HTTP TTS service.
 
     Parameters:
         pitch: Voice pitch adjustment (e.g., "+2st", "-50%").
@@ -505,7 +505,7 @@ class GoogleHttpTTSSettings(TTSSettings):
 
 @dataclass
 class GoogleStreamTTSSettings(TTSSettings):
-    """Typed settings for Google streaming TTS service.
+    """Settings for Google streaming TTS service.
 
     Parameters:
         language: Language for synthesis. Defaults to English.
@@ -518,7 +518,7 @@ class GoogleStreamTTSSettings(TTSSettings):
 
 @dataclass
 class GeminiTTSSettings(TTSSettings):
-    """Typed settings for Gemini TTS service.
+    """Settings for Gemini TTS service.
 
     Parameters:
         language: Language for synthesis. Defaults to English.
@@ -680,11 +680,11 @@ class GoogleHttpTTSService(TTSService):
         """
         return language_to_google_tts_language(language)
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
         """Override to handle speaking_rate validation.
 
         Args:
-            update: Typed settings delta. Can include 'speaking_rate' (float).
+            update: Settings delta. Can include 'speaking_rate' (float).
         """
         if isinstance(update, GoogleHttpTTSSettings) and is_given(update.speaking_rate):
             rate_value = float(update.speaking_rate)
@@ -693,7 +693,7 @@ class GoogleHttpTTSService(TTSService):
                     f"Invalid speaking_rate value: {rate_value}. Must be between 0.25 and 2.0"
                 )
                 update.speaking_rate = NOT_GIVEN
-        return await super()._update_settings_from_typed(update)
+        return await super()._update_settings(update)
 
     def _construct_ssml(self, text: str) -> str:
         ssml = "<speak>"
@@ -1024,11 +1024,11 @@ class GoogleTTSService(GoogleBaseTTSService):
             credentials, credentials_path
         )
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
         """Override to handle speaking_rate validation.
 
         Args:
-            update: Typed settings delta. Can include 'speaking_rate' (float).
+            update: Settings delta. Can include 'speaking_rate' (float).
         """
         if isinstance(update, GoogleStreamTTSSettings) and is_given(update.speaking_rate):
             rate_value = float(update.speaking_rate)
@@ -1037,7 +1037,7 @@ class GoogleTTSService(GoogleBaseTTSService):
                     f"Invalid speaking_rate value: {rate_value}. Must be between 0.25 and 2.0"
                 )
                 update.speaking_rate = NOT_GIVEN
-        return await super()._update_settings_from_typed(update)
+        return await super()._update_settings(update)
 
     @traced_tts
     async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
@@ -1259,11 +1259,11 @@ class GeminiTTSService(GoogleBaseTTSService):
                 f"Current rate of {self.sample_rate}Hz may cause issues."
             )
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update with voice validation.
+        """Apply a settings update with voice validation.
 
         Args:
-            update: Typed settings delta. Can include 'voice', 'prompt', etc.
+            update: Settings delta. Can include 'voice', 'prompt', etc.
 
         Returns:
             Set of field names whose values actually changed.
@@ -1271,7 +1271,7 @@ class GeminiTTSService(GoogleBaseTTSService):
         if is_given(update.voice) and update.voice not in self.AVAILABLE_VOICES:
             logger.warning(f"Voice '{update.voice}' not in known voices list. Using anyway.")
 
-        return await super()._update_settings_from_typed(update)
+        return await super()._update_settings(update)
 
     @traced_tts
     async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:

src/pipecat/services/gradium/stt.py

@@ -68,7 +68,7 @@ def language_to_gradium_language(language: Language) -> Optional[str]:
 
 @dataclass
 class GradiumSTTSettings(STTSettings):
-    """Typed settings for the Gradium STT service.
+    """Settings for the Gradium STT service.
 
     Parameters:
         delay_in_frames: Delay in audio frames (80ms each) before text is
@@ -171,8 +171,8 @@ class GradiumSTTService(WebsocketSTTService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update, sync params, and reconnect.
+        """Apply a settings update, sync params, and reconnect.
 
         Args:
             update: A :class:`STTSettings` (or ``GradiumSTTSettings``) delta.
@@ -180,7 +180,7 @@ class GradiumSTTService(WebsocketSTTService):
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if not changed:
             return changed
 

src/pipecat/services/gradium/tts.py

@@ -41,7 +41,7 @@ SAMPLE_RATE = 48000
 
 @dataclass
 class GradiumTTSSettings(TTSSettings):
-    """Typed settings for the Gradium TTS service.
+    """Settings for the Gradium TTS service.
 
     Parameters:
         output_format: Audio output format.
@@ -119,8 +119,8 @@ class GradiumTTSService(InterruptibleWordTTSService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update and reconnect if voice changed.
+        """Apply a settings update and reconnect if voice changed.
 
         Args:
             update: A :class:`TTSSettings` (or ``GradiumTTSSettings``) delta.
@@ -129,7 +129,7 @@ class GradiumTTSService(InterruptibleWordTTSService):
             Set of field names whose values actually changed.
         """
         prev_voice = self._voice_id
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if self._voice_id != prev_voice:
             await self._disconnect()
             await self._connect()

src/pipecat/services/grok/realtime/llm.py

@@ -88,7 +88,7 @@ class CurrentAudioResponse:
 
 @dataclass
 class GrokRealtimeLLMSettings(LLMSettings):
-    """Typed settings for Grok Realtime LLM services.
+    """Settings for Grok Realtime LLM services.
 
     Parameters:
         session_properties: Grok Realtime session configuration.
@@ -349,13 +349,13 @@ class GrokRealtimeLLMService(LLMService):
             frame: The frame to process.
             direction: The direction of frame flow in the pipeline.
         """
-        # Legacy dict path: frame.settings contains SessionProperties fields,
+        # Backward-compatible dict path: frame.settings contains SessionProperties
-        # not our Settings fields, so we construct SessionProperties directly.
+        # fields, not our Settings fields, so we construct SessionProperties
-        # The new typed path (frame.update) falls through to super, which calls
+        # directly. The frame.update path falls through to super, which calls
-        # _update_settings_from_typed → our override handles the rest.
+        # _update_settings → our override handles the rest.
         if isinstance(frame, LLMUpdateSettingsFrame) and frame.update is None:
             self._settings.session_properties = events.SessionProperties(**frame.settings)
-            await self._update_settings()
+            await self._send_session_update()
             await self.push_frame(frame, direction)
             return
 
@@ -379,7 +379,7 @@ class GrokRealtimeLLMService(LLMService):
         elif isinstance(frame, LLMMessagesAppendFrame):
             await self._handle_messages_append(frame)
         elif isinstance(frame, LLMSetToolsFrame):
-            await self._update_settings()
+            await self._send_session_update()
 
         await self.push_frame(frame, direction)
 
@@ -456,14 +456,14 @@ class GrokRealtimeLLMService(LLMService):
                 return
             await self.push_error(error_msg=f"Error sending client event: {e}", exception=e)
 
-    async def _update_settings_from_typed(self, update):
+    async def _update_settings(self, update):
-        """Apply a typed settings update, sending a session update if needed."""
+        """Apply a settings update, sending a session update if needed."""
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if "session_properties" in changed:
-            await self._update_settings()
+            await self._send_session_update()
         return changed
 
-    async def _update_settings(self):
+    async def _send_session_update(self):
         """Update session settings on the server."""
         settings = self._settings.session_properties
         adapter: GrokRealtimeLLMAdapter = self.get_llm_adapter()
@@ -543,7 +543,7 @@ class GrokRealtimeLLMService(LLMService):
 
     async def _handle_evt_conversation_created(self, evt):
         """Handle conversation.created event - first event after connecting."""
-        await self._update_settings()
+        await self._send_session_update()
 
     async def _handle_evt_response_created(self, evt):
         """Handle response.created event - response generation started."""
@@ -746,7 +746,7 @@ class GrokRealtimeLLMService(LLMService):
                 self._messages_added_manually[evt.item.id] = True
                 await self.send_client_event(evt)
 
-            await self._update_settings()
+            await self._send_session_update()
             self._llm_needs_conversation_setup = False
 
         logger.debug("Creating Grok response")

src/pipecat/services/groq/tts.py

@@ -36,7 +36,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class GroqTTSSettings(TTSSettings):
-    """Typed settings for the Groq TTS service.
+    """Settings for the Groq TTS service.
 
     Parameters:
         output_format: Audio output format.

src/pipecat/services/hathora/stt.py

@@ -31,7 +31,7 @@ from .utils import ConfigOption
 
 @dataclass
 class HathoraSTTSettings(STTSettings):
-    """Typed settings for the Hathora STT service.
+    """Settings for the Hathora STT service.
 
     Parameters:
         config: Some models support additional config, refer to

src/pipecat/services/hathora/tts.py

@@ -49,7 +49,7 @@ def _decode_audio_payload(
 
 @dataclass
 class HathoraTTSSettings(TTSSettings):
-    """Typed settings for Hathora TTS service.
+    """Settings for Hathora TTS service.
 
     Parameters:
         speed: Speech speed multiplier (if supported by model).

src/pipecat/services/inworld/tts.py

@@ -52,7 +52,7 @@ from pipecat.utils.tracing.service_decorators import traced_tts
 
 @dataclass
 class InworldTTSSettings(TTSSettings):
-    """Typed settings for Inworld TTS services.
+    """Settings for Inworld TTS services.
 
     Parameters:
         audio_encoding: Audio encoding format (e.g. LINEAR16).

src/pipecat/services/kokoro/tts.py

@@ -91,7 +91,7 @@ def language_to_kokoro_language(language: Language) -> str:
 
 @dataclass
 class KokoroTTSSettings(TTSSettings):
-    """Typed settings for the Kokoro TTS service.
+    """Settings for the Kokoro TTS service.
 
     Parameters:
         lang_code: Kokoro language code for synthesis.

src/pipecat/services/llm_service.py

@@ -59,7 +59,7 @@ from pipecat.processors.aggregators.llm_response import (
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
-from pipecat.services.settings import LLMSettings, ServiceSettings, is_given
+from pipecat.services.settings import LLMSettings, is_given
 from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionLLMServiceMixin
 from pipecat.utils.context.llm_context_summarization import (
     LLMContextSummarizationUtil,
@@ -313,16 +313,16 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
             await self._cancel_sequential_runner_task()
         await self._cancel_summary_task()
 
-    async def _update_settings_from_typed(self, update: LLMSettings) -> set[str]:
+    async def _update_settings(self, update: LLMSettings) -> set[str]:
-        """Apply a typed settings update, handling turn-completion fields.
+        """Apply a settings update, handling turn-completion fields.
 
         Args:
-            update: A typed LLM settings delta.
+            update: An LLM settings delta.
 
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if "filter_incomplete_user_turns" in changed:
             self._filter_incomplete_user_turns = self._settings.filter_incomplete_user_turns
@@ -336,35 +336,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
 
         return changed
 
-    async def _update_settings(self, settings: Mapping[str, Any]):
-        """Update LLM service settings.
-
-        Handles turn completion settings specially since they are not model
-        parameters and should not be passed to the underlying LLM API.
-
-        Args:
-            settings: Dictionary of settings to update.
-        """
-        # Turn completion settings to extract (not model parameters)
-        turn_completion_keys = {"filter_incomplete_user_turns", "user_turn_completion_config"}
-
-        # Handle turn completion settings
-        if "filter_incomplete_user_turns" in settings:
-            self._filter_incomplete_user_turns = settings["filter_incomplete_user_turns"]
-            logger.info(
-                f"{self}: Incomplete turn filtering {'enabled' if self._filter_incomplete_user_turns else 'disabled'}"
-            )
-
-            # Configure the mixin with config object
-            if self._filter_incomplete_user_turns and "user_turn_completion_config" in settings:
-                self.set_user_turn_completion_config(settings["user_turn_completion_config"])
-
-        # Remove turn completion settings before passing to parent
-        settings = {k: v for k, v in settings.items() if k not in turn_completion_keys}
-
-        # Let the parent handle remaining model parameters
-        await super()._update_settings(settings)
-
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process a frame.
 
@@ -379,16 +350,12 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
         elif isinstance(frame, LLMConfigureOutputFrame):
             self._skip_tts = frame.skip_tts
         elif isinstance(frame, LLMUpdateSettingsFrame):
-            # New path: typed settings update object.
             if frame.update is not None:
-                await self._update_settings_from_typed(frame.update)
+                await self._update_settings(frame.update)
-            # Legacy path: plain dict, but service uses typed settings — convert.
+            elif frame.settings:
-            elif isinstance(self._settings, ServiceSettings):
+                # Backward-compatible path: convert legacy dict to settings object.
                 update = type(self._settings).from_mapping(frame.settings)
-                await self._update_settings_from_typed(update)
+                await self._update_settings(update)
-            # Legacy path: plain dict, service still uses dict-based settings.
-            else:
-                await self._update_settings(frame.settings)
         elif isinstance(frame, LLMContextSummaryRequestFrame):
             await self._handle_summary_request(frame)
 

src/pipecat/services/lmnt/tts.py

@@ -75,7 +75,7 @@ def language_to_lmnt_language(language: Language) -> Optional[str]:
 
 @dataclass
 class LmntTTSSettings(TTSSettings):
-    """Typed settings for LMNT TTS service.
+    """Settings for LMNT TTS service.
 
     Parameters:
         format: Audio output format. Defaults to "raw".

src/pipecat/services/minimax/tts.py

@@ -89,7 +89,7 @@ def language_to_minimax_language(language: Language) -> Optional[str]:
 
 @dataclass
 class MiniMaxTTSSettings(TTSSettings):
-    """Typed settings for MiniMax TTS service.
+    """Settings for MiniMax TTS service.
 
     Parameters:
         stream: Whether to use streaming mode.

src/pipecat/services/neuphonic/tts.py

@@ -76,7 +76,7 @@ def language_to_neuphonic_lang_code(language: Language) -> Optional[str]:
 
 @dataclass
 class NeuphonicTTSSettings(TTSSettings):
-    """Typed settings for Neuphonic TTS service.
+    """Settings for Neuphonic TTS service.
 
     Parameters:
         lang_code: Neuphonic language code.
@@ -181,9 +181,9 @@ class NeuphonicTTSService(InterruptibleTTSService):
         """
         return language_to_neuphonic_lang_code(language)
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update and reconnect with new configuration."""
+        """Apply a settings update and reconnect with new configuration."""
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if changed:
             await self._disconnect()
             await self._connect()

src/pipecat/services/nvidia/stt.py

@@ -93,14 +93,14 @@ def language_to_nvidia_riva_language(language: Language) -> Optional[str]:
 
 @dataclass
 class NvidiaSTTSettings(STTSettings):
-    """Typed settings for the NVIDIA Riva streaming STT service."""
+    """Settings for the NVIDIA Riva streaming STT service."""
 
     pass
 
 
 @dataclass
 class NvidiaSegmentedSTTSettings(STTSettings):
-    """Typed settings for the NVIDIA Riva segmented STT service.
+    """Settings for the NVIDIA Riva segmented STT service.
 
     Parameters:
         profanity_filter: Whether to filter profanity from results.
@@ -579,8 +579,8 @@ class NvidiaSegmentedSTTService(SegmentedSTTService):
         self._config = self._create_recognition_config()
         logger.debug(f"Initialized NvidiaSegmentedSTTService with model: {self.model_name}")
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update and sync internal state.
+        """Apply a settings update and sync internal state.
 
         Args:
             update: A :class:`STTSettings` (or ``NvidiaSegmentedSTTSettings``) delta.
@@ -588,7 +588,7 @@ class NvidiaSegmentedSTTService(SegmentedSTTService):
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if changed:
             self._config = self._create_recognition_config()

src/pipecat/services/openai/base_llm.py

@@ -49,7 +49,7 @@ from pipecat.utils.tracing.service_decorators import traced_llm
 
 @dataclass
 class OpenAILLMSettings(LLMSettings):
-    """Typed settings for OpenAI-compatible LLM services.
+    """Settings for OpenAI-compatible LLM services.
 
     Parameters:
         max_completion_tokens: Maximum completion tokens to generate.

src/pipecat/services/openai/realtime/llm.py

@@ -93,7 +93,7 @@ class CurrentAudioResponse:
 
 @dataclass
 class OpenAIRealtimeLLMSettings(LLMSettings):
-    """Typed settings for OpenAI Realtime LLM services.
+    """Settings for OpenAI Realtime LLM services.
 
     Parameters:
         session_properties: OpenAI Realtime session configuration.
@@ -411,13 +411,13 @@ class OpenAIRealtimeLLMService(LLMService):
             frame: The frame to process.
             direction: The direction of frame flow in the pipeline.
         """
-        # Legacy dict path: frame.settings contains SessionProperties fields,
+        # Backward-compatible dict path: frame.settings contains SessionProperties
-        # not our Settings fields, so we construct SessionProperties directly.
+        # fields, not our Settings fields, so we construct SessionProperties
-        # The new typed path (frame.update) falls through to super, which calls
+        # directly. The frame.update path falls through to super, which calls
-        # _update_settings_from_typed → our override handles the rest.
+        # _update_settings → our override handles the rest.
         if isinstance(frame, LLMUpdateSettingsFrame) and frame.update is None:
             self._settings.session_properties = events.SessionProperties(**frame.settings)
-            await self._update_settings()
+            await self._send_session_update()
             await self.push_frame(frame, direction)
             return
 
@@ -449,7 +449,7 @@ class OpenAIRealtimeLLMService(LLMService):
         elif isinstance(frame, LLMMessagesAppendFrame):
             await self._handle_messages_append(frame)
         elif isinstance(frame, LLMSetToolsFrame):
-            await self._update_settings()
+            await self._send_session_update()
 
         await self.push_frame(frame, direction)
 
@@ -534,14 +534,14 @@ class OpenAIRealtimeLLMService(LLMService):
             # treat a send-side error as fatal.
             await self.push_error(error_msg=f"Error sending client event: {e}", exception=e)
 
-    async def _update_settings_from_typed(self, update):
+    async def _update_settings(self, update):
-        """Apply a typed settings update, sending a session update if needed."""
+        """Apply a settings update, sending a session update if needed."""
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if "session_properties" in changed:
-            await self._update_settings()
+            await self._send_session_update()
         return changed
 
-    async def _update_settings(self):
+    async def _send_session_update(self):
         settings = self._settings.session_properties
         adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
 
@@ -613,7 +613,7 @@ class OpenAIRealtimeLLMService(LLMService):
     async def _handle_evt_session_created(self, evt):
         # session.created is received right after connecting. Send a message
         # to configure the session properties.
-        await self._update_settings()
+        await self._send_session_update()
 
     async def _handle_evt_session_updated(self, evt):
         # If this is our first context frame, run the LLM
@@ -896,7 +896,7 @@ class OpenAIRealtimeLLMService(LLMService):
                 await self.send_client_event(evt)
 
             # Send new settings if needed
-            await self._update_settings()
+            await self._send_session_update()
 
             # We're done configuring the LLM for this session
             self._llm_needs_conversation_setup = False

src/pipecat/services/openai/stt.py

@@ -127,7 +127,7 @@ _OPENAI_SAMPLE_RATE = 24000
 
 @dataclass
 class OpenAIRealtimeSTTSettings(STTSettings):
-    """Typed settings for the OpenAI Realtime STT service.
+    """Settings for the OpenAI Realtime STT service.
 
     Parameters:
         prompt: Optional prompt text to guide transcription style.
@@ -268,10 +268,10 @@ class OpenAIRealtimeSTTService(WebsocketSTTService):
         """
         return True
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update and send session update if needed.
+        """Apply a settings update and send session update if needed.
 
-        Keeps ``_language_code`` and ``_prompt`` in sync with typed settings
+        Keeps ``_language_code`` and ``_prompt`` in sync with settings
         and sends a ``session.update`` to the server when the session is active.
 
         Args:
@@ -280,7 +280,7 @@ class OpenAIRealtimeSTTService(WebsocketSTTService):
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/openai/tts.py

@@ -64,7 +64,7 @@ VALID_VOICES: Dict[str, ValidVoice] = {
 
 @dataclass
 class OpenAITTSSettings(TTSSettings):
-    """Typed settings for OpenAI TTS service.
+    """Settings for OpenAI TTS service.
 
     Parameters:
         instructions: Instructions to guide voice synthesis behavior.

src/pipecat/services/openai_realtime_beta/openai.py

@@ -94,7 +94,7 @@ class CurrentAudioResponse:
 
 @dataclass
 class OpenAIRealtimeBetaLLMSettings(LLMSettings):
-    """Typed settings for OpenAI Realtime Beta LLM services.
+    """Settings for OpenAI Realtime Beta LLM services.
 
     Parameters:
         session_properties: OpenAI Realtime session configuration.
@@ -357,13 +357,13 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             frame: The frame to process.
             direction: The direction of frame flow in the pipeline.
         """
-        # Legacy dict path: frame.settings contains SessionProperties fields,
+        # Backward-compatible dict path: frame.settings contains SessionProperties
-        # not our Settings fields, so we construct SessionProperties directly.
+        # fields, not our Settings fields, so we construct SessionProperties
-        # The new typed path (frame.update) falls through to super, which calls
+        # directly. The frame.update path falls through to super, which calls
-        # _update_settings_from_typed → our override handles the rest.
+        # _update_settings → our override handles the rest.
         if isinstance(frame, LLMUpdateSettingsFrame) and frame.update is None:
             self._settings.session_properties = events.SessionProperties(**frame.settings)
-            await self._update_settings()
+            await self._send_session_update()
             await self.push_frame(frame, direction)
             return
 
@@ -403,7 +403,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
         elif isinstance(frame, RealtimeMessagesUpdateFrame):
             self._context = frame.context
         elif isinstance(frame, LLMSetToolsFrame):
-            await self._update_settings()
+            await self._send_session_update()
         elif isinstance(frame, RealtimeFunctionCallResultFrame):
             await self._handle_function_call_result(frame.result_frame)
 
@@ -478,14 +478,14 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             # treat a send-side error as fatal.
             await self.push_error(error_msg=f"Error sending client event: {e}", exception=e)
 
-    async def _update_settings_from_typed(self, update):
+    async def _update_settings(self, update):
-        """Apply a typed settings update, sending a session update if needed."""
+        """Apply a settings update, sending a session update if needed."""
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if "session_properties" in changed:
-            await self._update_settings()
+            await self._send_session_update()
         return changed
 
-    async def _update_settings(self):
+    async def _send_session_update(self):
         settings = self._settings.session_properties
         # tools given in the context override the tools in the session properties
         if self._context and self._context.tools:
@@ -540,7 +540,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
     async def _handle_evt_session_created(self, evt):
         # session.created is received right after connecting. Send a message
         # to configure the session properties.
-        await self._update_settings()
+        await self._send_session_update()
 
     async def _handle_evt_session_updated(self, evt):
         # If this is our first context frame, run the LLM
@@ -779,7 +779,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             self._context.llm_needs_initial_messages = False
 
         if self._context.llm_needs_settings_update:
-            await self._update_settings()
+            await self._send_session_update()
             self._context.llm_needs_settings_update = False
 
         logger.debug(f"Creating response: {self._context.get_messages_for_logging()}")

src/pipecat/services/playht/tts.py

@@ -101,7 +101,7 @@ def language_to_playht_language(language: Language) -> Optional[str]:
 
 @dataclass
 class PlayHTTTSSettings(TTSSettings):
-    """Typed settings for PlayHT TTS services.
+    """Settings for PlayHT TTS services.
 
     Parameters:
         output_format: Audio output format.

src/pipecat/services/resembleai/tts.py

@@ -42,7 +42,7 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class ResembleAITTSSettings(TTSSettings):
-    """Typed settings for Resemble AI TTS service.
+    """Settings for Resemble AI TTS service.
 
     Parameters:
         precision: PCM bit depth (PCM_32, PCM_24, PCM_16, or MULAW).

src/pipecat/services/rime/tts.py

@@ -72,7 +72,7 @@ def language_to_rime_language(language: Language) -> str:
 
 @dataclass
 class RimeTTSSettings(TTSSettings):
-    """Typed settings for Rime WS JSON and HTTP TTS services.
+    """Settings for Rime WS JSON and HTTP TTS services.
 
     Parameters:
         speaker: Voice speaker ID.
@@ -101,7 +101,7 @@ class RimeTTSSettings(TTSSettings):
 
 @dataclass
 class RimeNonJsonTTSSettings(TTSSettings):
-    """Typed settings for Rime non-JSON WS TTS service.
+    """Settings for Rime non-JSON WS TTS service.
 
     Parameters:
         speaker: Voice speaker ID.
@@ -271,10 +271,10 @@ class RimeTTSService(AudioContextWordTTSService):
         self._extra_msg_fields["inlineSpeedAlpha"] = ",".join(speed_vals + [str(speed)])
         return f"[{text}]"
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update and reconnect if voice changed."""
+        """Apply a settings update and reconnect if voice changed."""
         prev_voice = self._voice_id
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if "voice" in changed:
             self._settings.speaker = self._voice_id
             await self._disconnect()
@@ -975,13 +975,13 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
         except Exception as e:
             yield ErrorFrame(error=f"Unknown error occurred: {e}")
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update and reconnect if necessary.
+        """Apply a settings update and reconnect if necessary.
 
         Since all settings are WebSocket URL query parameters,
         any setting change requires reconnecting to apply the new values.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         # Sync voice and model to settings dict fields
         if "voice" in changed:

src/pipecat/services/sarvam/stt.py

@@ -133,7 +133,7 @@ MODEL_CONFIGS: Dict[str, ModelConfig] = {
 
 @dataclass
 class SarvamSTTSettings(STTSettings):
-    """Typed settings for the Sarvam STT service.
+    """Settings for the Sarvam STT service.
 
     Parameters:
         prompt: Optional prompt to guide transcription/translation style.
@@ -306,8 +306,8 @@ class SarvamSTTService(STTService):
                 if self._socket_client:
                     await self._socket_client.flush()
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update, validate, sync state, and reconnect.
+        """Apply a settings update, validate, sync state, and reconnect.
 
         Args:
             update: A :class:`STTSettings` (or ``SarvamSTTSettings``) delta.
@@ -336,7 +336,7 @@ class SarvamSTTService(STTService):
                 if not self._config.supports_mode:
                     raise ValueError(f"Model '{self.model_name}' does not support mode parameter.")
 
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/sarvam/tts.py

@@ -247,7 +247,7 @@ def language_to_sarvam_language(language: Language) -> Optional[str]:
 
 @dataclass
 class SarvamHttpTTSSettings(TTSSettings):
-    """Typed settings for Sarvam HTTP TTS service.
+    """Settings for Sarvam HTTP TTS service.
 
     Parameters:
         language: Sarvam language code.
@@ -277,7 +277,7 @@ class SarvamHttpTTSSettings(TTSSettings):
 
 @dataclass
 class SarvamWSTTSSettings(TTSSettings):
-    """Typed settings for Sarvam WebSocket TTS service.
+    """Settings for Sarvam WebSocket TTS service.
 
     Parameters:
         target_language_code: Sarvam language code.
@@ -953,9 +953,9 @@ class SarvamTTSService(InterruptibleTTSService):
         if isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
             await self.flush_audio()
 
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        """Apply a typed settings update and resend config if voice changed."""
+        """Apply a settings update and resend config if voice changed."""
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if "voice" in changed:
             await self._send_config()
         return changed

src/pipecat/services/settings.py

@@ -4,13 +4,12 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Typed settings infrastructure for Pipecat AI services.
+"""Settings infrastructure for Pipecat AI services.
 
-This module provides typed dataclass-based settings objects that replace the
+This module provides dataclass-based settings objects for service configuration.
-stringly-typed ``Mapping[str, Any]`` dictionaries previously used for service
+Each service type has a corresponding settings class (e.g. ``TTSSettings``,
-configuration. Each service type has a corresponding settings class (e.g.
+``LLMSettings``) whose fields use the ``NOT_GIVEN`` sentinel to distinguish
-``TTSSettings``, ``LLMSettings``) whose fields use the ``NOT_GIVEN`` sentinel
+"leave unchanged" from an explicit ``None``.
-to distinguish "leave unchanged" from an explicit ``None``.
 
 Key concepts:
 
@@ -21,7 +20,7 @@ Key concepts:
   ``NOT_GIVEN`` are simply skipped when applying an update.
 - **apply_update**: Applies a delta onto a target settings object and returns
   the set of field names that actually changed.
-- **from_mapping**: Constructs a typed settings object from a plain dict,
+- **from_mapping**: Constructs a settings object from a plain dict,
   supporting field aliases (e.g. ``"voice_id"`` → ``"voice"``).
 - **Extras**: Unknown keys land in the ``extra`` dict so services that have
   non-standard settings don't lose data.
@@ -91,7 +90,7 @@ _S = TypeVar("_S", bound="ServiceSettings")
 
 @dataclass
 class ServiceSettings:
-    """Base class for typed service settings.
+    """Base class for service settings.
 
     Every AI service type (LLM, TTS, STT) extends this with its own fields.
     Fields default to ``NOT_GIVEN`` so that an instance can represent either
@@ -188,7 +187,10 @@ class ServiceSettings:
 
     @classmethod
     def from_mapping(cls: Type[_S], settings: Mapping[str, Any]) -> _S:
-        """Construct a typed settings object from a plain dictionary.
+        """Construct a settings object from a plain dictionary.
+
+        This exists for backward compatibility with code that passes plain
+        dicts via ``*UpdateSettingsFrame(settings={...})``.
 
         Keys are matched to dataclass fields by name.  Keys listed in
         ``_aliases`` are translated to their canonical name first.  Any
@@ -250,7 +252,7 @@ class ServiceSettings:
 
 @dataclass
 class LLMSettings(ServiceSettings):
-    """Typed settings for LLM services.
+    """Settings for LLM services.
 
     Parameters:
         model: LLM model identifier.
@@ -285,7 +287,7 @@ class LLMSettings(ServiceSettings):
 
 @dataclass
 class TTSSettings(ServiceSettings):
-    """Typed settings for TTS services.
+    """Settings for TTS services.
 
     Parameters:
         model: TTS model identifier.
@@ -301,7 +303,7 @@ class TTSSettings(ServiceSettings):
 
 @dataclass
 class STTSettings(ServiceSettings):
-    """Typed settings for STT services.
+    """Settings for STT services.
 
     Parameters:
         model: STT model identifier.

src/pipecat/services/soniox/stt.py

@@ -138,7 +138,7 @@ def _prepare_language_hints(
 
 @dataclass
 class SonioxSTTSettings(STTSettings):
-    """Typed settings for Soniox STT service.
+    """Settings for Soniox STT service.
 
     Parameters:
         input_params: Soniox ``SonioxInputParams`` for detailed configuration.
@@ -217,8 +217,8 @@ class SonioxSTTService(WebsocketSTTService):
         await super().start(frame)
         await self._connect()
 
-    async def _update_settings_from_typed(self, update: SonioxSTTSettings) -> set[str]:
+    async def _update_settings(self, update: SonioxSTTSettings) -> set[str]:
-        """Apply a typed settings update, keeping ``input_params`` in sync.
+        """Apply a settings update, keeping ``input_params`` in sync.
 
         Top-level ``model`` is the source of truth.  When it is given in
         *update* its value is propagated into ``input_params``.  When only
@@ -228,14 +228,14 @@ class SonioxSTTService(WebsocketSTTService):
         Any change triggers a WebSocket reconnect.
 
         Args:
-            update: A typed settings delta.
+            update: A settings delta.
 
         Returns:
             Set of field names whose values actually changed.
         """
         model_given = is_given(getattr(update, "model", NOT_GIVEN))
 
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/speechmatics/stt.py

@@ -85,7 +85,7 @@ class TurnDetectionMode(str, Enum):
 
 @dataclass
 class SpeechmaticsSTTSettings(STTSettings):
-    """Typed settings for Speechmatics STT service.
+    """Settings for Speechmatics STT service.
 
     See ``SpeechmaticsSTTService.InputParams`` for detailed descriptions of each field.
 
@@ -415,7 +415,7 @@ class SpeechmaticsSTTService(STTService):
             )
         speaker_passive_format = params.speaker_passive_format or speaker_active_format
 
-        # Typed settings — seeded from InputParams
+        # Settings — seeded from InputParams
         self._settings = SpeechmaticsSTTSettings(
             language=params.language,
             domain=params.domain,
@@ -480,8 +480,8 @@ class SpeechmaticsSTTService(STTService):
         await super().start(frame)
         await self._connect()
 
-    async def _update_settings_from_typed(self, update: SpeechmaticsSTTSettings) -> set[str]:
+    async def _update_settings(self, update: SpeechmaticsSTTSettings) -> set[str]:
-        """Apply typed settings update, reconnecting only when necessary.
+        """Apply settings update, reconnecting only when necessary.
 
         Fields are classified into three categories (see
         ``SpeechmaticsSTTSettings``):
@@ -494,12 +494,12 @@ class SpeechmaticsSTTService(STTService):
           time and therefore require a full disconnect / reconnect.
 
         Args:
-            update: A typed settings delta.
+            update: A settings delta.
 
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if not changed:
             return changed

src/pipecat/services/stt_service.py

@@ -12,7 +12,7 @@ import time
 import warnings
 import wave
 from abc import abstractmethod
-from typing import Any, AsyncGenerator, Dict, Mapping, Optional
+from typing import Any, AsyncGenerator, Optional
 
 from loguru import logger
 from websockets.protocol import State
@@ -35,7 +35,7 @@ from pipecat.frames.frames import (
 from pipecat.metrics.metrics import TTFBMetricsData
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
-from pipecat.services.settings import ServiceSettings, STTSettings
+from pipecat.services.settings import STTSettings
 from pipecat.services.stt_latency import DEFAULT_TTFS_P99
 from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
@@ -183,11 +183,8 @@ class STTService(AIService):
                 stacklevel=2,
             )
         logger.info(f"Switching STT model to: [{model}]")
-        if isinstance(self._settings, ServiceSettings):
+        settings_cls = type(self._settings)
-            settings_cls = type(self._settings)
+        await self._update_settings(settings_cls(model=model))
-            await self._update_settings_from_typed(settings_cls(model=model))
-        else:
-            self.set_model_name(model)
 
     async def set_language(self, language: Language):
         """Set the language for speech recognition.
@@ -206,11 +203,8 @@ class STTService(AIService):
                 stacklevel=2,
             )
         logger.info(f"Switching STT language to: [{language}]")
-        if isinstance(self._settings, ServiceSettings):
+        settings_cls = type(self._settings)
-            settings_cls = type(self._settings)
+        await self._update_settings(settings_cls(language=language))
-            await self._update_settings_from_typed(settings_cls(language=language))
-        else:
-            pass
 
     @abstractmethod
     async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
@@ -242,23 +236,8 @@ class STTService(AIService):
         await super().cleanup()
         await self._cancel_ttfb_timeout()
 
-    async def _update_settings(self, settings: Mapping[str, Any]):
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        logger.info(f"Updating STT settings: {self._settings}")
+        """Apply an STT settings update.
-        for key, value in settings.items():
-            if key in self._settings:
-                logger.info(f"Updating STT setting {key} to: [{value}]")
-                self._settings[key] = value
-                if key == "language":
-                    await self.set_language(value)
-            elif key == "language":
-                await self.set_language(value)
-            elif key == "model":
-                self.set_model_name(value)
-            else:
-                logger.warning(f"Unknown setting for STT service: {key}")
-
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
-        """Apply a typed STT settings update.
 
         Handles ``model`` (via parent). Does **not** call ``set_language``
         — concrete services should override this method and handle language
@@ -266,12 +245,12 @@ class STTService(AIService):
         changed-field set.
 
         Args:
-            update: A typed STT settings delta.
+            update: An STT settings delta.
 
         Returns:
             Set of field names whose values actually changed.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         return changed
 
     async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
@@ -335,16 +314,12 @@ class STTService(AIService):
             await self._handle_vad_user_stopped_speaking(frame)
             await self.push_frame(frame, direction)
         elif isinstance(frame, STTUpdateSettingsFrame):
-            # New path: typed settings update object.
             if frame.update is not None:
-                await self._update_settings_from_typed(frame.update)
+                await self._update_settings(frame.update)
-            # Legacy path: plain dict, but service uses typed settings — convert.
+            elif frame.settings:
-            elif isinstance(self._settings, ServiceSettings):
+                # Backward-compatible path: convert legacy dict to settings object.
                 update = type(self._settings).from_mapping(frame.settings)
-                await self._update_settings_from_typed(update)
+                await self._update_settings(update)
-            # Legacy path: plain dict, service still uses dict-based settings.
-            else:
-                await self._update_settings(frame.settings)
         elif isinstance(frame, STTMuteFrame):
             self._muted = frame.mute
             logger.debug(f"STT service {'muted' if frame.mute else 'unmuted'}")

src/pipecat/services/tts_service.py

@@ -19,7 +19,6 @@ from typing import (
     Callable,
     Dict,
     List,
-    Mapping,
     Optional,
     Sequence,
     Tuple,
@@ -53,7 +52,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
-from pipecat.services.settings import ServiceSettings, TTSSettings, is_given
+from pipecat.services.settings import TTSSettings, is_given
 from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
@@ -280,11 +279,8 @@ class TTSService(AIService):
                 stacklevel=2,
             )
         logger.info(f"Switching TTS model to: [{model}]")
-        if isinstance(self._settings, ServiceSettings):
+        settings_cls = type(self._settings)
-            settings_cls = type(self._settings)
+        await self._update_settings(settings_cls(model=model))
-            await self._update_settings_from_typed(settings_cls(model=model))
-        else:
-            self.set_model_name(model)
 
     async def set_voice(self, voice: str):
         """Set the voice for speech synthesis.
@@ -303,11 +299,8 @@ class TTSService(AIService):
                 stacklevel=2,
             )
         logger.info(f"Switching TTS voice to: [{voice}]")
-        if isinstance(self._settings, ServiceSettings):
+        settings_cls = type(self._settings)
-            settings_cls = type(self._settings)
+        await self._update_settings(settings_cls(voice=voice))
-            await self._update_settings_from_typed(settings_cls(voice=voice))
-        else:
-            self._voice_id = voice
 
     def create_context_id(self) -> str:
         """Generate a unique context ID for a TTS request.
@@ -439,25 +432,8 @@ class TTSService(AIService):
             if not (agg_type == aggregation_type and func == transform_function)
         ]
 
-    async def _update_settings(self, settings: Mapping[str, Any]):
+    async def _update_settings(self, update: TTSSettings) -> set[str]:
-        for key, value in settings.items():
+        """Apply a TTS settings update.
-            if key in self._settings:
-                logger.info(f"Updating TTS setting {key} to: [{value}]")
-                self._settings[key] = value
-                if key == "language":
-                    self._settings[key] = self.language_to_service_language(value)
-            elif key == "model":
-                self.set_model_name(value)
-            elif key == "voice" or key == "voice_id":
-                self._voice_id = value
-            elif key == "text_filter":
-                for filter in self._text_filters:
-                    await filter.update_settings(value)
-            else:
-                logger.warning(f"Unknown setting for TTS service: {key}")
-
-    async def _update_settings_from_typed(self, update: TTSSettings) -> set[str]:
-        """Apply a typed TTS settings update.
 
         Handles ``model`` (via parent) and syncs ``_voice_id`` when voice
         changes.  Translates language values before applying.  Does **not**
@@ -466,7 +442,7 @@ class TTSService(AIService):
         returned changed-field set.
 
         Args:
-            update: A typed TTS settings delta.
+            update: A TTS settings delta.
 
         Returns:
             Set of field names whose values actually changed.
@@ -477,10 +453,10 @@ class TTSService(AIService):
             if converted is not None:
                 update.language = converted
 
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         # Keep _voice_id in sync for code that reads it directly
-        if "voice" in changed and isinstance(self._settings, TTSSettings):
+        if "voice" in changed:
             self._voice_id = self._settings.voice
 
         return changed
@@ -566,16 +542,12 @@ class TTSService(AIService):
             await self.flush_audio()
             self._processing_text = processing_text
         elif isinstance(frame, TTSUpdateSettingsFrame):
-            # New path: typed settings update object.
             if frame.update is not None:
-                await self._update_settings_from_typed(frame.update)
+                await self._update_settings(frame.update)
-            # Legacy path: plain dict, but service uses typed settings — convert.
+            elif frame.settings:
-            elif isinstance(self._settings, ServiceSettings):
+                # Backward-compatible path: convert legacy dict to settings object.
                 update = type(self._settings).from_mapping(frame.settings)
-                await self._update_settings_from_typed(update)
+                await self._update_settings(update)
-            # Legacy path: plain dict, service still uses dict-based settings.
-            else:
-                await self._update_settings(frame.settings)
         elif isinstance(frame, BotStoppedSpeakingFrame):
             await self._maybe_resume_frame_processing()
             await self.push_frame(frame, direction)

src/pipecat/services/ultravox/llm.py

@@ -325,8 +325,8 @@ class UltravoxRealtimeLLMService(LLMService):
             await self.cancel_task(self._receive_task, timeout=1.0)
             self._receive_task = None
 
-    async def _update_settings_from_typed(self, update: UltravoxRealtimeLLMSettings):
+    async def _update_settings(self, update: UltravoxRealtimeLLMSettings):
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
         if "output_medium" in changed:
             await self._update_output_medium(self._settings.output_medium)
         return changed

src/pipecat/services/whisper/base_stt.py

@@ -28,7 +28,7 @@ from pipecat.utils.tracing.service_decorators import traced_stt
 
 @dataclass
 class BaseWhisperSTTSettings(STTSettings):
-    """Typed settings for Whisper API-based STT services.
+    """Settings for Whisper API-based STT services.
 
     Parameters:
         base_url: API base URL.
@@ -174,13 +174,13 @@ class BaseWhisperSTTService(SegmentedSTTService):
     def _create_client(self, api_key: Optional[str], base_url: Optional[str]):
         return AsyncOpenAI(api_key=api_key, base_url=base_url)
 
-    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+    async def _update_settings(self, update: STTSettings) -> set[str]:
-        """Apply a typed settings update, syncing instance variables.
+        """Apply a settings update, syncing instance variables.
 
         Keeps ``_language``, ``_prompt``, and ``_temperature`` in sync with
-        the typed settings fields.
+        the settings fields.
         """
-        changed = await super()._update_settings_from_typed(update)
+        changed = await super()._update_settings(update)
 
         if "language" in changed:
             self._language = self.language_to_service_language(Language(self._settings.language))

src/pipecat/services/whisper/stt.py

@@ -176,7 +176,7 @@ def language_to_whisper_language(language: Language) -> Optional[str]:
 
 @dataclass
 class WhisperSTTSettings(STTSettings):
-    """Typed settings for the local Whisper (Faster Whisper) STT service.
+    """Settings for the local Whisper (Faster Whisper) STT service.
 
     Parameters:
         device: Inference device ('cpu', 'cuda', or 'auto').
@@ -191,7 +191,7 @@ class WhisperSTTSettings(STTSettings):
 
 @dataclass
 class WhisperMLXSTTSettings(STTSettings):
-    """Typed settings for the MLX Whisper STT service.
+    """Settings for the MLX Whisper STT service.
 
     Parameters:
         no_speech_prob: Probability threshold for filtering non-speech segments.

src/pipecat/services/xtts/tts.py

@@ -72,7 +72,7 @@ def language_to_xtts_language(language: Language) -> Optional[str]:
 
 @dataclass
 class XTTSTTSSettings(TTSSettings):
-    """Typed settings for XTTS TTS service.
+    """Settings for XTTS TTS service.
 
     Parameters:
         base_url: Base URL of the XTTS streaming server.