MCPClient

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. The client maintains a persistent connection to the MCP server. The connection opens automatically on the first call to tools() and closes at pipeline teardown:

mcp = MCPClient(server_params=...)
context = LLMContext(messages=[...], tools=await mcp.tools())

You can also scope the client with async with MCPClient(...) or call start() and close() explicitly.

Installation

To use MCPClient, install the required dependencies:

uv add "pipecat-ai[mcp]"

MCPClient works with MCP SDK versions 1.24.0+ and 2.x. You may also need to set environment variables as required by the specific MCP server to which you are connecting.

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.

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
)

List[str] | None

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.

Dict[str, Callable] | None

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
    },
)

Dict[str, Dict[str, Any]] | None

default:"None"

Optional dictionary mapping tool names to fixed arguments that are merged into every call of that tool. Use this for arguments the LLM shouldn’t choose — a fixed search mode, an account ID, or a caller-supplied filter. Fixed arguments override any model-supplied values and are hidden from the schema the LLM sees.

mcp = MCPClient(
    server_params=server_params,
    tools_arguments={
        "search": {"mode": "realtime"},  # LLM only sees search(query=...), server gets search(query=..., mode="realtime")
    },
)

Input Parameters

See more information regarding server params here.

Usage Examples

MCP Stdio Transport

import shutil

from mcp import StdioServerParameters
from pipecat.services.mcp_service import MCPClient

# Initialize an LLM
llm = ...

# Initialize 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 context with developer message and tools
context = LLMContext(
    messages=[
        {
            "role": "developer",
            "content": "You are a helpful assistant in a voice conversation. You have access to MCP tools. Keep responses concise."
        }
    ],
    tools=await mcp.tools()
)

MCP SSE Transport

from mcp.client.session_group import SseServerParameters
from pipecat.services.mcp_service import MCPClient

# Initialize an LLM
llm = ...

# Initialize MCPClient with SSE server parameters
mcp = MCPClient(
    server_params=SseServerParameters(
        url="https://your.mcp.server/sse",
    )
)

# Create context with developer message and tools
context = LLMContext(
    messages=[
        {
            "role": "developer",
            "content": "You are a helpful assistant in a voice conversation. You have access to MCP tools. Keep responses concise."
        }
    ],
    tools=await mcp.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 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 context with tools
context = LLMContext(tools=await mcp.tools())

Tools for LLMs with Construction-Time Tool Parameters

Some LLM services (e.g. Gemini Live) require tools to be passed at construction time. The tools() method returns a ToolsSchema with handlers already attached, so you can pass it directly to the LLM constructor.

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')}"},
    )
)

# Get tools schema (handlers are attached automatically)
tools = await mcp.tools()

# Create LLM with tools
llm = GeminiLiveLLMService(
    api_key=os.getenv("GOOGLE_API_KEY"),
    system_instruction="You are a helpful assistant.",
    tools=tools,
)

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')}"},
    )
)

# Get tools from each server
tools_a = await server_a.tools()
tools_b = await server_b.tools()

# 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

async method

Gets the available MCP tools with handlers attached, ready for LLM auto-registration. Starts the server connection if needed. The connection closes automatically at pipeline teardown. This is the recommended way to use MCP tools — pass the result to LLMContext(tools=...) and the handlers register automatically.

async def tools(self) -> ToolsSchema

async method

Opens a persistent connection to the MCP server and initializes the session. The session is reused for all subsequent tool calls until close() is called. Idempotent, and called automatically by tools(). Can also be used via async context manager instead of calling this directly.

async def start(self) -> None

async method

Closes the persistent MCP connection. Called automatically at pipeline teardown once the client’s tools have been registered with an LLM service. Call it directly only to release the connection earlier (e.g. on client disconnect). Safe to call multiple times, without having called start(), and from a different task than the one that called start().

async def close(self) -> None

async method

Deprecated since 1.8.0. Use tools() instead. Discovers available tools from the active session, converts their schemas to Pipecat format, and registers them with the LLM service. Will be removed in 2.0.0.

async def register_tools(self, llm: LLMService | LLMSwitcher) -> ToolsSchema

async method

Deprecated since 1.8.0. Use tools() instead. Discovers available tools from the active session and converts their schemas to Pipecat format. Will be removed in 2.0.0.

async def get_tools_schema(self) -> ToolsSchema

async method

Deprecated since 1.8.0. Use tools() instead. Registers a previously obtained ToolsSchema with an LLM service. Will be removed in 2.0.0.

async def register_tools_schema(self, tools_schema: ToolsSchema, llm: LLMService | LLMSwitcher) -> None

Additional documentation

See MCP’s docs for MCP related updates.

Get Started

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Overview

Installation

Configuration

Constructor Parameters

Input Parameters

Usage Examples

MCP Stdio Transport

MCP SSE Transport

MCP Streamable HTTP Transport

Tools for LLMs with Construction-Time Tool Parameters

Multiple MCP Servers

Methods

Additional documentation

​Overview

​Installation

​Configuration

​Constructor Parameters

​Input Parameters

​Usage Examples

​MCP Stdio Transport

​MCP SSE Transport

​MCP Streamable HTTP Transport

​Tools for LLMs with Construction-Time Tool Parameters

​Multiple MCP Servers

​Methods

​Additional documentation

Overview

Installation

Configuration

Constructor Parameters

Input Parameters

Usage Examples

MCP Stdio Transport

MCP SSE Transport

MCP Streamable HTTP Transport

Tools for LLMs with Construction-Time Tool Parameters

Multiple MCP Servers

Methods

Additional documentation