Remove the "needs alternate schema" mechanism in MCPClient, moving the necessary schema massaging into GeminiLLMAdapter instead.

This does a couple of things:
- Makes the `MCPClient` LLM agnostic, setting us up for some upcoming improvements (like making it possible to use with `LLMSwitcher`)
- Makes `GeminiLLMAdapter` more robust, as the schema massaging that was previously only done in `MCPClient` is useful for all tools, not just for MCP-provided ones

CHANGELOG.md

@@ -68,6 +68,11 @@ reason")`.
 - `GeminiLiveLLMService` now properly supports context-provided system
   instruction and tools.
 
+### Removed
+
+- Removed `needs_mcp_alternate_schema()` from `LLMService`. The mechanism that
+  relied on it went away.
+
 ## [0.0.92] - 2025-10-31 🎃 "The Haunted Edition" 👻
 
 ### Added

src/pipecat/adapters/services/gemini_adapter.py

@@ -80,12 +80,48 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
             List of tool definitions formatted for Gemini's function-calling API.
             Includes both converted standard tools and any custom Gemini-specific tools.
         """
+
+        def _strip_additional_properties(schema: Dict[str, Any]) -> Dict[str, Any]:
+            """Recursively remove "additionalProperties" fields from JSON schema, as they're not supported by Gemini.
+
+            Args:
+                schema: The JSON schema dict to process.
+
+            Returns:
+                JSON schema dict with "additionalProperties" stripped out.
+            """
+            if not isinstance(schema, dict):
+                return schema
+
+            result = {}
+
+            for key, value in schema.items():
+                if key == "additionalProperties":
+                    continue
+                elif isinstance(value, dict):
+                    result[key] = _strip_additional_properties(value)
+                elif isinstance(value, list):
+                    result[key] = [
+                        _strip_additional_properties(item) if isinstance(item, dict) else item
+                        for item in value
+                    ]
+                else:
+                    result[key] = value
+
+            return result
+
         functions_schema = tools_schema.standard_tools
-        formatted_standard_tools = (
-            [{"function_declarations": [func.to_default_dict() for func in functions_schema]}]
-            if functions_schema
-            else []
-        )
+        if functions_schema:
+            formatted_functions = []
+            for func in functions_schema:
+                func_dict = func.to_default_dict()
+                func_dict["parameters"]["properties"] = _strip_additional_properties(
+                    func_dict["parameters"]["properties"]
+                )
+                formatted_functions.append(func_dict)
+            formatted_standard_tools = [{"function_declarations": formatted_functions}]
+        else:
+            formatted_standard_tools = []
         custom_gemini_tools = []
         if tools_schema.custom_tools:
             custom_gemini_tools = tools_schema.custom_tools.get(AdapterType.GEMINI, [])

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

@@ -772,17 +772,6 @@ class GeminiLiveLLMService(LLMService):
         """
         return True
 
-    def needs_mcp_alternate_schema(self) -> bool:
-        """Check if this LLM service requires alternate MCP schema.
-
-        Google/Gemini has stricter JSON schema validation and requires
-        certain properties to be removed or modified for compatibility.
-
-        Returns:
-            True for Google/Gemini services.
-        """
-        return True
-
     def set_audio_input_paused(self, paused: bool):
         """Set the audio input pause state.
 

src/pipecat/services/google/llm.py

@@ -778,17 +778,6 @@ class GoogleLLMService(LLMService):
 
         return None
 
-    def needs_mcp_alternate_schema(self) -> bool:
-        """Check if this LLM service requires alternate MCP schema.
-
-        Google/Gemini has stricter JSON schema validation and requires
-        certain properties to be removed or modified for compatibility.
-
-        Returns:
-            True for Google/Gemini services.
-        """
-        return True
-
     def _maybe_unset_thinking_budget(self, generation_params: Dict[str, Any]):
         try:
             # There's no way to introspect on model capabilities, so

src/pipecat/services/llm_service.py

@@ -419,17 +419,6 @@ class LLMService(AIService):
             return True
         return function_name in self._functions.keys()
 
-    def needs_mcp_alternate_schema(self) -> bool:
-        """Check if this LLM service requires alternate MCP schema.
-
-        Some LLM services have stricter JSON schema validation and require
-        certain properties to be removed or modified for compatibility.
-
-        Returns:
-            True if MCP schemas should be cleaned for this service, False otherwise.
-        """
-        return False
-
     async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]):
         """Execute a sequence of function calls from the LLM.
 

src/pipecat/services/mcp_service.py

@@ -56,7 +56,6 @@ class MCPClient(BaseObject):
         super().__init__(**kwargs)
         self._server_params = server_params
         self._session = ClientSession
-        self._needs_alternate_schema = False
 
         if isinstance(server_params, StdioServerParameters):
             self._client = stdio_client
@@ -84,48 +83,9 @@ class MCPClient(BaseObject):
         Returns:
             A ToolsSchema containing all successfully registered tools.
         """
-        # Check once if the LLM needs alternate strict schema
-        self._needs_alternate_schema = llm and llm.needs_mcp_alternate_schema()
         tools_schema = await self._register_tools(llm)
         return tools_schema
 
-    def _get_alternate_schema_for_strict_validation(self, schema: Dict[str, Any]) -> Dict[str, Any]:
-        """Get an alternate JSON schema to be compatible with LLMs that have strict validation.
-
-        Some LLMs have stricter validation and don't allow certain schema properties
-        that are valid in standard JSON Schema.
-
-        Args:
-            schema: The JSON schema to get an alternate schema for
-
-        Returns:
-            An alternate schema compatible with strict validation
-        """
-        if not isinstance(schema, dict):
-            return schema
-
-        alternate_schema = {}
-
-        for key, value in schema.items():
-            # Skip additionalProperties as some LLMs don't like additionalProperties: false
-            if key == "additionalProperties":
-                continue
-
-            # Recursively get alternate schema for nested objects
-            if isinstance(value, dict):
-                alternate_schema[key] = self._get_alternate_schema_for_strict_validation(value)
-            elif isinstance(value, list):
-                alternate_schema[key] = [
-                    self._get_alternate_schema_for_strict_validation(item)
-                    if isinstance(item, dict)
-                    else item
-                    for item in value
-                ]
-            else:
-                alternate_schema[key] = value
-
-        return alternate_schema
-
     def _convert_mcp_schema_to_pipecat(
         self, tool_name: str, tool_schema: Dict[str, Any]
     ) -> FunctionSchema:
@@ -143,11 +103,6 @@ class MCPClient(BaseObject):
         properties = tool_schema["input_schema"].get("properties", {})
         required = tool_schema["input_schema"].get("required", [])
 
-        # Only get alternate schema for LLMs that need strict schema validation
-        if self._needs_alternate_schema:
-            logger.debug("Getting alternate schema for strict validation")
-            properties = self._get_alternate_schema_for_strict_validation(properties)
-
         schema = FunctionSchema(
             name=tool_name,
             description=tool_schema["description"],