Merge pull request #2970 from pipecat-ai/pk/tool-registration-improvements

Assorted tool registration improvements

CHANGELOG.md

@@ -39,6 +39,23 @@ reason")`.
   extract probability metrics from `TranscriptionFrame` objects for Whisper-based,
   OpenAI GPT-4o-transcribe, and Deepgram STT services respectively.
 
+- Added `LLMSwitcher.register_direct_function()`. It works much like
+  `LLMSwitcher.register_function()` in that it's a shorthand for registering
+  functions on all LLMs in the switcher, but for direct functions.
+
+- Added `LLMSwitcher.register_direct_function()`. It works much like
+  `LLMSwitcher.register_function()` in that it's a shorthand for registering
+  a function on all LLMs in the switcher, except this new method takes a direct
+  function (a `FunctionSchema`-less function).
+
+- Added `MCPClient.get_tools_schema()` and `MCPClient.register_tools_schema()`
+  as a two-step alternative to `MCPClient.register_tools()`, to allow users to
+  pass MCP tools to, say, `GeminiLiveLLMService` (as well as other
+  speech-to-speech services) in the constructor.
+
+- Added support for passing in an `LLMSwicher` to `MCPClient.register_tools()`
+  (as well as the new `MCPClient.register_tools_schema()`).
+
 ### Changed
 
 - Bumped the `fastapi` dependency's upperbound to `<0.122.0`.
@@ -59,11 +76,19 @@ reason")`.
   arbitrary request data from client due to camelCase typing. This fixes data
   passthrough for JS clients where `APIRequest` is used.
 
+- Fixed a bug in `GeminiLiveLLMService` where in some circumstances it wouldn't
+  respond after a tool call.
+
 - Fixed `GeminiLiveLLMService` session resumption after a connection timeout.
 
 - `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

examples/foundational/39d-mcp-run-http-gemini-live.py

@@ -0,0 +1,165 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+from mcp.client.session_group import StreamableHttpParameters
+
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import NOT_GIVEN, LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
+from pipecat.services.mcp_service import MCPClient
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    try:
+        # Github MCP docs: https://github.com/github/github-mcp-server
+        # Enable Github Copilot on your GitHub account. Free tier is ok. (https://github.com/settings/copilot)
+        # Generate a personal access token. It must be a Fine-grained token, classic tokens are not supported. (https://github.com/settings/personal-access-tokens)
+        # Set permissions you want to use (eg. "all repositories", "profile: read/write", etc)
+        mcp = MCPClient(
+            server_params=StreamableHttpParameters(
+                url="https://api.githubcopilot.com/mcp/",
+                headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"},
+            )
+        )
+    except Exception as e:
+        logger.error(f"error setting up mcp")
+        logger.exception("error trace:")
+
+    tools = {}
+    try:
+        tools = await mcp.get_tools_schema()
+    except Exception as e:
+        logger.error(f"error registering tools")
+        logger.exception("error trace:")
+
+    system = f"""
+    You are a helpful LLM in a WebRTC call.
+    Your goal is to answer questions about the user's GitHub repositories and account.
+    You have access to a number of tools provided by Github. Use any and all tools to help users.
+    Your output will be converted to audio so don't include special characters in your answers.
+    Don't overexplain what you are doing.
+    Just respond with short sentences when you are carrying out tool calls.
+    """
+
+    llm = GeminiLiveLLMService(
+        api_key=os.getenv("GOOGLE_API_KEY"),
+        system_instruction=system,
+        tools=tools,
+    )
+
+    await mcp.register_tools_schema(tools, llm)
+
+    context = LLMContext([{"role": "user", "content": "Please introduce yourself."}])
+    context_aggregator = LLMContextAggregatorPair(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            context_aggregator.user(),  # User spoken responses
+            llm,  # LLM
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses and tool context
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected: {client}")
+        # Kick off the conversation.
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    if not os.getenv("GITHUB_PERSONAL_ACCESS_TOKEN"):
+        logger.error(
+            f"Please set GITHUB_PERSONAL_ACCESS_TOKEN environment variable for this example."
+        )
+        import sys
+
+        sys.exit(1)
+
+    from pipecat.runner.run import main
+
+    main()

examples/foundational/48-service-switcher.py

@@ -10,11 +10,14 @@ import os
 from dotenv import load_dotenv
 from loguru import logger
 
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
 from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.frames.frames import LLMRunFrame, ManuallySwitchServiceFrame
+from pipecat.pipeline.llm_switcher import LLMSwitcher
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyManual
@@ -28,6 +31,7 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.deepgram.tts import DeepgramTTSService
 from pipecat.services.google.llm import GoogleLLMService
+from pipecat.services.llm_service import FunctionCallParams
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
@@ -35,6 +39,23 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
 
 load_dotenv(override=True)
 
+
+# "Classic" function
+async def fetch_weather_from_api(params: FunctionCallParams):
+    await params.result_callback({"conditions": "nice", "temperature": "75"})
+
+
+# "Direct" function
+async def get_restaurant_recommendation(params: FunctionCallParams, location: str):
+    """
+    Get a restaurant recommendation.
+
+    Args:
+        location (str): The city and state, e.g. "San Francisco, CA".
+    """
+    await params.result_callback({"name": "The Golden Dragon"})
+
+
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
 # instantiated. The function will be called when the desired transport gets
 # selected.
@@ -63,6 +84,23 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
+    weather_function = FunctionSchema(
+        name="get_current_weather",
+        description="Get the current weather",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+            "format": {
+                "type": "string",
+                "enum": ["celsius", "fahrenheit"],
+                "description": "The temperature unit to use. Infer this from the user's location.",
+            },
+        },
+        required=["location", "format"],
+    )
+
     stt_cartesia = CartesiaSTTService(api_key=os.getenv("CARTESIA_API_KEY"))
     stt_deepgram = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
     stt_switcher = ServiceSwitcher(
@@ -80,9 +118,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     llm_openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
     llm_google = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
-    llm_switcher = ServiceSwitcher(
-        services=[llm_openai, llm_google], strategy_type=ServiceSwitcherStrategyManual
+    llm_switcher = LLMSwitcher(
+        llms=[llm_openai, llm_google], strategy_type=ServiceSwitcherStrategyManual
     )
+    # Register a "classic" function
+    llm_switcher.register_function("get_current_weather", fetch_weather_from_api)
+    # Register a "direct" function
+    llm_switcher.register_direct_function(get_restaurant_recommendation)
 
     messages = [
         {
@@ -90,8 +132,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
         },
     ]
+    tools = ToolsSchema(standard_tools=[weather_function, get_restaurant_recommendation])
 
-    context = LLMContext(messages)
+    context = LLMContext(messages, tools)
     context_aggregator = LLMContextAggregatorPair(context)
 
     pipeline = Pipeline(

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/pipeline/llm_switcher.py

@@ -8,6 +8,7 @@
 
 from typing import Any, List, Optional, Type
 
+from pipecat.adapters.schemas.direct_function import DirectFunction
 from pipecat.pipeline.service_switcher import ServiceSwitcher, StrategyType
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.services.llm_service import LLMService
@@ -95,3 +96,22 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
                 start_callback=start_callback,
                 cancel_on_interruption=cancel_on_interruption,
             )
+
+    def register_direct_function(
+        self,
+        handler: DirectFunction,
+        *,
+        cancel_on_interruption: bool = True,
+    ):
+        """Register a direct function handler for LLM function calls, on all LLMs, active or not.
+
+        Args:
+            handler: The direct function to register. Must follow DirectFunction protocol.
+            cancel_on_interruption: Whether to cancel this function call when an
+                interruption occurs. Defaults to True.
+        """
+        for llm in self.llms:
+            llm.register_direct_function(
+                handler=handler,
+                cancel_on_interruption=cancel_on_interruption,
+            )

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

@@ -13,8 +13,6 @@ voice transcription, streaming responses, and tool usage.
 
 import base64
 import io
-import json
-import random
 import time
 import uuid
 import warnings
@@ -772,17 +770,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.
 
@@ -995,7 +982,24 @@ class GeminiLiveLLMService(LLMService):
             await self._process_completed_function_calls(send_new_results=False)
 
             # Create initial response if needed, based on conversation history
-            # in context
+            # in context.
+            # (If the context has no messages but we do have a system
+            # instruction — meaning it was provided at init time — doctor our
+            # context now so that we'll have something to send to the service
+            # to trigger a response).
+            messages = params["messages"]
+            if not messages and self._inference_on_context_initialization:
+                if self._system_instruction_from_init:
+                    logger.debug(
+                        "No messages found in initial context; seeding with system instruction to trigger bot response."
+                    )
+                    self._context.add_message(
+                        {"role": "system", "content": self._system_instruction_from_init}
+                    )
+                else:
+                    logger.warning(
+                        "No messages found in initial context; cannot trigger initial bot response without messages or system instruction."
+                    )
             await self._create_initial_response()
         else:
             # We got an updated context.
@@ -1023,7 +1027,13 @@ class GeminiLiveLLMService(LLMService):
                     if part.function_response:
                         tool_call_id = part.function_response.id
                         tool_name = part.function_response.name
-                        if tool_call_id and tool_call_id not in self._completed_tool_calls:
+                        response = part.function_response.response
+                        if (
+                            tool_call_id
+                            and tool_call_id not in self._completed_tool_calls
+                            and response
+                            and response.get("value") != "IN_PROGRESS"
+                        ):
                             # Found a newly-completed function call - send the result to the service
                             if send_new_results:
                                 await self._tool_result(
@@ -1149,8 +1159,9 @@ class GeminiLiveLLMService(LLMService):
                 params = adapter.get_llm_invocation_params(self._context)
                 system_instruction = params["system_instruction"]
                 tools = params["tools"]
-            else:
+            if not system_instruction:
                 system_instruction = self._system_instruction_from_init
+            if not tools:
                 tools = adapter.from_standard_tools(self._tools_from_init)
             if system_instruction:
                 logger.debug(f"Setting system instruction: {system_instruction}")

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

@@ -13,7 +13,8 @@ from loguru import logger
 
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.services.llm_service import FunctionCallParams
+from pipecat.pipeline.llm_switcher import LLMSwitcher
+from pipecat.services.llm_service import FunctionCallParams, LLMService
 from pipecat.utils.base_object import BaseObject
 
 try:
@@ -56,75 +57,67 @@ 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
-            self._register_tools = self._stdio_register_tools
+            self._list_tools = self._stdio_list_tools
+            self._tool_wrapper = self._stdio_tool_wrapper
         elif isinstance(server_params, SseServerParameters):
             self._client = sse_client
-            self._register_tools = self._sse_register_tools
+            self._list_tools = self._sse_list_tools
+            self._tool_wrapper = self._sse_tool_wrapper
         elif isinstance(server_params, StreamableHttpParameters):
             self._client = streamablehttp_client
-            self._register_tools = self._streamable_http_register_tools
+            self._list_tools = self._streamable_http_list_tools
+            self._tool_wrapper = self._streamable_http_tool_wrapper
         else:
             raise TypeError(
                 f"{self} invalid argument type: `server_params` must be either StdioServerParameters, SseServerParameters, or StreamableHttpParameters."
             )
 
-    async def register_tools(self, llm) -> ToolsSchema:
+    async def register_tools(self, llm: LLMService | LLMSwitcher) -> ToolsSchema:
         """Register all available MCP tools with an LLM service.
 
         Connects to the MCP server, discovers available tools, converts their
         schemas to Pipecat format, and registers them with the LLM service.
 
+        This is the equivalent of calling get_tools_schema() followed by
+        register_tools_schema().
+
         Args:
             llm: The Pipecat LLM service to register tools with.
 
         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)
+        tools_schema = await self.get_tools_schema()
+        await self.register_tools_schema(tools_schema, 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.
+    async def get_tools_schema(self) -> ToolsSchema:
+        """Get the schema of all available MCP tools without registering them.
 
-        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
+        Connects to the MCP server, discovers available tools, and converts their
+        schemas to Pipecat format.
 
         Returns:
-            An alternate schema compatible with strict validation
+            A ToolsSchema containing all available tools. This can be used for
+            subsequent registration using register_tools_schema().
         """
-        if not isinstance(schema, dict):
-            return schema
+        tools_schema = await self._list_tools()
+        return tools_schema
 
-        alternate_schema = {}
+    async def register_tools_schema(
+        self, tools_schema: ToolsSchema, llm: LLMService | LLMSwitcher
+    ) -> None:
+        """Register the MCP tools (previously obtained from get_tools_schema()) with the LLM service.
 
-        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
+        Args:
+            tools_schema: The ToolsSchema to register with the LLM service.
+            llm: The Pipecat LLM service to register tools with.
+        """
+        for function_schema in tools_schema.standard_tools:
+            llm.register_function(function_schema.name, self._tool_wrapper)
 
     def _convert_mcp_schema_to_pipecat(
         self, tool_name: str, tool_schema: Dict[str, Any]
@@ -143,11 +136,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"],
@@ -159,112 +147,76 @@ class MCPClient(BaseObject):
 
         return schema
 
-    async def _sse_register_tools(self, llm) -> ToolsSchema:
-        """Register all available mcp tools with the LLM service.
+    async def _sse_list_tools(self) -> ToolsSchema:
+        """List all available mcp tools with the LLM service.
 
-        Args:
-            llm: The Pipecat LLM service to register tools with
         Returns:
             A ToolsSchema containing all registered tools
         """
-
-        async def mcp_tool_wrapper(params: FunctionCallParams) -> None:
-            """Wrapper for mcp tool calls to match Pipecat's function call interface."""
-            logger.debug(
-                f"Executing tool '{params.function_name}' with call ID: {params.tool_call_id}"
-            )
-            logger.trace(f"Tool arguments: {json.dumps(params.arguments, indent=2)}")
-            try:
-                async with self._client(**self._server_params.model_dump()) as (read, write):
-                    async with self._session(read, write) as session:
-                        await session.initialize()
-                        await self._call_tool(
-                            session, params.function_name, params.arguments, params.result_callback
-                        )
-            except Exception as e:
-                error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}"
-                logger.error(error_msg)
-                logger.exception("Full exception details:")
-                await params.result_callback(error_msg)
-
         logger.debug(f"SSE server parameters: {self._server_params}")
-        logger.debug("Starting registration of mcp tools")
+        logger.debug(f"Starting reading mcp tools")
 
         async with self._client(**self._server_params.model_dump()) as (read, write):
             async with self._session(read, write) as session:
                 await session.initialize()
-                tools_schema = await self._list_tools(session, mcp_tool_wrapper, llm)
+                tools_schema = await self._list_tools_helper(session)
                 return tools_schema
 
-    async def _stdio_register_tools(self, llm) -> ToolsSchema:
-        """Register all available mcp tools with the LLM service.
+    async def _sse_tool_wrapper(self, params: FunctionCallParams) -> None:
+        """Wrapper for mcp tool calls to match Pipecat's function call interface."""
+        logger.debug(f"Executing tool '{params.function_name}' with call ID: {params.tool_call_id}")
+        logger.trace(f"Tool arguments: {json.dumps(params.arguments, indent=2)}")
+        try:
+            async with self._client(**self._server_params.model_dump()) as (read, write):
+                async with self._session(read, write) as session:
+                    await session.initialize()
+                    await self._call_tool(
+                        session, params.function_name, params.arguments, params.result_callback
+                    )
+        except Exception as e:
+            error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}"
+            logger.error(error_msg)
+            logger.exception("Full exception details:")
+            await params.result_callback(error_msg)
+
+    async def _stdio_list_tools(self) -> ToolsSchema:
+        """List all available mcp tools with the LLM service.
 
-        Args:
-            llm: The Pipecat LLM service to register tools with
         Returns:
-            A ToolsSchema containing all registered tools
+            A ToolsSchema containing all available tools.
         """
-
-        async def mcp_tool_wrapper(params: FunctionCallParams) -> None:
-            """Wrapper for mcp tool calls to match Pipecat's function call interface."""
-            logger.debug(
-                f"Executing tool '{params.function_name}' with call ID: {params.tool_call_id}"
-            )
-            logger.trace(f"Tool arguments: {json.dumps(params.arguments, indent=2)}")
-            try:
-                async with self._client(self._server_params) as streams:
-                    async with self._session(streams[0], streams[1]) as session:
-                        await session.initialize()
-                        await self._call_tool(
-                            session, params.function_name, params.arguments, params.result_callback
-                        )
-            except Exception as e:
-                error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}"
-                logger.error(error_msg)
-                logger.exception("Full exception details:")
-                await params.result_callback(error_msg)
-
-        logger.debug("Starting registration of mcp tools")
+        logger.debug(f"Starting reading mcp tools")
 
         async with self._client(self._server_params) as streams:
             async with self._session(streams[0], streams[1]) as session:
                 await session.initialize()
-                tools_schema = await self._list_tools(session, mcp_tool_wrapper, llm)
+                tools_schema = await self._list_tools_helper(session)
                 return tools_schema
 
-    async def _streamable_http_register_tools(self, llm) -> ToolsSchema:
-        """Register all available mcp tools with the LLM service using streamable HTTP.
+    async def _stdio_tool_wrapper(self, params: FunctionCallParams) -> None:
+        """Wrapper for mcp tool calls to match Pipecat's function call interface."""
+        logger.debug(f"Executing tool '{params.function_name}' with call ID: {params.tool_call_id}")
+        logger.trace(f"Tool arguments: {json.dumps(params.arguments, indent=2)}")
+        try:
+            async with self._client(self._server_params) as streams:
+                async with self._session(streams[0], streams[1]) as session:
+                    await session.initialize()
+                    await self._call_tool(
+                        session, params.function_name, params.arguments, params.result_callback
+                    )
+        except Exception as e:
+            error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}"
+            logger.error(error_msg)
+            logger.exception("Full exception details:")
+            await params.result_callback(error_msg)
+
+    async def _streamable_http_list_tools(self) -> ToolsSchema:
+        """List all available mcp tools with the LLM service using streamable HTTP.
 
-        Args:
-            llm: The Pipecat LLM service to register tools with
         Returns:
-            A ToolsSchema containing all registered tools
+            A ToolsSchema containing all available tools.
         """
-
-        async def mcp_tool_wrapper(params: FunctionCallParams) -> None:
-            """Wrapper for mcp tool calls to match Pipecat's function call interface."""
-            logger.debug(
-                f"Executing tool '{params.function_name}' with call ID: {params.tool_call_id}"
-            )
-            logger.trace(f"Tool arguments: {json.dumps(params.arguments, indent=2)}")
-            try:
-                async with self._client(**self._server_params.model_dump()) as (
-                    read_stream,
-                    write_stream,
-                    _,
-                ):
-                    async with self._session(read_stream, write_stream) as session:
-                        await session.initialize()
-                        await self._call_tool(
-                            session, params.function_name, params.arguments, params.result_callback
-                        )
-            except Exception as e:
-                error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}"
-                logger.error(error_msg)
-                logger.exception("Full exception details:")
-                await params.result_callback(error_msg)
-
-        logger.debug("Starting registration of mcp tools using streamable HTTP")
+        logger.debug(f"Starting reading mcp tools using streamable HTTP")
 
         async with self._client(**self._server_params.model_dump()) as (
             read_stream,
@@ -273,9 +225,30 @@ class MCPClient(BaseObject):
         ):
             async with self._session(read_stream, write_stream) as session:
                 await session.initialize()
-                tools_schema = await self._list_tools(session, mcp_tool_wrapper, llm)
+                tools_schema = await self._list_tools_helper(session)
                 return tools_schema
 
+    async def _streamable_http_tool_wrapper(self, params: FunctionCallParams) -> None:
+        """Wrapper for mcp tool calls to match Pipecat's function call interface."""
+        logger.debug(f"Executing tool '{params.function_name}' with call ID: {params.tool_call_id}")
+        logger.trace(f"Tool arguments: {json.dumps(params.arguments, indent=2)}")
+        try:
+            async with self._client(**self._server_params.model_dump()) as (
+                read_stream,
+                write_stream,
+                _,
+            ):
+                async with self._session(read_stream, write_stream) as session:
+                    await session.initialize()
+                    await self._call_tool(
+                        session, params.function_name, params.arguments, params.result_callback
+                    )
+        except Exception as e:
+            error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}"
+            logger.error(error_msg)
+            logger.exception("Full exception details:")
+            await params.result_callback(error_msg)
+
     async def _call_tool(self, session, function_name, arguments, result_callback):
         logger.debug(f"Calling mcp tool '{function_name}'")
         try:
@@ -302,7 +275,7 @@ class MCPClient(BaseObject):
         final_response = response if len(response) else "Sorry, could not call the mcp tool"
         await result_callback(final_response)
 
-    async def _list_tools(self, session, mcp_tool_wrapper, llm):
+    async def _list_tools_helper(self, session):
         available_tools = await session.list_tools()
         tool_schemas: List[FunctionSchema] = []
 
@@ -323,20 +296,16 @@ class MCPClient(BaseObject):
                     {"description": tool.description, "input_schema": tool.inputSchema},
                 )
 
-                # Register the wrapped function
-                logger.debug(f"Registering function handler for '{tool_name}'")
-                llm.register_function(tool_name, mcp_tool_wrapper)
-
                 # Add to list of schemas
                 tool_schemas.append(function_schema)
-                logger.debug(f"Successfully registered tool '{tool_name}'")
+                logger.debug(f"Successfully read tool '{tool_name}'")
 
             except Exception as e:
-                logger.error(f"Failed to register tool '{tool_name}': {str(e)}")
+                logger.error(f"Failed to read tool '{tool_name}': {str(e)}")
                 logger.exception("Full exception details:")
                 continue
 
-        logger.debug(f"Completed registration of {len(tool_schemas)} tools")
+        logger.debug(f"Completed reading {len(tool_schemas)} tools")
         tools_schema = ToolsSchema(standard_tools=tool_schemas)
 
         return tools_schema