Fluent MCP

一个用于构建模型上下文协议（MCP）服务器和客户端的链式流畅接口，只需少量代码即可实现。此库提供了一个类似jQuery的API，用于创建具有内置CRUD操作和资源管理的MCP服务器，以及与MCP服务器通信的轻量级MCP客户端。

特性

• 链式API：使用流畅的链式接口创建和配置MCP服务器和客户端。
• 内置CRUD操作：自动为资源生成CRUD工具。
• 资源管理：轻松使用内置存储管理资源。
• 客户端支持：使用相同的熟悉API创建MCP客户端。
• 多种传输方式：支持stdio、SSE和HTTP传输。
• 传输链式运行：在同一服务器上同时运行多种传输方式。
• 灵活配置：默认简单，但需要时可自定义。
• TypeScript支持：完整的TypeScript声明以确保类型安全。
• JavaScript兼容性：在TypeScript和JavaScript环境中均可工作。

安装

npm install @jasonkneen/fluent-mcp

快速开始

# 运行演示服务器
npm start

配置选项

createMCP函数接受一个可选的options参数用于自定义：

import { createMCP } from '@jasonkneen/fluent-mcp';

// 使用默认设置的简单用法
const simpleServer = createMCP('Notes API', '1.0.0');

// 具有自定义选项的高级用法
const advancedServer = createMCP('Notes API', '1.0.0', {
  autoGenerateIds: false,     // 默认：true - 设置为false手动管理ID
  timestampEntries: false,    // 默认：true - 设置为false禁用自动时间戳
  customOption: 'value'       // 任何额外的自定义选项
});

可用选项：

• autoGenerateIds（布尔值，默认：true）：为CRUD操作自动生成随机ID。
• timestampEntries（布尔值，默认：true）：自动添加createdAt/updatedAt时间戳。
• 自定义选项：您特定用例所需的任何额外选项。

使用Fluent MCP服务器接口

简单用法

有多种方法可以使用Zod与FluentMCP进行模式验证：

选项1：单独导入Zod（传统方法）

import { createMCP, z } from '@jasonkneen/fluent-mcp';

// 创建一个新的具有流畅接口的MCP服务器
const server = createMCP('Notes API', '1.0.0')
  // 定义Notes资源及其CRUD操作
  .resource('Notes', {})
  .crud('Note', {
    title: z.string().describe('笔记的标题'),
    content: z.string().describe('笔记的内容'),
    tags: z.array(z.string()).optional().describe('笔记的可选标签')
  })

选项2：使用内置的.z属性

import { createMCP } from '@jasonkneen/fluent-mcp';

// 创建一个新的具有流畅接口的MCP服务器
const server = createMCP('Notes API', '1.0.0');

// 使用内置的.z属性进行模式验证
server
  .resource('Notes', {})
  .crud('Note', {
    title: server.z.string().describe('笔记的标题'),
    content: server.z.string().describe('笔记的内容'),
    tags: server.z.array(server.z.string()).optional().describe('笔记的可选标签')
  })

选项3：使用内置的.schema属性（替代名称）

import { createMCP } from '@jasonkneen/fluent-mcp';

// 创建一个新的具有流畅接口的MCP服务器
const server = createMCP('Notes API', '1.0.0');

// 使用内置的.schema属性进行模式验证
server
  .resource('Notes', {})
  .crud('Note', {
    title: server.schema.string().describe('笔记的标题'),
    content: server.schema.string().describe('笔记的内容'),
    tags: server.schema.array(server.schema.string()).optional().describe('笔记的可选标签')
  })

选项4：解构以获得更简洁的代码

import { createMCP } from '@jasonkneen/fluent-mcp';

// 创建一个新的具有流畅接口的MCP服务器
const server = createMCP('Notes API', '1.0.0');
const { z } = server;  // 或 const { schema } = server;

server
  .resource('Notes', {})
  .crud('Note', {
    title: z.string().describe('笔记的标题'),
    content: z.string().describe('笔记的内容'),
    tags: z.array(z.string()).optional().describe('笔记的可选标签')
  })
  
  // 添加一个自定义搜索工具
  .tool(
    'searchNotes',
    {
      query: z.string().describe('搜索查询')
    },
    async ({ query }) => {
      // 实现...
    }
  )
  
  // 启用stdio传输并启动服务器
  .stdio()
  .start();

高级用法

import { createMCP, z } from '@jasonkneen/fluent-mcp';

// 使用高级选项创建一个新的MCP服务器
const server = createMCP('Task Manager API', '1.0.0', {
  autoGenerateIds: false,  // 我们将自行管理ID
  timestampEntries: false  // 不使用自动时间戳
})
  // 使用自定义选项定义资源
  .resource('Tasks', {})
  .crud('Task', {
    id: z.string().describe('任务的唯一ID'),
    title: z.string().describe('任务的标题'),
    // ...更多字段
  }, {
    singularName: 'Task',
    pluralName: 'Tasks'  // 显式的复数形式
  })
  
  // 启动服务器
  .start();

传输选项

FluentMCP支持多种传输机制，并且可以同时运行它们。

Stdio传输

Stdio传输是默认的，用于命令行MCP服务器。

import { createMCP } from '@jasonkneen/fluent-mcp';

const server = createMCP('My Server', '1.0.0')
  .tool('hello', { name: server.z.string() }, async ({ name }) => ({
    content: [{ type: 'text', text: `Hello, ${name}!` }]
  }))
  .stdio()
  .start();

SSE传输（服务器发送事件）

SSE传输允许通过HTTP进行实时通信。这对于基于Web的MCP客户端来说是理想的。

import express from 'express';
import { createMCP } from '@jasonkneen/fluent-mcp';

const app = express();
app.use(express.json());

