FastMCP

FastMCP 是一个 TypeScript 框架，用于构建能够进行客户端会话管理的 MCP 服务器。

[!NOTE]

Python 实现版本请参考 FastMCP Python。

主要功能

FastMCP 提供以下功能：

• 简单的工具、资源和提示定义
• 认证功能
• 会话管理
• 图像内容支持
• 日志记录
• 错误处理
• SSE(Server-Sent Events)
• CORS（默认启用）
• 进度通知
• 类型化的服务器事件
• 提示参数的自动补全
• 采样请求
• 自动 SSE Ping
• 路由管理
• 用于测试和调试的 CLI

安装方法

npm install fastmcp

快速开始

[!NOTE]

FastMCP 有很多实际使用案例。请参考案例介绍。

import { FastMCP } from "fastmcp";
import { z } from "zod"; // 或其他验证库（支持 Standard Schema 的库）

const server = new FastMCP({
  name: "我的服务器",
  version: "1.0.0",
});

server.addTool({
  name: "add",
  description: "将两个数值相加",
  parameters: z.object({
    a: z.number(),
    b: z.number(),
  }),
  execute: async (args) => {
    return String(args.a + args.b);
  },
});

server.start({
  transportType: "stdio",
});

这样就创建了一个可以运行的 MCP 服务器！

可以在终端中进行如下测试：

git clone https://github.com/punkpeye/fastmcp.git
cd fastmcp

pnpm install
pnpm build

# 使用 CLI 测试加法服务器的例子：
npx fastmcp dev src/examples/addition.ts
# 使用 MCP Inspector 检查加法服务器的例子：
npx fastmcp inspect src/examples/addition.ts

SSE

Server-Sent Events（SSE）是一种服务器通过 HTTPS 连接向客户端发送实时更新的机制。在 MCP 中，SSE 主要用于实现远程 MCP 通信，允许访问远程机器上托管的 MCP 并通过网络中继更新。

也可以使用 SSE 支持运行服务器：

server.start({
  transportType: "sse",
  sse: {
    endpoint: "/sse",
    port: 8080,
  },
});

这将启动服务器并在 http://localhost:8080/sse 上监听 SSE 连接。

之后，可以使用 SSEClientTransport 连接到服务器：

import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";

const client = new Client(
  {
    name: "example-client",
    version: "1.0.0",
  },
  {
    capabilities: {},
  },
);

const transport = new SSEClientTransport(new URL(`http://localhost:8080/sse`));

await client.connect(transport);

基本概念

工具

MCP 的工具公开了服务器可以执行的函数，客户端或 LLM 可以调用这些函数来执行操作。

FastMCP 使用 Standard Schema 规范来定义工具参数。 这允许使用您喜欢的、实现了该规范的模式验证库，例如 Zod、ArkType、Valibot 等。

Zod 的例子：

import { z } from "zod";

server.addTool({
  name: "fetch-zod",
  description: "获取 URL 的内容（使用 Zod）",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return await fetchWebpageContent(args.url);
  },
});

ArkType 的例子：

import { type } from "arktype";

server.addTool({
  name: "fetch-arktype",
  description: "获取 URL 的内容（使用 ArkType）",
  parameters: type({
    url: "string",
  }),
  execute: async (args) => {
    return await fetchWebpageContent(args.url);
  },
});

Valibot 的例子：

Valibot 需要对等依赖项 @valibot/to-json-schema。

import * as v from "valibot";

server.addTool({
  name: "fetch-valibot",
  description: "获取 URL 的内容（使用 Valibot）",
  parameters: v.object({
    url: v.string(),
  }),
  execute: async (args) => {
    return await fetchWebpageContent(args.url);
  },
});

返回字符串

execute 可以返回字符串：

server.addTool({
  name: "download",
  description: "下载文件",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return "你好，世界！";
  },
});

这等同于：

server.addTool({
  name: "download",
  description: "下载文件",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return {
      content: [
        {
          type: "text",
          text: "你好，世界！",
        },
      ],
    };
  },
});

返回列表

如果想返回消息列表，可以返回一个带有 content 属性的对象：

server.addTool({
  name: "download",
  description: "下载文件",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return {
      content: [
        { type: "text", text: "第一条消息" },
        { type: "text", text: "第二条消息" },
      ],
    };
  },
});

图像的返还

要创建图像内容对象，请使用 imageContent：

import { imageContent } from "fastmcp";

server.addTool({
  name: "download",
  description: "下载文件",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return imageContent({
      url: "https://example.com/image.png",
    });

    // 或者...
    // return imageContent({
    //   path: "/path/to/image.png",
    // });

    // 或者...
    // return imageContent({
    //   buffer: Buffer.from("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=", "base64"),
    // });

    // 或者...
    // return {
    //   content: [
    //     await imageContent(...)
    //   ],
    // };
  },
});

