MCP 模型上下文协议入门教程
面向开发者 | 全中文讲解 | 从零到一掌握 MCP 协议
目录
- MCP 简介
- MCP 架构详解
- MCP 核心概念
- 开发环境搭建
- MCP Server 开发实战
- MCP Client 开发
- 工具系统深入
- 资源系统
- 安全与权限
- 主流 MCP 集成
- MCP 生态系统
- 实战项目：文件系统+数据库 MCP Server
1. MCP 简介
1.1 什么是 MCP
MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 于 2024 年 11 月发布的一项开放协议标准。它定义了一种统一的、标准化的方式，让 AI 模型（大语言模型，LLM）能够与外部数据源、工具和系统进行交互。
用一句话概括：MCP 是 AI 应用的"USB-C 接口"。
就像 USB-C 让各种设备通过统一接口连接电脑一样，MCP 让各种外部工具和数据源通过统一协议连接到 AI 模型。开发者只需要实现一次 MCP 协议，就可以被所有支持 MCP 的 AI 应用使用。
1.2 诞生背景
在 MCP 出现之前，AI 应用开发者面临一个严峻的问题——集成碎片化。
想象一个场景：你正在开发一个 AI 助手，它需要：
- 读取 GitHub 上的 Issue 列表
- 查询公司的 PostgreSQL 数据库
- 搜索 Elasticsearch 中的日志
- 读取本地文件系统中的配置文件
- 调用内部的 REST API 服务
在没有 MCP 的情况下，你需要为每种数据源编写专用的集成代码。更糟糕的是，如果你要开发另一个 AI 应用，这些集成代码几乎无法复用。每个 AI 应用都需要重新实现与外部系统的连接逻辑。
这种碎片化带来了几个核心问题：
- 重复开发：每个 AI 应用都需要独立实现工具集成
- 维护困难：外部系统 API 变更时，所有集成代码都需要更新
- 生态割裂：开发者创建的工具无法跨平台共享
- 安全隐患：每个集成都需要单独处理认证和授权
1.3 解决什么问题
MCP 协议旨在解决以下核心问题：
（1）标准化工具接口
MCP 为 AI 模型调用外部工具定义了标准的消息格式和通信协议。无论底层是数据库、文件系统还是 Web API，对 AI 模型来说都是统一的"工具调用"。
（2）解耦 AI 应用与工具实现
AI 应用（Client）只需要了解 MCP 协议，不需要关心工具的具体实现细节。工具开发者（Server）只需要遵循 MCP 协议暴露能力，不需要关心哪个 AI 应用会调用它。
（3）促进生态系统建设
通过标准化，开发者可以创建可复用的 MCP Server，构建丰富的工具生态系统。一个 MCP Server 可以被多个 AI 应用共享，极大地降低了重复开发的成本。
（4）提升安全性
MCP 协议内置了安全机制，包括能力协商、权限控制、输入验证等，让工具集成更加安全可靠。
1.4 MCP 与 Function Calling 的区别
很多开发者会问：OpenAI 的 Function Calling 不也是让 AI 调用工具吗？MCP 和它有什么区别？
这是一个非常好的问题。让我们从几个维度来对比：
| 特性 | Function Calling | MCP |
|---|---|---|
| 通信模型 | 紧耦合，工具定义嵌入在 API 请求中 | 松耦合，独立的 Client-Server 通信 |
| 工具发现 | 需要在请求中预先定义所有工具 | 动态发现，Client 可以查询 Server 有哪些工具 |
| 资源访问 | 不支持 | 支持，Server 可以暴露资源供 Client 读取 |
| 提示模板 | 不支持 | 支持，Server 可以提供预定义的提示模板 |
| 传输协议 | 嵌入在 HTTP API 调用中 | 支持 stdio、SSE、Streamable HTTP 等多种传输方式 |
| 可复用性 | 工具定义与特定 AI API 耦合 | 工具实现与 AI 应用解耦，可跨平台复用 |
| 实时通信 | 不支持 | 支持，Server 可以主动推送通知 |
| 会话管理 | 无状态 | 支持有状态会话 |
简单来说：
- Function Calling 是"我告诉你有哪些函数，你来调用"——这是一种请求级别的集成。
- MCP 是"我建立一个连接，你可以发现和使用我的所有能力"——这是一种协议级别的集成。
打个比方：Function Calling 就像在简历里写上你会的技能，而 MCP 就像建立了一个完整的 API 网关，对方可以查询你有哪些能力、获取你的资源、使用你的服务。
1.5 MCP 的设计理念
MCP 协议的设计遵循几个关键理念：
- 标准化优先：定义统一的协议规范，而不是私有的 API 接口
- 能力协商：Client 和 Server 在连接时协商各自支持的能力
- 渐进增强：基础功能简单易用，高级功能按需启用
- 安全第一：协议设计内置安全考虑，而非事后补救
- 语言无关：协议本身不限制实现语言，官方提供 TypeScript 和 Python SDK
2. MCP 架构详解
2.1 Client-Server 架构
MCP 采用经典的 Client-Server 架构，但它的角色划分与传统 Web 开发有所不同。
在 MCP 中：
- MCP Host（宿主）：运行 AI 模型的应用程序，如 Claude Desktop、Cursor IDE、自定义 AI 应用等
- MCP Client（客户端）：Host 内部的 MCP 协议实现，负责与 Server 通信
- MCP Server（服务端）：提供工具、资源和提示的外部服务
┌──────────────────────────────────────────────────────────────┐
│ MCP Host │
│ (如 Claude Desktop) │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ AI Model │ │ MCP Client │ │ MCP Client │ │
│ │ (LLM) │ │ (实例1) │ │ (实例2) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ └────────────────┼────────────────┘ │
│ │ │
└──────────────────────────┼───────────────────────────────────┘
│
┌────────────┼────────────┐
│ │ │
┌──────┴──────┐ ┌──┴─────────┐ ┌┴────────────┐
│ MCP Server │ │ MCP Server │ │ MCP Server │
│ (GitHub) │ │ (数据库) │ │ (文件系统) │
└─────────────┘ └────────────┘ └─────────────┘
一个 Host 可以包含多个 MCP Client 实例，每个 Client 实例连接到一个独立的 MCP Server。这种设计允许一个 AI 应用同时访问多个外部系统。
2.2 传输层
MCP 协议支持三种传输方式，每种都有不同的适用场景：
2.2.1 stdio（标准输入/输出）
stdio 是最简单的传输方式，通过进程的标准输入（stdin）和标准输出（stdout）进行通信。
┌─────────┐ stdin/stdout ┌──────────┐
│ Client │ ◄──────────────────► │ Server │
│ (父进程) │ │ (子进程) │
└─────────┘ └──────────┘
适用场景：本地工具、命令行应用、Claude Desktop 集成
特点：
- 启动简单，Client 直接 fork Server 进程
- 延迟低，适合本地通信
- 无需网络配置
- Server 作为 Client 的子进程运行
2.2.2 SSE（Server-Sent Events）
SSE 传输使用 HTTP 协议，Client 向 Server 发送 POST 请求，Server 通过 SSE 流返回响应。
┌─────────┐ POST /messages ┌──────────┐
│ Client │ ────────────────► │ Server │
│ │ │ │
│ │ ◄── SSE Stream ── │ │
└─────────┘ GET /sse └──────────┘
适用场景：远程服务、Web 应用、需要跨网络通信的场景
特点：
- 基于 HTTP，天然支持跨网络
- Server 可以主动推送通知
- 需要管理 HTTP 连接
2.2.3 Streamable HTTP
Streamable HTTP 是 MCP 协议最新引入的传输方式，是对 SSE 的改进。
┌─────────┐ HTTP POST /mcp ┌──────────┐
│ Client │ ─────────────────► │ Server │
│ │ │ │
│ │ ◄─ HTTP Response ─ │ │
│ │ (JSON or SSE) │ │
└─────────┘ └──────────┘
适用场景：生产环境部署、需要无状态特性的场景
特点：
- 支持无状态请求
- 响应可以是普通 JSON 或 SSE 流
- 向后兼容 SSE 传输
- 更灵活的会话管理
2.3 消息格式：JSON-RPC 2.0
MCP 协议的所有消息都使用 JSON-RPC 2.0 格式。这是一个轻量级的远程过程调用（RPC）协议。
三种消息类型
（1）请求（Request）
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "get_weather",
"arguments": {
"city": "北京"
}
}
}
（2）响应（Response）
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "北京今天天气：晴，气温 25°C"
}
]
}
}
（3）通知（Notification）
{
"jsonrpc": "2.0",
"method": "notifications/resources/updated",
"params": {
"uri": "file:///config/app.json"
}
}
通知没有 id 字段，发送方不期望收到响应。
错误响应
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32602,
"message": "Invalid params",
"data": {
"detail": "参数 'city' 是必需的"
}
}
}
MCP 定义的错误码包括：
| 错误码 | 含义 | 说明 |
|---|---|---|
| -32700 | Parse error | JSON 解析失败 |
| -32600 | Invalid Request | 请求格式无效 |
| -32601 | Method not found | 方法不存在 |
| -32602 | Invalid params | 参数无效 |
| -32603 | Internal error | 内部错误 |
| -32002 | Resource not found | 资源未找到 |
| -32001 | Tool execution error | 工具执行错误 |
2.4 连接生命周期
一个 MCP 连接的完整生命周期如下：
Client Server
│ │
│──── initialize ────────────────────►│ 1. 能力协商
│◄─── initialize result ─────────────│
│ │
│──── initialized (通知) ────────────►│ 2. 连接确认
│ │
│◄─── notifications/initialized ─────│ 3. 可选通知
│ │
│──── tools/list ────────────────────►│ 4. 工具发现
│◄─── tools/list result ─────────────│
│ │
│──── resources/list ────────────────►│ 5. 资源发现
│◄─── resources/list result ─────────│
│ │
│──── tools/call ────────────────────►│ 6. 工具调用
│◄─── tools/call result ─────────────│
│ │
│──── resources/read ────────────────►│ 7. 资源读取
│◄─── resources/read result ─────────│
│ │
│◄─── notifications/resources/updated │ 8. 资源变更通知
│ │
│──── shutdown ──────────────────────►│ 9. 关闭连接
│◄─── shutdown result ───────────────│
3. MCP 核心概念
MCP 协议定义了五个核心概念（原语），它们构成了 MCP 的能力模型：
3.1 Resources（资源）
资源是 MCP Server 暴露给 Client 的数据。资源通过 URI（统一资源标识符） 进行标识。
// 资源示例
{
uri: "file:///home/user/config.json", // 文件资源
name: "应用配置文件",
description: "当前应用的 JSON 配置文件",
mimeType: "application/json"
}
// 数据库资源
{
uri: "postgres://localhost:5432/mydb/users", // 数据库表
name: "用户表",
description: "用户信息表",
mimeType: "application/json"
}
// API 资源
{
uri: "api://internal.company.com/v1/metrics", // API 端点
name: "系统指标",
description: "当前系统运行指标",
mimeType: "application/json"
}
资源可以是静态的（如文件内容）或动态的（如实时数据）。Client 可以订阅资源变更，当资源内容更新时，Server 会主动通知 Client。
3.2 Tools（工具）
工具是 MCP Server 暴露给 AI 模型的可执行操作。这是 MCP 最核心的功能。
// 工具定义示例
{
name: "query_database",
description: "执行 SQL 查询并返回结果",
inputSchema: {
type: "object",
properties: {
sql: {
type: "string",
description: "要执行的 SQL 查询语句"
},
database: {
type: "string",
description: "目标数据库名称",
default: "main"
}
},
required: ["sql"]
}
}
工具与资源的关键区别：
- 资源是数据，供 Client（或 AI 模型）读取和理解
- 工具是操作，供 AI 模型执行以产生副作用或获取动态结果
3.3 Prompts（提示）
提示是 MCP Server 提供的预定义提示模板。它们可以帮助 AI 模型更好地理解如何使用 Server 的能力。
// 提示模板示例
{
name: "debug_error",
description: "帮助调试错误日志",
arguments: [
{
name: "error_message",
description: "错误消息文本",
required: true
},
{
name: "context",
description: "相关上下文信息",
required: false
}
]
}
当 Client 请求这个提示时，Server 会返回一组结构化的消息：
// 提示返回结果
{
messages: [
{
role: "user",
content: {
type: "text",
text: "请分析以下错误信息并提供调试建议：\n\n错误：TypeError: Cannot read property 'id' of undefined\n上下文：用户登录流程"
}
}
]
}
3.4 Sampling（采样）
采样是一个相对特殊的概念——它允许 MCP Server 反过来请求 Client 中的 AI 模型进行推理。
Server Client
│ │
│──── sampling/createMessage ───►│ Server 请求 AI 推理
│ │──► AI Model ──►
│◄─── sampling result ──────────│ 返回 AI 推理结果
│ │
这在以下场景中非常有用：
- 智能工具：工具需要 AI 来理解非结构化输入
- 多步推理：工具执行过程中需要 AI 做决策
- 内容生成：工具需要 AI 生成文本内容
// Server 发起的采样请求
{
method: "sampling/createMessage",
params: {
messages: [
{
role: "user",
content: {
type: "text",
text: "分析这段代码的潜在问题：\n" + codeSnippet
}
}
],
maxTokens: 1000
}
}
3.5 Roots（根目录）
根目录告诉 MCP Server 它可以访问哪些文件系统路径或 URL。这为 Server 提供了访问范围的上下文。
// 根目录定义
{
roots: [
{
uri: "file:///home/user/projects",
name: "项目目录"
},
{
uri: "file:///home/user/documents",
name: "文档目录"
}
]
}
根目录的主要作用：
- 范围限制：限制 Server 可以访问的文件系统范围
- 上下文提供：帮助 Server 理解工作环境
- 安全控制：防止 Server 访问授权范围外的资源
3.6 能力协商
在连接建立时，Client 和 Server 会交换各自支持的能力：
// Client 的 initialize 请求
{
method: "initialize",
params: {
protocolVersion: "2025-03-26",
capabilities: {
roots: {
listChanged: true
},
sampling: {}
},
clientInfo: {
name: "MyAIApp",
version: "1.0.0"
}
}
}
// Server 的 initialize 响应
{
capabilities: {
tools: {
listChanged: true
},
resources: {
subscribe: true,
listChanged: true
},
prompts: {
listChanged: true
}
},
serverInfo: {
name: "FileSystemServer",
version: "1.0.0"
}
}
4. 开发环境搭建
4.1 前置要求
在开始 MCP 开发之前，确保你的开发环境满足以下要求：
- Node.js：v18 或更高版本（推荐 v20 LTS）
- Python：3.10 或更高版本（如果使用 Python SDK）
- 包管理器：npm、yarn 或 pnpm（Node.js）；pip 或 uv（Python）
- TypeScript：5.0 或更高版本（如果使用 TypeScript 开发）
- IDE：推荐 VS Code 或 Cursor
4.2 Node.js/TypeScript SDK 安装
MCP 官方提供 TypeScript SDK，这是最成熟的 SDK 之一。
初始化项目：
# 创建项目目录
mkdir my-mcp-server
cd my-mcp-server
# 初始化 Node.js 项目
npm init -y
# 安装 MCP SDK
npm install @modelcontextprotocol/sdk
# 安装 TypeScript 相关依赖
npm install -D typescript @types/node tsx
配置 TypeScript：
// tsconfig.json
{
"compilerOptions": {
"target": "ES2022",
"module": "Node16",
"moduleResolution": "Node16",
"outDir": "./build",
"rootDir": "./src",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"declaration": true,
"declarationMap": true,
"sourceMap": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "build"]
}
配置 package.json：
{
"name": "my-mcp-server",
"version": "1.0.0",
"type": "module",
"main": "build/index.js",
"bin": {
"my-mcp-server": "build/index.js"
},
"scripts": {
"build": "tsc",
"dev": "tsx src/index.ts",
"start": "node build/index.js"
},
"dependencies": {
"@modelcontextprotocol/sdk": "^1.0.0"
},
"devDependencies": {
"typescript": "^5.4.0",
"@types/node": "^20.0.0",
"tsx": "^4.0.0"
}
}
4.3 Python SDK 安装
Python SDK 同样功能完善，适合 Python 开发者。
使用 uv（推荐）：
# 安装 uv（如果尚未安装）
curl -LsSf https://astral.sh/uv/install.sh | sh
# 创建项目
uv init my-mcp-server
cd my-mcp-server
# 添加 MCP SDK
uv add mcp[cli]
使用 pip：
# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate # Linux/macOS
# .venv\Scripts\activate # Windows
# 安装 MCP SDK
pip install mcp[cli]
项目结构：
my-mcp-server/
├── src/
│ └── my_mcp_server/
│ ├── __init__.py
│ └── server.py
├── pyproject.toml
└── README.md
pyproject.toml 配置：
[project]
name = "my-mcp-server"
version = "0.1.0"
description = "My first MCP server"
requires-python = ">=3.10"
dependencies = [
"mcp[cli]>=1.0.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project.scripts]
my-mcp-server = "my_mcp_server.server:main"
4.4 MCP Inspector 调试工具
MCP Inspector 是一个官方提供的可视化调试工具，用于测试和调试 MCP Server。
启动 Inspector：
# 使用 npx 直接运行（无需安装）
npx @modelcontextprotocol/inspector
# 或者全局安装后运行
npm install -g @modelcontextprotocol/inspector
mcp-inspector
Inspector 启动后会在浏览器中打开一个 Web 界面，提供以下功能：
- 连接管理：连接到任意 MCP Server（支持 stdio、SSE、Streamable HTTP）
- 工具测试：浏览和调用 Server 暴露的工具
- 资源查看：查看和读取 Server 暴露的资源
- 提示测试：获取和测试提示模板
- 消息日志：查看 Client-Server 之间的原始 JSON-RPC 消息
- 能力查看：查看 Server 声明的能力
使用 Inspector 测试 Server：
# 测试 stdio 模式的 Server
npx @modelcontextprotocol/inspector node build/index.js
# 测试 SSE 模式的 Server
npx @modelcontextprotocol/inspector --transport sse http://localhost:3000/sse
4.5 项目初始化脚手架
TypeScript 项目脚手架：
# 使用官方模板创建项目
npx create-mcp-server my-server
# 或手动初始化（见上文配置）
Python 项目脚手架：
# 使用 MCP CLI 创建项目
uv run mcp create my-server
# 或使用 fastmcp 快速创建
pip install fastmcp
fastmcp create my-server
5. MCP Server 开发实战
5.1 TypeScript 实现第一个 MCP Server
让我们从一个最简单的 MCP Server 开始，逐步添加功能。
文件结构：
my-mcp-server/
├── src/
│ ├── index.ts # 入口文件
│ ├── tools/ # 工具实现
│ │ └── calculator.ts
│ ├── resources/ # 资源实现
│ │ └── config.ts
│ └── prompts/ # 提示模板
│ └── code-review.ts
├── package.json
└── tsconfig.json
入口文件 src/index.ts：
#!/usr/bin/env node
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { registerCalculatorTools } from "./tools/calculator.js";
import { registerConfigResources } from "./resources/config.js";
import { registerCodeReviewPrompts } from "./prompts/code-review.js";
// 创建 MCP Server 实例
const server = new McpServer({
name: "my-first-mcp-server",
version: "1.0.0",
});
// 注册工具、资源和提示
registerCalculatorTools(server);
registerConfigResources(server);
registerCodeReviewPrompts(server);
// 启动服务器
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("MCP Server 已启动，等待连接...");
}
main().catch((error) => {
console.error("启动失败:", error);
process.exit(1);
});
5.2 工具注册
文件 src/tools/calculator.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
export function registerCalculatorTools(server: McpServer) {
// 基础计算器工具
server.tool(
"calculate",
"执行基础数学运算（加减乘除、幂运算、开方等）",
{
expression: z.string().describe("数学表达式，如 '2 + 3 * 4'"),
precision: z.number().optional().describe("结果精度（小数位数），默认为 2"),
},
async ({ expression, precision = 2 }) => {
try {
// 安全的表达式求值（仅允许数字和基本运算符）
const sanitized = expression.replace(/[^0-9+\-*/().^% sqrt]/g, "");
const result = evaluateExpression(sanitized);
const formatted = Number(result.toFixed(precision));
return {
content: [
{
type: "text" as const,
text: `${expression} = ${formatted}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `计算错误: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 单位转换工具
server.tool(
"convert_unit",
"在不同单位之间进行转换",
{
value: z.number().describe("要转换的数值"),
from_unit: z.string().describe("源单位，如 'km', 'kg', 'celsius'"),
to_unit: z.string().describe("目标单位，如 'miles', 'pounds', 'fahrenheit'"),
},
async ({ value, from_unit, to_unit }) => {
const result = convertUnit(value, from_unit, to_unit);
return {
content: [
{
type: "text" as const,
text: `${value} ${from_unit} = ${result} ${to_unit}`,
},
],
};
}
);
// 时间戳转换工具
server.tool(
"timestamp_convert",
"在时间戳和人类可读日期之间转换",
{
input: z.string().describe("时间戳（秒或毫秒）或日期字符串"),
format: z.enum(["iso", "unix", "human"]).optional().describe("输出格式"),
},
async ({ input, format = "iso" }) => {
let date: Date;
if (/^\d+$/.test(input)) {
const ts = parseInt(input);
date = new Date(ts > 1e12 ? ts : ts * 1000);
} else {
date = new Date(input);
}
let result: string;
switch (format) {
case "unix":
result = Math.floor(date.getTime() / 1000).toString();
break;
case "human":
result = date.toLocaleString("zh-CN", { timeZone: "Asia/Shanghai" });
break;
default:
result = date.toISOString();
}
return {
content: [
{
type: "text" as const,
text: `转换结果: ${result}`,
},
],
};
}
);
}
// 辅助函数：安全的表达式求值
function evaluateExpression(expr: string): number {
// 替换 ^ 为 **（幂运算）
const normalized = expr.replace(/\^/g, "**");
// 使用 Function 构造器进行安全求值
// 注意：生产环境中应使用更安全的数学表达式解析器
try {
const fn = new Function(`"use strict"; return (${normalized});`);
return fn();
} catch {
throw new Error(`无效的表达式: ${expr}`);
}
}
// 辅助函数：单位转换
function convertUnit(value: number, from: string, to: string): number {
const conversions: Record<string, Record<string, number>> = {
km: { miles: 0.621371, m: 1000, cm: 100000 },
miles: { km: 1.60934, m: 1609.34 },
kg: { pounds: 2.20462, grams: 1000 },
pounds: { kg: 0.453592 },
celsius: { fahrenheit: (v: number) => v * 9 / 5 + 32 },
fahrenheit: { celsius: (v: number) => (v - 32) * 5 / 9 },
};
const fromConversions = conversions[from.toLowerCase()];
if (!fromConversions) throw new Error(`不支持的源单位: ${from}`);
const converter = fromConversions[to.toLowerCase()];
if (!converter) throw new Error(`不支持的目标单位: ${to}`);
return typeof converter === "function" ? converter(value) : value * converter;
}
5.3 资源暴露
文件 src/resources/config.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import * as fs from "fs/promises";
import * as path from "path";
export function registerConfigResources(server: McpServer) {
// 静态配置资源
server.resource(
"app-config",
"config://app/settings",
async (uri) => {
const config = {
appName: "My MCP Server",
version: "1.0.0",
features: {
calculator: true,
unitConversion: true,
timestampConversion: true,
},
limits: {
maxExpressionLength: 1000,
maxPrecision: 10,
},
};
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(config, null, 2),
mimeType: "application/json",
},
],
};
}
);
// 动态文件资源（带通配符）
server.resource(
"project-file",
"file:///{path}",
async (uri) => {
const filePath = uri.href.replace("file:///", "");
const fullPath = path.resolve("/", filePath);
try {
const content = await fs.readFile(fullPath, "utf-8");
const ext = path.extname(fullPath).toLowerCase();
const mimeTypes: Record<string, string> = {
".json": "application/json",
".ts": "text/typescript",
".js": "text/javascript",
".py": "text/python",
".md": "text/markdown",
".txt": "text/plain",
};
return {
contents: [
{
uri: uri.href,
text: content,
mimeType: mimeTypes[ext] || "text/plain",
},
],
};
} catch (error) {
throw new Error(`无法读取文件: ${fullPath}`);
}
}
);
// 系统信息资源
server.resource(
"system-info",
"system://info",
async (uri) => {
const info = {
platform: process.platform,
arch: process.arch,
nodeVersion: process.version,
uptime: process.uptime(),
memory: {
total: Math.round(process.memoryUsage().heapTotal / 1024 / 1024) + " MB",
used: Math.round(process.memoryUsage().heapUsed / 1024 / 1024) + " MB",
},
cwd: process.cwd(),
env: process.env.NODE_ENV || "development",
};
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(info, null, 2),
mimeType: "application/json",
},
],
};
}
);
}
5.4 提示模板
文件 src/prompts/code-review.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
export function registerCodeReviewPrompts(server: McpServer) {
// 代码审查提示
server.prompt(
"code-review",
"生成代码审查提示，帮助 AI 分析代码质量",
{
code: z.string().describe("要审查的代码"),
language: z.string().optional().describe("编程语言"),
focus: z.enum(["general", "security", "performance", "style"]).optional()
.describe("审查重点"),
},
async ({ code, language = "unknown", focus = "general" }) => {
const focusInstructions: Record<string, string> = {
general: "请从代码质量、可读性、可维护性、错误处理等方面进行全面审查。",
security: "请重点审查安全问题，包括输入验证、注入攻击、敏感信息泄露等。",
performance: "请重点审查性能问题，包括算法复杂度、内存使用、并发处理等。",
style: "请重点审查代码风格，包括命名规范、代码组织、注释质量等。",
};
return {
messages: [
{
role: "user",
content: {
type: "text",
text: `请审查以下 ${language} 代码：\n\n\`\`\`${language}\n${code}\n\`\`\`\n\n${focusInstructions[focus]}\n\n请提供：\n1. 发现的问题列表（按严重程度排序）\n2. 每个问题的具体位置和说明\n3. 改进建议和示例代码\n4. 总体评价`,
},
},
],
};
}
);
// 调试助手提示
server.prompt(
"debug-helper",
"帮助分析和调试错误",
{
error_message: z.string().describe("错误消息"),
stack_trace: z.string().optional().describe("堆栈跟踪信息"),
context: z.string().optional().describe("相关上下文（代码片段、操作描述等）"),
},
async ({ error_message, stack_trace, context }) => {
let promptText = `我遇到了一个错误，请帮我分析并提供解决方案：\n\n`;
promptText += `**错误消息**：\n${error_message}\n\n`;
if (stack_trace) {
promptText += `**堆栈跟踪**：\n\`\`\`\n${stack_trace}\n\`\`\`\n\n`;
}
if (context) {
promptText += `**相关上下文**：\n${context}\n\n`;
}
promptText += `请提供：\n`;
promptText += `1. 错误原因分析\n`;
promptText += `2. 可能的解决方案（按可能性排序）\n`;
promptText += `3. 预防此类错误的建议\n`;
promptText += `4. 相关的最佳实践`;
return {
messages: [
{
role: "user",
content: {
type: "text",
text: promptText,
},
},
],
};
}
);
}
5.5 运行与测试
编译和运行：
# 编译 TypeScript
npm run build
# 直接运行
npm start
# 或使用 tsx 进行开发模式运行
npm run dev
使用 MCP Inspector 测试：
npx @modelcontextprotocol/inspector node build/index.js
在 Claude Desktop 中配置：
编辑 claude_desktop_config.json（macOS 位于 ~/Library/Application Support/Claude/）：
{
"mcpServers": {
"my-first-server": {
"command": "node",
"args": ["/path/to/my-mcp-server/build/index.js"]
}
}
}
6. MCP Client 开发
6.1 创建 MCP Client
MCP Client 负责连接到 MCP Server 并调用其提供的能力。
TypeScript Client 实现：
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
async function main() {
// 创建传输层（连接到子进程）
const transport = new StdioClientTransport({
command: "node",
args: ["./build/index.js"],
});
// 创建 Client 实例
const client = new Client({
name: "my-mcp-client",
version: "1.0.0",
});
// 连接到 Server
await client.connect(transport);
console.log("已连接到 MCP Server");
// 列出可用工具
const tools = await client.listTools();
console.log("可用工具:", tools.tools.map((t) => t.name));
// 列出可用资源
const resources = await client.listResources();
console.log("可用资源:", resources.resources.map((r) => r.name));
// 调用工具
const result = await client.callTool({
name: "calculate",
arguments: {
expression: "2 + 3 * 4",
},
});
console.log("计算结果:", result.content);
// 读取资源
const resourceContent = await client.readResource({
uri: "config://app/settings",
});
console.log("配置内容:", resourceContent.contents[0].text);
// 获取提示模板
const prompt = await client.getPrompt({
name: "code-review",
arguments: {
code: 'console.log("Hello World")',
language: "javascript",
},
});
console.log("提示模板:", prompt.messages);
// 断开连接
await client.close();
}
main().catch(console.error);
6.2 会话管理
在实际应用中，你需要管理多个 MCP 连接会话。
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
interface McpSession {
id: string;
client: Client;
transport: StdioClientTransport;
serverName: string;
connectedAt: Date;
}
class McpSessionManager {
private sessions: Map<string, McpSession> = new Map();
// 创建新会话
async createSession(
sessionId: string,
command: string,
args: string[]
): Promise<McpSession> {
const transport = new StdioClientTransport({ command, args });
const client = new Client({
name: "session-manager",
version: "1.0.0",
});
await client.connect(transport);
const session: McpSession = {
id: sessionId,
client,
transport,
serverName: command,
connectedAt: new Date(),
};
this.sessions.set(sessionId, session);
return session;
}
// 获取会话
getSession(sessionId: string): McpSession | undefined {
return this.sessions.get(sessionId);
}
// 关闭会话
async closeSession(sessionId: string): Promise<void> {
const session = this.sessions.get(sessionId);
if (session) {
await session.client.close();
this.sessions.delete(sessionId);
}
}
// 关闭所有会话
async closeAll(): Promise<void> {
for (const [id] of this.sessions) {
await this.closeSession(id);
}
}
// 列出所有活跃会话
listSessions(): McpSession[] {
return Array.from(this.sessions.values());
}
}
6.3 工具调用流程
完整的工具调用流程包括工具发现、参数验证、调用执行和结果处理。
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { z } from "zod";
interface ToolCallOptions {
name: string;
arguments: Record<string, unknown>;
timeout?: number;
retries?: number;
}
class McpToolCaller {
constructor(private client: Client) {}
// 发现可用工具
async discoverTools() {
const { tools } = await this.client.listTools();
return tools.map((tool) => ({
name: tool.name,
description: tool.description,
schema: tool.inputSchema,
}));
}
// 调用工具（带重试）
async callTool(options: ToolCallOptions) {
const { name, arguments: args, timeout = 30000, retries = 3 } = options;
let lastError: Error | null = null;
for (let attempt = 1; attempt <= retries; attempt++) {
try {
const result = await Promise.race([
this.client.callTool({ name, arguments: args }),
new Promise<never>((_, reject) =>
setTimeout(() => reject(new Error("工具调用超时")), timeout)
),
]);
return result;
} catch (error) {
lastError = error instanceof Error ? error : new Error(String(error));
console.warn(`工具调用失败 (尝试 ${attempt}/${retries}):`, lastError.message);
if (attempt < retries) {
await new Promise((resolve) => setTimeout(resolve, 1000 * attempt));
}
}
}
throw new Error(`工具调用最终失败: ${lastError?.message}`);
}
// 批量调用工具
async callToolsBatch(calls: ToolCallOptions[]) {
const results = await Promise.allSettled(
calls.map((call) => this.callTool(call))
);
return results.map((result, index) => ({
tool: calls[index].name,
success: result.status === "fulfilled",
result: result.status === "fulfilled" ? result.value : null,
error: result.status === "rejected" ? result.reason.message : null,
}));
}
}
6.4 错误处理
健壮的错误处理是 MCP Client 的关键部分。
import {
McpError,
ErrorCode,
} from "@modelcontextprotocol/sdk/types.js";
class McpErrorHandler {
// 处理 MCP 错误
static handle(error: unknown): string {
if (error instanceof McpError) {
switch (error.code) {
case ErrorCode.ConnectionClosed:
return "连接已关闭，请重新连接";
case ErrorCode.RequestTimeout:
return "请求超时，Server 可能繁忙";
case ErrorCode.InvalidRequest:
return `无效请求: ${error.message}`;
case ErrorCode.MethodNotFound:
return `方法不存在: ${error.message}`;
case ErrorCode.InvalidParams:
return `参数无效: ${error.message}`;
case ErrorCode.InternalError:
return `Server 内部错误: ${error.message}`;
default:
return `MCP 错误 (${error.code}): ${error.message}`;
}
}
if (error instanceof Error) {
if (error.message.includes("ECONNREFUSED")) {
return "无法连接到 MCP Server，请检查 Server 是否运行";
}
if (error.message.includes("timeout")) {
return "操作超时，请稍后重试";
}
return `错误: ${error.message}`;
}
return `未知错误: ${String(error)}`;
}
// 带错误处理的工具调用
static async safeCall<T>(
operation: () => Promise<T>,
fallback?: T
): Promise<{ success: boolean; data?: T; error?: string }> {
try {
const data = await operation();
return { success: true, data };
} catch (error) {
const message = this.handle(error);
console.error("MCP 操作失败:", message);
return { success: false, error: message, data: fallback };
}
}
}
// 使用示例
async function example(client: Client) {
const result = await McpErrorHandler.safeCall(
() => client.callTool({
name: "calculate",
arguments: { expression: "1/0" },
})
);
if (!result.success) {
console.error("调用失败:", result.error);
} else {
console.log("结果:", result.data);
}
}
7. 工具系统深入
7.1 工具定义 Schema
MCP 使用 JSON Schema 来定义工具的输入参数。TypeScript SDK 集成了 Zod 库来简化 Schema 定义。
基本类型定义：
import { z } from "zod";
// 字符串类型
const stringSchema = z.string()
.min(1, "不能为空")
.max(100, "最长 100 字符")
.describe("用户名");
// 数字类型
const numberSchema = z.number()
.min(0, "不能为负数")
.max(100, "最大值为 100")
.describe("年龄");
// 枚举类型
const enumSchema = z.enum(["low", "medium", "high"])
.describe("优先级");
// 数组类型
const arraySchema = z.array(z.string())
.min(1, "至少需要一个标签")
.describe("标签列表");
// 对象类型
const objectSchema = z.object({
name: z.string().describe("名称"),
age: z.number().describe("年龄"),
tags: z.array(z.string()).optional().describe("标签"),
});
复杂工具定义示例：
server.tool(
"create_task",
"创建一个新任务",
{
title: z.string().min(1).max(200).describe("任务标题"),
description: z.string().optional().describe("任务详细描述"),
priority: z.enum(["low", "medium", "high", "urgent"]).describe("优先级"),
assignee: z.string().optional().describe("负责人"),
due_date: z.string().regex(/^\d{4}-\d{2}-\d{2}$/).optional()
.describe("截止日期，格式：YYYY-MM-DD"),
tags: z.array(z.string()).optional().describe("标签列表"),
metadata: z.record(z.string()).optional()
.describe("自定义元数据键值对"),
},
async ({ title, description, priority, assignee, due_date, tags, metadata }) => {
const task = {
id: generateId(),
title,
description: description || "",
priority,
assignee: assignee || "unassigned",
dueDate: due_date || null,
tags: tags || [],
metadata: metadata || {},
status: "todo",
createdAt: new Date().toISOString(),
};
// 保存任务...
await saveTask(task);
return {
content: [
{
type: "text" as const,
text: `任务创建成功！\nID: ${task.id}\n标题: ${task.title}\n优先级: ${task.priority}\n负责人: ${task.assignee}`,
},
],
};
}
);
7.2 输入验证
除了使用 Zod Schema 进行类型验证外，你还可以添加自定义业务验证逻辑。
server.tool(
"execute_query",
"执行数据库查询",
{
sql: z.string().describe("SQL 查询语句"),
database: z.string().default("main").describe("数据库名称"),
timeout: z.number().min(1000).max(60000).default(10000)
.describe("查询超时时间（毫秒）"),
},
async ({ sql, database, timeout }) => {
// 安全验证：禁止危险操作
const dangerousPatterns = [
/\b(DROP|DELETE|TRUNCATE|ALTER)\b/i,
/\bINSERT\b.*\bINTO\b/i,
/\bUPDATE\b.*\bSET\b/i,
];
for (const pattern of dangerousPatterns) {
if (pattern.test(sql)) {
return {
content: [
{
type: "text" as const,
text: `安全拒绝: 不允许执行修改数据的 SQL 语句。仅支持 SELECT 查询。`,
},
],
isError: true,
};
}
}
// 验证数据库名称
const allowedDatabases = ["main", "analytics", "logs"];
if (!allowedDatabases.includes(database)) {
return {
content: [
{
type: "text" as const,
text: `错误: 不允许访问数据库 '${database}'。允许的数据库: ${allowedDatabases.join(", ")}`,
},
],
isError: true,
};
}
// 执行查询...
try {
const results = await executeQuery(sql, database, { timeout });
return {
content: [
{
type: "text" as const,
text: `查询返回 ${results.length} 条记录:\n${JSON.stringify(results, null, 2)}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `查询执行失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
7.3 异步工具
对于需要较长时间执行的工具，应该支持异步执行和进度报告。
import { CallToolResult } from "@modelcontextprotocol/sdk/types.js";
server.tool(
"process_large_file",
"处理大型文件（支持进度报告）",
{
file_path: z.string().describe("文件路径"),
operation: z.enum(["analyze", "transform", "compress"]).describe("处理类型"),
callback: z.function()
.args(z.object({ progress: z.number(), message: z.string() }))
.returns(z.void())
.optional()
.describe("进度回调函数"),
},
async ({ file_path, operation, callback }): Promise<CallToolResult> => {
const steps = ["读取文件", "分析内容", "执行操作", "保存结果"];
const totalSteps = steps.length;
for (let i = 0; i < totalSteps; i++) {
// 报告进度
const progress = Math.round(((i + 1) / totalSteps) * 100);
callback?.({ progress, message: `步骤 ${i + 1}/${totalSteps}: ${steps[i]}` });
// 模拟异步处理
await new Promise((resolve) => setTimeout(resolve, 1000));
}
return {
content: [
{
type: "text" as const,
text: `文件处理完成！\n文件: ${file_path}\n操作: ${operation}\n结果: 处理成功`,
},
],
};
}
);
7.4 工具组合编排
在实际应用中，一个复杂的操作可能需要调用多个工具组合完成。
class ToolOrchestrator {
constructor(private client: Client) {}
// 顺序执行多个工具
async sequential(
steps: Array<{ tool: string; args: Record<string, unknown> }>
) {
const results: unknown[] = [];
let context: Record<string, unknown> = {};
for (const step of steps) {
// 支持参数中的变量替换
const resolvedArgs = this.resolveVariables(step.args, context);
const result = await this.client.callTool({
name: step.tool,
arguments: resolvedArgs,
});
results.push(result);
// 将结果添加到上下文，供后续步骤使用
context[`${step.tool}_result`] = result;
}
return results;
}
// 并行执行多个工具
async parallel(
calls: Array<{ tool: string; args: Record<string, unknown> }>
) {
return Promise.allSettled(
calls.map((call) =>
this.client.callTool({ name: call.tool, arguments: call.args })
)
);
}
// 条件执行
async conditional(
condition: () => Promise<boolean>,
trueBranch: { tool: string; args: Record<string, unknown> },
falseBranch?: { tool: string; args: Record<string, unknown> }
) {
const shouldExecute = await condition();
if (shouldExecute) {
return this.client.callTool({
name: trueBranch.tool,
arguments: trueBranch.args,
});
} else if (falseBranch) {
return this.client.callTool({
name: falseBranch.tool,
arguments: falseBranch.args,
});
}
return null;
}
// 变量替换
private resolveVariables(
args: Record<string, unknown>,
context: Record<string, unknown>
): Record<string, unknown> {
const resolved: Record<string, unknown> = {};
for (const [key, value] of Object.entries(args)) {
if (typeof value === "string" && value.startsWith("$.")) {
const varName = value.slice(2);
resolved[key] = context[varName] ?? value;
} else {
resolved[key] = value;
}
}
return resolved;
}
}
// 使用示例
async function example(client: Client) {
const orchestrator = new ToolOrchestrator(client);
// 数据处理流程：读取 -> 分析 -> 存储
const results = await orchestrator.sequential([
{ tool: "read_file", args: { path: "/data/input.csv" } },
{ tool: "analyze_data", args: { data: "$.read_file_result" } },
{ tool: "save_results", args: { results: "$.analyze_data_result" } },
]);
console.log("流程执行结果:", results);
}
7.5 进度报告与取消
对于长时间运行的工具，MCP 支持进度报告和取消操作。
import { CallToolResult, ProgressToken } from "@modelcontextprotocol/sdk/types.js";
server.tool(
"long_running_task",
"执行长时间运行的任务",
{
task_type: z.string().describe("任务类型"),
data: z.record(z.unknown()).describe("任务数据"),
},
async ({ task_type, data }, extra): Promise<CallToolResult> => {
const progressToken = extra._meta?.progressToken;
const totalSteps = 100;
for (let i = 0; i < totalSteps; i++) {
// 检查是否被取消
if (extra.signal?.aborted) {
return {
content: [
{
type: "text" as const,
text: `任务已被取消（已完成 ${i}%）`,
},
],
};
}
// 报告进度
if (progressToken) {
await extra.sendNotification({
method: "notifications/progress",
params: {
progressToken,
progress: i + 1,
total: totalSteps,
message: `处理中... ${i + 1}%`,
},
});
}
// 模拟工作
await new Promise((resolve) => setTimeout(resolve, 50));
}
return {
content: [
{
type: "text" as const,
text: `任务完成！类型: ${task_type}`,
},
],
};
}
);
8. 资源系统
8.1 资源 URI 设计
资源 URI 是 MCP 中标识资源的唯一方式。良好的 URI 设计对于资源管理至关重要。
URI 模式设计原则：
<scheme>://<authority>/<path>
常用的 URI 模式：
| 模式 | 示例 | 用途 |
|---|---|---|
file:// |
file:///home/user/data.txt |
本地文件 |
config:// |
config://app/settings |
配置资源 |
db:// |
db://postgres/users |
数据库资源 |
api:// |
api://weather/current |
API 资源 |
memory:// |
memory://cache/session |
内存资源 |
// URI 设计示例
const resourceUris = {
// 文件系统资源
file: "file:///home/user/projects/myapp/src/index.ts",
// 数据库资源
databaseTable: "db://postgres/mydb/users",
databaseRow: "db://postgres/mydb/users/123",
databaseQuery: "db://postgres/mydb/users?role=admin",
// API 资源
apiEndpoint: "api://github/repos/anthropics/mcp/issues",
apiItem: "api://github/repos/anthropics/mcp/issues/42",
// 配置资源
appConfig: "config://myapp/settings",
userConfig: "config://myapp/users/john/preferences",
// 缓存资源
cache: "memory://cache/session/abc123",
};
8.2 文本与二进制资源
MCP 资源可以包含文本内容或二进制（Base64 编码）内容。
文本资源：
server.resource(
"readme",
"file:///project/README.md",
async (uri) => {
const content = await fs.readFile("/project/README.md", "utf-8");
return {
contents: [
{
uri: uri.href,
text: content,
mimeType: "text/markdown",
},
],
};
}
);
二进制资源：
server.resource(
"logo",
"file:///project/assets/logo.png",
async (uri) => {
const buffer = await fs.readFile("/project/assets/logo.png");
const base64 = buffer.toString("base64");
return {
contents: [
{
uri: uri.href,
blob: base64,
mimeType: "image/png",
},
],
};
}
);
多内容资源：
server.resource(
"project-overview",
"file:///project/overview",
async (uri) => {
const readme = await fs.readFile("/project/README.md", "utf-8");
const packageJson = await fs.readFile("/project/package.json", "utf-8");
return {
contents: [
{
uri: `${uri.href}/README.md`,
text: readme,
mimeType: "text/markdown",
},
{
uri: `${uri.href}/package.json`,
text: packageJson,
mimeType: "application/json",
},
],
};
}
);
8.3 动态资源
动态资源的 URI 包含可变部分，可以根据参数返回不同内容。
// 使用 URI 模板
server.resource(
"user-profile",
"api://users/{userId}",
async (uri) => {
// 从 URI 中提取 userId
const match = uri.href.match(/api:\/\/users\/(.+)/);
if (!match) {
throw new Error("无效的用户 ID");
}
const userId = match[1];
const user = await fetchUserFromDatabase(userId);
if (!user) {
throw new Error(`用户 ${userId} 不存在`);
}
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(user, null, 2),
mimeType: "application/json",
},
],
};
}
);
// 使用查询参数
server.resource(
"search-results",
"api://search",
async (uri) => {
const url = new URL(uri.href);
const query = url.searchParams.get("q");
const limit = parseInt(url.searchParams.get("limit") || "10");
if (!query) {
throw new Error("搜索参数 'q' 是必需的");
}
const results = await performSearch(query, limit);
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(results, null, 2),
mimeType: "application/json",
},
],
};
}
);
8.4 资源订阅与通知
Client 可以订阅资源变更，当资源内容更新时接收通知。
// Server 端：发送资源变更通知
class ResourceManager {
private subscribers: Map<string, Set<string>> = new Map(); // uri -> sessionIds
// 订阅资源
subscribe(uri: string, sessionId: string) {
if (!this.subscribers.has(uri)) {
this.subscribers.set(uri, new Set());
}
this.subscribers.get(uri)!.add(sessionId);
}
// 取消订阅
unsubscribe(uri: string, sessionId: string) {
this.subscribers.get(uri)?.delete(sessionId);
}
// 通知资源变更
async notifyUpdate(uri: string) {
// MCP SDK 会自动处理通知发送
// 你只需要调用 server.sendResourceNotification
console.log(`资源 ${uri} 已更新，通知订阅者`);
}
}
// Server 端：注册资源更新通知
server.resource(
"live-data",
"api://live/metrics",
async (uri) => {
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(await getLatestMetrics()),
mimeType: "application/json",
},
],
};
}
);
// Client 端：订阅资源变更
async function subscribeToResource(client: Client, uri: string) {
// 发送订阅请求
await client.subscribeResource({ uri });
// 设置通知处理器
client.setNotificationHandler(
"notifications/resources/updated",
(notification) => {
console.log(`资源 ${notification.params.uri} 已更新`);
// 重新读取资源
client.readResource({ uri: notification.params.uri })
.then((result) => {
console.log("更新后的内容:", result.contents[0].text);
});
}
);
}
9. 安全与权限
9.1 认证授权
MCP 本身不定义具体的认证机制，但提供了能力协商框架来支持各种认证方式。
基于 Token 的认证（SSE/HTTP 传输）：
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from "express";
const app = express();
// 认证中间件
function authenticateToken(req: express.Request, res: express.Response, next: express.NextFunction) {
const authHeader = req.headers["authorization"];
const token = authHeader && authHeader.split(" ")[1]; // Bearer TOKEN
if (!token) {
return res.status(401).json({ error: "缺少认证令牌" });
}
try {
const decoded = verifyToken(token);
(req as any).user = decoded;
next();
} catch (error) {
return res.status(403).json({ error: "无效的认证令牌" });
}
}
// 创建带认证的 MCP Server
const server = new McpServer({
name: "secure-server",
version: "1.0.0",
});
// 注册工具（带权限检查）
server.tool(
"admin_action",
"执行管理员操作",
{
action: z.string().describe("操作类型"),
},
async ({ action }, extra) => {
// 从请求上下文中获取用户信息
const user = (extra as any).user;
if (!user || user.role !== "admin") {
return {
content: [
{
type: "text" as const,
text: "权限不足：此操作需要管理员权限",
},
],
isError: true,
};
}
// 执行管理员操作...
return {
content: [
{
type: "text" as const,
text: `管理员操作 '${action}' 执行成功`,
},
],
};
}
);
// HTTP 端点（带认证）
app.get("/sse", authenticateToken, async (req, res) => {
const transport = new SSEServerTransport("/messages", res);
await server.connect(transport);
});
app.post("/messages", authenticateToken, async (req, res) => {
// 处理消息...
});
9.2 速率限制
防止滥用的重要安全措施。
class RateLimiter {
private requests: Map<string, number[]> = new Map();
constructor(
private maxRequests: number = 100,
private windowMs: number = 60000 // 1 分钟
) {}
// 检查是否超出限制
check(clientId: string): boolean {
const now = Date.now();
const windowStart = now - this.windowMs;
// 获取该客户端的请求记录
let timestamps = this.requests.get(clientId) || [];
// 清理过期记录
timestamps = timestamps.filter((t) => t > windowStart);
// 检查是否超出限制
if (timestamps.length >= this.maxRequests) {
return false; // 超出限制
}
// 记录新请求
timestamps.push(now);
this.requests.set(clientId, timestamps);
return true; // 允许请求
}
// 获取剩余配额
getRemaining(clientId: string): number {
const now = Date.now();
const windowStart = now - this.windowMs;
const timestamps = this.requests.get(clientId) || [];
const validRequests = timestamps.filter((t) => t > windowStart);
return Math.max(0, this.maxRequests - validRequests.length);
}
// 重置限制
reset(clientId: string) {
this.requests.delete(clientId);
}
}
// 使用示例
const limiter = new RateLimiter(50, 60000); // 每分钟最多 50 次请求
server.tool(
"rate_limited_tool",
"有速率限制的工具",
{
input: z.string(),
},
async ({ input }, extra) => {
const clientId = (extra as any).clientId || "default";
if (!limiter.check(clientId)) {
return {
content: [
{
type: "text" as const,
text: `速率限制：已超出请求限制。剩余配额: ${limiter.getRemaining(clientId)}`,
},
],
isError: true,
};
}
// 执行工具逻辑...
return {
content: [
{
type: "text" as const,
text: `处理完成: ${input}`,
},
],
};
}
);
9.3 输入验证与净化
import { z } from "zod";
import { escape } from "sql-escape-string"; // 假设已安装
// SQL 注入防护
const safeSqlInput = z.string().refine(
(val) => {
// 检查是否包含危险的 SQL 关键词
const dangerous = [
"DROP", "DELETE", "TRUNCATE", "ALTER",
"INSERT", "UPDATE", "EXEC", "EXECUTE",
"UNION", "INTO", "OUTFILE", "DUMPFILE",
];
const upper = val.toUpperCase();
return !dangerous.some((keyword) => upper.includes(keyword));
},
{ message: "输入包含不允许的 SQL 关键词" }
);
// 路径遍历防护
const safePath = z.string().refine(
(val) => {
// 禁止路径遍历
if (val.includes("..") || val.includes("~")) {
return false;
}
// 禁止绝对路径（可选）
if (val.startsWith("/")) {
return false;
}
return true;
},
{ message: "路径包含不允许的字符" }
);
// XSS 防护
function sanitizeHtml(input: string): string {
return input
.replace(/&/g, "&")
.replace(/</g, "<")
.replace(/>/g, ">")
.replace(/"/g, """)
.replace(/'/g, "'");
}
// 注册安全的工具
server.tool(
"safe_query",
"安全的数据库查询",
{
table: z.string().regex(/^[a-zA-Z_][a-zA-Z0-9_]*$/).describe("表名"),
conditions: z.record(z.string()).describe("查询条件"),
},
async ({ table, conditions }) => {
// 构建安全的参数化查询
const whereClauses: string[] = [];
const values: unknown[] = [];
let paramIndex = 1;
for (const [key, value] of Object.entries(conditions)) {
// 验证列名
if (!/^[a-zA-Z_][a-zA-Z0-9_]*$/.test(key)) {
throw new Error(`无效的列名: ${key}`);
}
whereClauses.push(`${key} = $${paramIndex++}`);
values.push(value);
}
const sql = `SELECT * FROM ${table} ${whereClauses.length > 0 ? "WHERE " + whereClauses.join(" AND ") : ""}`;
// 使用参数化查询，防止 SQL 注入
const results = await db.query(sql, values);
return {
content: [
{
type: "text" as const,
text: JSON.stringify(results, null, 2),
},
],
};
}
);
9.4 沙箱隔离
为工具执行提供安全的沙箱环境。
import { Worker } from "worker_threads";
class ToolSandbox {
private worker: Worker | null = null;
// 在沙箱中执行工具
async executeInSandbox(
code: string,
input: unknown,
options: {
timeout?: number;
memoryLimit?: number;
} = {}
): Promise<unknown> {
const { timeout = 5000, memoryLimit = 50 * 1024 * 1024 } = options;
return new Promise((resolve, reject) => {
// 创建 Worker 执行代码
this.worker = new Worker(
`
const { parentPort } = require('worker_threads');
parentPort.on('message', async ({ code, input }) => {
try {
// 在受限环境中执行
const fn = new Function('input', code);
const result = await fn(input);
parentPort.postMessage({ success: true, result });
} catch (error) {
parentPort.postMessage({
success: false,
error: error.message
});
}
});
`,
{ eval: true }
);
// 设置超时
const timer = setTimeout(() => {
this.worker?.terminate();
reject(new Error("工具执行超时"));
}, timeout);
// 监听结果
this.worker.on("message", (message) => {
clearTimeout(timer);
if (message.success) {
resolve(message.result);
} else {
reject(new Error(message.error));
}
});
this.worker.on("error", (error) => {
clearTimeout(timer);
reject(error);
});
// 发送执行请求
this.worker.postMessage({ code, input });
});
}
// 清理资源
async cleanup() {
if (this.worker) {
await this.worker.terminate();
this.worker = null;
}
}
}
9.5 安全最佳实践清单
以下是 MCP 安全开发的核心最佳实践：
传输层安全：
- 生产环境必须使用 HTTPS/WSS
- 使用 TLS 1.2 或更高版本
- 配置适当的 CORS 策略
认证与授权：
- 实施强认证机制（OAuth 2.0、JWT 等）
- 使用最小权限原则
- 实施基于角色的访问控制（RBAC）
输入验证：
- 对所有输入进行严格验证
- 使用参数化查询防止 SQL 注入
- 对文件路径进行规范化和验证
- 防止路径遍历攻击
输出安全：
- 对输出内容进行适当的编码
- 防止 XSS 攻击
- 不在响应中泄露敏感信息
速率限制：
- 对 API 调用实施速率限制
- 对资源密集型操作设置超时
- 监控和记录异常访问模式
日志与审计：
- 记录所有工具调用
- 记录认证和授权事件
- 保留审计日志用于安全分析
10. 主流 MCP 集成
10.1 Claude Desktop 配置
Claude Desktop 是 MCP 最早的支持者之一，配置简单直观。
配置文件位置：
- macOS:
~/Library/Application Support/Claude/claude_desktop_config.json - Windows:
%APPDATA%\Claude\claude_desktop_config.json
基本配置：
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/home/user/documents",
"/home/user/projects"
]
}
}
}
多 Server 配置：
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user"]
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {
"GITHUB_TOKEN": "your-github-token"
}
},
"postgres": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres"],
"env": {
"DATABASE_URL": "postgresql://user:pass@localhost:5432/mydb"
}
},
"brave-search": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-brave-search"],
"env": {
"BRAVE_API_KEY": "your-brave-api-key"
}
}
}
}
自定义 Server 配置：
{
"mcpServers": {
"my-custom-server": {
"command": "node",
"args": ["/path/to/my-server/build/index.js"],
"env": {
"NODE_ENV": "production",
"API_KEY": "your-api-key"
}
}
}
}
10.2 Cursor / Windsurf 集成
Cursor 和 Windsurf 是支持 MCP 的 AI 代码编辑器。
Cursor 配置：
在项目根目录创建 .cursor/mcp.json：
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"."
]
},
"database": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres"],
"env": {
"DATABASE_URL": "postgresql://localhost:5432/dev"
}
}
}
}
Windsurf 配置：
在项目根目录创建 .windsurfrules 或编辑全局配置：
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "."]
}
}
}
10.3 LangChain MCP 适配器
LangChain 提供了 MCP 适配器，可以将 MCP Server 集成到 LangChain 应用中。
安装：
pip install langchain-mcp-adapters
使用示例：
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import MCPTool
from langgraph.prebuilt import create_react_agent
async def main():
# 创建多 Server 客户端
async with MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/sse",
"transport": "sse",
}
}
) as client:
# 获取所有工具
tools = client.get_tools()
# 创建 AI Agent
agent = create_react_agent(
model="anthropic:claude-3-5-sonnet-20241022",
tools=tools,
)
# 调用 Agent
result = await agent.ainvoke(
{"messages": [{"role": "user", "content": "北京今天天气怎么样？"}]}
)
print(result["messages"][-1].content)
import asyncio
asyncio.run(main())
10.4 自定义集成：OpenAI 兼容适配器
如果你想将 MCP Server 集成到 OpenAI 兼容的 API 中：
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
class McpToOpenAIAdapter {
private client: Client;
constructor(mcpCommand: string, mcpArgs: string[]) {
const transport = new StdioClientTransport({
command: mcpCommand,
args: mcpArgs,
});
this.client = new Client({
name: "openai-adapter",
version: "1.0.0",
});
}
async connect() {
await this.client.connect(transport);
}
// 将 MCP 工具转换为 OpenAI 函数格式
async getOpenAITools() {
const { tools } = await this.client.listTools();
return tools.map((tool) => ({
type: "function",
function: {
name: tool.name,
description: tool.description,
parameters: tool.inputSchema,
},
}));
}
// 处理 OpenAI 的函数调用
async handleFunctionCall(name: string, args: Record<string, unknown>) {
const result = await this.client.callTool({
name,
arguments: args,
});
// 转换为 OpenAI 格式
return {
role: "function",
name,
content: result.content
.filter((c) => c.type === "text")
.map((c) => c.text)
.join("\n"),
};
}
async disconnect() {
await this.client.close();
}
}
11. MCP 生态系统
11.1 官方 Server 列表
Anthropic 和社区提供了多个官方参考实现：
| Server | 功能 | 语言 |
|---|---|---|
@modelcontextprotocol/server-filesystem |
文件系统访问 | TypeScript |
@modelcontextprotocol/server-github |
GitHub API 集成 | TypeScript |
@modelcontextprotocol/server-gitlab |
GitLab API 集成 | TypeScript |
@modelcontextprotocol/server-google-drive |
Google Drive 集成 | TypeScript |
@modelcontextprotocol/server-postgres |
PostgreSQL 数据库 | TypeScript |
@modelcontextprotocol/server-slack |
Slack 消息集成 | TypeScript |
@modelcontextprotocol/server-memory |
知识图谱记忆 | TypeScript |
@modelcontextprotocol/server-puppeteer |
浏览器自动化 | TypeScript |
@modelcontextprotocol/server-brave-search |
Brave 搜索引擎 | TypeScript |
@modelcontextprotocol/server-google-maps |
Google Maps 集成 | TypeScript |
@modelcontextprotocol/server-fetch |
网页内容抓取 | TypeScript |
@modelcontextprotocol/server-everything |
测试/参考实现 | TypeScript |
11.2 社区热门 Server
社区贡献了大量高质量的 MCP Server：
开发工具类：
mcp-server-docker- Docker 容器管理mcp-server-kubernetes- Kubernetes 集群管理mcp-server-jenkins- Jenkins CI/CD 集成mcp-server-vercel- Vercel 部署管理
数据库类：
mcp-server-mysql- MySQL 数据库mcp-server-mongodb- MongoDB 数据库mcp-server-redis- Redis 缓存mcp-server-elasticsearch- Elasticsearch 搜索引擎mcp-server-sqlite- SQLite 数据库
云服务类：
mcp-server-aws- AWS 服务集成mcp-server-gcp- Google Cloud 服务mcp-server-azure- Azure 服务
生产力工具类：
mcp-server-notion- Notion 笔记mcp-server-linear- Linear 项目管理mcp-server-jira- Jira 任务管理mcp-server-confluence- Confluence 文档
AI/ML 工具类：
mcp-server-huggingface- Hugging Face 模型mcp-server-ollama- Ollama 本地模型mcp-server-pinecone- Pinecone 向量数据库
11.3 如何选择和评估 MCP Server
选择 MCP Server 时，考虑以下因素：
功能性评估：
- 是否满足你的具体需求
- API 覆盖范围是否全面
- 是否支持你常用的功能
质量评估：
- 代码质量和文档完善程度
- 测试覆盖率
- 社区活跃度和维护状态
安全评估：
- 是否有安全审计
- 是否遵循最小权限原则
- 是否有已知的安全漏洞
兼容性评估：
- 支持的 MCP 协议版本
- 支持的传输方式
- SDK 兼容性
评估清单模板：
## MCP Server 评估清单
### 基本信息
- [ ] Server 名称：________
- [ ] 版本：________
- [ ] 维护者：________
- [ ] 许可证：________
### 功能性
- [ ] 核心功能是否满足需求
- [ ] 工具数量和覆盖范围
- [ ] 资源暴露能力
- [ ] 提示模板支持
### 质量
- [ ] 文档完善程度
- [ ] 测试覆盖率
- [ ] 社区活跃度
- [ ] 最近更新时间
### 安全性
- [ ] 安全审计状态
- [ ] 输入验证机制
- [ ] 认证授权支持
- [ ] 已知漏洞
### 兼容性
- [ ] MCP 协议版本
- [ ] 传输方式支持
- [ ] SDK 兼容性
- [ ] 平台支持
11.4 发布自己的 MCP Server
如果你想将自己的工具分享给社区：
准备发布清单：
- 完善文档：README、API 文档、使用示例
- 添加测试：单元测试和集成测试
- 安全审查：检查敏感信息泄露、输入验证
- 版本管理：使用语义化版本号
- 许可证：选择合适的开源许可证
发布到 npm：
# 构建项目
npm run build
# 更新版本
npm version patch # 或 minor / major
# 发布到 npm
npm publish
发布到 PyPI：
# 构建
python -m build
# 上传
twine upload dist/*
12. 实战项目：文件系统+数据库 MCP Server
在这一章中，我们将构建一个完整的 MCP Server，集成文件系统操作和 SQLite 数据库功能。
12.1 项目结构
file-db-mcp-server/
├── src/
│ ├── index.ts # 入口文件
│ ├── server.ts # Server 配置
│ ├── tools/
│ │ ├── file-tools.ts # 文件操作工具
│ │ ├── directory-tools.ts # 目录操作工具
│ │ └── database-tools.ts # 数据库工具
│ ├── resources/
│ │ ├── file-resources.ts # 文件资源
│ │ └── db-resources.ts # 数据库资源
│ ├── prompts/
│ │ └── analysis-prompts.ts # 分析提示模板
│ └── utils/
│ ├── path-utils.ts # 路径工具函数
│ ├── db-utils.ts # 数据库工具函数
│ └── security.ts # 安全验证函数
├── package.json
├── tsconfig.json
└── README.md
12.2 安装依赖
mkdir file-db-mcp-server
cd file-db-mcp-server
npm init -y
# MCP SDK
npm install @modelcontextprotocol/sdk zod
# SQLite
npm install better-sqlite3
npm install -D @types/better-sqlite3
# TypeScript
npm install -D typescript @types/node tsx
# 工具库
npm install glob mime-types
npm install -D @types/mime-types
12.3 实现路径工具函数
文件 src/utils/path-utils.ts：
import * as path from "path";
import * as os from "os";
// 配置允许访问的根目录
const ALLOWED_ROOTS: string[] = [];
export function setAllowedRoots(roots: string[]) {
ALLOWED_ROOTS.length = 0;
ALLOWED_ROOTS.push(...roots.map((r) => path.resolve(r)));
}
export function addAllowedRoot(root: string) {
const resolved = path.resolve(root);
if (!ALLOWED_ROOTS.includes(resolved)) {
ALLOWED_ROOTS.push(resolved);
}
}
// 验证路径是否在允许范围内
export function validatePath(filePath: string): string {
const resolved = path.resolve(filePath);
// 检查路径遍历
if (filePath.includes("..")) {
throw new Error("不允许路径遍历（..）");
}
// 如果没有配置允许的根目录，使用默认限制
if (ALLOWED_ROOTS.length === 0) {
// 默认只允许访问用户目录
const homeDir = os.homedir();
if (!resolved.startsWith(homeDir)) {
throw new Error(`路径必须在用户目录下: ${homeDir}`);
}
return resolved;
}
// 检查是否在允许的根目录下
const isAllowed = ALLOWED_ROOTS.some(
(root) => resolved.startsWith(root) || resolved === root
);
if (!isAllowed) {
throw new Error(
`路径不在允许范围内。允许的根目录: ${ALLOWED_ROOTS.join(", ")}`
);
}
return resolved;
}
// 获取相对路径
export function getRelativePath(filePath: string, basePath: string): string {
return path.relative(basePath, filePath);
}
// 规范化路径
export function normalizePath(filePath: string): string {
return path.normalize(filePath);
}
// 获取文件扩展名
export function getExtension(filePath: string): string {
return path.extname(filePath).toLowerCase();
}
// 判断是否为文本文件
export function isTextFile(filePath: string): boolean {
const textExtensions = [
".txt", ".md", ".json", ".js", ".ts", ".jsx", ".tsx",
".py", ".rb", ".java", ".c", ".cpp", ".h", ".hpp",
".css", ".scss", ".less", ".html", ".xml", ".yaml",
".yml", ".toml", ".ini", ".cfg", ".conf", ".sh",
".bash", ".zsh", ".fish", ".ps1", ".bat", ".cmd",
".sql", ".graphql", ".gql", ".proto", ".go", ".rs",
".swift", ".kt", ".scala", ".r", ".m", ".mm",
];
const ext = getExtension(filePath);
return textExtensions.includes(ext);
}
12.4 实现数据库工具函数
文件 src/utils/db-utils.ts：
import Database from "better-sqlite3";
import * as path from "path";
import * as fs from "fs/promises";
let db: Database.Database | null = null;
export async function initDatabase(dbPath: string): Promise<void> {
// 确保目录存在
const dir = path.dirname(dbPath);
await fs.mkdir(dir, { recursive: true });
// 创建或打开数据库
db = new Database(dbPath);
// 启用 WAL 模式提高并发性能
db.pragma("journal_mode = WAL");
// 创建示例表
db.exec(`
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY AUTOINCREMENT,
path TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
extension TEXT,
size INTEGER,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
modified_at DATETIME DEFAULT CURRENT_TIMESTAMP,
indexed_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS tags (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE
);
CREATE TABLE IF NOT EXISTS file_tags (
file_id INTEGER NOT NULL,
tag_id INTEGER NOT NULL,
PRIMARY KEY (file_id, tag_id),
FOREIGN KEY (file_id) REFERENCES files(id),
FOREIGN KEY (tag_id) REFERENCES tags(id)
);
CREATE TABLE IF NOT EXISTS notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
file_id INTEGER,
content TEXT NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (file_id) REFERENCES files(id)
);
CREATE INDEX IF NOT EXISTS idx_files_path ON files(path);
CREATE INDEX IF NOT EXISTS idx_files_extension ON files(extension);
CREATE INDEX IF NOT EXISTS idx_tags_name ON tags(name);
`);
console.error("数据库初始化完成");
}
export function getDatabase(): Database.Database {
if (!db) {
throw new Error("数据库未初始化，请先调用 initDatabase");
}
return db;
}
export function closeDatabase(): void {
if (db) {
db.close();
db = null;
}
}
// 安全执行 SQL
export function safeQuery(sql: string, params: unknown[] = []): unknown[] {
const database = getDatabase();
// 只允许 SELECT 查询
const trimmedSql = sql.trim().toUpperCase();
if (!trimmedSql.startsWith("SELECT")) {
throw new Error("只允许 SELECT 查询");
}
try {
const stmt = database.prepare(sql);
return stmt.all(...params);
} catch (error) {
throw new Error(`SQL 执行失败: ${error instanceof Error ? error.message : "未知错误"}`);
}
}
// 索引文件到数据库
export function indexFile(filePath: string, name: string, extension: string, size: number): void {
const database = getDatabase();
const stmt = database.prepare(`
INSERT INTO files (path, name, extension, size)
VALUES (?, ?, ?, ?)
ON CONFLICT(path) DO UPDATE SET
name = excluded.name,
extension = excluded.extension,
size = excluded.size,
modified_at = CURRENT_TIMESTAMP,
indexed_at = CURRENT_TIMESTAMP
`);
stmt.run(filePath, name, extension, size);
}
// 添加标签
export function addTag(fileId: number, tagName: string): void {
const database = getDatabase();
// 确保标签存在
const insertTag = database.prepare("INSERT OR IGNORE INTO tags (name) VALUES (?)");
insertTag.run(tagName);
// 获取标签 ID
const getTag = database.prepare("SELECT id FROM tags WHERE name = ?");
const tag = getTag.get(tagName) as { id: number };
// 关联文件和标签
const linkStmt = database.prepare(
"INSERT OR IGNORE INTO file_tags (file_id, tag_id) VALUES (?, ?)"
);
linkStmt.run(fileId, tag.id);
}
// 添加笔记
export function addNote(fileId: number | null, content: string): number {
const database = getDatabase();
const stmt = database.prepare(
"INSERT INTO notes (file_id, content) VALUES (?, ?)"
);
const result = stmt.run(fileId, content);
return result.lastInsertRowid as number;
}
// 获取文件统计
export function getFileStats(): Record<string, unknown> {
const database = getDatabase();
const totalFiles = database.prepare("SELECT COUNT(*) as count FROM files").get() as { count: number };
const totalSize = database.prepare("SELECT COALESCE(SUM(size), 0) as total FROM files").get() as { total: number };
const byExtension = database.prepare(`
SELECT extension, COUNT(*) as count, SUM(size) as total_size
FROM files
GROUP BY extension
ORDER BY count DESC
`).all();
return {
totalFiles: totalFiles.count,
totalSize: totalSize.total,
totalSizeFormatted: formatBytes(totalSize.total),
byExtension,
};
}
function formatBytes(bytes: number): string {
if (bytes === 0) return "0 B";
const k = 1024;
const sizes = ["B", "KB", "MB", "GB"];
const i = Math.floor(Math.log(bytes) / Math.log(k));
return parseFloat((bytes / Math.pow(k, i)).toFixed(2)) + " " + sizes[i];
}
12.5 实现文件操作工具
文件 src/tools/file-tools.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import * as fs from "fs/promises";
import * as path from "path";
import { validatePath, isTextFile } from "../utils/path-utils.js";
import { indexFile } from "../utils/db-utils.js";
export function registerFileTools(server: McpServer) {
// 读取文件
server.tool(
"read_file",
"读取文件内容。支持文本文件和二进制文件（以 Base64 返回）",
{
path: z.string().describe("文件路径"),
encoding: z.enum(["utf-8", "base64"]).optional().default("utf-8")
.describe("编码方式"),
offset: z.number().optional().describe("起始行号（从 0 开始）"),
limit: z.number().optional().describe("读取行数限制"),
},
async ({ path: filePath, encoding, offset, limit }) => {
try {
const fullPath = validatePath(filePath);
const stat = await fs.stat(fullPath);
if (!stat.isFile()) {
return {
content: [{ type: "text" as const, text: `错误: ${filePath} 不是文件` }],
isError: true,
};
}
if (encoding === "base64") {
const buffer = await fs.readFile(fullPath);
return {
content: [
{
type: "text" as const,
text: `文件大小: ${stat.size} 字节\nMIME 类型: ${getMimeType(filePath)}\n\nBase64 内容:\n${buffer.toString("base64")}`,
},
],
};
}
let content = await fs.readFile(fullPath, "utf-8");
// 处理行偏移和限制
if (offset !== undefined || limit !== undefined) {
const lines = content.split("\n");
const start = offset || 0;
const end = limit ? start + limit : lines.length;
content = lines.slice(start, end).join("\n");
}
return {
content: [
{
type: "text" as const,
text: content,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `读取文件失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 写入文件
server.tool(
"write_file",
"写入文件内容。如果文件不存在则创建，存在则覆盖",
{
path: z.string().describe("文件路径"),
content: z.string().describe("要写入的内容"),
create_dirs: z.boolean().optional().default(true)
.describe("是否自动创建父目录"),
},
async ({ path: filePath, content, create_dirs }) => {
try {
const fullPath = validatePath(filePath);
if (create_dirs) {
const dir = path.dirname(fullPath);
await fs.mkdir(dir, { recursive: true });
}
await fs.writeFile(fullPath, content, "utf-8");
// 索引到数据库
const stat = await fs.stat(fullPath);
indexFile(fullPath, path.basename(fullPath), path.extname(fullPath), stat.size);
return {
content: [
{
type: "text" as const,
text: `文件写入成功: ${filePath} (${stat.size} 字节)`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `写入文件失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 编辑文件（查找替换）
server.tool(
"edit_file",
"在文件中查找并替换内容",
{
path: z.string().describe("文件路径"),
old_text: z.string().describe("要查找的文本"),
new_text: z.string().describe("替换后的文本"),
all: z.boolean().optional().default(false).describe("是否替换所有匹配项"),
},
async ({ path: filePath, old_text, new_text, all }) => {
try {
const fullPath = validatePath(filePath);
let content = await fs.readFile(fullPath, "utf-8");
const count = content.split(old_text).length - 1;
if (count === 0) {
return {
content: [
{
type: "text" as const,
text: `未找到匹配的文本: "${old_text}"`,
},
],
isError: true,
};
}
if (all) {
content = content.split(old_text).join(new_text);
} else {
content = content.replace(old_text, new_text);
}
await fs.writeFile(fullPath, content, "utf-8");
return {
content: [
{
type: "text" as const,
text: `文件编辑成功。替换了 ${all ? count : 1} 处匹配。`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `编辑文件失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 删除文件
server.tool(
"delete_file",
"删除指定文件（移动到回收站）",
{
path: z.string().describe("文件路径"),
},
async ({ path: filePath }) => {
try {
const fullPath = validatePath(filePath);
const stat = await fs.stat(fullPath);
if (!stat.isFile()) {
return {
content: [
{
type: "text" as const,
text: `错误: ${filePath} 不是文件`,
},
],
isError: true,
};
}
// 创建回收站目录
const trashDir = path.join(path.dirname(fullPath), ".trash");
await fs.mkdir(trashDir, { recursive: true });
// 移动到回收站
const trashPath = path.join(trashDir, `${Date.now()}_${path.basename(fullPath)}`);
await fs.rename(fullPath, trashPath);
return {
content: [
{
type: "text" as const,
text: `文件已移动到回收站: ${trashPath}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `删除文件失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 文件搜索
server.tool(
"search_files",
"在目录中搜索文件",
{
directory: z.string().describe("搜索目录"),
pattern: z.string().describe("文件名模式（支持通配符 * 和 ?）"),
content_search: z.string().optional().describe("在文件内容中搜索的文本"),
max_results: z.number().optional().default(50).describe("最大结果数"),
},
async ({ directory, pattern, content_search, max_results }) => {
try {
const fullPath = validatePath(directory);
const results: Array<{ path: string; name: string; size: number; modified: string }> = [];
async function searchDir(dirPath: string) {
if (results.length >= max_results) return;
const entries = await fs.readdir(dirPath, { withFileTypes: true });
for (const entry of entries) {
if (results.length >= max_results) break;
const entryPath = path.join(dirPath, entry.name);
if (entry.isDirectory()) {
// 跳过隐藏目录和 node_modules
if (!entry.name.startsWith(".") && entry.name !== "node_modules") {
await searchDir(entryPath);
}
} else if (entry.isFile()) {
// 文件名匹配
if (matchPattern(entry.name, pattern)) {
// 内容搜索
if (content_search && isTextFile(entryPath)) {
try {
const content = await fs.readFile(entryPath, "utf-8");
if (!content.includes(content_search)) continue;
} catch {
continue;
}
}
const stat = await fs.stat(entryPath);
results.push({
path: entryPath,
name: entry.name,
size: stat.size,
modified: stat.mtime.toISOString(),
});
}
}
}
}
await searchDir(fullPath);
return {
content: [
{
type: "text" as const,
text: `找到 ${results.length} 个文件:\n${JSON.stringify(results, null, 2)}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `搜索失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
}
// 辅助函数：通配符匹配
function matchPattern(filename: string, pattern: string): boolean {
const regexPattern = pattern
.replace(/\./g, "\\.")
.replace(/\*/g, ".*")
.replace(/\?/g, ".");
return new RegExp(`^${regexPattern}$`, "i").test(filename);
}
// 辅助函数：获取 MIME 类型
function getMimeType(filePath: string): string {
const ext = path.extname(filePath).toLowerCase();
const mimeTypes: Record<string, string> = {
".json": "application/json",
".js": "application/javascript",
".ts": "application/typescript",
".html": "text/html",
".css": "text/css",
".md": "text/markdown",
".txt": "text/plain",
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".svg": "image/svg+xml",
".pdf": "application/pdf",
};
return mimeTypes[ext] || "application/octet-stream";
}
12.6 实现目录操作工具
文件 src/tools/directory-tools.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import * as fs from "fs/promises";
import * as path from "path";
import { validatePath } from "../utils/path-utils.js";
export function registerDirectoryTools(server: McpServer) {
// 列出目录内容
server.tool(
"list_directory",
"列出目录中的文件和子目录",
{
path: z.string().describe("目录路径"),
recursive: z.boolean().optional().default(false).describe("是否递归列出"),
show_hidden: z.boolean().optional().default(false).describe("是否显示隐藏文件"),
max_depth: z.number().optional().default(3).describe("最大递归深度"),
},
async ({ path: dirPath, recursive, show_hidden, max_depth }) => {
try {
const fullPath = validatePath(dirPath);
const entries = await listDirectoryEntries(fullPath, recursive, show_hidden, 0, max_depth);
return {
content: [
{
type: "text" as const,
text: `目录内容 (${entries.length} 项):\n${formatDirectoryTree(entries)}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `列出目录失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 创建目录
server.tool(
"create_directory",
"创建新目录（包括父目录）",
{
path: z.string().describe("目录路径"),
},
async ({ path: dirPath }) => {
try {
const fullPath = validatePath(dirPath);
await fs.mkdir(fullPath, { recursive: true });
return {
content: [
{
type: "text" as const,
text: `目录创建成功: ${dirPath}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `创建目录失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 获取目录信息
server.tool(
"directory_info",
"获取目录的详细信息，包括大小、文件数量等统计",
{
path: z.string().describe("目录路径"),
},
async ({ path: dirPath }) => {
try {
const fullPath = validatePath(dirPath);
const stats = await getDirectoryStats(fullPath);
return {
content: [
{
type: "text" as const,
text: `目录信息:\n${JSON.stringify(stats, null, 2)}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `获取目录信息失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
}
// 辅助函数：列出目录内容
interface DirectoryEntry {
name: string;
path: string;
type: "file" | "directory" | "symlink";
size?: number;
modified?: string;
children?: DirectoryEntry[];
}
async function listDirectoryEntries(
dirPath: string,
recursive: boolean,
showHidden: boolean,
currentDepth: number,
maxDepth: number
): Promise<DirectoryEntry[]> {
const entries: DirectoryEntry[] = [];
const items = await fs.readdir(dirPath, { withFileTypes: true });
for (const item of items) {
// 跳过隐藏文件
if (!showHidden && item.name.startsWith(".")) continue;
const itemPath = path.join(dirPath, item.name);
const stat = await fs.stat(itemPath);
const entry: DirectoryEntry = {
name: item.name,
path: itemPath,
type: item.isDirectory() ? "directory" : item.isSymbolicLink() ? "symlink" : "file",
size: stat.size,
modified: stat.mtime.toISOString(),
};
// 递归处理子目录
if (recursive && item.isDirectory() && currentDepth < maxDepth) {
entry.children = await listDirectoryEntries(
itemPath,
recursive,
showHidden,
currentDepth + 1,
maxDepth
);
}
entries.push(entry);
}
// 排序：目录在前，文件在后
return entries.sort((a, b) => {
if (a.type === "directory" && b.type !== "directory") return -1;
if (a.type !== "directory" && b.type === "directory") return 1;
return a.name.localeCompare(b.name);
});
}
// 格式化目录树
function formatDirectoryTree(entries: DirectoryEntry[], indent: string = ""): string {
let result = "";
for (const entry of entries) {
const icon = entry.type === "directory" ? "📁" : entry.type === "symlink" ? "🔗" : "📄";
const size = entry.size !== undefined ? ` (${formatSize(entry.size)})` : "";
result += `${indent}${icon} ${entry.name}${size}\n`;
if (entry.children) {
result += formatDirectoryTree(entry.children, indent + " ");
}
}
return result;
}
// 格式化文件大小
function formatSize(bytes: number): string {
if (bytes === 0) return "0 B";
const k = 1024;
const sizes = ["B", "KB", "MB", "GB"];
const i = Math.floor(Math.log(bytes) / Math.log(k));
return parseFloat((bytes / Math.pow(k, i)).toFixed(2)) + " " + sizes[i];
}
// 获取目录统计
async function getDirectoryStats(dirPath: string): Promise<Record<string, unknown>> {
let totalFiles = 0;
let totalDirs = 0;
let totalSize = 0;
const extensions: Record<string, number> = {};
async function walkDir(currentPath: string) {
const items = await fs.readdir(currentPath, { withFileTypes: true });
for (const item of items) {
if (item.name.startsWith(".")) continue;
const itemPath = path.join(currentPath, item.name);
if (item.isDirectory()) {
totalDirs++;
await walkDir(itemPath);
} else if (item.isFile()) {
totalFiles++;
const stat = await fs.stat(itemPath);
totalSize += stat.size;
const ext = path.extname(item.name).toLowerCase() || "(无扩展名)";
extensions[ext] = (extensions[ext] || 0) + 1;
}
}
}
await walkDir(dirPath);
return {
path: dirPath,
totalFiles,
totalDirectories: totalDirs,
totalSize: formatSize(totalSize),
totalSizeBytes: totalSize,
extensionBreakdown: extensions,
};
}
12.7 实现数据库工具
文件 src/tools/database-tools.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import {
safeQuery,
indexFile,
addTag,
addNote,
getFileStats,
getDatabase,
} from "../utils/db-utils.js";
export function registerDatabaseTools(server: McpServer) {
// 执行 SQL 查询（只读）
server.tool(
"db_query",
"执行 SQL 查询（仅支持 SELECT）",
{
sql: z.string().describe("SQL 查询语句（仅 SELECT）"),
params: z.array(z.unknown()).optional().default([]).describe("查询参数"),
limit: z.number().optional().default(100).describe("返回行数限制"),
},
async ({ sql, params, limit }) => {
try {
// 添加 LIMIT 子句（如果没有）
let query = sql;
if (!query.toUpperCase().includes("LIMIT")) {
query += ` LIMIT ${limit}`;
}
const results = safeQuery(query, params);
return {
content: [
{
type: "text" as const,
text: `查询返回 ${results.length} 条记录:\n${JSON.stringify(results, null, 2)}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `查询失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 获取表结构
server.tool(
"db_schema",
"获取数据库表结构信息",
{
table: z.string().optional().describe("表名（可选，不提供则返回所有表）"),
},
async ({ table }) => {
try {
const db = getDatabase();
if (table) {
// 获取指定表的结构
const columns = db.prepare(`PRAGMA table_info(${table})`).all();
const indexes = db.prepare(`PRAGMA index_list(${table})`).all();
return {
content: [
{
type: "text" as const,
text: `表 ${table} 的结构:\n\n列:\n${JSON.stringify(columns, null, 2)}\n\n索引:\n${JSON.stringify(indexes, null, 2)}`,
},
],
};
} else {
// 获取所有表
const tables = db.prepare(
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
).all();
const schema: Record<string, unknown> = {};
for (const t of tables as Array<{ name: string }>) {
schema[t.name] = db.prepare(`PRAGMA table_info(${t.name})`).all();
}
return {
content: [
{
type: "text" as const,
text: `数据库包含 ${tables.length} 张表:\n${JSON.stringify(schema, null, 2)}`,
},
],
};
}
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `获取表结构失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 索引文件到数据库
server.tool(
"db_index_file",
"将文件信息索引到数据库",
{
file_path: z.string().describe("文件路径"),
tags: z.array(z.string()).optional().describe("文件标签"),
},
async ({ file_path, tags }) => {
try {
const path = await import("path");
const fs = await import("fs/promises");
const stat = await fs.stat(file_path);
const name = path.default.basename(file_path);
const ext = path.default.extname(file_path);
indexFile(file_path, name, ext, stat.size);
// 获取文件 ID
const db = getDatabase();
const file = db.prepare("SELECT id FROM files WHERE path = ?").get(file_path) as { id: number };
// 添加标签
if (tags) {
for (const tag of tags) {
addTag(file.id, tag);
}
}
return {
content: [
{
type: "text" as const,
text: `文件索引成功:\n路径: ${file_path}\n大小: ${stat.size} 字节\n标签: ${tags?.join(", ") || "无"}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `索引文件失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 添加笔记
server.tool(
"db_add_note",
"为文件或全局添加笔记",
{
content: z.string().describe("笔记内容"),
file_path: z.string().optional().describe("关联的文件路径（可选）"),
},
async ({ content, file_path }) => {
try {
let fileId: number | null = null;
if (file_path) {
const db = getDatabase();
const file = db.prepare("SELECT id FROM files WHERE path = ?").get(file_path) as { id: number } | undefined;
fileId = file?.id || null;
}
const noteId = addNote(fileId, content);
return {
content: [
{
type: "text" as const,
text: `笔记添加成功！ID: ${noteId}${file_path ? `\n关联文件: ${file_path}` : ""}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `添加笔记失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
// 获取数据库统计
server.tool(
"db_stats",
"获取数据库统计信息",
{},
async () => {
try {
const stats = getFileStats();
return {
content: [
{
type: "text" as const,
text: `数据库统计:\n${JSON.stringify(stats, null, 2)}`,
},
],
};
} catch (error) {
return {
content: [
{
type: "text" as const,
text: `获取统计失败: ${error instanceof Error ? error.message : "未知错误"}`,
},
],
isError: true,
};
}
}
);
}
12.8 实现资源和提示
文件 src/resources/file-resources.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import * as fs from "fs/promises";
import * as path from "path";
import { validatePath } from "../utils/path-utils.js";
export function registerFileResources(server: McpServer) {
// 工作目录资源
server.resource(
"working-directory",
"file:///",
async (uri) => {
const cwd = process.cwd();
const entries = await fs.readdir(cwd, { withFileTypes: true });
const listing = entries.map((entry) => ({
name: entry.name,
type: entry.isDirectory() ? "directory" : "file",
}));
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(listing, null, 2),
mimeType: "application/json",
},
],
};
}
);
// 文件信息资源
server.resource(
"file-info",
"file-info://{path}",
async (uri) => {
const filePath = uri.href.replace("file-info://", "");
const fullPath = validatePath(filePath);
const stat = await fs.stat(fullPath);
const info = {
path: fullPath,
name: path.basename(fullPath),
extension: path.extname(fullPath),
size: stat.size,
created: stat.birthtime.toISOString(),
modified: stat.mtime.toISOString(),
accessed: stat.atime.toISOString(),
isDirectory: stat.isDirectory(),
isFile: stat.isFile(),
permissions: stat.mode.toString(8),
};
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(info, null, 2),
mimeType: "application/json",
},
],
};
}
);
}
文件 src/resources/db-resources.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { getDatabase, getFileStats } from "../utils/db-utils.js";
export function registerDbResources(server: McpServer) {
// 数据库统计资源
server.resource(
"db-stats",
"db://stats",
async (uri) => {
const stats = getFileStats();
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(stats, null, 2),
mimeType: "application/json",
},
],
};
}
);
// 最近索引的文件
server.resource(
"recent-files",
"db://files/recent",
async (uri) => {
const db = getDatabase();
const files = db.prepare(
"SELECT * FROM files ORDER BY indexed_at DESC LIMIT 20"
).all();
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(files, null, 2),
mimeType: "application/json",
},
],
};
}
);
// 标签列表
server.resource(
"tags-list",
"db://tags",
async (uri) => {
const db = getDatabase();
const tags = db.prepare(`
SELECT t.name, COUNT(ft.file_id) as file_count
FROM tags t
LEFT JOIN file_tags ft ON t.id = ft.tag_id
GROUP BY t.id
ORDER BY file_count DESC
`).all();
return {
contents: [
{
uri: uri.href,
text: JSON.stringify(tags, null, 2),
mimeType: "application/json",
},
],
};
}
);
}
文件 src/prompts/analysis-prompts.ts：
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
export function registerAnalysisPrompts(server: McpServer) {
// 文件分析提示
server.prompt(
"analyze-file",
"分析文件内容并提供见解",
{
file_path: z.string().describe("文件路径"),
focus: z.enum(["structure", "quality", "security", "performance"]).optional()
.describe("分析重点"),
},
async ({ file_path, focus = "structure" }) => {
const focusInstructions: Record<string, string> = {
structure: "请分析文件的结构、组织方式和代码风格。",
quality: "请评估代码质量，包括可读性、可维护性和最佳实践。",
security: "请重点审查安全问题和潜在漏洞。",
performance: "请分析性能瓶颈和优化机会。",
};
return {
messages: [
{
role: "user",
content: {
type: "text",
text: `请分析文件 ${file_path} 的内容。\n\n${focusInstructions[focus]}\n\n请先读取文件内容，然后提供详细的分析报告。`,
},
},
],
};
}
);
// 项目总结提示
server.prompt(
"project-summary",
"生成项目结构和内容的总结",
{
directory: z.string().describe("项目根目录"),
},
async ({ directory }) => {
return {
messages: [
{
role: "user",
content: {
type: "text",
text: `请分析 ${directory} 目录下的项目结构。\n\n请完成以下任务：\n1. 列出目录结构\n2. 识别项目类型和使用的技术栈\n3. 总结主要功能模块\n4. 提供项目改进建议`,
},
},
],
};
}
);
}
12.9 主入口文件
文件 src/index.ts：
#!/usr/bin/env node
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import * as path from "path";
import { registerFileTools } from "./tools/file-tools.js";
import { registerDirectoryTools } from "./tools/directory-tools.js";
import { registerDatabaseTools } from "./tools/database-tools.js";
import { registerFileResources } from "./resources/file-resources.js";
import { registerDbResources } from "./resources/db-resources.js";
import { registerAnalysisPrompts } from "./prompts/analysis-prompts.js";
import { initDatabase, closeDatabase } from "./utils/db-utils.js";
import { setAllowedRoots, addAllowedRoot } from "./utils/path-utils.js";
// 创建 MCP Server
const server = new McpServer({
name: "file-db-mcp-server",
version: "1.0.0",
});
// 注册所有功能
registerFileTools(server);
registerDirectoryTools(server);
registerDatabaseTools(server);
registerFileResources(server);
registerDbResources(server);
registerAnalysisPrompts(server);
// 初始化并启动
async function main() {
// 解析命令行参数
const args = process.argv.slice(2);
const dbPath = args.find((a) => a.startsWith("--db="))?.split("=")[1] || "./data/file-index.db";
const allowedDirs = args.filter((a) => !a.startsWith("--"));
// 设置允许访问的目录
if (allowedDirs.length > 0) {
for (const dir of allowedDirs) {
addAllowedRoot(path.resolve(dir));
}
console.error(`允许访问的目录: ${allowedDirs.join(", ")}`);
} else {
// 默认允许访问当前目录
addAllowedRoot(process.cwd());
console.error(`允许访问的目录: ${process.cwd()}`);
}
// 初始化数据库
await initDatabase(dbPath);
console.error(`数据库路径: ${dbPath}`);
// 启动传输
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("MCP Server 已启动");
// 优雅关闭
process.on("SIGINT", () => {
console.error("正在关闭...");
closeDatabase();
process.exit(0);
});
process.on("SIGTERM", () => {
console.error("正在关闭...");
closeDatabase();
process.exit(0);
});
}
main().catch((error) => {
console.error("启动失败:", error);
process.exit(1);
});
12.10 构建与运行
构建项目：
npm run build
运行 Server：
# 允许访问 home 目录
node build/index.js /home/user
# 允许访问多个目录
node build/index.js /home/user/projects /home/user/documents
# 自定义数据库路径
node build/index.js --db=/tmp/mydb.sqlite /home/user
在 Claude Desktop 中配置：
{
"mcpServers": {
"file-db": {
"command": "node",
"args": [
"/path/to/file-db-mcp-server/build/index.js",
"/home/user/projects"
]
}
}
}
使用 MCP Inspector 测试：
npx @modelcontextprotocol/inspector node build/index.js /home/user
12.11 使用示例
配置完成后，你可以在 Claude Desktop 中这样使用：
文件操作：
请帮我读取 /home/user/projects/myapp/package.json 文件的内容
在 /home/user/projects 目录下搜索所有 .ts 文件
创建一个新文件 /home/user/projects/notes/todo.md，写入今天的待办事项
数据库操作：
帮我把 /home/user/projects 目录下的所有 .md 文件索引到数据库，并标记为 "文档"
查询数据库中最近修改的 10 个文件
添加一条笔记：今天的会议讨论了 MCP 协议的集成方案
分析操作：
分析 /home/user/projects/myapp/src/index.ts 文件的代码质量
总结 /home/user/projects/myapp 项目的结构和技术栈
总结
本教程全面介绍了 MCP（Model Context Protocol）模型上下文协议，从基础概念到高级开发，再到完整的实战项目。让我们回顾一下关键要点：
核心概念：
- MCP 是 AI 应用的"USB-C 接口"，提供标准化的工具集成方式
- 采用 Client-Server 架构，支持 stdio、SSE、Streamable HTTP 三种传输方式
- 基于 JSON-RPC 2.0 消息格式
- 五大核心概念：Resources、Tools、Prompts、Sampling、Roots
开发要点：
- 使用官方 TypeScript 或 Python SDK 进行开发
- 通过 MCP Inspector 进行调试和测试
- 工具定义使用 JSON Schema（Zod）进行输入验证
- 实施完善的安全措施：认证、授权、输入验证、速率限制
生态系统：
- 官方和社区提供了丰富的 MCP Server
- 支持 Claude Desktop、Cursor、Windsurf 等主流 AI 应用
- 可与 LangChain 等框架集成
下一步行动：
- 动手实践：按照教程中的示例代码，创建你自己的 MCP Server
- 探索生态：尝试使用社区提供的 MCP Server
- 贡献社区：将你的工具封装为 MCP Server 并开源
- 持续学习：关注 MCP 协议的更新和新特性
MCP 协议正处于快速发展阶段，随着越来越多的 AI 应用和工具支持这一标准，它将成为 AI 工具集成的事实标准。现在开始学习和实践 MCP，将为你在 AI 开发领域打下坚实的基础。
附录：常见问题解答（FAQ）
Q1: MCP 和 REST API 有什么区别？
A: REST API 是通用的 Web 服务接口，而 MCP 是专门为 AI 模型设计的工具集成协议。MCP 内置了工具发现、能力协商、资源订阅等 AI 特有的功能。
Q2: MCP Server 可以用哪些语言开发？
A: MCP 协议本身是语言无关的。官方提供 TypeScript 和 Python SDK，社区也有 Go、Rust、Java 等语言的实现。
Q3: MCP 如何处理并发请求？
A: MCP 基于 JSON-RPC 2.0，每个请求都有唯一的 ID，支持异步并发处理。Server 可以同时处理多个请求。
Q4: MCP 的性能如何？
A: stdio 传输延迟极低（微秒级），适合本地工具。SSE/HTTP 传输延迟取决于网络，通常在毫秒级。对于大多数应用场景，MCP 的性能完全足够。
Q5: 如何调试 MCP Server？
A: 使用 MCP Inspector 工具，它可以可视化展示 Client-Server 之间的所有消息，帮助你快速定位问题。
Q6: MCP 支持流式响应吗？
A: 是的，MCP 支持通过 SSE 传输实现流式响应，也支持进度通知机制。
Q7: 如何保证 MCP Server 的安全性？
A: 实施认证授权、输入验证、速率限制、路径规范化、最小权限原则等安全措施。详见本教程第 9 章。
本教程最后更新时间：2025 年
如有疑问或建议，欢迎在 GitHub 上提交 Issue 或 Pull Request。
