STARWARCLOUD STARWAR.AI
返回市场
MCP服务器

MCP服务器

作者：mastra-ai18452 星标更新：2025-11-24
安装使用来源：spider==github.com

项目介绍

@mastra/mcp

Model Context Protocol (MCP) 客户端实现，用于 Mastra，提供与兼容 MCP 的 AI 模型和工具无缝集成的功能。

安装

npm install @mastra/mcp@latest

概述

@mastra/mcp 包提供了 Model Context Protocol (MCP) 的客户端实现，使 Mastra 能够与兼容 MCP 的 AI 模型和工具进行通信。它封装了官方的 @modelcontextprotocol/sdk 并提供了特定于 Mastra 的功能。

客户端会根据您的服务器配置自动检测传输类型：

  • 如果您提供了 command，则使用 Stdio 传输。
  • 如果您提供了 url，则首先尝试使用可流式传输的 HTTP 传输（协议版本 2025-03-26），如果初始连接失败，则回退到旧版的 SSE 传输（协议版本 2024-11-05）。

使用方法

import { MCPClient } from '@mastra/mcp';

// 创建一个使用 Stdio 服务器的客户端
const stdioClient = new MCPClient({
  servers: {
    myStdioClient: {
      command: 'your-mcp-server-command',
      args: ['--your', 'args'],
      env: { API_KEY: 'your-api-key' }, // 可选环境变量
      capabilities: {}, // 可选客户端能力
      timeout: 60000, // 可选工具调用超时时间（毫秒）
    },
  },
});

// 创建一个使用 HTTP 服务器的客户端（尝试可流式传输的 HTTP，失败后回退到 SSE）
const httpClient = new MCPClient({
  servers: {
    myHttpClient: {
      url: new URL('https://your-mcp-server.com/mcp'), // 使用可流式传输的 HTTP 基础 URL
      requestInit: {
        // 可选的 fetch 请求配置
        headers: { Authorization: 'Bearer your-token' },
      },
      // eventSourceInit 仅在使用旧版 SSE 回退时需要自定义头部
      eventSourceInit: {
        /* ... */
      },
    },
  },
});

// 或创建一个使用 SSE 服务器的客户端
const sseClient = new MCPClient({
  servers: {
    mySseClient: {
      url: new URL('https://your-mcp-server.com/sse'),
      requestInit: {
        headers: { Authorization: 'Bearer your-token' },
      },
      eventSourceInit: {
        fetch(input: Request | URL | string, init?: RequestInit) {
          const headers = new Headers(init?.headers || {});
          headers.set('Authorization', 'Bearer your-token');
          return fetch(input, {
            ...init,
            headers,
          });
        },
      },
      timeout: 60000, // 可选工具调用超时时间（毫秒）
    },
  },
});

// 连接到 MCP 服务器（使用上述客户端之一）
await httpClient.connect();

// 列出可用资源
const resources = await httpClient.resources();

// 获取可用工具
const tools = await httpClient.tools();

// 完成后断开连接
await httpClient.disconnect();

管理多个 MCP 服务器

对于需要与多个 MCP 服务器交互的应用程序，MCPClient 类提供了一种方便的方式来管理多个服务器连接及其工具。它也基于 server 配置使用自动传输检测：

import { MCPClient } from '@mastra/mcp';

const mcp = new MCPClient({
  servers: {
    // 基于 Stdio 的服务器
    stockPrice: {
      command: 'npx',
      args: ['tsx', 'stock-price.ts'],
      env: {
        API_KEY: 'your-api-key',
      },
    },
    // 基于 HTTP 的服务器（尝试可流式传输的 HTTP，失败后回退到 SSE）
    weather: {
      url: new URL('http://localhost:8080/mcp'), // 使用可流式传输的 HTTP 基础 URL
      requestInit: {
        // 可选的 fetch 请求配置
        headers: { 'X-Api-Key': 'weather-key' },
      },
    },
  },
});

// 从所有已配置的服务器获取所有工具，并按服务器命名空间分组
const tools = await mcp.listTools();

// 按服务器分组获取工具集对象
const toolsets = await mcp.listToolsets();

日志记录

MCP 客户端提供了每个服务器的日志记录功能，允许您单独监控与每个 MCP 服务器的交互：

import { MCPClient, LogMessage, LoggingLevel } from '@mastra/mcp';

// 定义一个自定义日志处理器
const weatherLogger = (logMessage: LogMessage) => {
  console.log(`[${logMessage.level}] ${logMessage.serverName}: ${logMessage.message}`);

  // 日志数据包含有价值的信息
  console.log('详情:', logMessage.details);
  console.log('时间戳:', logMessage.timestamp);
};

// 初始化 MCP 配置并附加服务器特定的日志处理器
const mcp = new MCPClient({
  servers: {
    weatherService: {
      command: 'npx',
      args: ['tsx', 'weather-mcp.ts'],
      // 将日志处理器附加到此特定服务器
      logger: weatherLogger, // 使用 'logger' 键
    },

    stockPriceService: {
      command: 'npx',
      args: ['tsx', 'stock-mcp.ts'],
      // 此服务的不同日志处理器
      logger: logMessage => {
        // 使用 'logger' 键
        // 仅为此服务记录错误和关键事件
        if (['error', 'critical', 'alert', 'emergency'].includes(logMessage.level)) {
          console.error(`股票服务 ${logMessage.level}: ${logMessage.message}`);
        }
      },
    },
  },
});

日志消息结构

每个日志消息包含以下信息：

interface LogMessage {
  level: LoggingLevel; // MCP SDK 标准日志级别
  message: string;
  timestamp: Date;
  serverName: string;
  details?: Record<string, any>;
}

LoggingLevel 类型直接从 MCP SDK 导入，确保与所有标准 MCP 日志级别兼容：'debug' | 'info' | 'notice' | 'warning' | 'error' | 'critical' | 'alert' | 'emergency'。

创建可重用的日志处理器

您可以为常见模式创建可重用的日志处理器工厂：

import fs from 'node:fs';

// 带有不同严重性级别的彩色输出的文件日志处理器工厂
const createFileLogger = (filePath: string) => {
  return (logMessage: LogMessage) => {
    // 根据级别格式化消息
    const prefix =
      logMessage.level === 'emergency' ? '!!! EMERGENCY !!! ' : logMessage.level === 'alert' ? '! ALERT ! ' : '';

    // 写入带有时间戳、级别等的文件
    fs.appendFileSync(
      filePath,
      `[${logMessage.timestamp.toISOString()}] [${logMessage.level.toUpperCase()}] ${prefix}${logMessage.message}\n`,
    );
  };
};

// 在配置中使用工厂
const mcp = new MCPClient({
  servers: {
    weatherService: {
      command: 'npx',
      args: ['tsx', 'weather-mcp.ts'],
      logger: createFileLogger('./logs/weather.log'), // 使用 'logger' 键
    },
  },
});

请参阅 examples/server-logging.ts 文件以获得各种日志策略的全面示例。

工具 vs 工具集

MCPClient 类提供了两种访问 MCP 工具的方式：

工具 (listTools())

当满足以下条件时使用：

  • 您有一个单一的 MCP 连接
  • 工具由单个用户/上下文使用（CLI 工具、自动化脚本等）
  • 工具配置（API 密钥、凭据）保持不变
  • 您希望使用一组固定的工具初始化代理
import { Agent } from '@mastra/core/agent';
import { openai } from '@ai-sdk/openai';

const agent = new Agent({
  id: 'cli-assistant',
  name: 'CLI 助手',
  instructions: '您帮助用户完成 CLI 任务',
  model: openai('gpt-4'),
  tools: await mcp.listTools(), // 工具在代理创建时固定
});

工具集 (listToolsets())

当满足以下条件时使用：

  • 您需要按请求配置工具
  • 工具需要不同的用户凭据
  • 在多用户环境中运行（Web 应用、API 等）
  • 工具配置需要动态更改
import { MCPClient } from '@mastra/mcp';
import { Agent } from '@mastra/core/agent';
import { openai } from '@ai-sdk/openai';