// SSE端点
app.get('/sse', async (req, res) => {
  const server = createMCP('SSE Server', '1.0.0')
    .tool('getData', {}, async () => ({
      content: [{ type: 'text', text: '来自SSE传输的数据' }]
    }))
    .sse('/messages', res)
    .start();
});

app.listen(3000, () => {
  console.log('SSE服务器正在运行于 http://localhost:3000/sse');
});

多种传输（链式）

您可以在同一个MCP服务器实例上同时运行多种传输方式。这允许客户端通过不同的方式进行连接，同时共享相同的工具和资源。

import express from 'express';
import { createMCP } from '@jasonkneen/fluent-mcp';

const app = express();

// 创建一个单一的服务器实例
function createServerInstance() {
  return createMCP('Multi-Transport Server', '1.0.0')
    .resource('Data', {})
    .crud('Item', {
      name: server.z.string(),
      value: server.z.string()
    })
    .tool('customTool', { query: server.z.string() }, async ({ query }) => ({
      content: [{ type: 'text', text: `处理结果：${query}` }]
    }));
}

// SSE端点
app.get('/sse', async (req, res) => {
  await createServerInstance().sse('/messages', res).start();
});

// Stdio传输（并行运行）
if (!process.stdin.isTTY) {
  createServerInstance().stdio().start();
}

app.listen(3000);

两种传输方式都可以访问相同的工具和资源！

使用Fluent MCP客户端接口

简单用法

import { createMCPClient } from '@jasonkneen/fluent-mcp';

// 创建一个具有流畅API的MCP客户端
const client = createMCPClient('Notes Client', '1.0.0')
  // 配置通过stdio连接到MCP服务器
  .stdio('node', ['path/to/server.js'])
  // 连接到服务器
  .connect();
  
// 调用服务器工具
const result = await client.callTool('searchNotes', { query: '示例' });
console.log(client.parseToolResult(result)); // 解析JSON响应

HTTP客户端

import { createMCPClient } from '@jasonkneen/fluent-mcp';

// 创建一个HTTP客户端
const client = createMCPClient('Notes HTTP Client', '1.0.0')
  // 配置HTTP连接
  .http('http://localhost:3000/mcp')
  // 连接到服务器
  .connect();

// 列出可用工具
const tools = await client.listTools();
console.log(tools);

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

SSE客户端

import { createMCPClient } from '@jasonkneen/fluent-mcp';

// 创建一个SSE客户端
const client = createMCPClient('Notes SSE Client', '1.0.0')
  // 配置SSE连接
  .sse('http://localhost:3000/sse')
  // 连接到服务器
  .connect();

// 调用工具
const result = await client.callTool('getData', {});
console.log(client.parseToolResult(result));

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

高级客户端用法

import { createMCPClient, LoggingMessageNotificationSchema } from '@jasonkneen/fluent-mcp';

// 创建一个带有通知处理器的客户端
const client = createMCPClient('Notes Client', '1.0.0')
  // 注册通知处理器
  .onNotification(LoggingMessageNotificationSchema, (notification) => {
    console.log(`服务器通知：${notification.params.level} - ${notification.params.data}`);
  })
  // 注册错误处理器
  .onError((error)  => {
    console.error('客户端错误：', error);
  })
  // 配置连接
  .stdio('node', ['path/to/server.js']);

// 连接并使用客户端
await client.connect();

// 使用辅助方法
const resources = await client.listResources();
const prompts = await client.listPrompts();
const promptTemplate = await client.getPrompt('examplePrompt', { param: 'value' });

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

运行演示

服务器演示

npm start            # 运行JavaScript演示服务器
# 或
npm run demo         # 与上面相同
npm run demo:ts      # 运行TypeScript演示服务器
npm run demo:sse     # 运行SSE服务器演示（带多种传输方式）

客户端演示

npm run demo:client       # 运行stdio客户端演示
npm run demo:http-client  # 运行HTTP客户端演示
npm run demo:sse-client   # 运行SSE客户端演示

可用方法

核心方法

• createMCP(name, version, options)：使用灵活的配置选项创建一个新的FluentMCP服务器实例。
• createMCPClient(name, version, options)：使用灵活的配置选项创建一个新的FluentMCPClient实例。
• connectMCPClient(name, version, transportConfig)：一步创建并连接客户端。
• z：重新导出的Zod库用于模式定义（无需单独安装Zod）。

服务器方法

• resource(name, initialData)：初始化资源存储。
• getResource(name)：获取资源存储。
• setResource(name, id, data)：设置资源值。
• deleteResource(name, id)：删除资源值。
• crud(resourceName, schema, options)：为资源创建CRUD操作。
• tool(name, schema, handler)：向服务器添加工具。
• stdio()：启用stdio传输。
• sse(endpoint, res)：启用SSE传输。
• start()：使用配置的传输方式启动服务器。

客户端方法

• onError(handler)：注册错误处理器。
• onNotification(schema, handler)：注册通知处理器。
• stdio(command, args, options)：配置stdio传输。
• http(url, options)：配置HTTP传输。
• sse(url, options)：配置SSE传输。
• callTool(name, args, resultSchema)：调用MCP服务器上的工具。
• listTools()：列出服务器上的可用工具。
• listResources()：列出服务器上的可用资源。
• listPrompts()：列出服务器上的可用提示。
• getPrompt(name, args)：从服务器获取提示。
• parseToolResult(result)：将工具结果解析为JSON。
• connect()：连接到MCP服务器。
• disconnect()：从MCP服务器断开连接。

测试

运行测试：

npm test

或者监视模式：

npm run test:watch

自定义

您可以扩展FluentMCP类，添加自己的方法以增加额外的功能。链式设计使得添加新功能的同时保持API整洁变得容易。