diff --git a/CHANGELOG.md b/CHANGELOG.md index c58fb5cf5..e69e74b45 100644 --- a/CHANGELOG.md +++ b/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 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..b4dfbb01e --- /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."}]) + 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() diff --git a/examples/foundational/48-service-switcher.py b/examples/foundational/48-service-switcher.py index d0e15d2d3..8e0f8db85 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,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( 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/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, + ) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 70beed86c..cfa0cb0a4 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 @@ -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}") 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..ccfb47adb 100644 --- a/src/pipecat/services/mcp_service.py +++ b/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