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`.
This commit is contained in:
@@ -57,6 +57,12 @@ reason")`.
|
|||||||
supported languages before Pipecat's service classes are updated, while still
|
supported languages before Pipecat's service classes are updated, while still
|
||||||
providing guidance on verified languages.
|
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
|
||||||
|
|
||||||
- Fixed an issue where the `SmallWebRTCRequest` dataclass in runner would scrub
|
- Fixed an issue where the `SmallWebRTCRequest` dataclass in runner would scrub
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ from loguru import logger
|
|||||||
|
|
||||||
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
||||||
from pipecat.adapters.schemas.tools_schema import ToolsSchema
|
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
|
from pipecat.utils.base_object import BaseObject
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -59,13 +59,16 @@ class MCPClient(BaseObject):
|
|||||||
|
|
||||||
if isinstance(server_params, StdioServerParameters):
|
if isinstance(server_params, StdioServerParameters):
|
||||||
self._client = stdio_client
|
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):
|
elif isinstance(server_params, SseServerParameters):
|
||||||
self._client = sse_client
|
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):
|
elif isinstance(server_params, StreamableHttpParameters):
|
||||||
self._client = streamablehttp_client
|
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:
|
else:
|
||||||
raise TypeError(
|
raise TypeError(
|
||||||
f"{self} invalid argument type: `server_params` must be either StdioServerParameters, SseServerParameters, or StreamableHttpParameters."
|
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
|
Connects to the MCP server, discovers available tools, converts their
|
||||||
schemas to Pipecat format, and registers them with the LLM service.
|
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:
|
Args:
|
||||||
llm: The Pipecat LLM service to register tools with.
|
llm: The Pipecat LLM service to register tools with.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
A ToolsSchema containing all successfully registered tools.
|
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
|
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(
|
def _convert_mcp_schema_to_pipecat(
|
||||||
self, tool_name: str, tool_schema: Dict[str, Any]
|
self, tool_name: str, tool_schema: Dict[str, Any]
|
||||||
) -> FunctionSchema:
|
) -> FunctionSchema:
|
||||||
@@ -114,112 +144,76 @@ class MCPClient(BaseObject):
|
|||||||
|
|
||||||
return schema
|
return schema
|
||||||
|
|
||||||
async def _sse_register_tools(self, llm) -> ToolsSchema:
|
async def _sse_list_tools(self) -> ToolsSchema:
|
||||||
"""Register all available mcp tools with the LLM service.
|
"""List all available mcp tools with the LLM service.
|
||||||
|
|
||||||
Args:
|
|
||||||
llm: The Pipecat LLM service to register tools with
|
|
||||||
Returns:
|
Returns:
|
||||||
A ToolsSchema containing all registered tools
|
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(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._client(**self._server_params.model_dump()) as (read, write):
|
||||||
async with self._session(read, write) as session:
|
async with self._session(read, write) as session:
|
||||||
await session.initialize()
|
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
|
return tools_schema
|
||||||
|
|
||||||
async def _stdio_register_tools(self, llm) -> ToolsSchema:
|
async def _sse_tool_wrapper(self, params: FunctionCallParams) -> None:
|
||||||
"""Register all available mcp tools with the LLM service.
|
"""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:
|
Returns:
|
||||||
A ToolsSchema containing all registered tools
|
A ToolsSchema containing all available tools.
|
||||||
"""
|
"""
|
||||||
|
logger.debug(f"Starting reading mcp 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")
|
|
||||||
|
|
||||||
async with self._client(self._server_params) as streams:
|
async with self._client(self._server_params) as streams:
|
||||||
async with self._session(streams[0], streams[1]) as session:
|
async with self._session(streams[0], streams[1]) as session:
|
||||||
await session.initialize()
|
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
|
return tools_schema
|
||||||
|
|
||||||
async def _streamable_http_register_tools(self, llm) -> ToolsSchema:
|
async def _stdio_tool_wrapper(self, params: FunctionCallParams) -> None:
|
||||||
"""Register all available mcp tools with the LLM service using streamable HTTP.
|
"""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:
|
Returns:
|
||||||
A ToolsSchema containing all registered tools
|
A ToolsSchema containing all available tools.
|
||||||
"""
|
"""
|
||||||
|
logger.debug(f"Starting reading mcp tools using streamable HTTP")
|
||||||
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")
|
|
||||||
|
|
||||||
async with self._client(**self._server_params.model_dump()) as (
|
async with self._client(**self._server_params.model_dump()) as (
|
||||||
read_stream,
|
read_stream,
|
||||||
@@ -228,9 +222,30 @@ class MCPClient(BaseObject):
|
|||||||
):
|
):
|
||||||
async with self._session(read_stream, write_stream) as session:
|
async with self._session(read_stream, write_stream) as session:
|
||||||
await session.initialize()
|
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
|
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):
|
async def _call_tool(self, session, function_name, arguments, result_callback):
|
||||||
logger.debug(f"Calling mcp tool '{function_name}'")
|
logger.debug(f"Calling mcp tool '{function_name}'")
|
||||||
try:
|
try:
|
||||||
@@ -257,7 +272,7 @@ class MCPClient(BaseObject):
|
|||||||
final_response = response if len(response) else "Sorry, could not call the mcp tool"
|
final_response = response if len(response) else "Sorry, could not call the mcp tool"
|
||||||
await result_callback(final_response)
|
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()
|
available_tools = await session.list_tools()
|
||||||
tool_schemas: List[FunctionSchema] = []
|
tool_schemas: List[FunctionSchema] = []
|
||||||
|
|
||||||
@@ -278,20 +293,16 @@ class MCPClient(BaseObject):
|
|||||||
{"description": tool.description, "input_schema": tool.inputSchema},
|
{"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
|
# Add to list of schemas
|
||||||
tool_schemas.append(function_schema)
|
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:
|
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:")
|
logger.exception("Full exception details:")
|
||||||
continue
|
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)
|
tools_schema = ToolsSchema(standard_tools=tool_schemas)
|
||||||
|
|
||||||
return tools_schema
|
return tools_schema
|
||||||
|
|||||||
Reference in New Issue
Block a user