FastMCP

一个用于构建 MCP 服务器的 TypeScript 框架，支持处理客户端会话。

特性

• 简洁的 Tool、Resource、Prompt 定义方式
• OpenAPI 到 MCP 的转换
• 身份认证
• 通过上下文传递请求头
• Session ID 与 Request ID 追踪
• 会话管理
• 图片内容
• 音频内容
• 内嵌资源
• 日志
• 错误处理
• HTTP 流式传输（兼容 SSE）
• HTTPS 支持，用于安全连接
• 自定义 HTTP 路由，适用于 REST API、Webhook 和管理界面
• 边缘运行时支持，适用于 Cloudflare Workers、Deno Deploy 等
• 无状态模式，适用于无服务器部署
• CORS（默认启用）
• 进度通知
• 流式输出
• 类型化的服务器事件
• Prompt 参数自动补全
• 采样
• 信息征询
• 可配置的 Ping 行为
• 健康检查端点
• Roots 管理
• 内存传输层，用于无需绑定端口的单元测试
• CLI 工具，支持测试和调试

何时选择 FastMCP 而非官方 SDK？

FastMCP 基于官方 SDK 构建。

官方 SDK 提供了构建 MCP 的基础模块，但将大量实现细节留给了开发者：

• 初始化并配置所有服务端组件
• 连接管理
• 工具处理
• 响应处理
• 资源处理
• 添加 prompts、resources、resource templates
• 嵌入 resources、image 和 audio 内容块

FastMCP 通过提供一个强约定框架来消除上述复杂性：

• 自动处理所有样板代码
• 为常见任务提供简洁直观的 API
• 内置最佳实践和错误处理
• 让你专注于 MCP 的核心功能

何时选择 FastMCP： 当你希望快速构建 MCP 服务器，而不想处理底层实现细节时。

何时使用官方 SDK： 当你需要最大程度的控制权或有特定的架构需求时。在这种情况下，我们建议参考 FastMCP 的实现来规避常见陷阱。

安装

npm install fastmcp

快速入门

import { FastMCP } from "fastmcp";
import { z } from "zod"; // Or any validation library that supports Standard Schema

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
});

server.addTool({
  name: "add",
  description: "Add two numbers",
  parameters: z.object({
    a: z.number(),
    b: z.number(),
  }),
  execute: async (args) => {
    return String(args.a + args.b);
  },
});

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

搞定！ 现在你已经拥有一个可用的 MCP server 了。

你可以在终端中通过以下方式测试该 server：

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

pnpm install
pnpm build

# Test the addition server example using CLI:
npx fastmcp dev src/examples/addition.ts
# Test the addition server example using MCP Inspector:
npx fastmcp inspect src/examples/addition.ts

如果你正在寻找一个模板仓库来构建自己的 MCP 服务器，可以参考 fastmcp-boilerplate。

远程服务器选项

FastMCP 支持多种传输方式用于远程通信，使得部署在远程机器上的 MCP 服务器可以通过网络进行访问。

HTTP 流式传输

HTTP 流式传输 在支持该特性的环境中，相比 SSE 提供了更高效的选择，对于较大的数据负载可能具有更优的性能表现。

你可以使用以下命令以 HTTP 流式传输模式运行服务器：

server.start({
  transportType: "httpStream",
  httpStream: {
    port: 8080,
  },
});

这将启动服务器，并在 http://localhost:8080/mcp 上监听 HTTP 流式连接。

注意： 你还可以通过 httpStream.endpoint 选项自定义端点路径（默认为 /mcp）。

注意： 若要在某个 issuer 路径下提供 HTTP 流式服务及内置 OAuth 路由，请设置 httpStream.basePath（例如 /issuer1）。按照 RFC 8414，这将在 /.well-known/oauth-authorization-server/issuer1 处暴露授权服务器元数据。

注意： 这同时会在 http://localhost:8080/sse 上启动一个 SSE 服务器。

你可以使用相应的客户端传输方式连接到这些服务器。

对于 HTTP 流式连接：

import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

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

const transport = new StreamableHTTPClientTransport(
  new URL(`http://localhost:8080/mcp`),
);

await client.connect(transport);

对于 SSE 连接：

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);

HTTPS 支持

FastMCP 通过提供 SSL 证书配置选项来支持 HTTPS 安全连接：

server.start({
  transportType: "httpStream",
  httpStream: {
    port: 8443,
    sslCert: "./path/to/cert.pem",
    sslKey: "./path/to/key.pem",
    sslCa: "./path/to/ca.pem", // Optional: for client certificate authentication
  },
});

这将通过 HTTPS 在 https://localhost:8443/mcp 启动服务器。

SSL 选项：

• sslCert - SSL 证书文件路径
• sslKey - SSL 私钥文件路径
• sslCa - （可选）用于双向 TLS 认证的 CA 证书路径

测试时，可以生成自签名证书：

openssl req -x509 -newkey rsa:2048 -keyout key.pem -out cert.pem -days 365 -nodes -subj "/CN=localhost"

生产环境 请从受信任的证书颁发机构（CA，如 Let's Encrypt）获取证书。

完整的示例请参见 https-server 示例。

CORS 配置

默认情况下，FastMCP 会启用 CORS 并包含一组标准的允许请求头。你可以通过传入 cors 选项来自定义 CORS 行为：

server.start({
  transportType: "httpStream",
  httpStream: {
    port: 8080,
    cors: {
      origin: "http://localhost:3000",
      allowedHeaders: [
        "Content-Type",
        "Authorization",
        "Accept",
        "Mcp-Session-Id",
        "Mcp-Protocol-Version",
        "Last-Event-Id",
        "X-Custom-Header",
      ],
      credentials: true,
    },
  },
});

cors 选项接受以下值：

• true（默认值）- 以默认配置启用 CORS
• false - 完全禁用 CORS
• 包含以下字段的对象：
  • origin - 字符串、字符串数组，或函数 (origin: string) => boolean
  • allowedHeaders - 字符串或字符串数组
  • methods - 允许的 HTTP 方法数组
  • exposedHeaders - 需要暴露的响应头数组
  • credentials - 布尔值，是否允许携带凭证
  • maxAge - 预检请求缓存时长（秒）

CorsOptions 类型已从 fastmcp 中导出，方便直接使用。

自定义 HTTP 路由

FastMCP 支持在 MCP 端点之外添加自定义 HTTP 路由，使你能够构建包含 REST API、Webhook、管理界面等在内的完整 HTTP 服务——全部运行在同一个服务进程中。

const app = server.getApp();

// Add REST API endpoints with Hono's native API
app.get("/api/users", async (c) => {
  return c.json({ users: [] });
});

// Handle path parameters
app.get("/api/users/:id", async (c) => {
  return c.json({
    userId: c.req.param("id"),
    query: c.req.query(), // Access query parameters
  });
});

// Handle POST requests with body parsing
app.post("/api/users", async (c) => {
  const body = await c.req.json();
  return c.json({ created: body }, 201);
});

// Serve HTML content
app.get("/admin", async (c) => {
  return c.html("<html><body><h1>Admin Panel</h1></body></html>");
});

// Handle webhooks
app.post("/webhook/github", async (c) => {
  const payload = await c.req.json();
  const event = c.req.header("x-github-event");

  // Process webhook...
  return c.json({ received: true });
});

自定义路由基于 server.getApp() 返回的底层 Hono 应用，支持以下能力：

• Hono 的 HTTP 方法：get、post、put、delete、patch、options 等
• 路径参数（:param）与通配符（*）
• 查询字符串解析
• 来自 c.req 的 JSON、文本、表单及其他请求体辅助方法
• 自定义状态码与响应头
• 通过 Hono 实现中间件与路由分组

路由按注册顺序依次匹配，因此建议将具体路由定义在通配模式之前。

公开路由与受保护路由

自定义 Hono 路由默认为公开路由，除非你添加了自定义的路由中间件或身份验证检查。对于需要保护的自定义路由，建议将认证逻辑封装为可复用的辅助函数，并在 FastMCP 的 authenticate 选项和 Hono 路由处理函数中统一调用：

import type { IncomingMessage } from "node:http";
import type { Context } from "hono";
import { FastMCP } from "fastmcp";

async function authenticateRequest(request: IncomingMessage) {
  const apiKey = request.headers["x-api-key"];
  return apiKey === "123" ? { userId: "123" } : undefined;
}

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  authenticate: authenticateRequest,
});

