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