imageContent 函数接受以下选项：

• url: 图像的 URL
• path: 图像文件的路径
• buffer: 作为缓冲区的图像数据

必须仅指定 url、path 或 buffer 中的一个。

上面的例子等同于：

server.addTool({
  name: "download",
  description: "下载文件",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return {
      content: [
        {
          type: "image",
          data: "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=",
          mimeType: "image/png",
        },
      ],
    };
  },
});

日志记录

工具可以使用上下文对象的 log 向客户端输出消息日志：

server.addTool({
  name: "download",
  description: "下载文件",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args, { log }) => {
    log.info("正在下载文件...", {
      url: args.url,
    });

    // ...

    log.info("文件已下载");

    return "完成";
  },
});

log 对象具有以下方法：

• debug(message: string, data?: SerializableValue)
• error(message: string, data?: SerializableValue)
• info(message: string, data?: SerializableValue)
• warn(message: string, data?: SerializableValue)

错误

应该向用户显示错误，必须作为 UserError 实例抛出：

import { UserError } from "fastmcp";

server.addTool({
  name: "download",
  description: "下载文件",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    if (args.url.startsWith("https://example.com")) {
      throw new UserError("不允许使用此 URL");
    }

    return "完成";
  },
});

进度通知

工具可以通过调用上下文对象的 reportProgress 来报告进度：

server.addTool({
  name: "download",
  description: "下载文件",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args, { reportProgress }) => {
    reportProgress({
      progress: 0,
      total: 100,
    });

    // ...

    reportProgress({
      progress: 100,
      total: 100,
    });

    return "完成";
  },
});

资源

资源表示 MCP 服务器想要提供给客户端的任何类型的数据。 这包括：

• 文件的内容
• 屏幕截图或图像
• 日志文件
• 还有很多

每个资源都由唯一的 URI 标识，并且可以包含文本或二进制数据。

server.addResource({
  uri: "file:///logs/app.log",
  name: "应用程序日志",
  mimeType: "text/plain",
  async load() {
    return {
      text: await readLogFile(),
    };
  },
});

[!NOTE]

load 可以返回多个资源。 例如，当加载目录时，可以使用它来返回目录中文件的列表。

async load() {
  return [
    {
      text: "第一个文件的内容",
    },
    {
      text: "第二个文件的内容",
    },
  ];
}

也可以在 load 中返回二进制内容：

async load() {
  return {
    blob: 'base64 编码的数据'
  };
}

资源模板

也可以定义资源模板：

server.addResourceTemplate({
  uriTemplate: "file:///logs/{name}.log",
  name: "应用程序日志",
  mimeType: "text/plain",
  arguments: [
    {
      name: "name",
      description: "日志的名称",
      required: true,
    },
  ],
  async load({ name }) {
    return {
      text: `${name} 的示例日志内容`,
    };
  },
});

资源模板参数的自动补全

为了启用资源模板参数的自动补全，请提供 complete 函数：

server.addResourceTemplate({
  uriTemplate: "file:///logs/{name}.log",
  name: "应用程序日志",
  mimeType: "text/plain",
  arguments: [
    {
      name: "name",
      description: "日志的名称",
      required: true,
      complete: async (value) => {
        if (value === "示例") {
          return {
            values: ["示例日志"],
          };
        }

        return {
          values: [],
        };
      },
    },
  ],
  async load({ name }) {
    return {
      text: `${name} 的示例日志内容`,
    };
  },
});

提示

提示允许服务器定义可重用的提示模板和工作流程，以便客户端可以轻松地将其呈现给用户或 LLM。 这提供了一种标准化和共享常见 LLM 交互的强大方法。

server.addPrompt({
  name: "git-commit",
  description: "生成 Git 提交消息",
  arguments: [
    {
      name: "changes",
      description: "Git 差异或更改的描述",
      required: true,
    },
  ],
  load: async (args) => {
    return `请为这些更改生成简洁且描述性的提交消息：\n\n${args.changes}`;
  },
});

提示参数的自动补全

提示可以提供参数的自动补全：

server.addPrompt({
  name: "countryPoem",
  description: "写一首关于国家的诗",
  load: async ({ name }) => {
    return `你好，${name}！`;
  },
  arguments: [
    {
      name: "name",
      description: "国家的名称",
      required: true,
      complete: async (value) => {
        if (value === "日") {
          return {
            values: ["日本"],
          };
        }

        return {
          values: [],
        };
      },
    },
  ],
});

使用 enum 进行提示参数的自动补全

