MCP Server 开发实战完全教程
SEO 元信息
- 名称:MCP Server 开发实战完全教程
- 描述:零基础 MCP Server 开发实战完全教程,涵盖 MCP 协议架构、TypeScript/Python SDK、Resources/Tools/Prompts 核心能力开发、SSE 与 Stdio 传输、Claude/Cursor 集成、调试技巧、生产部署等核心技能,适合 AI 开发者系统学习。
- 关键词:MCP Server, 模型上下文协议, AI 工具开发, Claude 集成, TypeScript SDK
- 长尾关键词:MCP Server 开发入门教程, MCP Server TypeScript 开发, MCP 协议工具开发实战, Claude MCP Server 开发指南
目录
- 什么是 MCP?协议架构回顾
- MCP Server 三大核心概念
- TypeScript SDK 详解
- Python SDK 详解
- 实战:从零开发一个文件系统 MCP Server
- 实战:数据库查询 MCP Server
- SSE 与 Stdio 传输模式
- 工具注册、资源暴露与 Prompt 模板
- 与 Claude Desktop / Cursor 集成测试
- 调试技巧与常见问题
- 生产部署方案
- 总结与进阶资源
1. 什么是 MCP?协议架构回顾
MCP(Model Context Protocol,模型上下文协议)是一个开放协议,旨在为大语言模型(LLM)提供标准化的方式来连接外部数据源和工具。你可以把它理解为 AI 领域的"USB-C 接口"——不论你使用什么模型、什么客户端,只要双方都支持 MCP,就能无缝对接。
1.1 协议架构
MCP 采用经典的 客户端-服务器(Client-Server) 架构:
┌─────────────────┐ JSON-RPC 2.0 ┌─────────────────┐
│ MCP Host │ ◄──────────────────────────► │ MCP Server │
│ (Claude Desktop │ Stdio / SSE / HTTP │ (你的自定义服务) │
│ Cursor, etc.) │ │ │
└─────────────────┘ └─────────────────┘
关键角色说明:
| 角色 | 说明 |
|---|---|
| MCP Host | 宿主应用,如 Claude Desktop、Cursor IDE 等,负责发起连接 |
| MCP Client | Host 内部的协议客户端,维护与 Server 的 1:1 连接 |
| MCP Server | 轻量级服务,暴露 Tools、Resources、Prompts 三类能力 |
| 传输层 | 支持 Stdio(本地进程)和 SSE(HTTP 流)两种模式 |
通信协议基于 JSON-RPC 2.0,所有消息都是标准的请求-响应模式,也支持服务端主动通知(notification)。
1.2 MCP 的价值
在没有 MCP 之前,每个 AI 应用都需要为每个外部服务编写定制的集成代码。MCP 的出现将这个 O(M×N) 的问题降为 O(M+N):
- 工具开发者只需编写一个 MCP Server
- 应用开发者只需实现 MCP Client 协议
- 所有 Server 与所有 Client 自动兼容
2. MCP Server 三大核心概念
MCP Server 的能力通过三类原语(Primitives)暴露给客户端:
2.1 Resources(资源)
资源是 MCP Server 向客户端暴露数据的方式。它们类似于 REST API 中的 GET 端点——提供只读的数据访问。
特点:
- 使用 URI 标识,如
file:///path/to/file、db://table/users - 支持文本和二进制(Base64)内容
- 可包含 MIME 类型信息
- 可动态列举(list)和读取(read)
典型场景: 文件系统浏览、数据库表结构、配置文件内容、日志文件等。
2.2 Tools(工具)
工具是 MCP 中最核心的能力——允许 LLM 执行操作、调用外部服务。它们类似于 POST 请求,可以产生副作用。
特点:
- 有明确的名称和描述(供 LLM 理解何时使用)
- 接受 JSON Schema 格式的参数定义
- 返回结构化的执行结果
- 是"模型可控"的(由 LLM 决定何时调用)
典型场景: 执行数据库查询、发送邮件、创建文件、调用第三方 API 等。
2.3 Prompts(提示模板)
提示模板是预定义的、可复用的对话模板,用于引导 LLM 完成特定任务。
特点:
- 由用户主动选择使用(非模型自动调用)
- 接受参数,动态生成提示内容
- 返回一组结构化的消息列表
- 适合封装复杂工作流
典型场景: 代码审查模板、数据分析流程、文档生成模板等。
3. TypeScript SDK 详解
MCP 官方提供了 @modelcontextprotocol/sdk 这个 TypeScript SDK,是目前最成熟的实现。
3.1 安装与初始化
# 使用 npm
npm install @modelcontextprotocol/sdk
# 使用 pnpm
pnpm add @modelcontextprotocol/sdk
# 使用 bun
bun add @modelcontextprotocol/sdk
TypeScript 配置要求:
// tsconfig.json
{
"compilerOptions": {
"target": "ES2022",
"module": "Node16",
"moduleResolution": "Node16",
"strict": true,
"esModuleInterop": true,
"outDir": "./build",
"rootDir": "./src"
}
}
3.2 核心类
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
McpServer 是服务端核心类,提供以下方法:
| 方法 | 作用 |
|---|---|
server.tool(name, schema, handler) |
注册工具 |
server.resource(name, uri, handler) |
注册资源 |
server.prompt(name, schema, handler) |
注册提示模板 |
server.connect(transport) |
启动传输连接 |
3.3 创建基础 Server
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
const server = new McpServer({
name: "my-first-server",
version: "1.0.0",
});
// 注册一个简单的工具
server.tool(
"calculate-bmi",
"计算身体质量指数(BMI)",
{
weightKg: z.number().describe("体重,单位:千克"),
heightM: z.number().describe("身高,单位:米"),
},
async ({ weightKg, heightM }) => {
const bmi = weightKg / (heightM * heightM);
let category = "正常";
if (bmi < 18.5) category = "偏瘦";
else if (bmi >= 25) category = "偏胖";
else if (bmi >= 30) category = "肥胖";
return {
content: [
{
type: "text",
text: `BMI = ${bmi.toFixed(1)},分类:${category}`,
},
],
};
}
);
// 启动
const transport = new StdioServerTransport();
await server.connect(transport);
4. Python SDK 详解
Python SDK 包名为 mcp,支持同步和异步两种编程模式。
4.1 安装
# 使用 pip
pip install mcp
# 使用 uv(推荐)
uv pip install mcp
4.2 核心组件
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
name="my-server",
version="1.0.0",
)
FastMCP 是高级封装,通过装饰器即可注册工具、资源和提示:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("demo-server")
# 注册工具
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> str:
"""计算身体质量指数(BMI)"""
bmi = weight_kg / (height_m ** 2)
if bmi < 18.5:
category = "偏瘦"
elif bmi < 25:
category = "正常"
elif bmi < 30:
category = "偏胖"
else:
category = "肥胖"
return f"BMI = {bmi:.1f},分类:{category}"
# 注册资源
@mcp.resource("config://app")
def get_config() -> str:
"""返回应用配置信息"""
return '{"app_name": "Demo", "version": "1.0"}'
# 注册提示模板
@mcp.prompt()
def review_code(code: str) -> str:
"""生成代码审查提示"""
return f"请审查以下代码并提供改进建议:\n\n```\n{code}\n```"
# 启动
if __name__ == "__main__":
mcp.run(transport="stdio")
4.3 Python SDK vs TypeScript SDK 对比
| 特性 | TypeScript SDK | Python SDK |
|---|---|---|
| 包名 | @modelcontextprotocol/sdk |
mcp |
| 服务端类 | McpServer |
FastMCP |
| 参数校验 | Zod schema | Python type hints |
| 注册方式 | 方法调用 | 装饰器 / 方法调用 |
| 异步支持 | 原生 async/await | async/await |
| 成熟度 | 较高 | 高 |
5. 实战:从零开发一个文件系统 MCP Server
下面我们用 TypeScript 开发一个功能完整的文件系统 MCP Server,让 LLM 能够浏览目录、读取文件和搜索内容。
5.1 项目初始化
mkdir fs-mcp-server && cd fs-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
npx tsc --init
5.2 完整代码
// src/index.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import * as fs from "node:fs/promises";
import * as path from "node:path";
const ALLOWED_DIR = process.argv[2] || ".";
const server = new McpServer({
name: "filesystem-server",
version: "1.0.0",
});
// ========== 工具 1:列出目录内容 ==========
server.tool(
"list-directory",
"列出指定目录下的所有文件和子目录",
{
dirPath: z.string().describe("相对于根目录的路径,如 '.' 或 'src/utils'"),
},
async ({ dirPath }) => {
const fullPath = path.resolve(ALLOWED_DIR, dirPath);
// 安全检查:防止路径穿越
if (!fullPath.startsWith(path.resolve(ALLOWED_DIR))) {
return { content: [{ type: "text", text: "错误:路径越界" }] };
}
try {
const entries = await fs.readdir(fullPath, { withFileTypes: true });
const listing = entries.map((e) => ({
name: e.name,
type: e.isDirectory() ? "directory" : "file",
}));
return {
content: [{ type: "text", text: JSON.stringify(listing, null, 2) }],
};
} catch (err: any) {
return { content: [{ type: "text", text: `错误:${err.message}` }] };
}
}
);
// ========== 工具 2:读取文件内容 ==========
server.tool(
"read-file",
"读取指定文件的文本内容",
{
filePath: z.string().describe("相对于根目录的文件路径"),
maxLines: z.number().optional().describe("最大读取行数,默认 500"),
},
async ({ filePath, maxLines }) => {
const fullPath = path.resolve(ALLOWED_DIR, filePath);
if (!fullPath.startsWith(path.resolve(ALLOWED_DIR))) {
return { content: [{ type: "text", text: "错误:路径越界" }] };
}
try {
const content = await fs.readFile(fullPath, "utf-8");
const lines = content.split("\n").slice(0, maxLines ?? 500);
return {
content: [{ type: "text", text: lines.join("\n") }],
};
} catch (err: any) {
return { content: [{ type: "text", text: `错误:${err.message}` }] };
}
}
);
// ========== 工具 3:写入文件 ==========
server.tool(
"write-file",
"创建或覆盖写入文件内容",
{
filePath: z.string().describe("相对于根目录的文件路径"),
content: z.string().describe("要写入的文本内容"),
},
async ({ filePath, content }) => {
const fullPath = path.resolve(ALLOWED_DIR, filePath);
if (!fullPath.startsWith(path.resolve(ALLOWED_DIR))) {
return { content: [{ type: "text", text: "错误:路径越界" }] };
}
try {
await fs.mkdir(path.dirname(fullPath), { recursive: true });
await fs.writeFile(fullPath, content, "utf-8");
return {
content: [{ type: "text", text: `文件已写入:${filePath}` }],
};
} catch (err: any) {
return { content: [{ type: "text", text: `错误:${err.message}` }] };
}
}
);
// ========== 资源:暴露目录结构 ==========
server.resource(
"directory-tree",
"dir://tree",
async (uri) => {
async function buildTree(dir: string, depth: number): Promise<string> {
if (depth > 3) return "";
const entries = await fs.readdir(dir, { withFileTypes: true });
let result = "";
for (const entry of entries) {
if (entry.name.startsWith(".")) continue;
const prefix = " ".repeat(depth);
if (entry.isDirectory()) {
result += `${prefix}📁 ${entry.name}/\n`;
result += await buildTree(path.join(dir, entry.name), depth + 1);
} else {
result += `${prefix}📄 ${entry.name}\n`;
}
}
return result;
}
const tree = await buildTree(path.resolve(ALLOWED_DIR), 0);
return {
contents: [{ uri: uri.href, text: tree, mimeType: "text/plain" }],
};
}
);
// 启动
const transport = new StdioServerTransport();
server.connect(transport).then(() => {
console.error("文件系统 MCP Server 已启动");
});
5.3 构建与运行
npx tsc
node build/index.js /path/to/your/project
5.4 项目配置文件
// package.json
{
"name": "fs-mcp-server",
"version": "1.0.0",
"type": "module",
"scripts": {
"build": "tsc",
"start": "node build/index.js"
},
"dependencies": {
"@modelcontextprotocol/sdk": "^1.0.0",
"zod": "^3.22.0"
},
"devDependencies": {
"typescript": "^5.4.0",
"@types/node": "^20.0.0"
}
}
6. 实战:数据库查询 MCP Server
用 Python 开发一个 SQLite 数据库查询 Server,让 LLM 能查询表结构、执行只读 SQL。
6.1 完整代码
# db_server.py
import sqlite3
import json
from mcp.server.fastmcp import FastMCP
DB_PATH = "data.db"
mcp = FastMCP("database-server")
def get_connection():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
@mcp.tool()
def list_tables() -> str:
"""列出数据库中所有表名"""
conn = get_connection()
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
)
tables = [row["name"] for row in cursor.fetchall()]
conn.close()
return json.dumps(tables, ensure_ascii=False)
@mcp.tool()
def describe_table(table_name: str) -> str:
"""获取指定表的列定义信息(列名、类型、是否可为空等)"""
conn = get_connection()
try:
cursor = conn.execute(f"PRAGMA table_info({table_name})")
columns = [
{
"name": row["name"],
"type": row["type"],
"nullable": not row["notnull"],
"default": row["dflt_value"],
}
for row in cursor.fetchall()
]
return json.dumps(columns, ensure_ascii=False, indent=2)
finally:
conn.close()
@mcp.tool()
def execute_query(sql: str) -> str:
"""执行只读 SQL 查询(仅 SELECT 语句)。返回查询结果的 JSON 数组。"""
# 安全检查:只允许 SELECT
sql_stripped = sql.strip().upper()
if not sql_stripped.startswith("SELECT"):
return json.dumps({"error": "仅支持 SELECT 查询"}, ensure_ascii=False)
conn = get_connection()
try:
cursor = conn.execute(sql)
rows = [dict(row) for row in cursor.fetchall()]
return json.dumps(rows, ensure_ascii=False, indent=2, default=str)
except Exception as e:
return json.dumps({"error": str(e)}, ensure_ascii=False)
finally:
conn.close()
@mcp.resource("schema://database")
def get_full_schema() -> str:
"""返回数据库的完整表结构"""
conn = get_connection()
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
)
tables = {}
for row in cursor.fetchall():
table_name = row["name"]
cols = conn.execute(f"PRAGMA table_info({table_name})").fetchall()
tables[table_name] = [
{"name": c["name"], "type": c["type"]} for c in cols
]
conn.close()
return json.dumps(tables, ensure_ascii=False, indent=2)
@mcp.prompt()
def analyze_table(table_name: str) -> str:
"""生成表分析提示模板"""
return f"""请分析数据库表 `{table_name}` 的结构和数据特征:
1. 首先使用 describe_table 工具查看表结构
2. 然后使用 execute_query 执行 `SELECT * FROM {table_name} LIMIT 10` 查看样本数据
3. 分析各列的数据分布、可能的数据质量问题
4. 给出该表的用途总结和改进建议"""
if __name__ == "__main__":
mcp.run(transport="stdio")
6.2 测试
python db_server.py
7. SSE 与 Stdio 传输模式
MCP 支持两种传输方式,适用于不同场景。
7.1 Stdio 传输
Stdio(标准输入/输出)是最常用的传输方式,适合本地开发和桌面应用集成。
工作原理:
Host 进程 Server 进程
│ │
├── stdin (JSON-RPC) ──────►│
│ │
│◄── stdout (JSON-RPC) ──────┤
│ │
│◄── stderr (日志) ──────────┤
TypeScript 实现:
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const transport = new StdioServerTransport();
await server.connect(transport);
Python 实现:
mcp.run(transport="stdio")
优点: 简单直接、无网络依赖、天然进程隔离 缺点: 仅限本地、无法远程访问
7.2 SSE 传输
SSE(Server-Sent Events)基于 HTTP,适合远程部署和 Web 场景。
TypeScript 实现:
import express from "express";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
const app = express();
let transport: SSEServerTransport;
app.get("/sse", async (req, res) => {
transport = new SSEServerTransport("/messages", res);
await server.connect(transport);
});
app.post("/messages", async (req, res) => {
await transport.handlePostMessage(req, res);
});
app.listen(3000, () => {
console.log("MCP SSE Server running on http://localhost:3000");
});
Python 实现:
# 使用内置的 SSE 传输
mcp.run(transport="sse", host="0.0.0.0", port=3000)
优点: 支持远程访问、可部署到服务器、兼容 Web 生态 缺点: 需要网络、需要处理认证和安全
7.3 如何选择?
| 场景 | 推荐传输方式 |
|---|---|
| Claude Desktop 本地集成 | Stdio |
| Cursor IDE 本地集成 | Stdio |
| 团队共享的 Server | SSE |
| 部署到云端 | SSE |
| Docker 容器 | SSE 或 Stdio |
8. 工具注册、资源暴露与 Prompt 模板
8.1 工具注册最佳实践
工具命名: 使用清晰的动词-名词格式,如 search-documents、create-user。
参数描述至关重要: LLM 依靠描述来理解参数含义。
// 好的描述
server.tool(
"search-knowledge",
"在知识库中搜索相关文档。当用户提问时使用此工具查找相关信息。",
{
query: z.string().describe("搜索查询文本,应为自然语言问题"),
maxResults: z
.number()
.optional()
.describe("返回结果数量上限,默认 5"),
},
handler
);
// 差的描述
server.tool(
"search", // 太简短
"搜索", // 缺乏上下文
{
q: z.string(), // 参数名不清晰
},
handler
);
错误处理: 工具应始终返回结构化结果,不要抛出异常。
server.tool(
"safe-tool",
"一个安全的工具示例",
{ input: z.string() },
async ({ input }) => {
try {
const result = await doSomething(input);
return {
content: [{ type: "text", text: JSON.stringify(result) }],
};
} catch (error) {
// 返回错误信息,而非抛出异常
return {
isError: true,
content: [
{
type: "text",
text: `操作失败:${error instanceof Error ? error.message : "未知错误"}`,
},
],
};
}
}
);
8.2 资源暴露
资源使用 URI 模式标识,支持动态 URI 参数:
// 静态资源
server.resource("app-config", "config://app", async (uri) => ({
contents: [
{ uri: uri.href, text: JSON.stringify(config), mimeType: "application/json" },
],
}));
// 带参数的动态资源
server.resource(
"user-profile",
new ResourceTemplate("users://{userId}/profile", { list: undefined }),
async (uri, { userId }) => {
const user = await getUser(userId);
return {
contents: [
{ uri: uri.href, text: JSON.stringify(user), mimeType: "application/json" },
],
};
}
);
8.3 Prompt 模板
Prompt 模板可以接受参数,返回一组结构化消息:
import { z } from "zod";
server.prompt(
"code-review",
"生成代码审查提示",
{
language: z.string().describe("编程语言"),
code: z.string().describe("待审查的代码"),
},
async ({ language, code }) => ({
messages: [
{
role: "user",
content: {
type: "text",
text: `请审查以下 ${language} 代码,关注:
1. 潜在的 bug 和边界情况
2. 性能问题
3. 代码风格和可读性
4. 安全隐患
\`\`\`${language}
${code}
\`\`\``,
},
},
],
})
);
9. 与 Claude Desktop / Cursor 集成测试
9.1 Claude Desktop 集成
编辑 Claude Desktop 的配置文件:
macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json
{
"mcpServers": {
"filesystem": {
"command": "node",
"args": ["/absolute/path/to/fs-mcp-server/build/index.js", "/Users/yourname/projects"],
"env": {}
},
"database": {
"command": "python",
"args": ["/absolute/path/to/db_server.py"],
"env": {
"DB_PATH": "/path/to/your/data.db"
}
}
}
}
注意: 所有路径必须是绝对路径。配置后重启 Claude Desktop 即可看到新增的工具。
9.2 Cursor IDE 集成
在项目根目录创建 .cursor/mcp.json:
{
"mcpServers": {
"filesystem": {
"command": "node",
"args": ["./build/index.js", "."]
}
}
}
在 Cursor Settings 中启用 MCP 后,Agent 模式下即可使用注册的工具。
9.3 验证连接
在 Claude Desktop 中,你应该能在对话框的工具图标处看到注册的工具列表。试着让 Claude 列出目录或读取文件,确认工具被正确调用。
如果工具没有出现,检查:
- 配置文件路径是否正确
- 命令路径是否为绝对路径
- 重启客户端
- 查看 stderr 日志输出
10. 调试技巧与常见问题
10.1 调试工具
MCP Inspector: 官方提供的可视化调试工具。
npx @modelcontextprotocol/inspector node build/index.js
它会在浏览器中打开一个调试界面,你可以:
- 查看已注册的工具、资源和提示
- 手动调用工具并查看返回结果
- 查看 JSON-RPC 消息日志
日志输出: MCP Server 的日志应输出到 stderr(而非 stdout),因为 stdout 被用于协议通信。
// TypeScript
console.error("[MCP] 工具被调用:", toolName);
# Python
import sys
print("[MCP] 工具被调用:", file=sys.stderr)
10.2 常见问题
Q1: 工具在 Claude Desktop 中不显示
- 检查配置文件中的路径是否为绝对路径
- 确认 JSON 格式正确(无尾逗号)
- 重启 Claude Desktop
Q2: 连接成功但调用工具超时
- 检查工具处理函数是否有异步问题
- 确认没有死锁(如在 Stdio 传输中执行了大量同步操作)
- 检查网络连接(SSE 模式)
Q3: 参数传递错误
- 确认 Zod schema 或 Python type hints 定义正确
- 检查参数名是否与 LLM 传递的一致
- 使用 MCP Inspector 手动测试参数
Q4: "Tool execution failed" 错误
- 查看 stderr 日志获取详细错误信息
- 确认工具函数正确返回了
{ content: [...] }格式 - 检查异常是否被捕获
Q5: SSE 模式连接断开
- 检查网络稳定性
- 实现重连逻辑
- 考虑增加心跳机制
10.3 调试工作流
1. 先用 MCP Inspector 测试工具是否正常注册和执行
2. 检查参数 schema 是否匹配实际输入
3. 在 stderr 中输出详细日志
4. 用 curl 测试 SSE 端点是否可达
5. 在客户端中观察错误消息
11. 生产部署方案
11.1 Stdio 模式的部署
Stdio 模式适用于桌面应用,部署通常由客户端自动管理。确保:
{
"scripts": {
"build": "tsc",
"start": "node build/index.js"
}
}
客户端配置中引用编译后的文件即可。
11.2 SSE 模式的部署
SSE 模式需要独立运行服务,适合部署到服务器或 Docker。
Dockerfile 示例:
FROM node:20-slim
WORKDIR /app
COPY package*.json ./
RUN npm ci --production
COPY build/ ./build/
EXPOSE 3000
CMD ["node", "build/index.js"]
Docker Compose:
version: "3.8"
services:
mcp-server:
build: .
ports:
- "3000:3000"
environment:
- NODE_ENV=production
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:3000/health"]
interval: 30s
timeout: 10s
retries: 3
11.3 安全考虑
- 认证:在 SSE 模式中添加 API Key 或 JWT 认证
app.use((req, res, next) => {
const apiKey = req.headers["x-api-key"];
if (apiKey !== process.env.MCP_API_KEY) {
return res.status(401).json({ error: "Unauthorized" });
}
next();
});
路径限制:文件系统 Server 必须限制访问范围(如前面代码中的路径穿越检查)
输入验证:对所有工具参数进行严格校验,防止 SQL 注入、路径穿越等攻击
速率限制:对 SSE 端点添加速率限制
CORS 配置:如果从浏览器访问,正确配置 CORS
11.4 监控与日志
// 结构化日志
function log(level: string, message: string, meta?: object) {
console.error(JSON.stringify({
timestamp: new Date().toISOString(),
level,
message,
...meta,
}));
}
建议在生产环境中:
- 记录所有工具调用及其耗时
- 监控错误率和响应时间
- 设置告警阈值
12. 总结与进阶资源
核心要点回顾
- MCP 是 AI 工具集成的标准化协议,基于 JSON-RPC 2.0,采用客户端-服务器架构
- 三大核心能力:Resources(数据暴露)、Tools(操作执行)、Prompts(模板引导)
- 两种 SDK:TypeScript 和 Python,各有优势
- 两种传输:Stdio(本地)和 SSE(远程),按需选择
- 生产部署需关注安全、认证、监控
进阶学习资源
- 官方文档:https://modelcontextprotocol.io
- TypeScript SDK 仓库:https://github.com/modelcontextprotocol/typescript-sdk
- Python SDK 仓库:https://github.com/modelcontextprotocol/python-sdk
- 官方 Server 仓库:https://github.com/modelcontextprotocol/servers(包含大量参考实现)
- MCP Inspector:https://github.com/modelcontextprotocol/inspector
下一步建议
- 从官方 Server 仓库中找到对你有用的现成 Server
- 基于本教程的代码,开发你自己的业务 MCP Server
- 在 Claude Desktop 或 Cursor 中测试你的 Server
- 将你的 Server 开源分享给社区
本教程由 AI 辅助生成,内容基于 MCP 协议公开文档整理,仅供学习参考。