// 使用用户特定设置配置 MCP 服务器，然后获取工具集
const mcp = new MCPClient({
  servers: {
    stockPrice: {
      command: 'npx',
      args: ['tsx', 'weather-mcp.ts'],
      env: {
        // 这些会因用户而异
        API_KEY: 'user-1-api-key',
      },
    },
    weather: {
      url: new URL('http://localhost:8080/mcp'), // 使用可流式传输的 HTTP 基础 URL
      requestInit: {
        headers: {
          // 这些会因用户而异
          Authorization: 'Bearer user-1-token',
        },
      },
      // eventSourceInit 仅在使用旧版 SSE 回退时需要自定义头部
      eventSourceInit: {
        /* ... */
      },
    },
  },
});

// 获取当前配置的用户工具集
const toolsets = await mcp.listToolsets();

// 使用具有用户特定工具配置的代理
const response = await agent.generate('伦敦的天气如何？', {
  toolsets,
});

console.log(response.text);

MCPClient 类自动：

  • 管理与多个 MCP 服务器的连接
  • 按服务器命名空间工具以防止命名冲突
  • 处理连接生命周期和清理
  • 提供对工具的扁平和分组访问

访问 MCP 资源

MCP 服务器可以公开资源——可以在应用程序中检索和使用的数据或内容。MCPClient 类提供了跨多个服务器访问这些资源的方法：

import { MCPClient } from '@mastra/mcp';

const mcp = new MCPClient({
  servers: {
    weather: {
      url: new URL('http://localhost:8080/mcp'),
    },
    dataService: {
      command: 'npx',
      args: ['tsx', 'data-service.ts'],
    },
  },
});

// 从所有连接的 MCP 服务器获取资源
const resources = await mcp.resources.get();

// 资源按服务器名称分组
console.log(Object.keys(resources)); // ['weather', 'dataService']

// 每个服务器条目包含资源数组
if (resources.weather) {
  // 访问天气服务器的资源
  const weatherResources = resources.weather;

  // 每个资源都有 uri、name、description 和 mimeType
  weatherResources.forEach(resource => {
    console.log(`${resource.uri}: ${resource.name} (${resource.mimeType})`);
  });

  // 通过 URI 查找特定资源
  const forecast = weatherResources.find(r => r.uri === 'weather://forecast');
  if (forecast) {
    console.log(`找到预报资源: ${forecast.description}`);
  }
}

getResources() 方法优雅地处理错误——如果服务器失败或不支持资源，它将被省略而不导致整个操作失败。

提示

MCP 服务器还可以公开提示，这些提示代表结构化的消息模板或代理的对话上下文。

列出提示

const prompts = await mcp.prompts.list();
console.log(prompts.weather); // [ { name: 'current', ... }, ... ]

获取提示和消息

const { prompt, messages } = await mcp.prompts.get({ serverName: 'weather', name: 'current' });
console.log(prompt); // { name: 'current', version: 'v1', ... }
console.log(messages); // [ { role: 'assistant', content: { type: 'text', text: '...' } }, ... ]

处理提示列表变更通知

mcp.prompts.onListChanged({
  serverName: 'weather',
  handler: () => {
    // 刷新提示列表或更新 UI
  },
});

提示通知通过 SSE 或兼容传输发送。在期望接收通知之前注册处理器。

认证

使用 AuthProvider 自动刷新 OAuth 令牌

对于需要 OAuth 认证且自动刷新令牌的基于 HTTP 的 MCP 服务器，您可以使用 authProvider 选项：

const httpClient = new MCPClient({
  servers: {
    myOAuthClient: {
      url: new URL('https://your-mcp-server.com/mcp'),
      authProvider: {
        tokens: async () => {
          // 您的令牌刷新逻辑在此处
          const refreshedToken = await refreshAccessToken();
          return {
            token: refreshedToken,
            type: 'Bearer',
          };
        },
        // 根据需要添加其他 OAuth 提供者方法
        redirectUrl: 'https://your-app.com/oauth/callback',
        clientMetadata: {
          /* ... */
        },
        // ... 其他 OAuth 提供者属性
      },
    },
  },
});

authProvider 自动传递给可流式传输的 HTTP 和 SSE 传输。

SSE 认证和头部（旧版回退）

