# MCP client

> Give the bot the tools of an MCP server (stdio, SSE or streamable HTTP) with MCPClient.

Source: https://polvallverdu.github.io/pipedog/docs/services/mcp-client/



## Overview [#overview]

`MCPClient` connects to a [Model Context Protocol](https://modelcontextprotocol.io) server, lists
its tools, and gives them to the LLM as function-call tools. When the model calls one, the client
runs it on the server and returns the text to the model.

* **`MCPClient`**: the client. It keeps one persistent session with the server, opened by the first
  call to `tools()` (or by `start()`).
* **`StreamableHttpParameters`, `SseServerParameters`, `StdioServerParameters`**: how to reach the
  server.

This is the MCP *client*. For the MCP server that serves these docs to coding agents, see
[`@pipedog/mcp`](https://polvallverdu.github.io/pipedog/docs/guides/llms/#mcp-server).

<Cards>
  <Card title="Upstream source" href="https://github.com/pipecat-ai/pipecat/blob/main/src/pipecat/services/mcp_service.py">
    Pipecat's `MCPClient` (`services/mcp_service.py`).
  </Card>

  <Card title="Source" href="https://github.com/polvallverdu/pipedog/tree/main/packages/mcp-client/src">
    `packages/mcp-client/src/mcp_service.ts`.
  </Card>

  <Card title="MCP docs" href="https://modelcontextprotocol.io/docs/getting-started/intro">
    The protocol, and servers to connect to.
  </Card>
</Cards>

## Installation [#installation]

```bash
bun add @pipedog/mcp-client
```

```ts
import { MCPClient, StreamableHttpParameters } from "@pipedog/mcp-client";
```

It runs on the official [`@modelcontextprotocol/sdk`](https://github.com/modelcontextprotocol/typescript-sdk),
which is a dependency.

## Runtimes [#runtimes]

| Runtime            | Works     | What it needs                                                                                                                                                                                |
| ------------------ | --------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Bun                | ✅ yes     | An MCP server to connect to: a URL (streamable HTTP or SSE), or a command (stdio). Stdio servers start as child processes.                                                                   |
| Node.js            | ✅ yes     | An MCP server to connect to: a URL (streamable HTTP or SSE), or a command (stdio). Stdio servers start as child processes.                                                                   |
| Deno               | ✅ yes     | An MCP server to connect to: a URL (streamable HTTP or SSE), or a command (stdio). Stdio servers start as child processes: run with --allow-run. HTTP servers need --allow-net.              |
| Cloudflare Workers | ⚠️ partly | An MCP server to connect to: a URL (streamable HTTP or SSE), or a command (stdio). Streamable HTTP and SSE only. Workers cannot start processes, so a stdio server is refused with an error. |

Generated from `pipecat.runtimes` and `pipecat.compat` in `packages/mcp-client/package.json`: ✅ means the compat smoke test passes on that runtime.

| Transport       | Bun | Node | Deno | Cloudflare Workers |
| --------------- | --- | ---- | ---- | ------------------ |
| Streamable HTTP | ✅   | ✅    | ✅    | ✅                  |
| SSE             | ✅   | ✅    | ✅    | ✅                  |
| stdio           | ✅   | ✅    | ✅    | no                 |

Streamable HTTP and SSE go through `fetch`. A stdio server is a child process, which Workers cannot
start: there `tools()` rejects with an error that says to use an HTTP server. The stdio code is a
separate module that the `workerd` export condition replaces with that error, so a Worker bundle
does not include the SDK's process spawner.

## Prerequisites [#prerequisites]

* An MCP server to connect to: its URL (streamable HTTP or SSE), or the command that starts it
  (stdio). Pass credentials as `headers` (HTTP) or `env` (stdio).

## MCPClient [#mcpclient]

### Configuration [#configuration]

| Option | Type | Required | Description |
|---|---|---|---|
| `serverParams` | `ServerParameters` | yes | Server connection parameters (stdio, SSE, or streamable HTTP). |
| `toolsFilter` | `string[] \| null \| undefined` | no | Optional list of tool names to register. If null, all tools are registered. |
| `toolsOutputFilters` | `Record<string, (output: any) => any> \| null \| undefined` | no | Optional map from tool names to filter functions that process tool outputs. Each filter receives the tool's text output and returns the processed output. |
| `toolsArguments` | `Record<string, Record<string, unknown>> \| null \| undefined` | no | Optional map from tool names to fixed arguments that are merged into every call of that tool (overriding any model-supplied values). The fixed parameter names are removed from the advertised tool schema so the LLM never sees them. |
| `name` | `string \| null \| undefined` | no | Custom name; defaults to `ClassName#count`. |
| `taskManager` | `BaseTaskManager \| null \| undefined` | no | Task manager for handling tasks. |

Server parameters:

| Option | Type | Required | Description |
|---|---|---|---|
| `url` | `string \| URL` | yes | The MCP endpoint of the server. |
| `headers` | `Record<string, string> \| null \| undefined` | no | Extra HTTP headers, for example `Authorization`. |
| `timeout` | `number \| undefined` | no | Seconds to wait for the `initialize` handshake. Defaults to 30. |
| `sseReadTimeout` | `number \| undefined` | no | Seconds to wait for the response headers of a request. Defaults to 300. |
| `terminateOnClose` | `boolean \| undefined` | no | Whether closing the client also ends the server's session (an HTTP DELETE). Defaults to true. |
| `fetch` | `{ (input: RequestInfo \| URL, init?: RequestInit): Promise<Response>; (input: string \| URL \| Request, init?: RequestInit): Promise<Response>; } \| null \| undefined` | no | The `fetch` to use, instead of the global one. DIVERGENCE: not in upstream (which uses httpx). It lets a test, or a Worker with a service binding, serve the connection in process. |

| Option | Type | Required | Description |
|---|---|---|---|
| `url` | `string \| URL` | yes | The SSE endpoint of the server. |
| `headers` | `Record<string, string> \| null \| undefined` | no | Extra HTTP headers, for example `Authorization`. |
| `timeout` | `number \| undefined` | no | Seconds to wait for the `initialize` handshake. Defaults to 5. |
| `sseReadTimeout` | `number \| undefined` | no | Seconds to wait for the response headers of a request. Defaults to 300. |
| `fetch` | `{ (input: RequestInfo \| URL, init?: RequestInit): Promise<Response>; (input: string \| URL \| Request, init?: RequestInit): Promise<Response>; } \| null \| undefined` | no | The `fetch` to use, instead of the global one. DIVERGENCE: not in upstream (which uses httpx). It lets a test, or a Worker with a service binding, serve the connection in process. |

| Option | Type | Required | Description |
|---|---|---|---|
| `command` | `string` | yes | The executable to run to start the server. |
| `args` | `string[] \| undefined` | no | Command line arguments to pass to the executable. |
| `env` | `Record<string, string> \| null \| undefined` | no | The environment to use when spawning the process. It is merged over a default environment of safe variables (`PATH`, `HOME` and so on), not over the whole of `process.env`. |
| `cwd` | `string \| null \| undefined` | no | The working directory to use when spawning the process. |

### Usage [#usage]

Create the client, call `tools()`, and put the result in the context. There is no
`registerFunction` call: each schema carries its handler, and the LLM service registers it when the
context advertises the tools.

```ts title="mcp-client.ts"
import { LLMContext } from "@pipedog/core/processors/aggregators/llm_context";
import { LLMContextAggregatorPair } from "@pipedog/core/processors/aggregators/llm_response_universal";
import { MCPClient, StreamableHttpParameters } from "@pipedog/mcp-client";
import { OpenAILLMService } from "@pipedog/openai/llm";

const mcp = new MCPClient({
  serverParams: new StreamableHttpParameters({
    url: "https://mcp.example.com/mcp",
    headers: { authorization: `Bearer ${process.env.MCP_TOKEN ?? ""}` },
  }),
});

const llm = new OpenAILLMService({ apiKey: process.env.OPENAI_API_KEY ?? "" });

// tools() connects, and returns schemas that carry their handlers. The LLM service registers them
// when the context advertises them, and closes the connection when the pipeline is torn down.
const context = new LLMContext({
  messages: [{ role: "system", content: "You are a helpful voice assistant." }],
  tools: await mcp.tools(),
});
const aggregators = new LLMContextAggregatorPair(context);

```

The connection closes by itself when the pipeline is torn down, because the handlers tell the LLM
service to close the client at cleanup.

### Transports [#transports]

```ts title="transports.ts"
import { SseServerParameters, StdioServerParameters } from "@pipedog/mcp-client";

// A server that is a child process (Bun, Node and Deno).
const stdio = new MCPClient({
  serverParams: new StdioServerParameters({
    command: "npx",
    args: ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
    env: { LOG_LEVEL: "warn" },
  }),
});

// The older HTTP transport of MCP.
const sse = new MCPClient({
  serverParams: new SseServerParameters({ url: "https://mcp.example.com/sse" }),
});
```

### Filtering tools [#filtering-tools]

`toolsFilter` limits which tools are registered. `toolsOutputFilters` processes the text a tool
returns. `toolsArguments` fixes arguments: they are removed from the schema the model sees, merged
into every call, and win over anything the model sends. That is how to pin a setting, or pass an
identifier the model should not choose.

```ts title="filters.ts"
const filtered = new MCPClient({
  serverParams: new StreamableHttpParameters({ url: "https://mcp.example.com/mcp" }),
  // Register only these tools.
  toolsFilter: ["search", "get_page"],
  // Shorten what a tool returns before the model reads it.
  toolsOutputFilters: { get_page: (text: string) => text.slice(0, 2000) },
  // Fixed arguments: removed from the schema the model sees, and sent on every call.
  toolsArguments: { search: { mode: "realtime" } },
});
```

### Sessions and errors [#sessions-and-errors]

One session serves every call, as in Pipecat.

* A call that returns an error returns the line `Error calling mcp tool <name>: <reason>` to the
  model, cut to 200 characters. An empty result returns `Sorry, could not call the mcp tool`.
* A call that finds the connection gone returns its error to the model and drops the session. The
  next call connects again.
* `close()` releases the connection early. A closed client stays closed until `start()` or
  `tools()`; a tool call on it throws (it is a wiring mistake, not something for the model).
* `start()` and `close()` can be called from different tasks, and from several at once.

```ts title="lifecycle.ts"
// Open the connection ahead of the first call, so the first user turn does not pay for it.
await mcp.start();

// Release it earlier than pipeline teardown, for example when the client disconnects.
await mcp.close();

// Or scope it: `await using` closes the client at the end of the block.
{
  await using scoped = new MCPClient({
    serverParams: new StreamableHttpParameters({ url: "https://mcp.example.com/mcp" }),
  });
  await scoped.start();
}
```

### Serving a connection in process [#serving-a-connection-in-process]

`StreamableHttpParameters` and `SseServerParameters` take a `fetch`, so a test, or a Worker with a
service binding, can serve the connection without a network. Pipecat has no such option.

```ts title="fetch.ts"
// Serve the connection from a function, with no network: a test, or a service binding on Workers.
declare const handler: (request: Request) => Promise<Response>;

const inProcess = new MCPClient({
  serverParams: new StreamableHttpParameters({
    url: "http://in.process/mcp",
    fetch: ((input, init) => handler(new Request(input, init))) as typeof fetch,
  }),
});

```

## Differences from Pipecat [#differences-from-pipecat]

* **The constructor takes one options object**: `new MCPClient({ serverParams, toolsFilter, ... })`.
* **`async with` is `await using`**, or `await mcp.scoped(async (mcp) => { ... })`.
* **Deprecated methods are not ported**: `register_tools`, `get_tools_schema` and
  `register_tools_schema`. Use `tools()`.
* **Timeouts.** With `fetch` there is no separate connect timeout. `timeout` bounds the `initialize`
  handshake and `sseReadTimeout` the wait for the response headers of a request; an idle SSE stream
  is not timed out.
* **`tools()` follows the server's pagination cursor.** Pipecat reads the first page.
* **Lost connections** are recognised from what the SDK raises (`McpError` with code
  `ConnectionClosed`, or `Not connected`), where Pipecat reads anyio's stream errors. A request that
  fails at the network level is reported to the model and keeps the session.
* **The SDK's `MCP error -32000:` prefix is left off** the line the model reads.
* **On Workers the SDK does not check a result against the tool's output schema.** The check needs
  code generation, which Workers forbid.

## Runtime notes [#runtime-notes]

* Calls are cancellation points: when a function call is cancelled (an interruption, a timeout),
  the request to the server is cancelled with it.
* The SDK's own request timeout is off, as Pipecat's session has none, so a slow tool is waited for.
  Set `functionCallTimeoutSecs` on the LLM service if the bot should not wait.
* On Deno, stdio servers need `--allow-run`, and HTTP ones `--allow-net`.