如果在参数中提供 enum 数组，服务器将自动提供参数补全。

server.addPrompt({
  name: "countryPoem",
  description: "写一首关于国家的诗",
  load: async ({ name }) => {
    return `你好，${name}！`;
  },
  arguments: [
    {
      name: "name",
      description: "国家的名称",
      required: true,
      enum: ["日本", "法国", "意大利"],
    },
  ],
});

认证

FastMCP 允许使用自定义函数 authenticate 客户端：

import { AuthError } from "fastmcp";

const server = new FastMCP({
  name: "我的服务器",
  version: "1.0.0",
  authenticate: ({request}) => {
    const apiKey = request.headers["x-api-key"];

    if (apiKey !== '123') {
      throw new Response(null, {
        status: 401,
        statusText: "Unauthorized",
      });
    }

    // 在这里返回的内容可以通过 `context.session` 对象访问
    return {
      id: 1,
    }
  },
});

现在可以在工具中访问经过身份验证的会话数据：

server.addTool({
  name: "sayHello",
  execute: async (args, { session }) => {
    return `你好，${session.id}！`;
  },
});

会话

session 对象是 FastMCPSession 的实例，它描述了活动的客户端会话。

server.sessions;

为了实现客户端和服务器之间的一对一通信，为每个客户端连接分配一个新的服务器实例。

类型化的服务器事件

可以使用 on 方法监听从服务器发出的事件：

server.on("connect", (event) => {
  console.log("客户端连接:", event.session);
});

server.on("disconnect", (event) => {
  console.log("客户端断开连接:", event.session);
});

FastMCPSession

FastMCPSession 表示客户端会话，并提供与客户端交互的方法。

有关如何获取 FastMCPSession 实例的信息，请参阅会话示例。

requestSampling

requestSampling 创建一个采样请求并返回响应。

await session.requestSampling({
  messages: [
    {
      role: "user",
      content: {
        type: "text",
        text: "当前目录中有哪些文件？",
      },
    },
  ],
  systemPrompt: "你是一个有用的文件系统助手。",
  includeContext: "thisServer",
  maxTokens: 100,
});

clientCapabilities

clientCapabilities 属性包含客户端功能。

session.clientCapabilities;

loggingLevel

loggingLevel 属性描述了客户端设置的日志记录级别。

session.loggingLevel;

roots

roots 属性包含客户端设置的根目录。

session.roots;

server

server 属性包含与会话关联的 MCP 服务器的实例。

session.server;

类型化的会话事件

可以使用 on 方法监听从会话发出的事件：

session.on("rootsChanged", (event) => {
  console.log("根目录已更改:", event.roots);
});

session.on("error", (event) => {
  console.error("错误:", event.error);
});

服务器的运行

MCP-CLI でテスト

测试和调试服务器的最快方法是使用 fastmcp dev：

npx fastmcp dev server.js
npx fastmcp dev server.ts

这将运行一个服务器，用于使用 mcp-cli 在终端中测试和调试 MCP 服务器。

MCP Inspector で検査

另一种方法是使用官方的 MCP Inspector 在 WebUI 中检查服务器：

npx fastmcp inspect server.ts

常见问题

如何在 Claude Desktop 中使用？

按照指南 https://modelcontextprotocol.io/quickstart/user 添加以下配置：

{
  "mcpServers": {
    "my-mcp-server": {
      "command": "npx",
      "args": [
        "tsx",
        "/项目路径/src/index.ts"
      ],
      "env": {
        "环境变量名": "值"
      }
    }
  }
}

案例介绍

[!NOTE]

如果您使用 FastMCP 开发了服务器，请务必提交 PR 以作为案例进行介绍！

• apinetwork/piapi-mcp-server - 使用 Midjourney/Flux/Kling/LumaLabs/Udio/Chrip/Trellis 生成媒体
• domdomegg/computer-use-mcp - 控制计算机
• LiterallyBlah/Dradis-MCP – 在 Dradis 中管理项目和漏洞
• Meeting-Baas/meeting-mcp - 创建会议机器人、搜索会议记录、管理录制数据
• drumnation/unsplash-smart-mcp-server – 使 AI 代理能够无缝地从 Unsplash 搜索、推荐和交付专业照片
• ssmanji89/halopsa-workflows-mcp - HaloPSA 工作流程和 AI 助手的集成
• aiamblichus/mcp-chat-adapter – 提供 LLM 使用聊天完成的干净接口

鸣谢

• FastMCP 的灵感来自 Jonathan Lowin 的 Python 实现。
• 代码库的一部分采用自 LiteMCP。
• 代码库的一部分采用自 Model Context protocolでSSEをやってみる。