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 and 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)

Installation

To use MCPClient, install the required dependencies:

uv add "pipecat-ai[mcp]"

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

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 and configure MCPClient with server parameters
async with MCPClient(
    server_params=StdioServerParameters(
        command=shutil.which("npx"),
        args=["-y", "@name/mcp-server-name@latest"],
        env={"ENV_API_KEY": "<env_api_key>"},
    )
) as mcp:
    # Create tools schema from the MCP server and register them with llm
    tools = await mcp.register_tools(llm)

    # 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=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
async with MCPClient(
    server_params=SseServerParameters(
        url="https://your.mcp.server/sse",
    )
) as mcp:
    # Create tools schema from the MCP server and register them with llm
    tools = await mcp.register_tools(llm)

    # 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=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
async with MCPClient(
    server_params=StreamableHttpParameters(
        url="https://api.githubcopilot.com/mcp/",
        headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"},
    )
) as mcp:
    # 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
async with MCPClient(
    server_params=StreamableHttpParameters(
        url="https://api.githubcopilot.com/mcp/",
        headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"},
    )
) as mcp:
    # 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
async with MCPClient(
    server_params=StdioServerParameters(
        command=shutil.which("npx"),
        args=["-y", "mcp-server-a"],
        env={"API_KEY": os.getenv("SERVER_A_API_KEY")},
    )
) as server_a, MCPClient(
    server_params=StreamableHttpParameters(
        url="https://api.githubcopilot.com/mcp/",
        headers={"Authorization": f"Bearer {os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')}"},
    )
) as server_b:
    # 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

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. 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. Safe to call multiple times or without having called start().

async def close(self) -> None

async method

Discovers available tools from the active session, 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(). Requires the client to be started via start() or async context manager.

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

async method

Discovers available tools from the active session 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. Requires the client to be started via start() or async context manager.

async def get_tools_schema(self) -> ToolsSchema

async method

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

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

Two-Step Registration

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

​Two-Step Registration

​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

Two-Step Registration

Multiple MCP Servers

Methods

Additional documentation