const app = server.getApp();

async function requireAuth(c: Context) {
  const auth = await authenticateRequest(c.env.incoming);

  if (!auth) {
    return c.json({ error: "Authentication required" }, 401);
  }

  return auth;
}

// Public route - no authentication required
app.get("/.well-known/openid-configuration", async (c) => {
  return c.json({
    issuer: "https://example.com",
    authorization_endpoint: "https://example.com/auth",
    token_endpoint: "https://example.com/token",
  });
});

// Private route - requires authentication
app.get("/api/users", async (c) => {
  const auth = await requireAuth(c);
  if (auth instanceof Response) {
    return auth;
  }

  return c.json({ users: [] });
});

// Public static files
app.get("/public/*", async (c) => {
  return c.text(`File: ${c.req.path}`);
});

公开路由非常适合以下场景：

• OAuth 发现端点（.well-known/*）
• 健康检查和状态页面
• 静态资源和文档
• 来自外部服务的 Webhook 端点
• 无需用户认证的公开 API

完整的示例请参考 custom-routes 示例。

边缘运行时支持

FastMCP 支持 Cloudflare Workers 等边缘运行时，可以将 MCP 服务器部署到边缘，在全球范围内实现极低延迟。

FastMCP 与 EdgeFastMCP 的选择

注意： EdgeFastMCP 的内置认证功能计划在未来的版本中推出。FastMCP 和 EdgeFastMCP 底层都使用 Hono，因此不存在技术障碍——EdgeFastMCP 只是在 FastMCP 加入 OAuth 之前编写的。欢迎提交 PR 来添加一个接受 Web Request 而非 Node.js http.IncomingMessage 的 authenticate 选项。

在此之前，可以使用 Hono 中间件：

const app = server.getApp();
app.use("/api/*", async (c, next) => {
  if (c.req.header("authorization") !== "Bearer secret") {
    return c.json({ error: "Unauthorized" }, 401);
  }
  await next();
});

Cloudflare Workers

要将 FastMCP 部署到 Cloudflare Workers，请使用 /edge 子路径中的 EdgeFastMCP 类：

import { EdgeFastMCP } from "fastmcp/edge";
import { z } from "zod";

const server = new EdgeFastMCP({
  name: "My Edge Server",
  version: "1.0.0",
  description: "MCP server running on Cloudflare Workers",
});

// Add tools, resources, prompts as usual
server.addTool({
  name: "greet",
  description: "Greet someone",
  parameters: z.object({
    name: z.string(),
  }),
  execute: async ({ name }) => {
    return `Hello, ${name}! Served from the edge.`;
  },
});

// Export the server as the default (required for Cloudflare Workers)
export default server;

边缘运行时差异

在边缘运行时环境下：

• 默认无状态：每个请求独立处理
• 无文件系统访问：使用 fetch API 获取外部数据
• V8 隔离环境：快速冷启动，资源利用高效
• 全球部署：自动分发至边缘节点

边缘自定义路由

你可以访问底层的 Hono 应用来添加自定义 HTTP 路由：

const app = server.getApp();

// Add a landing page
app.get("/", (c) => c.html("<h1>Welcome to my MCP server</h1>"));

// Add REST API endpoints
app.get("/api/status", (c) => c.json({ status: "ok" }));

部署

配置你的 wrangler.toml：

name = "my-mcp-server"
main = "src/index.ts"
compatibility_date = "2024-01-01"

使用以下方式部署：

wrangler deploy

参考 edge-cloudflare-worker 示例 了解完整的演示。

无状态模式

FastMCP 支持无状态运行模式，适用于 HTTP 流式传输场景，每个请求独立处理，无需维护持久化会话。这非常适合无服务器环境、负载均衡部署或无需会话状态的场景。

在无状态模式下：

• 服务器不跟踪任何会话
• 每个请求会创建一个临时会话，响应完成后即被丢弃
• 降低内存占用，提升可扩展性
• 完美适配无状态部署环境

你可以通过添加 stateless: true 选项来启用无状态模式：

server.start({
  transportType: "httpStream",
  httpStream: {
    port: 8080,
    stateless: true,
  },
});

注意： 无状态模式仅在 HTTP 流式传输中可用。依赖持久化会话的功能（如会话特定状态）在无状态模式下不可用。

你也可以通过 CLI 参数或环境变量来启用无状态模式：

# Via CLI argument
npx tsx src/server.ts --transport http-stream --port 8080 --stateless true

# Via environment variable
FASTMCP_TRANSPORT=httpStream FASTMCP_STATELESS=true npx tsx src/server.ts

/ready 健康检查端点将指示服务器是否正在以无状态模式运行：

{
  "mode": "stateless",
  "ready": 1,
  "status": "ready",
  "total": 1
}

核心概念

工具

Tools 是 MCP 中允许服务器暴露可执行函数的机制，客户端可以调用这些函数，LLM 也可以借助它们来执行操作。

FastMCP 使用 Standard Schema 规范来定义工具参数。只要你所使用的 schema 验证库（如 Zod、ArkType 或 Valibot）实现了该规范，就可以直接接入。

Zod 示例：

import { z } from "zod";

server.addTool({
  name: "fetch-zod",
  description: "Fetch the content of a url (using 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: "Fetch the content of a url (using 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: "Fetch the content of a url (using Valibot)",
  parameters: v.object({
    url: v.string(),
  }),
  execute: async (args) => {
    return await fetchWebpageContent(args.url);
  },
});

纯 JSON Schema 示例：

如果你已有 JSON Schema——无论是来自 OpenAPI 文档、配置文件还是其他服务器——jsonSchemaAdapter 会将其封装，使其可以直接使用，无需借助任何 Schema 库。

它依赖 peer dependency ajv 来执行校验；如果 Schema 中使用了 format 关键字（如 email 或 uri），则还需要 ajv-formats。两者都只在首次调用工具时才导入，因此不使用此功能的服务器不会产生任何额外开销。

npm install ajv ajv-formats

import { jsonSchemaAdapter } from "fastmcp";

server.addTool({
  name: "fetch-json-schema",
  description: "Fetch the content of a url (using plain JSON Schema)",
  parameters: jsonSchemaAdapter({
    type: "object",
    properties: {
      url: { type: "string", format: "uri" },
    },
    required: ["url"],
  }),
  execute: async (args) => {
    const { url } = args as { url: string };
    return await fetchWebpageContent(url);
  },
});

对 outputSchema 同样适用。需要注意的是，FastMCP 在发布输入 schema 时会统一添加 additionalProperties: false，无论你原始 schema 中如何声明——这与 Zod 和 Valibot schema 的处理方式一致。唯一的例外是仅由 additional properties 构成的对象：字典类型（没有 properties 但声明了 additionalProperties schema，例如 z.record()）会保留其值 schema，而自由格式参数（{ type: "object" } 或 additionalProperties: true）则保持开放状态。输出 schema 则会保留你在 schema 中声明的 additional-properties 规则。

与上述 schema 库不同，纯 JSON Schema 不携带任何 TypeScript 类型信息，因此 execute 接收到的参数类型为 unknown。你需要自行进行类型断言或收窄。

无参数的工具

创建不需要参数的工具时，有两种方式：

1. 完全省略 parameters 属性：

   server.addTool({
     name: "sayHello",
     description: "Say hello",
     // No parameters property
     execute: async () => {
       return "Hello, world!";
     },
   });
   

2. 显式定义空的 parameters：

   import { z } from "zod";
   
   server.addTool({
     name: "sayHello",
     description: "Say hello",
     parameters: z.object({}), // Empty object
     execute: async () => {
       return "Hello, world!";
     },
   });
   

结构化工具输出

工具可以声明 outputSchema 并返回结构化数据。FastMCP 会将其值暴露为 MCP 的 structuredContent，同时为仅支持文本内容渲染的客户端返回 JSON 文本作为回退。

server.addTool({
  name: "get-weather",
  description: "Get weather for a city",
  parameters: z.object({
    city: z.string(),
  }),
  outputSchema: z.object({
    temperature: z.number(),
    humidity: z.number(),
  }),
  execute: async ({ city }) => {
    const weather = await getWeather(city);

    return {
      temperature: weather.temperature,
      humidity: weather.humidity,
    };
  },
});

你也可以同时返回明确的文本内容和结构化内容：

server.addTool({
  name: "get-weather",
  description: "Get weather for a city",
  parameters: z.object({
    city: z.string(),
  }),
  outputSchema: z.object({
    temperature: z.number(),
    humidity: z.number(),
  }),
  execute: async ({ city }) => {
    const weather = await getWeather(city);

    return {
      content: [
        {
          type: "text",
          text: `${city}: ${weather.temperature}F`,
        },
      ],
      structuredContent: {
        temperature: weather.temperature,
        humidity: weather.humidity,
      },
    };
  },
});

当提供了 outputSchema 时，FastMCP 会在发送工具结果之前对 structuredContent 进行校验。无效的结构化输出将以工具错误的形式返回给客户端，而非静默地违反已声明的 schema。

工具授权

你可以在工具定义中添加一个可选的 canAccess 函数，来控制已认证用户可以使用哪些工具。该函数会接收认证上下文，当用户被允许访问该工具时应返回 true。

server.addTool({
  name: "admin-tool",
  description: "An admin-only tool",
  canAccess: (auth) => auth?.role === "admin",
  execute: async () => "Welcome, admin!",
});

没有 authenticate 函数的服务器无需进行认证检查，因此每个会话都能看到所有工具。而配置了认证的情况下，某个会话如果最终仍未通过认证——例如在 stdio 模式下，authenticate 抛出异常或返回空值时服务器仍会继续运行——则该会话将无法访问任何配置了 canAccess 的工具。

返回字符串

execute 可以返回字符串：

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return "Hello, world!";
  },
});

后者等价于：

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return {
      content: [
        {
          type: "text",
          text: "Hello, world!",
        },
      ],
    };
  },
});

返回列表

如果需要返回消息列表，可以返回一个包含 content 属性的对象：

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return {
      content: [
        { type: "text", text: "First message" },
        { type: "text", text: "Second message" },
      ],
    };
  },
});

返回图片

使用 imageContent 创建图片的内容对象：

import { imageContent } from "fastmcp";

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

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

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

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

imageContent 函数接受以下选项：

• url：图像的 URL 地址。
• timeoutMs：URL 下载的超时时间（毫秒），默认为 30 秒。
• path：图像文件的路径。
• buffer：以 Buffer 形式提供的图像数据。

url、path 和 buffer 三者中必须且只能指定其中一个。

上述示例等价于：

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

可配置的 Ping 行为

FastMCP 内置了可配置的 ping 机制，用于维持连接的健康状态。可以通过服务器选项自定义 ping 的行为：

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  ping: {
    // Explicitly enable or disable pings (defaults vary by transport)
    enabled: true,
    // Configure ping interval in milliseconds (default: 5000ms)
    intervalMs: 10000,
    // Set log level for ping-related messages (default: 'debug')
    logLevel: "debug",
  },
});

默认情况下，ping 行为会根据每种传输类型进行优化：

• 在 SSE 和 HTTP 流式连接中启用（这些连接可受益于保活机制）
• 在 stdio 连接中禁用（此场景下通常无需 ping）

这种可配置的方式有助于减少日志冗杂，并针对不同使用场景优化性能。

保持长时间工具调用的连接活跃（streamKeepalive）

ping 消息通过独立的服务器到客户端流传输，因此它们永远不会保持某个工具调用的连接处于打开状态——在 stateless 模式下，这条流甚至根本不存在。如果一个工具运行数分钟却不产生任何输出，其响应连接就会保持静默，而服务器前面的空闲连接超时机制（AWS ALB 默认值为 60 秒）会在结果写入之前将其关闭。

这一点适用于两种会话模式。有状态部署虽然通过 ping 保持独立流的活跃，但承载工具调用的连接仍会进入空闲状态并最终被关闭；而无状态部署则从一开始就没有独立流。

streamKeepalive 会定期向正在执行的工具调用自身的响应流写入数据，而该连接正是否则会被关闭的那条。此功能为可选开启：

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  streamKeepalive: {
    // Opt in; disabled by default
    enabled: true,
    // Keep comfortably below the shortest idle timeout on the path
    intervalMs: 20000,
  },
});

每个保活消息都是一条 notifications/message（级别可通过 logLevel 配置，默认为 debug），标记了日志记录器 fastmcp-keepalive，并关联到当前正在处理的工具调用。保活从工具开始执行时起，到请求不再等待其结果时止——无论是成功、报错、超时还是客户端取消——从而让空闲连接保持安静。由于这些消息与请求相关联，该机制在 stateful 和 stateless 模式下均适用，且除标准的日志通知外无需客户端额外支持。

需要了解的限制：

• 使用 httpStream.enableJsonResponse 时无效，因为该模式会将响应缓冲为单条 JSON 回复而非流式传输，因此没有可写入的开放流。
• 仅覆盖工具调用。其他长时间运行的请求（如 resources/read、prompts/get）在完成前不会写入任何内容。
• 在 stdio 模式下没有代理可供保活，因此在该模式下启用只会增加通知流量。

取消长时间运行的工具调用（context.signal）

每个 execute 都会收到一个 AbortSignal，当结果不再能送达任何接收方时触发。将其传递给实际执行工作的部分，以便工作随调用的取消而停止，而非在调用结束后仍继续运行：

server.addTool({
  name: "fetch_report",
  parameters: z.object({ url: z.string() }),
  timeoutMs: 30000,
  execute: async (args, { signal }) => {
    const response = await fetch(args.url, { signal });
    return await response.text();
  },
});

它会在以下三种情况之一发生时中止：

• 客户端取消了调用 — 客户端发送了 notifications/cancelled，这是 MCP SDK 在调用方主动中止自身请求时发出的通知。
• 会话结束 — 传输层已关闭，或会话被显式关闭。此场景中不涉及任何取消通知。
• timeoutMs 超时 — signal.reason 与调用方收到的 UserError 相同，因此工具可以据此区分超时和取消。

调用正常完成后，signal 不会被中止，因此将清理逻辑挂载到它是安全的。

需要了解的限制：

• 不会替你看管进程。 FastMCP 会停止等待工具的返回，但 execute 返回的 promise 仍会持续运行直到最终完成。如果工具忽略了 signal，它依然会一直运行到结束——只是届时已无处汇报结果。
• 无法检测请求中途消失的 HTTP 客户端。 MCP SDK 仅在收到显式的 notifications/cancelled 时才会中止请求对应的 signal，而 StreamableHTTPServerTransport 不会将未完成的响应流视为会话关闭。如果调用方未终止会话（DELETE）就直接断开连接，工具将继续运行直至完成或超时。对于耗时较长的操作，请设置 timeoutMs，而不要依赖断连检测。
• load 不会收到 signal。 资源、资源模板和提示词接收到的是一个更精简的上下文，其中不包含 signal。

健康检查端点

使用 httpStream 传输层运行 FastMCP 时，你可以选择暴露一个简化的 HTTP 端点，返回纯文本响应，供负载均衡器或容器编排系统做存活检查使用。

通过服务器选项中的 health 字段启用（或自定义）该端点：

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  health: {
    // Enable / disable (default: true)
    enabled: true,
    // Body returned by the endpoint (default: 'ok')
    message: "healthy",
    // Path that should respond (default: '/health')
    path: "/healthz",
    // HTTP status code to return (default: 200)
    status: 200,
  },
});

await server.start({
  transportType: "httpStream",
  httpStream: { port: 8080 },
});

现在，向 http://localhost:8080/healthz 发送请求将会返回：

HTTP/1.1 200 OK
content-type: text/plain

healthy

当服务器通过 stdio 传输方式启动时，endpoint 参数将被忽略。

Roots 管理

FastMCP 支持 Roots——一项允许客户端提供一组类似文件系统的根位置的功能，这些位置可以被列举并动态更新。Roots 功能可以通过服务器选项进行配置或禁用：

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  roots: {
    // Set to false to explicitly disable roots support
    enabled: false,
    // By default, roots support is enabled (true)
  },
});

这将带来以下优势：

• 与可能不支持 Roots 的不同客户端具有更好的兼容性
• 连接未实现 roots 能力的客户端时，减少错误日志的干扰
• 对 MCP 服务器能力进行更明确的控制
• 当 roots 功能不可用时实现优雅降级

你可以在服务器中监听 Roots 变更：

server.on("connect", (event) => {
  const session = event.session;

  // Access the current roots
  console.log("Initial roots:", session.roots);

  // Listen for changes to the roots
  session.on("rootsChanged", (event) => {
    console.log("Roots changed:", event.roots);
  });
});

当客户端不支持 roots 或 roots 功能被显式禁用时，这些操作会优雅地处理该情况，而不会抛出错误。

返回音频

使用 audioContent 为音频创建内容对象：

import { audioContent } from "fastmcp";

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return audioContent({
      url: "https://example.com/audio.mp3",
    });

    // or...
    // return audioContent({
    //   path: "/path/to/audio.mp3",
    // });

    // or...
    // return audioContent({
    //   buffer: Buffer.from("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=", "base64"),
    // });

    // or...
    // return {
    //   content: [
    //     await audioContent(...)
    //   ],
    // };
  },
});

audioContent 函数接受以下选项：

• url：音频的 URL。
• timeoutMs：可选，URL 下载的超时时间（毫秒），默认为 30 秒。
• path：音频文件的路径。
• buffer：以 buffer 形式提供的音频数据。

url、path 和 buffer 三者中必须且只能指定其中一个。

上述示例等价于：

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return {
      content: [
        {
          type: "audio",
          data: "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=",
          mimeType: "audio/mpeg",
        },
      ],
    };
  },
});

返回组合类型

你可以通过这种方式组合各种类型，并将其返回给 AI

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    return {
      content: [
        {
          type: "text",
          text: "Hello, world!",
        },
        {
          type: "image",
          data: "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=",
          mimeType: "image/png",
        },
        {
          type: "audio",
          data: "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=",
          mimeType: "audio/mpeg",
        },
      ],
    };
  },

  // or...
  // execute: async (args) => {
  //   const imgContent = await imageContent({
  //     url: "https://example.com/image.png",
  //   });
  //   const audContent = await audioContent({
  //     url: "https://example.com/audio.mp3",
  //   });
  //   return {
  //     content: [
  //       {
  //         type: "text",
  //         text: "Hello, world!",
  //       },
  //       imgContent,
  //       audContent,
  //     ],
  //   };
  // },
});

自定义日志记录器

FastMCP 允许你提供自定义的日志记录器实现，以控制服务器如何输出日志。这在对接现有日志基础设施或自定义日志格式时非常实用。

import { FastMCP, Logger } from "fastmcp";

class CustomLogger implements Logger {
  debug(...args: unknown[]): void {
    console.log("[DEBUG]", new Date().toISOString(), ...args);
  }

  error(...args: unknown[]): void {
    console.error("[ERROR]", new Date().toISOString(), ...args);
  }

  info(...args: unknown[]): void {
    console.info("[INFO]", new Date().toISOString(), ...args);
  }

  log(...args: unknown[]): void {
    console.log("[LOG]", new Date().toISOString(), ...args);
  }

  warn(...args: unknown[]): void {
    console.warn("[WARN]", new Date().toISOString(), ...args);
  }
}

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  logger: new CustomLogger(),
});

请参阅 src/examples/custom-logger.ts，其中包含使用 Winston、Pino 以及基于文件的日志记录示例。

日志记录

工具可以通过上下文对象中的 log 对象向客户端记录日志消息：

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args, { log }) => {
    log.info("Downloading file...", {
      url,
    });

    // ...

    log.info("Downloaded file");

    return "done";
  },
});

log 对象提供以下方法：

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

客户端可以通过 logging/setLevel 请求提升其所需的最低日志级别，FastMCP 会将低于该级别的消息直接丢弃而不发送——例如，当通过 logging/setLevel 设置为 error 后，log.debug 和 log.info 产生的消息将不会被送达任何客户端。在客户端发出请求之前，所有级别的消息都会正常发送。当前生效的级别可通过 session.loggingLevel 读取。

错误

需要向用户展示的错误，应当以 UserError 实例的形式抛出：

import { UserError } from "fastmcp";

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args) => {
    if (args.url.startsWith("https://example.com")) {
      throw new UserError("This URL is not allowed");
    }

    return "done";
  },
});

进度

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

server.addTool({
  name: "download",
  description: "Download a file",
  parameters: z.object({
    url: z.string(),
  }),
  execute: async (args, { reportProgress }) => {
    await reportProgress({
      progress: 0,
      total: 100,
    });

    // ...

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

    return "done";
  },
});

reportProgress 接受一个可选的人类可读 message 参数，伴随数值字段一起传递，客户端可将其展示在进度指示器旁：

await reportProgress({
  progress: 40,
  total: 100,
  message: "Downloading chunk 4 of 10…",
});

进度通知仅在客户端通过工具调用提供 progressToken 时才会发出，否则 reportProgress 将不会执行任何操作。由于 notifications/progress 是 MCP 规范的一部分（message 字段自 2025-03-26 修订版起支持），这是在长时间运行的工具调用期间发送增量更新的通用方式——关于其与流式输出的区别，请参阅下方的 Streaming Output 部分。

Streaming Output

FastMCP 支持在工具执行过程中流式输出部分结果，从而实现响应式界面和实时反馈。以下场景尤为适用：

• 逐步生成内容的长时间运行操作
• 文本、图像或其他媒体的渐进式生成
• 用户能够即时看到部分结果的操作

要从工具中流式输出内容，请使用 streamContent 方法：

server.addTool({
  name: "generateText",
  description: "Generate text incrementally",
  parameters: z.object({
    prompt: z.string(),
  }),
  annotations: {
    streamingHint: true, // Advisory only; see below
    readOnlyHint: true,
  },
  execute: async (args, { streamContent }) => {
    // Send initial content immediately
    await streamContent({ type: "text", text: "Starting generation...\n" });

    // Simulate incremental content generation
    const words = "The quick brown fox jumps over the lazy dog.".split(" ");
    for (const word of words) {
      await streamContent({ type: "text", text: word + " " });
      await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate delay
    }

    // Always return a final result. Returning nothing sends an empty tool
    // result, so clients that ignore the streamed notifications see no output
    // at all.
    return "The quick brown fox jumps over the lazy dog.";
  },
});

streamingHint 注解属于建议性元数据。它会原样转发给 tools/list 中的客户端，但既不会启用或控制 streamContent，FastMCP 自身也从不读取它。目前没有已知客户端会基于此注解采取行动，不过 SEP-2998 提议标准化同一注解名称。

消费流式内容

客户端只有在注册了该方法的处理器（或设置了 fallbackNotificationHandler）时，才能收到这些通知：

import { z } from "zod";

const StreamContentNotificationSchema = z.object({
  method: z.literal("notifications/tool/streamContent"),
  params: z.object({
    content: z.array(z.any()),
    toolName: z.string(),
  }),
});

client.setNotificationHandler(
  StreamContentNotificationSchema,
  (notification) => {
    const { content, toolName } = notification.params;
    // Render the partial content however you like.
  },
);

请注意，通知仅携带 toolName，而不包含请求令牌或进度令牌，因此同一会话中对同一工具的并发调用无法加以区分。

流式传输支持所有内容类型（文本、图片、音频），并且可以与进度报告结合使用：

server.addTool({
  name: "processData",
  description: "Process data with streaming updates",
  parameters: z.object({
    datasetSize: z.number(),
  }),
  annotations: {
    streamingHint: true,
  },
  execute: async (args, { streamContent, reportProgress }) => {
    const total = args.datasetSize;

    for (let i = 0; i < total; i++) {
      // Standard progress notification: reaches every spec-compliant client
      await reportProgress({
        progress: i,
        total,
        message: `Processed ${i} of ${total} items`,
      });

      // Richer partial content: only reaches clients that opt in
      if (i % 10 === 0) {
        await streamContent({
          type: "text",
          text: `Processed ${i} of ${total} items...\n`,
        });
      }

      await new Promise((resolve) => setTimeout(resolve, 50));
    }

    return "Processing complete!";
  },
});

信息征询

工具可在执行过程中通过信息征询向用户请求额外信息，即调用上下文对象中的 elicit 方法。客户端必须声明对应的 elicitation 能力模式——表单请求使用 elicitation: { form: {} }（默认），URL 请求使用 elicitation: { url: {} }。

server.addTool({
  name: "delete-file",
  description: "Delete a file",
  parameters: z.object({
    path: z.string(),
  }),
  execute: async (args, { elicit }) => {
    const response = await elicit({
      message: `Are you sure you want to delete ${args.path}?`,
      requestedSchema: {
        type: "object",
        properties: {
          confirmed: {
            type: "boolean",
          },
        },
        required: ["confirmed"],
      },
    });

    if (response.action !== "accept" || !response.content?.confirmed) {
      return "Deletion cancelled.";
    }

    // ...

    return `Deleted ${args.path}`;
  },
});

响应中的 action 值为 "accept"、"decline" 或 "cancel"；当值为 accept 时，content 中承载的是用户根据 requestedSchema 提供的回答。除工具外，还可以通过 session.requestElicitation 使用征询功能。

工具注解

根据 MCP 规范（2025-03-26），工具可以包含注解，通过添加关于工具行为的元数据，提供更丰富的上下文和控制能力：

server.addTool({
  name: "fetch-content",
  description: "Fetch content from a URL",
  parameters: z.object({
    url: z.string(),
  }),
  annotations: {
    title: "Web Content Fetcher", // Human-readable title for UI display
    readOnlyHint: true, // Tool doesn't modify its environment
    openWorldHint: true, // Tool interacts with external entities
  },
  execute: async (args) => {
    return await fetchWebpageContent(args.url);
  },
});

可用的注解如下：

这些注解有助于客户端和 LLM 更好地理解如何调用工具，以及调用时的预期行为。

资源

资源（Resources）代表 MCP 服务器希望向客户端提供的任意类型的数据，可包括：

• 文件内容
• 截图和图像
• 日志文件
• 等等

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

server.addResource({
  uri: "file:///logs/app.log",
  name: "Application Logs",
  mimeType: "text/plain",
  async load() {
    return {
      text: await readLogFile(),
    };
  },
});

async load() {
  return [
    {
      text: "First file content",
    },
    {
      text: "Second file content",
    },
  ];
}

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

async load() {
  return {
    blob: 'base64-encoded-data'
  };
}

load 还会接收 auth（即 authenticate 函数的返回值，如果有的话）和 context 对象，分别作为其第二个和第三个参数。context 与 tool.execute 中可用的 client、log、session 和 sessionId 字段相对应（参见 Session ID and Request ID Tracking）；reportProgress 和 streamContent 未被包含在内，因为它们绑定于工具调用的进度令牌（progress token）：

server.addResource({
  uri: "file:///logs/app.log",
  name: "Application Logs",
  mimeType: "text/plain",
  async load(auth, context) {
    context.log.info("loading application logs", { requestedBy: auth?.userId });

    return {
      text: await readLogFile(),
    };
  },
});

订阅资源更新

客户端可以通过 MCP 的 resources/subscribe 方法订阅资源，以便在其内容发生变化时收到通知。FastMCP 会为任何暴露资源的服务器自动声明 subscribe 能力，跟踪每个客户端的订阅状态，并允许你通过 sendResourceUpdated 发送更新：

server.addResource({
  uri: "file:///logs/app.log",
  name: "Application Logs",
  mimeType: "text/plain",
  async load() {
    return { text: await readLogFile() };
  },
});

// Whenever the underlying data changes, notify subscribed clients:
await server.sendResourceUpdated("file:///logs/app.log");

sendResourceUpdated 只会通知已订阅指定 URI 的客户端，因此无论你的数据何时变更，都可以安全地调用它。FastMCP 还为工具、资源和提示词声明了 listChanged 能力，并在运行时添加或移除工具、资源、资源模板或提示词时自动发出 notifications/tools/list_changed / notifications/resources/list_changed / notifications/prompts/list_changed 通知。

一个会话无法看到它从未协商过的原语类型。MCP 在 initialize 阶段确定能力，且这些能力在整个会话生命周期内保持不变，因此如果客户端在服务器尚无任何资源时就连入，它将看不到之后添加的资源——FastMCP 会记录一条包含该能力名称的警告日志，并保持该会话不受影响。请在调用 start() 之前注册每种你计划添加的原语，或者让客户端重新连接；之后连入的会话将能看到所有内容。

资源模板

你也可以定义资源模板：

server.addResourceTemplate({
  uriTemplate: "file:///logs/{name}.log",
  name: "Application Logs",
  mimeType: "text/plain",
  arguments: [
    {
      name: "name",
      description: "Name of the log",
      required: true,
    },
  ],
  async load({ name }) {
    return {
      text: `Example log content for ${name}`,
    };
  },
});

与普通资源一样，load 也会将 auth 和 context 作为第二和第三个参数接收（参见 Resources）。

{name} 表达式仅在单个路径段内匹配，遵循 RFC 6570 规范，因此 file:///logs/a/b.log 不会匹配上述模板。对于跨越多个路径段的值，请使用 {+name}，例如 file:///{+path}。接收到的值已经过百分号解码（file:///logs/..%2Fsecret.log 会得到 name: "../secret"），因此将其用作路径前务必进行校验。

资源模板参数自动补全

为资源模板参数提供 complete 函数以启用自动补全：

server.addResourceTemplate({
  uriTemplate: "file:///logs/{name}.log",
  name: "Application Logs",
  mimeType: "text/plain",
  arguments: [
    {
      name: "name",
      description: "Name of the log",
      required: true,
      complete: async (value) => {
        if (value === "Example") {
          return {
            values: ["Example Log"],
          };
        }

        return {
          values: [],
        };
      },
    },
  ],
  async load({ name }) {
    return {
      text: `Example log content for ${name}`,
    };
  },
});

内嵌资源

FastMCP 提供了便捷的 embedded() 方法，简化了在工具响应中嵌入资源的操作。该功能有效减少了代码重复，使从工具内部引用资源更加简便。

基本用法

server.addTool({
  name: "get_user_data",
  description: "Retrieve user information",
  parameters: z.object({
    userId: z.string(),
  }),
  execute: async (args) => {
    return {
      content: [
        {
          type: "resource",
          resource: await server.embedded(`user://profile/${args.userId}`),
        },
      ],
    };
  },
});

使用 Resource Templates

embedded() 方法与 Resource Templates 完美配合：

// Define a resource template
server.addResourceTemplate({
  uriTemplate: "docs://project/{section}",
  name: "Project Documentation",
  mimeType: "text/markdown",
  arguments: [
    {
      name: "section",
      required: true,
    },
  ],
  async load(args) {
    const docs = {
      "getting-started": "# Getting Started\n\nWelcome to our project!",
      "api-reference": "# API Reference\n\nAuthentication is required.",
    };
    return {
      text: docs[args.section] || "Documentation not found",
    };
  },
});

// Use embedded resources in a tool
server.addTool({
  name: "get_documentation",
  description: "Retrieve project documentation",
  parameters: z.object({
    section: z.enum(["getting-started", "api-reference"]),
  }),
  execute: async (args) => {
    return {
      content: [
        {
          type: "resource",
          resource: await server.embedded(`docs://project/${args.section}`),
        },
      ],
    };
  },
});

使用直接定义的资源

它同样适用于直接定义的资源：

// Define a direct resource
server.addResource({
  uri: "system://status",
  name: "System Status",
  mimeType: "text/plain",
  async load() {
    return {
      text: "System operational",
    };
  },
});

// Use in a tool
server.addTool({
  name: "get_system_status",
  description: "Get current system status",
  parameters: z.object({}),
  execute: async () => {
    return {
      content: [
        {
          type: "resource",
          resource: await server.embedded("system://status"),
        },
      ],
    };
  },
});

Prompts

Prompts 允许服务器定义可复用的提示模板与工作流，便于客户端将其呈现给用户及 LLM。这为标准化和共享常见的 LLM 交互提供了一种强有力的方式。

server.addPrompt({
  name: "git-commit",
  description: "Generate a Git commit message",
  arguments: [
    {
      name: "changes",
      description: "Git diff or description of changes",
      required: true,
    },
  ],
  load: async (args) => {
    return `Generate a concise but descriptive commit message for these changes:\n\n${args.changes}`;
  },
});

与资源类似，load 同样会接收 auth 和 context 作为第二个和第三个参数（参见 Resources）：

server.addPrompt({
  name: "git-commit",
  description: "Generate a Git commit message",
  arguments: [
    {
      name: "changes",
      description: "Git diff or description of changes",
      required: true,
    },
  ],
  load: async (args, auth, context) => {
    context.log.debug("generating git commit prompt", { user: auth?.userId });

    return `Generate a concise but descriptive commit message for these changes:\n\n${args.changes}`;
  },
});

Prompt 参数自动补全

Prompts 可以为其参数提供自动补全：

server.addPrompt({
  name: "countryPoem",
  description: "Writes a poem about a country",
  load: async ({ name }) => {
    return `Hello, ${name}!`;
  },
  arguments: [
    {
      name: "name",
      description: "Name of the country",
      required: true,
      complete: async (value) => {
        if (value === "Germ") {
          return {
            values: ["Germany"],
          };
        }

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

使用 enum 实现 Prompt 参数的自动补全

如果你为某个参数提供了 enum 数组，服务器将自动为该参数提供补全建议。

server.addPrompt({
  name: "countryPoem",
  description: "Writes a poem about a country",
  load: async ({ name }) => {
    return `Hello, ${name}!`;
  },
  arguments: [
    {
      name: "name",
      description: "Name of the country",
      required: true,
      enum: ["Germany", "France", "Italy"],
    },
  ],
});

OpenAPI

fromOpenAPI()（来自 fastmcp/openapi）可将 OpenAPI 3.x 文档转换为 FastMCP 服务器，每个操作对应一个工具——在此过程中自动处理外部 $ref 引用（多文件规范）、相对 servers[0].url 的解析以及参数扁平化冲突：

import { fromOpenAPI } from "fastmcp/openapi";

const server = await fromOpenAPI({
  spec: "https://petstore3.swagger.io/api/v3/openapi.json",
  include: (operation) => operation.tags.includes("pet"),
});

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

完整的选项参考、认证配置及已知限制，请参阅 OpenAPI to MCP。

认证

FastMCP 支持基于预配置提供商的 OAuth 2.1 认证，助你以最小的配置开销即可为服务器提供安全保障。

使用预配置提供商进行 OAuth 认证

使用 auth 选项并指定提供商即可启用 OAuth 认证：

import { FastMCP, getAuthSession, GoogleProvider, requireAuth } from "fastmcp";

const server = new FastMCP({
  auth: new GoogleProvider({
    baseUrl: "https://your-server.com",
    clientId: process.env.GOOGLE_CLIENT_ID!,
    clientSecret: process.env.GOOGLE_CLIENT_SECRET!,
  }),
  name: "My Server",
  version: "1.0.0",
});

server.addTool({
  canAccess: requireAuth,
  description: "Get user profile",
  execute: async (_args, { session }) => {
    const { accessToken } = getAuthSession(session);
    const response = await fetch(
      "https://www.googleapis.com/oauth2/v2/userinfo",
      {
        headers: { Authorization: `Bearer ${accessToken}` },
      },
    );
    return JSON.stringify(await response.json());
  },
  name: "get-profile",
});

可用提供程序：

通用 OAuth 提供程序（适用于 SAP、Auth0、Okta 等）：

import { FastMCP, OAuthProvider } from "fastmcp";

const server = new FastMCP({
  auth: new OAuthProvider({
    authorizationEndpoint: process.env.OAUTH_AUTH_ENDPOINT!,
    baseUrl: "https://your-server.com",
    clientId: process.env.OAUTH_CLIENT_ID!,
    clientSecret: process.env.OAUTH_CLIENT_SECRET!,
    scopes: ["openid", "profile"],
    tokenEndpoint: process.env.OAUTH_TOKEN_ENDPOINT!,
  }),
  name: "My Server",
  version: "1.0.0",
});

工具授权

通过 canAccess 属性配合内置辅助函数来控制工具访问权限：

import {
  requireAuth,
  requireScopes,
  requireRole,
  requireAll,
  requireAny,
  getAuthSession,
} from "fastmcp";

// Require any authenticated user
server.addTool({
  canAccess: requireAuth,
  name: "user-tool",
  // ...
});

// Require specific OAuth scopes
server.addTool({
  canAccess: requireScopes("read:user", "write:data"),
  name: "scoped-tool",
  // ...
});

// Require specific role
server.addTool({
  canAccess: requireRole("admin"),
  name: "admin-tool",
  // ...
});

// Combine with AND logic
server.addTool({
  canAccess: requireAll(requireAuth, requireRole("admin")),
  name: "admin-only",
  // ...
});

// Combine with OR logic
server.addTool({
  canAccess: requireAny(requireRole("admin"), requireRole("moderator")),
  name: "staff-tool",
  // ...
});

自定义授权：

对于自定义逻辑，直接传入一个函数：

server.addTool({
  name: "custom-auth-tool",
  canAccess: (auth) =>
    auth?.role === "admin" && auth?.department === "engineering",
  execute: async () => "Access granted!",
});

提取会话数据：

在工具执行函数中，使用 getAuthSession 以类型安全的方式访问 OAuth 会话：

import { getAuthSession, GoogleSession } from "fastmcp";

server.addTool({
  canAccess: requireAuth,
  name: "get-profile",
  execute: async (_args, { session }) => {
    // Type-safe destructuring (throws if not authenticated)
    const { accessToken } = getAuthSession(session);

    // Or with provider-specific typing:
    // const { accessToken } = getAuthSession<GoogleSession>(session);

    const response = await fetch("https://api.example.com/user", {
      headers: { Authorization: `Bearer ${accessToken}` },
    });
    return JSON.stringify(await response.json());
  },
});

注意： 你也可以直接访问 session.accessToken，但必须处理 session 未定义的情况。getAuthSession 辅助方法会在会话未通过身份验证时抛出明确的错误，因此与 canAccess: requireAuth 配合使用时更加安全。

自定义身份验证

对于非 OAuth 场景（如 API 密钥、自定义令牌），请使用 authenticate 选项：

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  authenticate: (request) => {
    const apiKey = request.headers["x-api-key"];

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

    return { id: 1, role: "user" };
  },
});

server.addTool({
  name: "sayHello",
  execute: async (args, { session }) => {
    return `Hello, ${session.id}!`;
  },
});

OAuth Proxy

auth 选项使用 FastMCP 内置的 OAuth Proxy，作为 MCP 客户端与上游 OAuth 提供方之间的安全中介。该代理完整处理 OAuth 2.1 授权流程，涵盖动态客户端注册（DCR）、PKCE、授权同意管理及令牌管理，并默认启用加密与令牌交换模式。

核心特性：

• 🔐 默认安全：自动加密（AES-256-GCM）与令牌交换模式
• 🚀 零配置：自动生成密钥，自动处理 OAuth 流程
• 🔌 预配置提供方：内置支持 Google、GitHub 和 Azure
• 🎯 符合 RFC 规范：实现 DCR（RFC 7591）、PKCE 及 OAuth 2.1
• 🔑 可选 JWKS：支持 RS256/ES256 令牌验证（通过可选的 jose 依赖）

快速开始：

import { FastMCP, getAuthSession, GoogleProvider, requireAuth } from "fastmcp";

const server = new FastMCP({
  auth: new GoogleProvider({
    baseUrl: "https://your-server.com",
    clientId: process.env.GOOGLE_CLIENT_ID!,
    clientSecret: process.env.GOOGLE_CLIENT_SECRET!,
  }),
  name: "My Server",
  version: "1.0.0",
});

server.addTool({
  canAccess: requireAuth,
  name: "protected-tool",
  execute: async (_args, { session }) => {
    const { accessToken } = getAuthSession(session);
    // Use accessToken to call upstream APIs
    return "Authenticated!";
  },
});

高级配置：

如需更精细地控制 OAuth 行为，可直接使用 oauth 选项：

import { FastMCP } from "fastmcp";
import { GoogleProvider } from "fastmcp/auth";

const authProvider = new GoogleProvider({
  baseUrl: "https://your-server.com",
  clientId: process.env.GOOGLE_CLIENT_ID!,
  clientSecret: process.env.GOOGLE_CLIENT_SECRET!,
  scopes: ["openid", "profile", "email"],
});

const server = new FastMCP({
  name: "My Server",
  oauth: {
    authorizationServer: authProvider
      .getProxy()
      .getAuthorizationServerMetadata(),
    enabled: true,
    proxy: authProvider.getProxy(),
  },
  version: "1.0.0",
});

文档：

• OAuth Proxy 功能特性 - 完整的功能列表与能力说明
• OAuth Proxy 实现指南 - 部署与配置
• Python 与 TypeScript 对比 - 功能对比

OAuth 发现端点

FastMCP 还支持 OAuth 发现端点，以便与 OAuth 提供商直接集成，同时兼容 MCP Specification 2025-03-26 和 MCP Specification 2025-06-18。该功能提供符合 RFC 8414（OAuth 2.0 授权服务器元数据）和 RFC 9470（OAuth 2.0 受保护资源元数据）的标准发现端点：

import { FastMCP } from "fastmcp";
import buildGetJwks from "get-jwks";
import fastJwt, { type DecodedJwt } from "fast-jwt";

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  oauth: {
    enabled: true,
    authorizationServer: {
      issuer: "https://auth.example.com",
      authorizationEndpoint: "https://auth.example.com/oauth/authorize",
      tokenEndpoint: "https://auth.example.com/oauth/token",
      jwksUri: "https://auth.example.com/.well-known/jwks.json",
      responseTypesSupported: ["code"],
    },
    protectedResource: {
      resource: "mcp://my-server",
      authorizationServers: ["https://auth.example.com"],
    },
  },
  authenticate: async (request) => {
    const authHeader = request.headers.authorization;

    if (!authHeader?.startsWith("Bearer ")) {
      throw new Response(null, {
        status: 401,
        statusText: "Missing or invalid authorization header",
      });
    }

    const token = authHeader.slice(7); // Remove 'Bearer ' prefix

    // Validate OAuth JWT access token using OpenID Connect discovery
    try {
      // Create JWKS client for token verification
      const getJwks = buildGetJwks();

      // Create JWT verifier
      const verify = fastJwt.createVerifier({
        async key({ header }: DecodedJwt) {
          const publicKey = await getJwks.getPublicKey({
            kid: header.kid,
            alg: header.alg,
            domain: "https://auth.example.com",
          });
          return publicKey;
        },
        algorithms: ["RS256"],
      });

      // Verify the JWT token
      const payload = await verify(token);

      return {
        userId: payload.sub,
        scope: payload.scope,
        email: payload.email,
        // Include other claims as needed
      };
    } catch (error) {
      throw new Response(null, {
        status: 401,
        statusText: "Invalid OAuth token",
      });
    }
  },
});

如果您的 MCP 服务器发布在 issuer 路径下，请同时配置 HTTP 流基础路径：

server.start({
  transportType: "httpStream",
  httpStream: {
    basePath: "/issuer1",
    endpoint: "/mcp",
    port: 8080,
  },
});

通过此配置，FastMCP 会在 /.well-known/oauth-authorization-server/issuer1 路径下提供授权服务器的 issuer-path 元数据，而受保护资源的元数据仍可在 /.well-known/oauth-protected-resource/issuer1/mcp 路径下供 MCP 端点访问。

此配置会自动暴露以下 OAuth 发现端点：

• /.well-known/oauth-authorization-server - 授权服务器元数据（RFC 8414）
• /.well-known/oauth-authorization-server<basePath> - 当设置 httpStream.basePath 时的授权服务器元数据（RFC 8414 第 3 节）
• /.well-known/oauth-protected-resource - 受保护资源元数据（RFC 9728）
• /.well-known/oauth-protected-resource<endpoint> - 子路径下的受保护资源元数据（MCP 2025-11-25）

发现机制（MCP 规范 2025-11-25）：

客户端按照以下搜索顺序发现受保护资源元数据：

1. WWW-Authenticate 响应头 - 首选方式（由 mcp-proxy 自动处理）
2. 子路径 well-known - /.well-known/oauth-protected-resource<endpoint>（例如 /.well-known/oauth-protected-resource/mcp）
3. 根路径 well-known - /.well-known/oauth-protected-resource（回退方案）

子路径和根路径端点返回完全相同的元数据，确保与所有 MCP 客户端实现兼容。

对于 JWT 令牌验证，可以使用 get-jwks 和 fast-jwt 等库来处理 OAuth JWT 令牌。

通过上下文传递请求头

如果你通过 HTTP 暴露 MCP 服务器，可能希望允许客户端通过请求头提供敏感的密钥，然后将这些密钥传递给你的工具所调用的 API，使每个客户端都能提供自己的 API 密钥。这可以通过在 authenticate 部分捕获 HTTP 请求头并将其存储在会话中来实现，以便工具后续引用。

import { FastMCP } from "fastmcp";
import { IncomingHttpHeaders } from "http";

// Define the session data type
interface SessionData {
  headers: IncomingHttpHeaders;
  [key: string]: unknown; // Add index signature to satisfy Record<string, unknown>
}

// Create a server instance
const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  authenticate: async (request: any): Promise<SessionData> => {
    // Authentication logic
    return {
      headers: request.headers,
    };
  },
});

// Tool to display HTTP headers
server.addTool({
  name: "headerTool",
  description: "Reads HTTP headers from the request",
  execute: async (args: any, context: any) => {
    const session = context.session as SessionData;
    const headers = session?.headers ?? {};

    const getHeaderString = (header: string | string[] | undefined) =>
      Array.isArray(header) ? header.join(", ") : (header ?? "N/A");

    const userAgent = getHeaderString(headers["user-agent"]);
    const authorization = getHeaderString(headers["authorization"]);
    return `User-Agent: ${userAgent}\nAuthorization: ${authorization}\nAll Headers: ${JSON.stringify(headers, null, 2)}`;
  },
});

// Start the server
server.start({
  transportType: "httpStream",
  httpStream: {
    port: 8080,
  },
});

连接到该服务的客户端大致如下所示：

import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";

const transport = new StreamableHTTPClientTransport(
  new URL(`http://localhost:8080/mcp`),
  {
    requestInit: {
      headers: {
        Authorization: "Test 123",
      },
    },
  },
);

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

(async () => {
  await client.connect(transport);

  // Call a tool
  const result = await client.callTool({
    name: "headerTool",
    arguments: {
      arg1: "value",
    },
  });

  console.log("Tool result:", result);
})().catch(console.error);

客户端运行后，控制台中显示的内容大致如下：

Tool result: {
  content: [
    {
      type: 'text',
      text: 'User-Agent: node\n' +
        'Authorization: Test 123\n' +
        'All Headers: {\n' +
        '  "host": "localhost:8080",\n' +
        '  "connection": "keep-alive",\n' +
        '  "authorization": "Test 123",\n' +
        '  "content-type": "application/json",\n' +
        '  "accept": "application/json, text/event-stream",\n' +
        '  "accept-language": "*",\n' +
        '  "sec-fetch-mode": "cors",\n' +
        '  "user-agent": "node",\n' +
        '  "accept-encoding": "gzip, deflate",\n' +
        '  "content-length": "163"\n' +
        '}'
    }
  ]
}

会话 ID 与请求 ID 追踪

FastMCP 通过 context 参数自动向工具处理器暴露会话 ID 和请求 ID，从而实现按会话的状态管理与请求追踪。

会话 ID（context.sessionId）：

• 仅在基于 HTTP 的传输方式（HTTP Stream、SSE）中可用
• 从 Mcp-Session-Id 请求头中提取
• 同一客户端的多次请求中保持不变
• 适用于维护会话级状态、计数器或用户专属数据

请求 ID（context.requestId）：

• 当客户端提供时，在所有传输方式中均可用
• 每个请求唯一
• 适用于请求追踪与调试

import { FastMCP } from "fastmcp";
import { z } from "zod";

const server = new FastMCP({
  name: "Session Counter Server",
  version: "1.0.0",
});

// Per-session counter storage
const sessionCounters = new Map<string, number>();

server.addTool({
  name: "increment_counter",
  description: "Increment a per-session counter",
  parameters: z.object({}),
  execute: async (args, context) => {
    if (!context.sessionId) {
      return "Session ID not available (requires HTTP transport)";
    }

    const counter = sessionCounters.get(context.sessionId) || 0;
    const newCounter = counter + 1;
    sessionCounters.set(context.sessionId, newCounter);

    return `Counter for session ${context.sessionId}: ${newCounter}`;
  },
});

server.addTool({
  name: "show_ids",
  description: "Display session and request IDs",
  parameters: z.object({}),
  execute: async (args, context) => {
    return `Session ID: ${context.sessionId || "N/A"}
Request ID: ${context.requestId || "N/A"}`;
  },
});

server.start({
  transportType: "httpStream",
  httpStream: {
    port: 8080,
  },
});

使用场景：

• 按 Session 管理状态：维护每个客户端 session 独有的计数器、缓存或临时数据
• 用户认证与授权：跨请求跟踪已认证的用户
• Session 级资源管理：按 session 分配和管理资源
• 多租户实现：按 session 隔离数据和操作
• 请求追踪：跟踪单个请求以用于调试和监控

示例：

完整示例请参见 src/examples/session-id-counter.ts，展示了基于 session 的计数器管理。

注意事项：

• Session ID 由 MCP transport 层自动生成
• 在无状态模式下，session ID 不会在请求之间持久化
• 对于 stdio transport，由于不存在 HTTP session 的概念，sessionId 将为 undefined

提供指令

你可以使用 instructions 选项向服务器提供指令：

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  instructions:
    'Instructions describing how to use the server and its features.\n\nThis can be used by clients to improve the LLM\'s understanding of available tools, resources, etc. It can be thought of like a "hint" to the model. For example, this information MAY be added to the system prompt.',
});

服务器图标与元数据

你可以为你的服务器声明图标及相关元数据。支持图标的客户端可以在其界面中展示这些图标：

const server = new FastMCP({
  name: "My Server",
  version: "1.0.0",
  websiteUrl: "https://example.com",
  icons: [
    {
      src: "https://example.com/icon.png",
      mimeType: "image/png",
      sizes: ["48x48"],
    },
  ],
});

可选的 title 也会被透传到 MCP initialize（serverInfo）中，供偏好使用显示名称而非服务器 name 的客户端读取。

会话

session 对象是 FastMCPSession 的实例，用于描述当前活跃的客户端会话。

server.sessions;

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

类型化的服务器事件

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

server.on("connect", (event) => {
  console.log("Client connected:", event.session);
});

server.on("disconnect", (event) => {
  console.log("Client disconnected:", event.session);
});

FastMCPSession

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

请参阅 Sessions 了解如何获取 FastMCPSession 实例的示例。

requestElicitation

requestElicitation 创建一个 elicitation 请求，通过客户端向用户收集额外信息并返回响应。客户端必须声明对应的 elicitation 能力模式——表单请求（默认）使用 elicitation: { form: {} }，和/或 URL 请求使用 elicitation: { url: {} }。

await session.requestElicitation({
  message: "What is your name?",
  requestedSchema: {
    type: "object",
    properties: {
      name: { type: "string" },
    },
    required: ["name"],
  },
});

在工具内部，优先使用来自上下文对象的 elicit 方法。

requestSampling

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

await session.requestSampling({
  messages: [
    {
      role: "user",
      content: {
        type: "text",
        text: "What files are in the current directory?",
      },
    },
  ],
  systemPrompt: "You are a helpful file system assistant.",
  includeContext: "thisServer",
  maxTokens: 100,
});

选项

requestSampling 接受一个可选的第二参数，用于指定请求选项：

await session.requestSampling(
  {
    messages: [
      {
        role: "user",
        content: {
          type: "text",
          text: "What files are in the current directory?",
        },
      },
    ],
    systemPrompt: "You are a helpful file system assistant.",
    includeContext: "thisServer",
    maxTokens: 100,
  },
  {
    // Progress callback - called when progress notifications are received
    onprogress: (progress) => {
      console.log(`Progress: ${progress.progress}/${progress.total}`);
    },

    // Abort signal for cancelling the request
    signal: abortController.signal,

    // Request timeout in milliseconds (default: DEFAULT_REQUEST_TIMEOUT_MSEC)
    timeout: 30000,

    // Whether progress notifications reset the timeout (default: false)
    resetTimeoutOnProgress: true,

    // Maximum total timeout regardless of progress (no default)
    maxTotalTimeout: 60000,
  },
);

选项：

• onprogress?: (progress: Progress) => void - 来自远端的进度通知回调
• signal?: AbortSignal - 用于取消请求的中止信号
• timeout?: number - 请求超时时间（毫秒）
• resetTimeoutOnProgress?: boolean - 进度通知是否重置超时计时
• maxTotalTimeout?: number - 最大总超时时间，不受进度通知影响

clientCapabilities

clientCapabilities 属性包含客户端的能力信息。

session.clientCapabilities;

loggingLevel

loggingLevel 属性描述客户端所设置的日志级别。在客户端发送 logging/setLevel 之前，该值默认为 info；严重程度低于客户端所设级别的日志消息将被丢弃，而不会被发送。

session.loggingLevel;

roots

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

session.roots;

server

server 属性包含一个与当前会话关联的 MCP server 实例。

session.server;

类型化的会话事件

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

session.on("rootsChanged", (event) => {
  console.log("Roots changed:", event.roots);
});

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

运行你的服务器

使用内存传输进行单元测试

server.connect(transport) 将服务器连接到你自行构建的传输，而不是由 start() 自动创建。配合 SDK 提供的 InMemoryTransport，你可以在进程内驱动服务器——无需绑定端口，也无需派生子进程——这通常是测试 stdio 服务器时的首选方案：

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";

async function createTestClient(server: FastMCP) {
  const [clientTransport, serverTransport] =
    InMemoryTransport.createLinkedPair();

  const client = new Client({ name: "test-client", version: "0.0.0" });

  const [session] = await Promise.all([
    server.connect(serverTransport),
    client.connect(clientTransport),
  ]);

  return { client, session };
}

test("adds two numbers", async () => {
  const { client } = await createTestClient(server);

  expect(
    await client.callTool({ arguments: { a: 2, b: 3 }, name: "add" }),
  ).toEqual({
    content: [{ text: "5", type: "text" }],
  });

  await client.close();
});

会话基于实例上注册的工具、资源和提示来构建，构建方式与 start() 完全一致，因此你的测试能够验证真实服务器所使用的同一套连接机制——包括 canAccess 过滤和 connect/disconnect 事件。

将会话认证信息作为第二个参数传入，其值应与你的 authenticate 函数的返回值等效：

await server.connect(serverTransport, { id: 7, role: "admin" });

connect 返回 FastMCPSession 实例，因此你可以对 session.clientCapabilities、session.roots 及其他属性进行断言。传输层的生命周期由你自行管理：stop() 不会关闭传入 connect 的传输，因此在测试结束时，你需要手动关闭客户端（如果需要触发 disconnect，则同时关闭会话）。

使用 mcp-cli 进行测试

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

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

该命令会通过 mcp-cli 在终端中运行你的服务器，方便测试和调试 MCP 服务器。

若要以非交互方式调用工具（例如在脚本或自动化测试中），请传入 --tool 以及可选的 JSON 格式 --args 参数：

npx fastmcp dev server.ts --tool add --args '{"a":1,"b":2}'

这会将工具结果以 JSON 格式输出并退出，而不会打开交互式检查器。在此模式下，--watch 不起作用，因为服务器仅为单次调用而启动。

使用 MCP Inspector 进行调试

另一种方式是使用官方的 MCP Inspector，通过 Web UI 来调试你的服务器：

npx fastmcp inspect server.ts

常见问题

如何与 Claude Desktop 配合使用？

请遵循该指南 https://modelcontextprotocol.io/quickstart/user 并添加以下配置：

{
  "mcpServers": {
    "my-mcp-server": {
      "command": "npx",
      "args": ["tsx", "/PATH/TO/YOUR_PROJECT/src/index.ts"],
      "env": {
        "YOUR_ENV_VAR": "value"
      }
    }
  }
}

如何在代理后面运行 FastMCP？

请参考此 issue 中关于使用 FastMCP 配合 express 和 http-proxy-middleware 的示例。

应用案例

• 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 提供简洁的对话补全接口
• eyaltoledano/claude-task-master – 由 FastMCP 驱动的先进 AI 项目/任务管理器
• cswkim/discogs-mcp-server - 连接 Discogs API，管理你的音乐收藏
• Panzer-Jack/feuse-mcp - 前端实用 MCP 工具集 - 为 Web 开发者提供 API 集成和代码生成的自动化工具
• sunra-ai/sunra-clients - Sunra.ai 是一个面向开发者构建的生成式媒体平台，提供高性能 AI 模型推理能力。
• foxtrottwist/shortcuts-mcp - 将 Claude 连接到 macOS 快捷指令，实现系统自动化、应用集成和交互式工作流

致谢

• FastMCP 受到 Jonathan Lowin 的 Python 实现 的启发。
• 部分代码借鉴自 LiteMCP。
• 部分代码借鉴自 Model Context protocolでSSEをやってみる。

本项目使用 BrowserStack 进行测试。