当客户端回退到使用旧版 SSE（服务器发送事件）传输并且您需要包含认证或其他自定义头部时，您需要以特定方式配置头部。标准的 requestInit 头部不足以单独使用，因为使用浏览器的 EventSource API 的 SSE 连接不直接支持自定义头部。

eventSourceInit 配置允许您自定义用于 SSE 连接的基础 fetch 请求，确保您的认证头部正确包含。

要正确包含 SSE 连接中的认证头部或其他自定义头部，当使用旧版回退时，您需要同时使用 requestInit 和 eventSourceInit：

const sseClient = new MCPClient({
  servers: {
    authenticatedSseClient: {
      url: new URL('https://your-mcp-server.com/sse'), // 注意典型的 /sse 路径用于旧版服务器
      // 单独的 requestInit 对 SSE 连接不够
      requestInit: {
        headers: { Authorization: 'Bearer your-token' },
      },
      // eventSourceInit 是必需的，以便在 SSE 连接中包含头部
      eventSourceInit: {
        fetch(input: Request | URL | string, init?: RequestInit) {
          const headers = new Headers(init?.headers || {});
          headers.set('Authorization', 'Bearer your-token');
          return fetch(input, {
            ...init,
            headers,
          });
        },
      },
    },
  },
});

此配置确保：

  1. 认证头部正确包含在 SSE 连接请求中
  2. 连接可以使用所需的凭据建立
  3. 后续消息可以通过经过身份验证的连接接收
const sseClient = new MastraMCPClient({
  name: 'authenticated-sse-client',
  server: {
    url: new URL('https://your-mcp-server.com/sse'), // 注意典型的 /sse 路径用于旧版服务器
    // 单独的 requestInit 对 SSE 连接不够
    requestInit: {
      headers: { Authorization: 'Bearer your-token' },
    },
    // eventSourceInit 是必需的，以便在 SSE 连接中包含头部
    eventSourceInit: {
      fetch(input: Request | URL | string, init?: RequestInit) {
        const headers = new Headers(init?.headers || {});
        headers.set('Authorization', 'Bearer your-token');
        return fetch(input, {
          ...init,
          headers,
        });
      },
    },
  },
});

配置 (MastraMCPServerDefinition)

server 参数用于 MastraMCPClient 和 MCPConfiguration，使用 MastraMCPServerDefinition 类型。客户端会根据提供的参数自动检测传输类型：

  • 如果提供了 command，则使用 Stdio 传输。
  • 如果提供了 url，则首先尝试使用可流式传输的 HTTP 传输，如果初始连接失败，则回退到旧版 SSE 传输。

以下是 MastraMCPServerDefinition 中可用的选项：

  • command: (可选，字符串) 对于 Stdio 服务器：要执行的命令。
  • args: (可选，字符串数组) 对于 Stdio 服务器：传递给命令的参数。
  • env: (可选，字符串键值对记录) 对于 Stdio 服务器：为命令设置的环境变量。
  • url: (可选，URL) 对于 HTTP 服务器（可流式传输的 HTTP 或 SSE）：服务器的 URL。
  • requestInit: (可选，RequestInit) 对于 HTTP 服务器：fetch API 的请求配置。用于初始可流式传输的 HTTP 连接尝试以及后续的 POST 请求。也用于初始 SSE 连接尝试。
  • eventSourceInit: (可选，EventSourceInit) 仅用于旧版 SSE 回退：自定义 SSE 连接的 fetch 配置。当使用 SSE 的自定义头部时需要。
  • authProvider: (可选，OAuthClientProvider) 对于 HTTP 服务器：用于自动令牌刷新的 OAuth 认证提供者。自动传递给可流式传输的 HTTP 和 SSE 传输。
  • logger: (可选，LogHandler) 可选的额外日志处理器。
  • timeout: (可选，数字) 服务器特定的超时时间（毫秒），覆盖全局客户端/配置超时。
  • capabilities: (可选，ClientCapabilities) 服务器特定的能力配置。
  • enableServerLogs: (可选，布尔，默认值：true) 是否启用此服务器的日志记录。

特性

  • 标准 MCP 客户端实现
  • 自动工具转换为 Mastra 格式
  • 资源发现和管理