wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #2967 from pipecat-ai/vp-39-bork-83

Browse Source 
update example 39-mcp-stdio.py to use different mcp server
This commit is contained in:
Vanessa Pyne
2025-11-04 09:02:29 -06:00
committed by GitHub
parent 5c1bd8cda2 b7a4d7371c
commit fb737ff671
4 changed files with 52 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
examples/foundational/39-mcp-stdio.py
View File

@@ -6,6 +6,7 @@
import asyncio import asyncio
import io import io
import json
import os import os
import re import re
import shutil import shutil
@@ -63,10 +64,12 @@ class UrlToImageProcessor(FrameProcessor):
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
def extract_url(self, text: str): def extract_url(self, text: str):
pattern = r"!\[[^\]]*\]\((https?://[^)]+\.(png|jpg|jpeg|PNG|JPG|JPEG))\)" data = json.loads(text)
match = re.search(pattern, text) if "artObject" in data:
if match: return data["artObject"]["webImage"]["url"]
return match.group(1) if "artworks" in data and len(data["artworks"]):
return data["artworks"][0]["webImage"]["url"]
return None return None
async def run_image_process(self, image_url: str): async def run_image_process(self, image_url: str):
@@ -130,9 +133,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
mcp = MCPClient( mcp = MCPClient(
server_params=StdioServerParameters( server_params=StdioServerParameters(
command=shutil.which("npx"), command=shutil.which("npx"),
args=["-y", "@programcomputer/nasa-mcp-server@latest"], # https://github.com/r-huijts/rijksmuseum-mcp
# https://api.nasa.gov args=["-y", "mcp-server-rijksmuseum"],
env={"NASA_API_KEY": os.getenv("NASA_API_KEY")}, env={"RIJKSMUSEUM_API_KEY": os.getenv("RIJKSMUSEUM_API_KEY")},
) )
) )
except Exception as e: except Exception as e:
@@ -141,15 +144,20 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
mcp_image = UrlToImageProcessor(aiohttp_session=session) mcp_image = UrlToImageProcessor(aiohttp_session=session)
tools = await mcp.register_tools(llm) tools = {}
try:
tools = await mcp.register_tools(llm)
except Exception as e:
logger.error(f"error registering tools")
logger.exception("error trace:")
system = f""" system = f"""
You are a helpful LLM in a WebRTC call. You are a helpful LLM in a WebRTC call.
Your goal is to demonstrate your capabilities in a succinct way. Your goal is to demonstrate your capabilities in a succinct way.
You have access to a number of tools provided by NASA MCP. Use any and all tools to help users. You have access to tools to search the Rijksmuseum collection.
When asked for the astronomy picture of the day, PASS in NO date to the API. Offer, for example, to show the earliest Rembrandt work from the museum. Use the `search_artwork` tool.
This ensures we get the latest picture available. If as specific date is asked for, you The tool may respond with a JSON object with an `artworks` array. Choose the art from that array.
can pass in that date to the API. Once the tool has responded, tell the user the title and use the `open_image_in_browser` tool.
Your output will be converted to audio so don't include special characters in your answers. 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. Respond to what the user said in a creative and helpful way.
Don't overexplain what you are doing. Don't overexplain what you are doing.
@@ -206,14 +214,13 @@ async def bot(runner_args: RunnerArguments):
if __name__ == "__main__": if __name__ == "__main__":
if not os.getenv("NASA_API_KEY"): if not os.getenv("RIJKSMUSEUM_API_KEY"):
logger.error( logger.error(
f"Please set NASA_API_KEY environment variable for this example. See https://api.nasa.gov" 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 import sys
sys.exit(1) sys.exit(1)
from pipecat.runner.run import main from pipecat.runner.run import main
main() main()

7
examples/foundational/39a-mcp-run-sse.py
View File

@@ -79,7 +79,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.error(f"error setting up mcp") logger.error(f"error setting up mcp")
logger.exception("error trace:") logger.exception("error trace:")
tools = await mcp.register_tools(llm) tools = {}
try:
tools = await mcp.register_tools(llm)
except Exception as e:
logger.error(f"error registering tools")
logger.exception("error trace:")
system = f""" system = f"""
You are a helpful LLM in a WebRTC call. You are a helpful LLM in a WebRTC call.

29
examples/foundational/39b-multiple-mcp.py
View File

@@ -132,9 +132,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
system = f""" system = f"""
You are a helpful LLM in a WebRTC call. You are a helpful LLM in a WebRTC call.
Your goal is to demonstrate your capabilities in a succinct way. Your goal is to demonstrate your capabilities in a succinct way.
You have access to a number of tools provided by NASA MCP. Use any and all tools to help users. You have access to tools to search the Rijksmuseum collection.
When asked for today's date, use 'https://www.datetoday.net/'. Offer, for example, to show the earliest Rembrandt work from the museum. Use the `search_artwork` tool.
When asked for the astronomy picture of the day, use 'https://www.datetoday.net/', to get today's date. 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 converted to audio so don't include special characters in your answers. 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. Respond to what the user said in a creative and helpful way.
Don't overexplain what you are doing. Don't overexplain what you are doing.
@@ -147,13 +148,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
mcp = MCPClient( mcp = MCPClient(
server_params=StdioServerParameters( server_params=StdioServerParameters(
command=shutil.which("npx"), command=shutil.which("npx"),
args=["-y", "@programcomputer/nasa-mcp-server@latest"], # https://github.com/r-huijts/rijksmuseum-mcp
# https://api.nasa.gov args=["-y", "mcp-server-error setting up mcp"],
env={"NASA_API_KEY": os.getenv("NASA_API_KEY")}, env={"RIJKSMUSEUM_API_KEY": os.getenv("RIJKSMUSEUM_API_KEY")},
) )
) )
except Exception as e: except Exception as e:
logger.error(f"error setting up nasa mcp") logger.error(f"error setting up rijksmuseum mcp")
logger.exception("error trace:") logger.exception("error trace:")
try: try:
# https://docs.mcp.run/integrating/tutorials/mcp-run-sse-openai-agents/ # https://docs.mcp.run/integrating/tutorials/mcp-run-sse-openai-agents/
@@ -164,8 +165,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.error(f"error setting up mcp.run") logger.error(f"error setting up mcp.run")
logger.exception("error trace:") logger.exception("error trace:")
tools = await mcp.register_tools(llm) tools = {}
run_tools = await mcp_run.register_tools(llm) run_tools = {}
try:
tools = await mcp.register_tools(llm)
run_tools = await mcp_run.register_tools(llm)
except Exception as e:
logger.error(f"error registering tools")
logger.exception("error trace:")
all_standard_tools = run_tools.standard_tools + tools.standard_tools all_standard_tools = run_tools.standard_tools + tools.standard_tools
all_tools = ToolsSchema(standard_tools=all_standard_tools) all_tools = ToolsSchema(standard_tools=all_standard_tools)
@@ -219,9 +226,9 @@ async def bot(runner_args: RunnerArguments):
if __name__ == "__main__": if __name__ == "__main__":
if not os.getenv("NASA_API_KEY") or not os.getenv("MCP_RUN_SSE_URL"): if not os.getenv("RIJKSMUSEUM_API_KEY") or not os.getenv("MCP_RUN_SSE_URL"):
logger.error( logger.error(
f"Please set NASA_API_KEY and MCP_RUN_SSE_URL environment variables. See https://api.nasa.gov and https://mcp.run" f"Please set RIJKSMUSEUM_API_KEY and MCP_RUN_SSE_URL environment variables. See https://github.com/r-huijts/rijksmuseum-mcp and https://mcp.run"
) )
import sys import sys

7
examples/foundational/39c-mcp-run-http.py
View File

@@ -85,7 +85,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.error(f"error setting up mcp") logger.error(f"error setting up mcp")
logger.exception("error trace:") logger.exception("error trace:")
tools = await mcp.register_tools(llm) tools = {}
try:
tools = await mcp.register_tools(llm)
except Exception as e:
logger.error(f"error registering tools")
logger.exception("error trace:")
system = f""" system = f"""
You are a helpful LLM in a WebRTC call. You are a helpful LLM in a WebRTC call.