From 018449371191cd48a5a4f8960eec15822d7504db Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 3 Nov 2025 16:40:23 -0500 Subject: [PATCH 01/12] Update the service switcher example to illustrate registering tools on all LLMs in a switcher --- examples/foundational/48-service-switcher.py | 34 ++++++++++++++++++-- 1 file changed, 31 insertions(+), 3 deletions(-) diff --git a/examples/foundational/48-service-switcher.py b/examples/foundational/48-service-switcher.py index d0e15d2d3..221a4ab7d 100644 --- a/examples/foundational/48-service-switcher.py +++ b/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,11 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams load_dotenv(override=True) + +async def fetch_weather_from_api(params: FunctionCallParams): + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + # 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 +72,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 +106,10 @@ 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 ) + llm_switcher.register_function("get_current_weather", fetch_weather_from_api) messages = [ { @@ -90,8 +117,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]) - context = LLMContext(messages) + context = LLMContext(messages, tools) context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( From bee4165ba4cff562052c8d24e289c1aed4d62cc8 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 3 Nov 2025 16:48:14 -0500 Subject: [PATCH 02/12] Add `LLMSwitcher.register_direct_function()` --- CHANGELOG.md | 4 ++++ examples/foundational/48-service-switcher.py | 17 ++++++++++++++++- src/pipecat/pipeline/llm_switcher.py | 20 ++++++++++++++++++++ 3 files changed, 40 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c58fb5cf5..fbc62fb33 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,10 @@ 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. + ### Changed - Bumped the `fastapi` dependency's upperbound to `<0.122.0`. diff --git a/examples/foundational/48-service-switcher.py b/examples/foundational/48-service-switcher.py index 221a4ab7d..8e0f8db85 100644 --- a/examples/foundational/48-service-switcher.py +++ b/examples/foundational/48-service-switcher.py @@ -40,10 +40,22 @@ 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. @@ -109,7 +121,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): 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 = [ { @@ -117,7 +132,7 @@ 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]) + tools = ToolsSchema(standard_tools=[weather_function, get_restaurant_recommendation]) context = LLMContext(messages, tools) context_aggregator = LLMContextAggregatorPair(context) diff --git a/src/pipecat/pipeline/llm_switcher.py b/src/pipecat/pipeline/llm_switcher.py index 50d919263..cbbb8c315 100644 --- a/src/pipecat/pipeline/llm_switcher.py +++ b/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, + ) From e6f881bb08ba22bfc6e16a960ac546ecaa4ece25 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 3 Nov 2025 18:03:48 -0500 Subject: [PATCH 03/12] 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 | 5 ++ .../adapters/services/gemini_adapter.py | 46 +++++++++++++++++-- .../services/google/gemini_live/llm.py | 11 ----- src/pipecat/services/google/llm.py | 11 ----- src/pipecat/services/llm_service.py | 11 ----- src/pipecat/services/mcp_service.py | 45 ------------------ 6 files changed, 46 insertions(+), 83 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fbc62fb33..9e5985218 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index dc5bae559..a4f70b1fa 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/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, []) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 70beed86c..56d81e12b 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/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. diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 7cbc82789..47877c4df 100644 --- a/src/pipecat/services/google/llm.py +++ b/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 diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 0a1a835f7..e342be7eb 100644 --- a/src/pipecat/services/llm_service.py +++ b/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. diff --git a/src/pipecat/services/mcp_service.py b/src/pipecat/services/mcp_service.py index 3d851e050..5d11cfa29 100644 --- a/src/pipecat/services/mcp_service.py +++ b/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"], From a9d78bd9561f71c63cfc95740728248288b623bc Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 3 Nov 2025 20:52:53 -0500 Subject: [PATCH 04/12] Make it possible to get a `ToolsSchema` out of an `MCPClient` without passing in an LLM service. This allows folks to use `MCPClient` alongside the pattern of passing in tools at LLM init time, a pattern supported by speech-to-speech services such as `GeminiLiveLLMService`. --- CHANGELOG.md | 6 + src/pipecat/services/mcp_service.py | 205 +++++++++++++++------------- 2 files changed, 114 insertions(+), 97 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e5985218..2f5aba6f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,6 +57,12 @@ reason")`. supported languages before Pipecat's service classes are updated, while still providing guidance on verified languages. +- Added the two-step `MCPClient.get_tools_schema()` and + `MCPClient.register_tools_schema()` as two-step alternative to + `MCPClient.register_tools()`, to allow users to use `MCPClient` alongside + the pattern of passing in tools to the LLM service constructor (a pattern + supported by speech-to-speech services such as `GeminiLiveLLMService`). + ### Fixed - Fixed an issue where the `SmallWebRTCRequest` dataclass in runner would scrub diff --git a/src/pipecat/services/mcp_service.py b/src/pipecat/services/mcp_service.py index 5d11cfa29..e38ea17b1 100644 --- a/src/pipecat/services/mcp_service.py +++ b/src/pipecat/services/mcp_service.py @@ -13,7 +13,7 @@ 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.services.llm_service import FunctionCallParams, LLMService from pipecat.utils.base_object import BaseObject try: @@ -59,13 +59,16 @@ class MCPClient(BaseObject): 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." @@ -77,15 +80,42 @@ class MCPClient(BaseObject): 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. """ - 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 + async def get_tools_schema(self) -> ToolsSchema: + """Get the schema of all available MCP tools without registering them. + + Connects to the MCP server, discovers available tools, and converts their + schemas to Pipecat format. + + Returns: + A ToolsSchema containing all available tools. This can be used for + subsequent registration using register_tools_schema(). + """ + tools_schema = await self._list_tools() + return tools_schema + + async def register_tools_schema(self, tools_schema: ToolsSchema, llm: LLMService) -> None: + """Register the MCP tools (previously obtained from get_tools_schema()) with the LLM service. + + 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] ) -> FunctionSchema: @@ -114,112 +144,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, @@ -228,9 +222,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: @@ -257,7 +272,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] = [] @@ -278,20 +293,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 From 29ef0f419f7a87ce6f0d48b6e434458ec327c20f Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 3 Nov 2025 22:19:07 -0500 Subject: [PATCH 05/12] Add typing formalizing `MCPClient` support for registering tools on an `LLMSwitcher` in addition to an `LLMService`. --- CHANGELOG.md | 20 ++++++++++++++------ src/pipecat/services/mcp_service.py | 7 +++++-- 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f5aba6f2..411e381af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,20 @@ reason")`. `LLMSwitcher.register_function()` in that it's a shorthand for registering functions on all LLMs in the switcher, but for direct functions. +- Added support for passing in an `LLMSwicher` to `MCPClient.register_tools()` + (as well as the new `MCPClient.register_tools_schema()`). + +- 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 it takes a direct function (a + `FunctionSchema`-less function). + +- Added the two-step `MCPClient.get_tools_schema()` and + `MCPClient.register_tools_schema()` as two-step alternative to + `MCPClient.register_tools()`, to allow users to use `MCPClient` alongside + the pattern of passing in tools to the LLM service constructor (a pattern + supported by speech-to-speech services such as `GeminiLiveLLMService`). + ### Changed - Bumped the `fastapi` dependency's upperbound to `<0.122.0`. @@ -57,12 +71,6 @@ reason")`. supported languages before Pipecat's service classes are updated, while still providing guidance on verified languages. -- Added the two-step `MCPClient.get_tools_schema()` and - `MCPClient.register_tools_schema()` as two-step alternative to - `MCPClient.register_tools()`, to allow users to use `MCPClient` alongside - the pattern of passing in tools to the LLM service constructor (a pattern - supported by speech-to-speech services such as `GeminiLiveLLMService`). - ### Fixed - Fixed an issue where the `SmallWebRTCRequest` dataclass in runner would scrub diff --git a/src/pipecat/services/mcp_service.py b/src/pipecat/services/mcp_service.py index e38ea17b1..ccfb47adb 100644 --- a/src/pipecat/services/mcp_service.py +++ b/src/pipecat/services/mcp_service.py @@ -13,6 +13,7 @@ from loguru import logger from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.pipeline.llm_switcher import LLMSwitcher from pipecat.services.llm_service import FunctionCallParams, LLMService from pipecat.utils.base_object import BaseObject @@ -74,7 +75,7 @@ class MCPClient(BaseObject): 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 @@ -106,7 +107,9 @@ class MCPClient(BaseObject): tools_schema = await self._list_tools() return tools_schema - async def register_tools_schema(self, tools_schema: ToolsSchema, llm: LLMService) -> None: + 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. Args: From 24365aeefe16946b29e621ad71f29ecd53d27cec Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 4 Nov 2025 09:29:30 -0500 Subject: [PATCH 06/12] CHANGELOG wording fix --- CHANGELOG.md | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 411e381af..b7a9f56f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,14 +48,13 @@ reason")`. - 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 it takes a direct function (a - `FunctionSchema`-less function). + a function on all LLMs in the switcher, except this new method takes a direct + function (a `FunctionSchema`-less function). -- Added the two-step `MCPClient.get_tools_schema()` and - `MCPClient.register_tools_schema()` as two-step alternative to - `MCPClient.register_tools()`, to allow users to use `MCPClient` alongside - the pattern of passing in tools to the LLM service constructor (a pattern - supported by speech-to-speech services such as `GeminiLiveLLMService`). +- 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. ### Changed From 75245e1daab939a64f27aed2f19968f4683692d8 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 4 Nov 2025 10:26:56 -0500 Subject: [PATCH 07/12] Fix a bug in `GeminiLiveLLMService` where in some circumstances it wouldn't respond after a tool call --- CHANGELOG.md | 3 +++ src/pipecat/services/google/gemini_live/llm.py | 10 +++++++--- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b7a9f56f0..63a3d5932 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -76,6 +76,9 @@ 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 diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 56d81e12b..c7e4827df 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/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 @@ -1012,7 +1010,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( From 9ce33f23b9cea025741017501005cbb85eec4ba1 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 4 Nov 2025 13:29:08 -0500 Subject: [PATCH 08/12] Add an example demonstrating MCP usage with a speech-to-speech service (`GeminiLiveLLMService`) using the pattern of passing in tools in the constructor --- .../39d-mcp-run-http-gemini-live.py | 165 ++++++++++++++++++ 1 file changed, 165 insertions(+) create mode 100644 examples/foundational/39d-mcp-run-http-gemini-live.py diff --git a/examples/foundational/39d-mcp-run-http-gemini-live.py b/examples/foundational/39d-mcp-run-http-gemini-live.py new file mode 100644 index 000000000..3b9cd503e --- /dev/null +++ b/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."}], NOT_GIVEN) + 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() From 84ba628dfb2bd155a911f9ea68ed9c52aeff0c8d Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 5 Nov 2025 11:40:36 -0500 Subject: [PATCH 09/12] Fix a bug in `GeminiLiveLLMService` where if only *one* of tools or system instruction was provided in the context, the other wouldn't fall back to using the value provided in the constructor. Not adding this fix to the CHANGELOG since `GeminiLiveLLMService`'s ability to properly handle context-provided tools and system instruction hasn't been published yet. --- src/pipecat/services/google/gemini_live/llm.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index c7e4827df..317ed034d 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -1142,8 +1142,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}") From 0f69d4aea33107b86186b7640ea1cdf89ad12146 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 5 Nov 2025 12:02:02 -0500 Subject: [PATCH 10/12] Fixed an issue where `GeminiLiveLLMService` wasn't consistent in what it would do if if it received an `LLMContextFrame` (triggered by an `LLMRunFrame`, say) and there were no user messages in the initial context: - If the context contained a system message, that message would be converted to a user message and the LLM would respond - If the system message was provided as a constructor argument, though, no user messages would be sent to the LLM, and the LLM would therefore not respond Not adding this fix to the CHANGELOG since `GeminiLiveLLMService`'s ability to properly handle context-provided tools and system instruction hasn't been published yet. --- .../services/google/gemini_live/llm.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 317ed034d..cfa0cb0a4 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -982,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. From 61aec087942a6f2ec52e9975564136f7cf349615 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 5 Nov 2025 14:33:44 -0500 Subject: [PATCH 11/12] CHANGELOG item ordering tweak --- CHANGELOG.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 63a3d5932..e69e74b45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,9 +43,6 @@ reason")`. `LLMSwitcher.register_function()` in that it's a shorthand for registering functions on all LLMs in the switcher, but for direct functions. -- Added support for passing in an `LLMSwicher` to `MCPClient.register_tools()` - (as well as the new `MCPClient.register_tools_schema()`). - - 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 @@ -56,6 +53,9 @@ reason")`. 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`. From 13d6078ea07b6ca592741b5cf1a4e5e4a3671d48 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 5 Nov 2025 14:35:21 -0500 Subject: [PATCH 12/12] Minor tweak to an example for clarity. --- examples/foundational/39d-mcp-run-http-gemini-live.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/foundational/39d-mcp-run-http-gemini-live.py b/examples/foundational/39d-mcp-run-http-gemini-live.py index 3b9cd503e..b4dfbb01e 100644 --- a/examples/foundational/39d-mcp-run-http-gemini-live.py +++ b/examples/foundational/39d-mcp-run-http-gemini-live.py @@ -107,7 +107,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await mcp.register_tools_schema(tools, llm) - context = LLMContext([{"role": "user", "content": "Please introduce yourself."}], NOT_GIVEN) + context = LLMContext([{"role": "user", "content": "Please introduce yourself."}]) context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline(