MCP client

OverviewCopy Anchor Link

MCPClient connects to a Model Context Protocol 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.

InstallationCopy Anchor Link

bun add @pipedog/mcp-client

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

It runs on the official @modelcontextprotocol/sdk, which is a dependency.

RuntimesCopy Anchor Link

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

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.

PrerequisitesCopy Anchor Link

• 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).

MCPClientCopy Anchor Link

ConfigurationCopy Anchor Link

Server parameters:

UsageCopy Anchor Link

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.

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.

TransportsCopy Anchor Link

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 toolsCopy Anchor Link

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.

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 errorsCopy Anchor Link

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.

// 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 processCopy Anchor Link

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.

// 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 PipecatCopy Anchor Link

• 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 notesCopy Anchor Link

• 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.