Overview
MCP is an open standard for enabling AI agents to interact with external data and tools. MCPClient provides a way to access and call tools via MCP. For example, instead of writing bespoke function call implementations for an external API, you may use an MCP server that provides a bridge to the API. Be aware there may be security implications. See MCP documenation for more details.
Installation
To use MCPClient, install the required dependencies:
pip install "pipecat-ai[mcp]"
Configuration
Constructor Parameters
You can connect to your MCP server via Stdio, SSE, or Streamable HTTP transport. See here for more documentation on MCP transports.
server_params
StdioServerParameters | SseServerParameters | StreamableHttpParameters
required
Connection parameters for the MCP server. Must be one of:
StdioServerParameters (from mcp): Connects to a local MCP server process via stdio.
from mcp import StdioServerParameters
StdioServerParameters(
command="python", # Executable
args=["server.py"], # Optional command line arguments
env=None, # Optional environment variables
)
SseServerParameters (from mcp.client.session_group): Connects to a remote MCP server via Server-Sent Events.
from mcp.client.session_group import SseServerParameters
SseServerParameters(
url="https://your.mcp.server/sse", # Server URL
headers=None, # Optional HTTP headers
timeout=5, # Connection timeout in seconds
sse_read_timeout=300, # SSE read timeout in seconds
)
StreamableHttpParameters (from mcp.client.session_group): Connects to a remote MCP server via Streamable HTTP.
from mcp.client.session_group import StreamableHttpParameters
StreamableHttpParameters(
url="https://your.mcp.server/mcp/", # Server URL
headers=None, # Optional HTTP headers
timeout=5, # Connection timeout in seconds
sse_read_timeout=300, # SSE read timeout in seconds
terminate_on_close=True, # Terminate session on close
)
tools_filter
Optional[List[str]]
default:"None"
Optional list of tool names to register. If None, all tools from the MCP
server are registered. Use this to limit which tools are exposed to the LLM.
tools_output_filters
Optional[Dict[str, Callable]]
default:"None"
Optional dictionary mapping tool names to filter functions that post-process tool outputs. Each filter function receives the raw tool output and returns the processed output.mcp = MCPClient(
server_params=server_params,
tools_output_filters={
"search": lambda result: result[:500], # Truncate long results
},
)
Usage Examples
MCP Stdio Transport
import shutil
from mcp import StdioServerParameters
from pipecat.services.mcp_service import MCPClient
# Initialize an LLM
llm = ...
# Initialize and configure MCPClient with server parameters
mcp = MCPClient(
server_params=StdioServerParameters(
command=shutil.which("npx"),
args=["-y", "@name/mcp-server-name@latest"],
env={"ENV_API_KEY": "<env_api_key>"},
)
)
# Create tools schema from the MCP server and register them with llm
tools = await mcp.register_tools(llm)
# Create context with system message and tools
context = LLMContext(
messages=[
{
"role": "system",
"content": "You are a helpful assistant in a voice conversation. You have access to MCP tools. Keep responses concise."
}
],
tools=tools
)
MCP SSE Transport
from mcp.client.session_group import SseServerParameters
from pipecat.services.mcp_service import MCPClient
# Initialize an LLM
llm = ...
# Initialize and configure MCPClient with SSE server parameters
mcp = MCPClient(
server_params=SseServerParameters(
url="https://your.mcp.server/sse",
)
)
# Create tools schema from the MCP server and register them with llm
tools = await mcp.register_tools(llm)
# Create context with system message and tools
context = LLMContext(
messages=[
{
"role": "system",
"content": "You are a helpful assistant in a voice conversation. You have access to MCP tools. Keep responses concise."
}
],
tools=tools
)
MCP Streamable HTTP Transport
import os
from mcp.client.session_group import StreamableHttpParameters
from pipecat.services.mcp_service import MCPClient
# Initialize an LLM
llm = ...
# Initialize and configure MCPClient with Streamable HTTP parameters
mcp = MCPClient(
server_params=StreamableHttpParameters(
url="https://api.githubcopilot.com/mcp/",
headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"},
)
)
# Create tools schema from the MCP server and register them with llm
tools = await mcp.register_tools(llm)
# Create context with tools
context = LLMContext(tools=tools)
Two-Step Registration
Some LLM services (e.g. Gemini Live) require tools to be passed at construction time. Use get_tools_schema() to obtain the schema first, then register_tools_schema() to register handlers after the LLM is created.
from mcp.client.session_group import StreamableHttpParameters
from pipecat.services.mcp_service import MCPClient
# Initialize MCPClient
mcp = MCPClient(
server_params=StreamableHttpParameters(
url="https://api.githubcopilot.com/mcp/",
headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"},
)
)
# Step 1: Get tools schema without registering
tools = await mcp.get_tools_schema()
# Step 2: Create LLM with tools
llm = GeminiLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
system_instruction="You are a helpful assistant.",
tools=tools,
)
# Step 3: Register tool handlers with the LLM
await mcp.register_tools_schema(tools, llm)
Multiple MCP Servers
You can combine tools from multiple MCP servers by merging their ToolsSchema objects.
from mcp import StdioServerParameters
from mcp.client.session_group import StreamableHttpParameters
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.services.mcp_service import MCPClient
# Initialize an LLM
llm = ...
# Set up multiple MCP clients
server_a = MCPClient(
server_params=StdioServerParameters(
command=shutil.which("npx"),
args=["-y", "mcp-server-a"],
env={"API_KEY": os.getenv("SERVER_A_API_KEY")},
)
)
server_b = MCPClient(
server_params=StreamableHttpParameters(
url="https://api.githubcopilot.com/mcp/",
headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"},
)
)
# Register tools from each server
tools_a = await server_a.register_tools(llm)
tools_b = await server_b.register_tools(llm)
# Merge tools into a single schema
all_tools = ToolsSchema(
standard_tools=tools_a.standard_tools + tools_b.standard_tools
)
# Create context with combined tools
context = LLMContext(tools=all_tools)
Methods
Connects to the MCP server, discovers available tools, converts their schemas
to Pipecat format, and registers them with the LLM service. This is equivalent
to calling get_tools_schema() followed by register_tools_schema().
async def register_tools(self, llm: LLMService | LLMSwitcher) -> ToolsSchema
Connects to the MCP server, discovers available tools, and converts their
schemas to Pipecat format — without registering them with an LLM. Use this
when you need the tools schema before the LLM is created.
async def get_tools_schema(self) -> ToolsSchema
Registers a previously obtained ToolsSchema with an LLM service. Use this
after get_tools_schema() once the LLM is available.
async def register_tools_schema(self, tools_schema: ToolsSchema, llm: LLMService | LLMSwitcher) -> None
Additional documentation
