diff --git a/changelog/4034.changed.md b/changelog/4034.changed.md new file mode 100644 index 000000000..4b4637ffe --- /dev/null +++ b/changelog/4034.changed.md @@ -0,0 +1 @@ +- MCPClient now requires async with MCPClient(...) as mcp: or explicit start()/close() calls to manage the connection lifecycle. diff --git a/changelog/4034.fixed.md b/changelog/4034.fixed.md new file mode 100644 index 000000000..18d000d07 --- /dev/null +++ b/changelog/4034.fixed.md @@ -0,0 +1 @@ +- Fixed MCPClient opening a new connection for every tool call instead of reusing the session. diff --git a/examples/mcp/mcp-multiple-mcp.py b/examples/mcp/mcp-multiple-mcp.py index 9d449251d..7bdfa1343 100644 --- a/examples/mcp/mcp-multiple-mcp.py +++ b/examples/mcp/mcp-multiple-mcp.py @@ -5,27 +5,17 @@ # -import asyncio -import io -import json import os import shutil -import aiohttp from dotenv import load_dotenv from loguru import logger from mcp import StdioServerParameters from mcp.client.session_group import StreamableHttpParameters -from PIL import Image from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import ( - Frame, - FunctionCallResultFrame, - LLMRunFrame, - URLImageRawFrame, -) +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 @@ -34,7 +24,6 @@ from pipecat.processors.aggregators.llm_response_universal import ( LLMContextAggregatorPair, LLMUserAggregatorParams, ) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.anthropic.llm import AnthropicLLMService @@ -47,66 +36,16 @@ from pipecat.transports.daily.transport import DailyParams load_dotenv(override=True) -class UrlToImageProcessor(FrameProcessor): - def __init__(self, aiohttp_session: aiohttp.ClientSession, **kwargs): - super().__init__(**kwargs) - self._aiohttp_session = aiohttp_session - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, FunctionCallResultFrame): - await self.push_frame(frame, direction) - image_url = self.extract_url(frame.result) - if image_url: - await self.run_image_process(image_url) - # sometimes we get multiple image urls- process 1 at a time - await asyncio.sleep(1) - else: - await self.push_frame(frame, direction) - - def extract_url(self, text: str): - try: - data = json.loads(text) - if "artObject" in data: - return data["artObject"]["webImage"]["url"] - if "artworks" in data and len(data["artworks"]): - return data["artworks"][0]["webImage"]["url"] - except (json.JSONDecodeError, KeyError, TypeError): - pass - - async def run_image_process(self, image_url: str): - try: - logger.debug(f"handling image from url: '{image_url}'") - async with self._aiohttp_session.get(image_url) as response: - image_stream = io.BytesIO(await response.content.read()) - image = Image.open(image_stream) - image = image.convert("RGB") - frame = URLImageRawFrame( - url=image_url, image=image.tobytes(), size=image.size, format="RGB" - ) - await self.push_frame(frame) - except Exception as e: - error_msg = f"Error handling image url {image_url}: {str(e)}" - logger.error(error_msg) - - # We use lambdas to defer transport parameter creation until the transport # type is selected at runtime. transport_params = { "daily": lambda: DailyParams( audio_in_enabled=True, audio_out_enabled=True, - video_out_enabled=True, - video_out_width=1024, - video_out_height=1024, ), "webrtc": lambda: TransportParams( audio_in_enabled=True, audio_out_enabled=True, - video_out_enabled=True, - video_out_width=1024, - video_out_height=1024, ), } @@ -114,85 +53,70 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - # Create an HTTP session for API calls - async with aiohttp.ClientSession() as session: - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - settings=CartesiaTTSService.Settings( - voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + settings=CartesiaTTSService.Settings( + voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ), + ) + + system_prompt = f""" + You are a helpful LLM in a voice call. + Your goal is to demonstrate your capabilities in a succinct way. + You have access to memory tools that let you store and recall information, + and tools to answer questions about the user's GitHub repositories and account. + Offer to remember things for the user, like their name, preferences, or anything they'd like. + You can also recall things you've previously stored. + You can also offer to answer users questions about their GitHub repositories and account. + Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. + Respond to what the user said in a creative and helpful way. + Don't overexplain what you are doing. + Just respond with short sentences when you are carrying out tool calls. + """ + + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), + settings=AnthropicLLMService.Settings( + system_instruction=system_prompt, + ), + ) + + async with ( + # https://github.com/modelcontextprotocol/servers/tree/main/src/memory + MCPClient( + server_params=StdioServerParameters( + command=shutil.which("npx"), + args=["-y", "@modelcontextprotocol/server-memory"], + # env={"MEMORY_FILE_PATH": "/tmp/pipecat_memory.jsonl"}, # Optional: specify MEMORY_FILE_PATH ), - ) - - system_prompt = f""" - You are a helpful LLM in a voice call. - Your goal is to demonstrate your capabilities in a succinct way. - You have access to tools to search the Rijksmuseum collection and the user's GitHub repositories and account. - Offer, for example, to show a floral still life, use the `search_artwork` tool. - The tool may respond with a JSON object with an `artworks` array. Choose the art from that array. - Once the tool has responded, tell the user the title and use the `open_image_in_browser` tool. - You can also offer to answer users questions about their GitHub repositories and account. - Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. - Respond to what the user said in a creative and helpful way. - Don't overexplain what you are doing. - Just respond with short sentences when you are carrying out tool calls. - """ - - llm = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), - settings=AnthropicLLMService.Settings( - system_instruction=system_prompt, + ) as memory_mcp, + # 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) + MCPClient( + server_params=StreamableHttpParameters( + url="https://api.githubcopilot.com/mcp/", + headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"}, ), - ) + ) as github_mcp, + ): + memory_tools = await memory_mcp.register_tools(llm) + github_tools = await github_mcp.register_tools(llm) - try: - rijksmuseum_mcp = MCPClient( - server_params=StdioServerParameters( - command=shutil.which("npx"), - # https://github.com/r-huijts/rijksmuseum-mcp - args=["-y", "mcp-server-rijksmuseum"], - env={"RIJKSMUSEUM_API_KEY": os.getenv("RIJKSMUSEUM_API_KEY")}, - ) - ) - except Exception as e: - logger.error(f"error setting up rijksmuseum mcp") - logger.exception("error trace:") - 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) - github_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.run") - logger.exception("error trace:") - - rijksmuseum_tools = {} - github_tools = {} - try: - rijksmuseum_tools = await rijksmuseum_mcp.register_tools(llm) - github_tools = await github_mcp.register_tools(llm) - except Exception as e: - logger.error(f"error registering tools") - logger.exception("error trace:") - - all_standard_tools = rijksmuseum_tools.standard_tools + github_tools.standard_tools + all_standard_tools = memory_tools.standard_tools + github_tools.standard_tools all_tools = ToolsSchema(standard_tools=all_standard_tools) - context = LLMContext(tools=all_tools) + context = LLMContext( + messages=[{"role": "user", "content": "Please introduce yourself."}], + tools=all_tools, + ) user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), ) - mcp_image_processor = UrlToImageProcessor(aiohttp_session=session) pipeline = Pipeline( [ @@ -201,7 +125,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): user_aggregator, # User spoken responses llm, # LLM tts, # TTS - mcp_image_processor, # URL image -> output transport.output(), # Transport bot output assistant_aggregator, # Assistant spoken responses and tool context ] @@ -239,10 +162,8 @@ async def bot(runner_args: RunnerArguments): if __name__ == "__main__": - if not os.getenv("RIJKSMUSEUM_API_KEY") or not os.getenv("GITHUB_PERSONAL_ACCESS_TOKEN"): - logger.error( - f"Please set `RIJKSMUSEUM_API_KEY` and `GITHUB_PERSONAL_ACCESS_TOKEN` environment variables. See https://github.com/r-huijts/rijksmuseum-mcp." - ) + if not os.getenv("GITHUB_PERSONAL_ACCESS_TOKEN"): + logger.error(f"Please set `GITHUB_PERSONAL_ACCESS_TOKEN` environment variable.") import sys sys.exit(1) diff --git a/examples/mcp/mcp-stdio.py b/examples/mcp/mcp-stdio.py index 48382046d..daae45c0c 100644 --- a/examples/mcp/mcp-stdio.py +++ b/examples/mcp/mcp-stdio.py @@ -4,26 +4,15 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio -import io -import json import os -import re import shutil -import aiohttp from dotenv import load_dotenv from loguru import logger from mcp import StdioServerParameters -from PIL import Image from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import ( - Frame, - FunctionCallResultFrame, - LLMRunFrame, - URLImageRawFrame, -) +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 @@ -32,7 +21,6 @@ from pipecat.processors.aggregators.llm_response_universal import ( LLMContextAggregatorPair, LLMUserAggregatorParams, ) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.anthropic.llm import AnthropicLLMService @@ -44,86 +32,16 @@ from pipecat.transports.daily.transport import DailyParams load_dotenv(override=True) - -class UrlToImageProcessor(FrameProcessor): - def __init__(self, aiohttp_session: aiohttp.ClientSession, **kwargs): - super().__init__(**kwargs) - self._aiohttp_session = aiohttp_session - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, FunctionCallResultFrame): - await self.push_frame(frame, direction) - image_url = self.extract_url(frame.result) - if image_url: - await self.run_image_process(image_url) - # sometimes we get multiple image urls- process 1 at a time - await asyncio.sleep(1) - else: - await self.push_frame(frame, direction) - - def extract_url(self, text: str): - try: - data = json.loads(text) - if "artObject" in data: - return data["artObject"]["webImage"]["url"] - if "artworks" in data and len(data["artworks"]): - return data["artworks"][0]["webImage"]["url"] - except (json.JSONDecodeError, KeyError, TypeError): - pass - - return None - - async def run_image_process(self, image_url: str): - try: - logger.debug(f"handling image from url: '{image_url}'") - async with self._aiohttp_session.get(image_url) as response: - image_stream = io.BytesIO(await response.content.read()) - image = Image.open(image_stream) - image = image.convert("RGB") - frame = URLImageRawFrame( - url=image_url, image=image.tobytes(), size=image.size, format="RGB" - ) - await self.push_frame(frame) - except Exception as e: - error_msg = f"Error handling image url {image_url}: {str(e)}" - logger.error(error_msg) - - -# full list of tools available from rijksmuseum MCP: -# - get_artwork_details -# - get_artwork_image -# - get_user_sets -# - get_user_set_details -# - open_image_in_browser -# - get_artist_timeline - -mcp_tools_filter = ["get_artwork_details", "get_artwork_image", "open_image_in_browser"] - - -def open_image_output_filter(output: str): - pattern = r"Successfully opened image in browser: " - text_to_print = re.sub(pattern, "", output) - print(f"🖼️ link to high resolution artwork: {text_to_print}") - - # We use lambdas to defer transport parameter creation until the transport # type is selected at runtime. transport_params = { "daily": lambda: DailyParams( audio_in_enabled=True, audio_out_enabled=True, - video_out_enabled=True, - video_out_width=1024, - video_out_height=1024, ), "webrtc": lambda: TransportParams( audio_in_enabled=True, audio_out_enabled=True, - video_out_enabled=True, - video_out_width=1024, - video_out_height=1024, ), } @@ -131,63 +49,48 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - # Create an HTTP session for API calls - async with aiohttp.ClientSession() as session: - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - settings=CartesiaTTSService.Settings( - voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady - ), + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + settings=CartesiaTTSService.Settings( + voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ), + ) + + system_prompt = f""" + You are a helpful LLM in a voice call. + Your goal is to demonstrate your capabilities in a succinct way. + You have access to memory tools that let you store and recall information. + Offer to remember things for the user, like their name, preferences, or anything they'd like. + You can also recall things you've previously stored. + Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. + Respond to what the user said in a creative and helpful way. + Don't overexplain what you are doing. + Just respond with short sentences when you are carrying out tool calls. + """ + + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), + settings=AnthropicLLMService.Settings( + system_instruction=system_prompt, + ), + ) + + # https://github.com/modelcontextprotocol/servers/tree/main/src/memory + async with MCPClient( + server_params=StdioServerParameters( + command=shutil.which("npx"), + args=["-y", "@modelcontextprotocol/server-memory"], + # env={"MEMORY_FILE_PATH": "/tmp/pipecat_memory.jsonl"}, # Optional: specify MEMORY_FILE_PATH + ), + ) as mcp: + tools = await mcp.register_tools(llm) + + context = LLMContext( + messages=[{"role": "user", "content": "Please introduce yourself."}], + tools=tools, ) - - system_prompt = f""" - You are a helpful LLM in a voice call. - Your goal is to demonstrate your capabilities in a succinct way. - You have access to tools to search the Rijksmuseum collection. - Offer, for example, to show a floral still life, use the `search_artwork` tool. - The tool may respond with a JSON object with an `artworks` array. Choose the art from that array. - Once the tool has responded, tell the user the title and use the `open_image_in_browser` tool. - Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. - Respond to what the user said in a creative and helpful way. - Don't overexplain what you are doing. - Just respond with short sentences when you are carrying out tool calls. - """ - - llm = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), - settings=AnthropicLLMService.Settings( - system_instruction=system_prompt, - ), - ) - - try: - mcp = MCPClient( - server_params=StdioServerParameters( - command=shutil.which("npx"), - # https://github.com/r-huijts/rijksmuseum-mcp - args=["-y", "mcp-server-rijksmuseum"], - env={"RIJKSMUSEUM_API_KEY": os.getenv("RIJKSMUSEUM_API_KEY")}, - ), - # Optional - tools_filter=mcp_tools_filter, # Optional - tools_output_filters={"open_image_in_browser": open_image_output_filter}, - ) - except Exception as e: - logger.error(f"error setting up mcp") - logger.exception("error trace:") - - mcp_image = UrlToImageProcessor(aiohttp_session=session) - - tools = {} - try: - tools = await mcp.register_tools(llm) - except Exception as e: - logger.error(f"error registering tools") - logger.exception("error trace:") - - context = LLMContext(tools=tools) user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), @@ -200,7 +103,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): user_aggregator, # User spoken responses llm, # LLM tts, # TTS - mcp_image, # URL image -> output transport.output(), # Transport bot output assistant_aggregator, # Assistant spoken responses and tool context ] @@ -238,13 +140,6 @@ async def bot(runner_args: RunnerArguments): if __name__ == "__main__": - if not os.getenv("RIJKSMUSEUM_API_KEY"): - logger.error( - f"Please set RIJKSMUSEUM_API_KEY environment variable for this example. See https://github.com/r-huijts/rijksmuseum-mcp and https://www.rijksmuseum.nl/en/register?redirectUrl=https://www.https://www.rijksmuseum.nl/en/rijksstudio/my/profile" - ) - import sys - - sys.exit(1) from pipecat.runner.run import main main() diff --git a/examples/mcp/mcp-streamable-http-gemini-live.py b/examples/mcp/mcp-streamable-http-gemini-live.py index ba9dba6e0..8b824454f 100644 --- a/examples/mcp/mcp-streamable-http-gemini-live.py +++ b/examples/mcp/mcp-streamable-http-gemini-live.py @@ -63,28 +63,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) - 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 voice call. Your goal is to answer questions about the user's GitHub repositories and account. @@ -94,53 +72,65 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): 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, - ) + # 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) + async with MCPClient( + server_params=StreamableHttpParameters( + url="https://api.githubcopilot.com/mcp/", + headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"}, + ) + ) as mcp: + tools = await mcp.get_tools_schema() - await mcp.register_tools_schema(tools, llm) + llm = GeminiLiveLLMService( + api_key=os.getenv("GOOGLE_API_KEY"), + system_instruction=system, + tools=tools, + ) - context = LLMContext([{"role": "developer", "content": "Please introduce yourself."}]) - user_aggregator, assistant_aggregator = LLMContextAggregatorPair( - context, - user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), - ) + await mcp.register_tools_schema(tools, llm) - pipeline = Pipeline( - [ - transport.input(), # Transport user input - user_aggregator, # User spoken responses - llm, # LLM - transport.output(), # Transport bot output - assistant_aggregator, # Assistant spoken responses and tool context - ] - ) + context = LLMContext([{"role": "user", "content": "Please introduce yourself."}]) + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - ) + pipeline = Pipeline( + [ + transport.input(), # Transport user input + user_aggregator, # User spoken responses + llm, # LLM + transport.output(), # Transport bot output + assistant_aggregator, # Assistant spoken responses and tool context + ] + ) - @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()]) + 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_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() + @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()]) - runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() - await runner.run(task) + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) async def bot(runner_args: RunnerArguments): diff --git a/examples/mcp/mcp-streamable-http.py b/examples/mcp/mcp-streamable-http.py index 5ddf53264..59860d307 100644 --- a/examples/mcp/mcp-streamable-http.py +++ b/examples/mcp/mcp-streamable-http.py @@ -63,83 +63,78 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) - system_prompt = f""" - You are a helpful LLM in a voice 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 spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. - Don't overexplain what you are doing. - Just respond with short sentences when you are carrying out tool calls. - """ + system_prompt = """\ +You are a helpful LLM in a voice 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 spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. +Don't overexplain what you are doing. +Just respond with short sentences when you are carrying out tool calls. +""" llm = GoogleLLMService( api_key=os.getenv("GOOGLE_API_KEY"), - system_instruction=system_prompt, - ) - - 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.register_tools(llm) - except Exception as e: - logger.error(f"error registering tools") - logger.exception("error trace:") - - context = LLMContext(tools=tools) - user_aggregator, assistant_aggregator = LLMContextAggregatorPair( - context, - user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), - ) - - pipeline = Pipeline( - [ - transport.input(), # Transport user input - stt, - user_aggregator, # User spoken responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - assistant_aggregator, # Assistant spoken responses and tool context - ] - ) - - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, + settings=GoogleLLMService.Settings( + system_instruction=system_prompt, ), - 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()]) + # 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) + async with MCPClient( + server_params=StreamableHttpParameters( + url="https://api.githubcopilot.com/mcp/", + headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"}, + ) + ) as mcp: + tools = await mcp.register_tools(llm) - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() + context = LLMContext( + messages=[{"role": "user", "content": "Please introduce yourself."}], + tools=tools, + ) + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) - runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, + user_aggregator, # User spoken responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + assistant_aggregator, # Assistant spoken responses and tool context + ] + ) - await runner.run(task) + 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): diff --git a/src/pipecat/services/mcp_service.py b/src/pipecat/services/mcp_service.py index d4f0807b8..275569714 100644 --- a/src/pipecat/services/mcp_service.py +++ b/src/pipecat/services/mcp_service.py @@ -7,6 +7,7 @@ """MCP (Model Context Protocol) client for integrating external tools with LLMs.""" import json +from contextlib import AsyncExitStack from typing import Any, Callable, Dict, List, Optional, TypeAlias from loguru import logger @@ -36,8 +37,14 @@ class MCPClient(BaseObject): """Client for Model Context Protocol (MCP) servers. Enables integration with MCP servers to provide external tools and resources - to LLMs. Supports both stdio and SSE server connections with automatic tool - registration and schema conversion. + to LLMs. Supports stdio, SSE, and streamable HTTP server connections with + automatic tool registration and schema conversion. + + The client maintains a persistent connection to the MCP server. It must + be used as an async context manager or explicitly started and closed:: + + async with MCPClient(server_params=...) as mcp: + tools = await mcp.register_tools(llm) Raises: TypeError: If server_params is not a supported parameter type. @@ -53,7 +60,7 @@ class MCPClient(BaseObject): """Initialize the MCP client with server parameters. Args: - server_params: Server connection parameters (stdio or SSE). + server_params: Server connection parameters (stdio, SSE, or streamable HTTP). tools_filter: Optional list of tool names to register. If None, all tools are registered. tools_output_filters: Optional dict mapping tool names to filter functions that process tool outputs. Each filter function receives the raw tool output (any type) and returns the processed output (any type). @@ -61,31 +68,84 @@ class MCPClient(BaseObject): """ super().__init__(**kwargs) self._server_params = server_params - self._session = ClientSession self._tools_filter = tools_filter self._tools_output_filters = tools_output_filters or {} + self._exit_stack: Optional[AsyncExitStack] = None + self._active_session: Optional[ClientSession] = None - if isinstance(server_params, StdioServerParameters): - self._client = stdio_client - self._list_tools = self._stdio_list_tools - self._tool_wrapper = self._stdio_tool_wrapper - elif isinstance(server_params, SseServerParameters): - self._client = sse_client - self._list_tools = self._sse_list_tools - self._tool_wrapper = self._sse_tool_wrapper - elif isinstance(server_params, StreamableHttpParameters): - self._client = streamablehttp_client - self._list_tools = self._streamable_http_list_tools - self._tool_wrapper = self._streamable_http_tool_wrapper - else: + if not isinstance( + server_params, + (StdioServerParameters, SseServerParameters, StreamableHttpParameters), + ): 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." ) + async def start(self) -> None: + """Start a persistent connection to the MCP server. + + Opens the transport and initializes the MCP session. The session + is reused for all subsequent tool calls and schema requests until + close() is called. + + Can also be used via async context manager:: + + async with MCPClient(server_params=...) as mcp: + ... + """ + if self._active_session: + return + + # We manage the exit stack manually (not via `async with`) so we can + # clean up partial resources on failure before assigning to self. + exit_stack = AsyncExitStack() + await exit_stack.__aenter__() + + try: + if isinstance(self._server_params, StdioServerParameters): + streams = await exit_stack.enter_async_context(stdio_client(self._server_params)) + read_stream, write_stream = streams[0], streams[1] + elif isinstance(self._server_params, SseServerParameters): + read_stream, write_stream = await exit_stack.enter_async_context( + sse_client(**self._server_params.model_dump()) + ) + else: # StreamableHttpParameters (validated in __init__) + read_stream, write_stream, _ = await exit_stack.enter_async_context( + streamablehttp_client(**self._server_params.model_dump()) + ) + + session = await exit_stack.enter_async_context(ClientSession(read_stream, write_stream)) + await session.initialize() + + self._exit_stack = exit_stack + self._active_session = session + + except Exception: + await exit_stack.aclose() + raise + + async def close(self) -> None: + """Close the persistent MCP connection. + + Safe to call multiple times or without having called start(). + """ + self._active_session = None + if self._exit_stack: + await self._exit_stack.aclose() + self._exit_stack = None + + async def __aenter__(self): + await self.start() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.close() + 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 + Discovers available tools from the active session, converts their schemas to Pipecat format, and registers them with the LLM service. This is the equivalent of calling get_tools_schema() followed by @@ -101,18 +161,26 @@ class MCPClient(BaseObject): await self.register_tools_schema(tools_schema, llm) return tools_schema + def _ensure_connected(self) -> ClientSession: + """Return the active session or raise if not connected.""" + if not self._active_session: + raise RuntimeError( + "MCPClient is not connected. Use 'async with MCPClient(...) as mcp:' " + "or call 'await mcp.start()' before using MCPClient." + ) + return self._active_session + 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. + Requires the client to be started via start() or async with. 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 + session = self._ensure_connected() + return await self._list_tools_helper(session) async def register_tools_schema( self, tools_schema: ToolsSchema, llm: LLMService | LLMSwitcher @@ -154,107 +222,21 @@ class MCPClient(BaseObject): return schema - async def _sse_list_tools(self) -> ToolsSchema: - """List all available mcp tools with the LLM service. - - Returns: - A ToolsSchema containing all registered tools - """ - logger.debug(f"SSE server parameters: {self._server_params}") - 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_helper(session) - return tools_schema - - async def _sse_tool_wrapper(self, params: FunctionCallParams) -> None: - """Wrapper for mcp tool calls to match Pipecat's function call interface.""" + async def _tool_wrapper(self, params: FunctionCallParams) -> None: + """Execute an MCP tool call using the persistent session.""" + session = self._ensure_connected() 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) - await params.result_callback(error_msg) - - async def _stdio_list_tools(self) -> ToolsSchema: - """List all available mcp tools with the LLM service. - - Returns: - A ToolsSchema containing all available 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_helper(session) - return tools_schema - - 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) - 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. - - Returns: - A ToolsSchema containing all available tools. - """ - logger.debug(f"Starting reading mcp tools using streamable HTTP") - - 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() - 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) - await params.result_callback(error_msg) + await self._call_tool( + session, + params.function_name, + params.arguments, + params.result_callback, + ) async def _call_tool(self, session, function_name, arguments, result_callback): logger.debug(f"Calling mcp tool '{function_name}'") + results = None try: results = await session.call_tool(function_name, arguments=arguments) except Exception as e: