> ## Documentation Index
> Fetch the complete documentation index at: https://docs.pipecat.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# MCPClient

> Service to connect to MCP (Model Context Protocol) servers

## 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](https://github.com/modelcontextprotocol) 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:

```python theme={null}
async with MCPClient(server_params=...) as mcp:
    tools = await mcp.register_tools(llm)
```

## Installation

To use `MCPClient`, install the required dependencies:

```bash theme={null}
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](https://modelcontextprotocol.io/docs/concepts/transports#built-in-transport-types) for more documentation on MCP transports.

<ParamField path="server_params" type="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.

  ```python theme={null}
  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.

  ```python theme={null}
  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.

  ```python theme={null}
  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
  )
  ```
</ParamField>

<ParamField path="tools_filter" type="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.
</ParamField>

<ParamField path="tools_output_filters" type="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.

  ```python theme={null}
  mcp = MCPClient(
      server_params=server_params,
      tools_output_filters={
          "search": lambda result: result[:500],  # Truncate long results
      },
  )
  ```
</ParamField>

### Input Parameters

See more information regarding server params [here](https://github.com/modelcontextprotocol/python-sdk?tab=readme-ov-file#writing-mcp-clients).

## Usage Examples

### MCP Stdio Transport

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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

<ResponseField name="start" type="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.
</ResponseField>

```python theme={null}
async def start(self) -> None
```

<ResponseField name="close" type="async method">
  Closes the persistent MCP connection. Safe to call multiple times or without
  having called `start()`.
</ResponseField>

```python theme={null}
async def close(self) -> None
```

<ResponseField name="register_tools" type="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.
</ResponseField>

```python theme={null}
async def register_tools(self, llm: LLMService | LLMSwitcher) -> ToolsSchema
```

<ResponseField name="get_tools_schema" type="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.
</ResponseField>

```python theme={null}
async def get_tools_schema(self) -> ToolsSchema
```

<ResponseField name="register_tools_schema" type="async method">
  Registers a previously obtained `ToolsSchema` with an LLM service. Use this
  after `get_tools_schema()` once the LLM is available.
</ResponseField>

```python theme={null}
async def register_tools_schema(self, tools_schema: ToolsSchema, llm: LLMService | LLMSwitcher) -> None
```

## Additional documentation

<Note>
  See [MCP's docs](https://github.com/modelcontextprotocol/python-sdk) for MCP
  related updates.
</Note>