MCP（Model Context Protocol）是Anthropic推出的开放协议，标准化了AI模型与外部工具、数据源的交互方式。本教程将带你从零构建一个完整的MCP服务器，深入掌握Tools、Resources、Prompts三大原语的实现，分别用TypeScript和Python两种主流SDK完成开发，并完成从调试、安全加固到Claude Code接入的全链路部署。
适用人群：有一定编程基础（Python或TypeScript）、已安装Claude Code CLI的开发者。
MCP采用经典的客户端-服务器（Client-Server）架构，基于JSON-RPC 2.0协议通信。它的核心价值在于：让任何AI客户端（Claude Code、Claude Desktop、Cursor等）能用统一的方式调用你的工具，就像USB-C接口统一了设备连接一样。
┌─────────────────┐ JSON-RPC 2.0 ┌─────────────────┐
│ MCP Client │◄──────────────────►│ MCP Server │
│ (Claude Code) │ stdio / HTTP │ (你的服务器) │
└─────────────────┘ └────────┬────────┘
│
┌───────────┼───────────┐
│ │ │
┌───┴──┐ ┌────┴────┐ ┌───┴───┐
│Tools │ │Resources│ │Prompts│
└──────┘ └─────────┘ └───────┘
| 原语 | 方向 | 核心特征 | 典型用途 |
|---|---|---|---|
| Tools | 模型→服务器（调用） | 有副作用，可执行操作 | 查询数据库、发送邮件、操作文件 |
| Resources | 模型→服务器（读取） | 只读，无副作用 | 获取文件内容、读取配置、查看状态 |
| Prompts | 用户→服务器（选择） | 模板化提示词 | 预设工作流、标准化交互模式 |
| 传输方式 | 适用场景 | 配置复杂度 | 多客户端 |
|---|---|---|---|
| stdio | 本地开发、Claude Code | 低 | 单客户端 |
| Streamable HTTP | 远程部署、多客户端 | 中 | 多客户端 |
graph TB
subgraph "本地模式 stdio"
CC1[Claude Code CLI] -->|stdin/stdout| MS1[MCP Server 进程]
end
subgraph "远程模式 HTTP"
CC2[Claude Code CLI] -->|HTTP POST| LB[负载均衡器]
LB --> MS2[MCP Server 实例1]
LB --> MS3[MCP Server 实例2]
end
style CC1 fill:#4A90D9,color:#fff
style CC2 fill:#4A90D9,color:#fff
style MS1 fill:#E67E22,color:#fff
style MS2 fill:#E67E22,color:#fff
style MS3 fill:#E67E22,color:#fff
# 检查Node.js版本（需要18+）
node --version
# 检查Python版本（需要3.10+）
python3 --version
# 确认Claude Code已安装
claude --version
TypeScript方案：
mkdir my-mcp-server && cd my-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
Python方案：
mkdir my-mcp-server && cd my-mcp-server
python3 -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install fastmcp
创建 tsconfig.json：
{
"compilerOptions": {
"target": "ES2022",
"module": "Node16",
"moduleResolution": "Node16",
"outDir": "./dist",
"strict": true,
"esModuleInterop": true
},
"include": ["src/**/*"]
}
在 package.json 中添加：
{
"type": "module",
"scripts": {
"build": "tsc",
"start": "node dist/index.js"
}
}
创建 src/index.ts：
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",
});
// 注册一个简单的Tool
server.tool(
"greet", // 工具名称
"向用户发送问候语", // 工具描述（重要！模型根据描述决定是否调用）
{ name: z.string().describe("用户姓名") }, // 输入Schema（zod）
async ({ name }) => { // 处理函数
return {
content: [{ type: "text", text: `你好，${name}！欢迎使用MCP服务器。` }],
};
}
);
// 启动服务器（stdio传输）
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("MCP服务器已启动"); // 注意：用stderr输出日志，stdout被协议占用
}
main().catch(console.error);
关键注意事项：
- console.log 会写入stdout，会被MCP协议占用，日志必须用 console.error 输出到stderr
- 工具描述（第二个参数）直接影响模型判断是否调用该工具，务必准确清晰
- zod schema中的 .describe() 会成为参数描述，帮助模型理解参数含义
编译并测试：
npm run build
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}}}' | node dist/index.js
预期输出类似：
{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-03-26","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"my-first-server","version":"1.0.0"}}}
创建 src/full-server.ts，包含Tools、Resources、Prompts的完整实现：
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: "project-assistant",
version: "1.0.0",
});
// ==========================================
// 一、Tools（工具）—— 模型可调用的函数
// ==========================================
// Tool 1: 项目健康检查
server.tool(
"health_check",
"检查项目健康状态，返回关键指标",
{
projectPath: z.string().describe("项目根目录路径"),
checks: z.array(z.enum(["dependencies", "tests", "lint", "build"]))
.optional()
.describe("要执行的检查项"),
},
async ({ projectPath, checks }) => {
// 模拟检查逻辑
const results = {
dependencies: { status: "healthy", details: "所有依赖已安装" },
tests: { status: "warning", details: "覆盖率 72%，建议提升至80%" },
lint: { status: "healthy", details: "无lint错误" },
build: { status: "healthy", details: "构建成功，耗时3.2s" },
};
const selectedChecks = checks ?? Object.keys(results) as string[];
const report = selectedChecks
.map((key) => {
const r = results[key as keyof typeof results];
const icon = r.status === "healthy" ? "✅" : r.status === "warning" ? "⚠️" : "❌";
return `${icon} ${key}: ${r.details}`;
})
.join("\n");
return {
content: [{ type: "text", text: `项目健康报告 (${projectPath}):\n\n${report}` }],
};
}
);
// Tool 2: 代码搜索（带分页）
server.tool(
"search_code",
"在项目中搜索代码片段",
{
query: z.string().describe("搜索关键词"),
filePattern: z.string().optional().describe("文件匹配模式，如 *.ts"),
maxResults: z.number().optional().default(10).describe("最大返回结果数"),
},
async ({ query, filePattern, maxResults }) => {
// 实际场景中这里调用grep/ripgrep等
const results = [
{ file: "src/index.ts", line: 15, snippet: `// ${query} 相关代码` },
{ file: "src/utils.ts", line: 42, snippet: `function process${query}() {}` },
].slice(0, maxResults);
return {
content: [{
type: "text",
text: results
.map((r) => `${r.file}:${r.line}\n ${r.snippet}`)
.join("\n\n"),
}],
};
}
);
// ==========================================
// 二、Resources（资源）—— 模型可读取的数据
// ==========================================
// Resource 1: 静态资源 — 项目配置
server.resource(
"project-config",
"config://project", // 资源URI
async (uri) => ({
contents: [{
uri: uri.href,
text: JSON.stringify({
name: "my-project",
version: "2.1.0",
framework: "Next.js",
nodeVersion: "20.x",
environment: "production",
}, null, 2),
}],
})
);
// Resource 2: 动态资源 — 系统状态
server.resource(
"system-status",
"status://system",
async (uri) => {
const status = {
uptime: process.uptime(),
memoryUsage: Math.round(process.memoryUsage().heapUsed / 1024 / 1024),
timestamp: new Date().toISOString(),
nodeVersion: process.version,
};
return {
contents: [{ uri: uri.href, text: JSON.stringify(status, null, 2) }],
};
}
);
// Resource 3: 模板资源 — 支持参数化URI
server.resource(
"user-profile",
new z.ZodTemplate( // 注意：实际用 resourceTemplate
"user://{userId}/profile"
),
async (uri, { userId }) => ({
contents: [{
uri: uri.href,
text: JSON.stringify({
id: userId,
name: `用户${userId}`,
role: "developer",
lastActive: new Date().toISOString(),
}, null, 2),
}],
})
);
// ==========================================
// 三、Prompts（提示词模板）
// ==========================================
// Prompt 1: 代码审查提示
server.prompt(
"code-review",
"生成代码审查提示词",
{
language: z.string().describe("编程语言"),
focus: z.enum(["security", "performance", "readability", "all"])
.default("all")
.describe("审查重点"),
code: z.string().describe("待审查的代码"),
},
async ({ language, focus, code }) => {
const focusMap = {
security: "重点关注安全漏洞：SQL注入、XSS、CSRF、敏感数据泄露等",
performance: "重点关注性能问题：时间复杂度、内存泄漏、不必要的计算等",
readability: "重点关注可读性：命名规范、代码结构、注释完整性等",
all: "全面审查：安全性、性能、可读性、最佳实践",
};
return {
messages: [{
role: "user",
content: {
type: "text",
text: `请对以下${language}代码进行${focus === 'all' ? '全面' : focus + '方面的'}审查。\n\n${focusMap[focus]}\n\n\`\`\`${language}\n${code}\n\`\`\`\n\n请给出：\n1. 发现的问题（按严重程度排列）\n2. 具体修改建议\n3. 修改后的代码片段`,
},
}],
};
}
);
// Prompt 2: 架构设计提示
server.prompt(
"architecture-design",
"生成系统架构设计提示词",
{
systemType: z.string().describe("系统类型，如微服务、单体、Serverless"),
requirements: z.string().describe("核心需求描述"),
constraints: z.string().optional().describe("约束条件"),
},
async ({ systemType, requirements, constraints }) => {
return {
messages: [{
role: "user",
content: {
type: "text",
text: `设计一个${systemType}架构方案。\n\n核心需求：${requirements}\n${constraints ? `约束条件：${constraints}\n` : ''}\n请包含：\n1. 架构图（用Mermaid描述）\n2. 核心组件及其职责\n3. 数据流设计\n4. 技术选型建议\n5. 扩展性考虑`,
},
}],
};
}
);
// 启动
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("✅ project-assistant MCP服务器已启动");
}
main().catch(console.error);
实际项目中，服务器通常需要管理状态。以下是一个带配置管理和缓存的服务器骨架：
// src/stateful-server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
// 服务器状态管理
interface ServerState {
cache: Map<string, { data: unknown; expiresAt: number }>;
config: Record<string, string>;
requestCount: number;
}
const state: ServerState = {
cache: new Map(),
config: {},
requestCount: 0,
};
// 缓存中间件
function withCache<T>(key: string, ttlMs: number, fetcher: () => Promise<T>): Promise<T> {
return (async () => {
const cached = state.cache.get(key);
if (cached && Date.now() < cached.expiresAt) {
return cached.data as T;
}
const data = await fetcher();
state.cache.set(key, { data, expiresAt: Date.now() + ttlMs });
return data;
})();
}
const server = new McpServer({
name: "stateful-assistant",
version: "1.0.0",
});
// 带缓存的API查询工具
server.tool(
"query_api",
"查询外部API并缓存结果",
{
endpoint: z.string().describe("API端点路径"),
cacheTtl: z.number().optional().default(60000).describe("缓存时间（毫秒）"),
},
async ({ endpoint, cacheTtl }) => {
state.requestCount++;
const data = await withCache(endpoint, cacheTtl, async () => {
// 实际场景中这里调用fetch
return { endpoint, result: "cached_response", queriedAt: new Date().toISOString() };
});
return {
content: [{
type: "text",
text: JSON.stringify({ data, totalRequests: state.requestCount }, null, 2),
}],
};
}
);
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("Stateful server running");
}
main().catch(console.error);
Python生态中，fastmcp（原名 mcp 官方SDK）是最主流的MCP开发框架。它的优势：
创建 server.py：
from fastmcp import FastMCP
# 创建服务器实例
mcp = FastMCP("my-first-python-server")
# 注册Tool — 装饰器风格
@mcp.tool()
def greet(name: str) -> str:
"""向用户发送问候语"""
return f"你好，{name}！欢迎使用Python MCP服务器。"
# 启动（默认stdio传输）
if __name__ == "__main__":
mcp.run()
FastMCP的魔法：注意函数的docstring会自动成为工具描述，类型注解会自动成为参数schema。你不需要手写任何JSON Schema。
运行：
# 直接运行（stdio模式）
python server.py
# 或使用fastmcp CLI
fastmcp run server.py
创建 project_assistant.py：
"""项目管理助手 MCP服务器 - 完整三原语示例"""
from fastmcp import FastMCP
from typing import Optional
from pydantic import BaseModel
mcp = FastMCP(
"project-assistant",
version="1.0.0",
)
# ==========================================
# 数据模型
# ==========================================
class HealthReport(BaseModel):
project_path: str
status: str
details: list[str]
# 模拟数据库
_projects_db = {
"frontend": {"name": "Web Frontend", "language": "TypeScript", "framework": "React"},
"backend": {"name": "API Server", "language": "Python", "framework": "FastAPI"},
"mobile": {"name": "Mobile App", "language": "Kotlin", "framework": "Compose"},
}
_cache: dict[str, tuple[object, float]] = {}
# ==========================================
# 一、Tools（工具）
# ==========================================
@mcp.tool()
def list_projects(tech_stack: Optional[str] = None) -> str:
"""列出所有项目，可按技术栈过滤
Args:
tech_stack: 可选的技术栈过滤条件，如 Python、TypeScript
"""
projects = list(_projects_db.values())
if tech_stack:
tech_lower = tech_stack.lower()
projects = [
p for p in projects
if tech_lower in p["language"].lower() or tech_lower in p["framework"].lower()
]
if not projects:
return f"未找到使用 {tech_stack} 的项目"
return "\n".join(
f"- {p['name']} ({p['language']} / {p['framework']})"
for p in projects
)
@mcp.tool()
async def health_check(project_path: str, checks: Optional[list[str]] = None) -> str:
"""检查项目健康状态
Args:
project_path: 项目根目录路径
checks: 要执行的检查项列表，支持 dependencies/tests/lint/build
"""
import asyncio
available_checks = ["dependencies", "tests", "lint", "build"]
selected = checks or available_checks
# 模拟异步检查
results = {}
for check in selected:
await asyncio.sleep(0.01) # 模拟IO
results[check] = {
"dependencies": ("✅", "所有依赖已安装"),
"tests": ("⚠️", "覆盖率 72%，建议提升至80%"),
"lint": ("✅", "无lint错误"),
"build": ("✅", "构建成功，耗时3.2s"),
}.get(check, ("❓", "未知检查项"))
report_lines = [f"{icon} {name}: {detail}" for name, (icon, detail) in results.items()]
return f"项目健康报告 ({project_path}):\n\n" + "\n".join(report_lines)
@mcp.tool()
def deploy_project(
project_name: str,
environment: str = "staging",
dry_run: bool = True,
) -> str:
"""部署项目到指定环境
Args:
project_name: 项目名称
environment: 目标环境 (staging/production)
dry_run: 是否为预演模式（不实际执行）
"""
if project_name not in _projects_db:
return f"❌ 项目 '{project_name}' 不存在"
mode = "预演" if dry_run else "实际部署"
proj = _projects_db[project_name]
return (
f"🚀 {mode}部署计划\n"
f"项目: {proj['name']}\n"
f"环境: {environment}\n"
f"技术栈: {proj['language']} / {proj['framework']}\n"
f"状态: {'计划已生成（未实际执行）' if dry_run else '部署已触发'}"
)
# ==========================================
# 二、Resources（资源）
# ==========================================
@mcp.resource("config://project")
def get_project_config() -> str:
"""获取项目全局配置"""
import json
config = {
"version": "2.1.0",
"defaultEnvironment": "staging",
"maxDeployRetries": 3,
"notificationEmail": "[email protected]",
"features": {
"autoRollback": True,
"canaryDeployment": True,
"blueGreenSwitch": False,
},
}
return json.dumps(config, indent=2, ensure_ascii=False)
@mcp.resource("status://system")
def get_system_status() -> str:
"""获取系统实时状态"""
import json
import platform
import psutil # pip install psutil
status = {
"hostname": platform.node(),
"python": platform.python_version(),
"platform": platform.system(),
"cpuPercent": psutil.cpu_percent(),
"memoryPercent": psutil.virtual_memory().percent,
"diskPercent": psutil.disk_usage("/").percent,
}
return json.dumps(status, indent=2, ensure_ascii=False)
@mcp.resource("project://{project_name}/info")
def get_project_info(project_name: str) -> str:
"""获取指定项目的详细信息（模板资源）"""
import json
if project_name not in _projects_db:
return json.dumps({"error": f"项目 '{project_name}' 不存在"}, ensure_ascii=False)
return json.dumps(_projects_db[project_name], indent=2, ensure_ascii=False)
# ==========================================
# 三、Prompts（提示词模板）
# ==========================================
@mcp.prompt()
def code_review(code: str, language: str = "python", focus: str = "all") -> str:
"""代码审查提示词模板
Args:
code: 待审查的代码
language: 编程语言
focus: 审查重点 (security/performance/readability/all)
"""
focus_instructions = {
"security": "重点关注安全漏洞：SQL注入、XSS、CSRF、敏感数据泄露",
"performance": "重点关注性能问题：时间复杂度、内存泄漏、不必要的计算",
"readability": "重点关注可读性：命名规范、代码结构、注释完整性",
"all": "全面审查：安全性、性能、可读性、最佳实践",
}
return (
f"请对以下{language}代码进行审查。\n\n"
f"审查重点：{focus_instructions.get(focus, focus_instructions['all'])}\n\n"
f"```{language}\n{code}\n```\n\n"
f"请给出：\n"
f"1. 发现的问题（按严重程度排列）\n"
f"2. 具体修改建议\n"
f"3. 修改后的代码片段"
)
@mcp.prompt()
def architecture_design(
system_type: str,
requirements: str,
constraints: str = "",
) -> str:
"""系统架构设计提示词模板
Args:
system_type: 系统类型（微服务/单体/Serverless等）
requirements: 核心需求描述
constraints: 约束条件
"""
constraint_text = f"\n约束条件：{constraints}" if constraints else ""
return (
f"设计一个{system_type}架构方案。\n\n"
f"核心需求：{requirements}{constraint_text}\n\n"
f"请包含：\n"
f"1. 架构图（用Mermaid描述）\n"
f"2. 核心组件及其职责\n"
f"3. 数据流设计\n"
f"4. 技术选型建议\n"
f"5. 扩展性考虑\n"
f"6. 容错与降级策略"
)
@mcp.prompt()
def debugging_session(
error_message: str,
code_context: str = "",
steps_reproduced: str = "",
) -> str:
"""调试辅助提示词模板
Args:
error_message: 错误信息
code_context: 相关代码片段
steps_reproduced: 已尝试的排查步骤
"""
parts = [f"我遇到了以下错误，请帮我分析原因和解决方案：\n\n错误信息：\n```\n{error_message}\n```"]
if code_context:
parts.append(f"\n相关代码：\n```\n{code_context}\n```")
if steps_reproduced:
parts.append(f"\n已尝试的排查步骤：\n{steps_reproduced}")
parts.append(
"\n请提供：\n"
"1. 错误原因分析\n"
"2. 可能的解决方案（按推荐优先级排列）\n"
"3. 预防类似问题的建议"
)
return "".join(parts)
# ==========================================
# 启动
# ==========================================
if __name__ == "__main__":
mcp.run()
对于远程部署场景，切换到HTTP传输只需改一行：
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)
或通过CLI启动：
fastmcp run server.py --transport streamable-http --port 8080
Claude Code支持三种方式注册MCP服务器，推荐按场景选择：
graph LR
A["注册MCP服务器"] --> B{使用范围?}
B -->|当前项目| C[".mcp.json<br>项目根目录"]
B -->|个人全局| D["claude mcp add<br>命令行注册"]
B -->|团队共享| E["~/.claude.json<br>用户级配置"]
style C fill:#27AE60,color:#fff
style D fill:#2980B9,color:#fff
style E fill:#8E44AD,color:#fff
.mcp.json（推荐）在项目根目录创建 .mcp.json：
{
"mcpServers": {
"project-assistant-ts": {
"type": "stdio",
"command": "node",
"args": ["/absolute/path/to/my-mcp-server/dist/index.js"],
"env": {
"NODE_ENV": "production"
}
},
"project-assistant-py": {
"type": "stdio",
"command": "/absolute/path/to/my-mcp-server/.venv/bin/python",
"args": ["/absolute/path/to/my-mcp-server/project_assistant.py"],
"env": {
"PYTHONUNBUFFERED": "1"
}
}
}
}
# 注册TypeScript服务器
claude mcp add project-assistant-ts -- node /path/to/dist/index.js
# 注册Python服务器
claude mcp add project-assistant-py -- /path/to/.venv/bin/python /path/to/server.py
# 注册远程HTTP服务器
claude mcp add --transport http remote-server https://api.example.com/mcp
# 查看已注册的服务器
claude mcp list
# 移除服务器
claude mcp remove project-assistant-ts
编辑 ~/.claude.json：
{
"mcpServers": {
"my-global-tools": {
"type": "stdio",
"command": "node",
"args": ["/path/to/global-mcp-server/dist/index.js"]
}
}
}
# 在项目目录中启动Claude Code
claude
# 在交互界面中输入测试指令
> 请列出所有可用的MCP工具
# 或直接使用工具
> 用health_check工具检查 ./src 目录的健康状态
MCP Inspector是官方调试工具，提供可视化界面直接测试服务器：
# 方法一：npx直接启动（推荐）
npx @modelcontextprotocol/inspector node dist/index.js
# 方法二：Python服务器
npx @modelcontextprotocol/inspector python server.py
# 方法三：带环境变量
npx @modelcontextprotocol/inspector \
-e API_KEY=your_key \
-- node dist/index.js
Inspector会在浏览器中打开一个界面，你可以： - 查看、调用所有注册的Tools - 浏览和读取Resources - 测试Prompts模板 - 查看JSON-RPC请求和响应的原始日志
sequenceDiagram
participant Dev as 开发者
participant Insp as MCP Inspector
participant Server as MCP Server
Dev->>Insp: npx @modelcontextprotocol/inspector
Insp->>Server: initialize (JSON-RPC)
Server-->>Insp: capabilities + serverInfo
Insp->>Server: tools/list
Server-->>Insp: 工具列表
Insp->>Server: tools/call {name: "health_check", ...}
Server-->>Insp: 执行结果
Insp-->>Dev: 可视化展示结果
| 问题 | 症状 | 解决方案 |
|---|---|---|
| 服务器无法启动 | claude mcp list 显示error |
检查命令路径是否为绝对路径；检查Node/Python版本 |
| 工具不显示 | 服务器连接成功但无工具 | 确认 server.tool() 注册在 connect() 之前 |
| stdout污染 | 协议解析错误 | 确保日志只用stderr（console.error / sys.stderr） |
| zod验证失败 | 工具调用报参数错误 | 检查schema定义和实际传入参数是否匹配 |
| 超时 | 工具执行时间过长 | 考虑异步处理或分批返回结果 |
不依赖Inspector，直接用管道测试stdio服务器：
# 发送初始化请求
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}' | node dist/index.js
# 发送工具列表请求（需要先initialize）
(echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}'; echo '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}') | node dist/index.js
graph TB
subgraph "MCP安全层次"
S1["第1层：输入验证<br>所有输入都不可信"]
S2["第2层：权限控制<br>最小权限原则"]
S3["第3层：输出过滤<br>不泄露敏感信息"]
S4["第4层：传输安全<br>加密+认证"]
end
S1 --> S2 --> S3 --> S4
style S1 fill:#E74C3C,color:#fff
style S2 fill:#E67E22,color:#fff
style S3 fill:#F1C40F,color:#333
style S4 fill:#27AE60,color:#fff
TypeScript示例：
import { z } from "zod";
import path from "path";
// 路径遍历攻击防护
const safePath = z.string().refine(
(p) => {
const resolved = path.resolve(p);
return resolved.startsWith(process.cwd());
},
{ message: "路径不允许超出项目目录" }
);
server.tool(
"read_file",
"读取项目内文件",
{ filePath: safePath.describe("文件路径（项目内）") },
async ({ filePath }) => {
// 安全：filePath已被验证在项目目录内
const content = await fs.readFile(filePath, "utf-8");
return { content: [{ type: "text", text: content }] };
}
);
Python示例：
import os
from pathlib import Path
def validate_path(file_path: str) -> Path:
"""验证路径在允许的目录内"""
allowed_root = Path(os.getenv("ALLOWED_ROOT", ".")).resolve()
resolved = Path(file_path).resolve()
if not str(resolved).startswith(str(allowed_root)):
raise ValueError(f"路径 '{file_path}' 超出允许范围")
return resolved
@mcp.tool()
def safe_read_file(file_path: str) -> str:
"""安全读取文件（限制在允许目录内）"""
try:
validated = validate_path(file_path)
return validated.read_text()
except ValueError as e:
return f"安全错误：{e}"
except FileNotFoundError:
return f"文件不存在：{file_path}"
// 不要在工具描述中暴露内部实现细节
server.tool(
"query_database",
"查询项目数据库", // ✅ 通用描述
// "直接执行SQL查询project_db", // ❌ 暴露数据库名称
{ query: z.string() },
async ({ query }) => {
// ✅ 过滤敏感字段
const result = await executeQuery(query);
const sanitized = result.map(row => {
const { password, api_key, secret, ...safe } = row;
return safe;
});
return { content: [{ type: "text", text: JSON.stringify(sanitized) }] };
}
);
以下是一个完整的、可直接部署的实战项目——API文档助手，它能自动扫描项目中的API端点、生成文档、校验接口一致性。
api-doc-assistant/
├── package.json
├── tsconfig.json
├── src/
│ └── index.ts # 主服务器
├── .mcp.json # Claude Code配置
└── README.md
src/index.ts：
#!/usr/bin/env node
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import fs from "fs/promises";
import path from "path";
// ==========================================
// 数据模型
// ==========================================
interface ApiEndpoint {
method: string;
path: string;
description: string;
parameters: string[];
responseExample?: string;
}
// ==========================================
// 服务器实例
// ==========================================
const server = new McpServer({
name: "api-doc-assistant",
version: "1.0.0",
});
// ==========================================
// 工具：扫描API端点
// ==========================================
server.tool(
"scan_endpoints",
"扫描项目中的API端点定义（支持Express/Fastify/Koa路由）",
{
projectPath: z.string().describe("项目根目录的绝对路径"),
framework: z.enum(["express", "fastify", "koa"]).optional().describe("Web框架类型"),
},
async ({ projectPath, framework }) => {
try {
// 安全校验：确保路径存在
const stat = await fs.stat(projectPath);
if (!stat.isDirectory()) {
return { content: [{ type: "text", text: `❌ ${projectPath} 不是目录` }] };
}
// 模拟扫描路由文件（实际实现需根据框架解析路由）
const routesDir = path.join(projectPath, "src", "routes");
let endpoints: ApiEndpoint[] = [];
try {
const files = await fs.readdir(routesDir);
for (const file of files.filter(f => f.endsWith(".ts") || f.endsWith(".js"))) {
const content = await fs.readFile(path.join(routesDir, file), "utf-8");
// 简单的正则匹配路由定义
const routeRegex = /(?:app|router)\.(get|post|put|delete|patch)\s*\(\s*['"`]([^'"`\s]+)/g;
let match;
while ((match = routeRegex.exec(content)) !== null) {
endpoints.push({
method: match[1].toUpperCase(),
path: match[2],
description: `来自 ${file}`,
parameters: [],
});
}
}
} catch {
endpoints = [
{ method: "GET", path: "/api/health", description: "健康检查端点", parameters: [] },
{ method: "GET", path: "/api/users", description: "用户列表", parameters: ["page", "limit"] },
{ method: "POST", path: "/api/users", description: "创建用户", parameters: ["name", "email"] },
{ method: "GET", path: "/api/users/:id", description: "获取用户详情", parameters: ["id"] },
];
}
const summary = endpoints
.map((e) => `${e.method.padEnd(7)} ${e.path.padEnd(25)} ${e.description}`)
.join("\n");
return {
content: [{
type: "text",
text: `📊 扫描结果（共 ${endpoints.length} 个端点）\n\n${"METHOD".padEnd(7)} ${"PATH".padEnd(25)} DESCRIPTION\n${"─".repeat(60)}\n${summary}`,
}],
};
} catch (error) {
return {
content: [{ type: "text", text: `❌ 扫描失败：${(error as Error).message}` }],
};
}
}
);
// ==========================================
// 工具：生成API文档
// ==========================================
server.tool(
"generate_docs",
"为扫描到的API端点生成Markdown文档",
{
format: z.enum(["markdown", "openapi"]).default("markdown").describe("输出格式"),
title: z.string().optional().default("API Documentation").describe("文档标题"),
},
async ({ format, title }) => {
if (format === "openapi") {
const spec = {
openapi: "3.1.0",
info: { title, version: "1.0.0" },
paths: {
"/api/health": { get: { summary: "健康检查", responses: { "200": { description: "OK" } } } },
"/api/users": {
get: { summary: "用户列表", parameters: [{ name: "page", in: "query" }] },
post: { summary: "创建用户", requestBody: { content: { "application/json": {} } } },
},
},
};
return { content: [{ type: "text", text: JSON.stringify(spec, null, 2) }] };
}
const markdown = `# ${title}
## 端点列表
### GET /api/health
健康检查端点。
**响应**：
\`\`\`json
{ "status": "ok", "uptime": 12345 }
\`\`\`
### GET /api/users
获取用户列表。
**参数**：
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| page | number | 否 | 页码，默认1 |
| limit | number | 否 | 每页数量，默认20 |
**响应**：
\`\`\`json
{ "users": [...], "total": 100, "page": 1 }
\`\`\`
### POST /api/users
创建新用户。
**请求体**：
\`\`\`json
{ "name": "string", "email": "string" }
\`\`\`
---
*由 api-doc-assistant MCP 服务器自动生成*
`;
return { content: [{ type: "text", text: markdown }] };
}
);
// ==========================================
// 工具：校验API一致性
// ==========================================
server.tool(
"validate_consistency",
"校验API实现与文档的一致性",
{
projectPath: z.string().describe("项目根目录"),
},
async ({ projectPath }) => {
// 模拟校验逻辑
const issues = [
{ severity: "warning", endpoint: "POST /api/users", issue: "文档中缺少 request body schema" },
{ severity: "error", endpoint: "GET /api/users/:id", issue: "路由参数 :id 类型未定义" },
{ severity: "info", endpoint: "DELETE /api/users/:id", issue: "端点已实现但未在文档中记录" },
];
const severityIcon = { error: "❌", warning: "⚠️", info: "ℹ️" };
const report = issues
.map((i) => `${severityIcon[i.severity]} [${i.severity.toUpperCase()}] ${i.endpoint}: ${i.issue}`)
.join("\n");
const stats = `统计：${issues.filter(i => i.severity === "error").length} 错误, ${issues.filter(i => i.severity === "warning").length} 警告, ${issues.filter(i => i.severity === "info").length} 提示`;
return {
content: [{ type: "text", text: `📋 API一致性校验报告\n\n${report}\n\n${stats}` }],
};
}
);
// ==========================================
// 资源：API统计
// ==========================================
server.resource(
"api-stats",
"stats://api",
async (uri) => ({
contents: [{
uri: uri.href,
text: JSON.stringify({
totalEndpoints: 4,
methods: { GET: 3, POST: 1, PUT: 0, DELETE: 0 },
documentedEndpoints: 3,
coveragePercent: 75,
lastScanned: new Date().toISOString(),
}, null, 2),
}],
})
);
// ==========================================
// 提示词：API审查
// ==========================================
server.prompt(
"api-review",
"API设计审查提示词",
{
apiSpec: z.string().describe("API规范（OpenAPI JSON或Markdown描述）"),
reviewFocus: z.enum(["restful", "security", "versioning", "all"]).default("all"),
},
async ({ apiSpec, reviewFocus }) => ({
messages: [{
role: "user",
content: {
type: "text",
text: `请审查以下API设计。\n\n重点：${reviewFocus}\n\n${apiSpec}\n\n请从以下维度评审：\n1. RESTful设计规范\n2. 安全性（认证、授权、输入验证）\n3. 版本管理策略\n4. 错误处理一致性\n5. 分页与过滤设计`,
},
}],
})
);
// ==========================================
// 启动
// ==========================================
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("✅ api-doc-assistant MCP服务器已启动");
}
main().catch(console.error);
# 编译TypeScript
npx tsc
# 本地测试
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}' | node dist/index.js
# 注册到Claude Code
claude mcp add api-doc-assistant -- node /absolute/path/to/api-doc-assistant/dist/index.js
flowchart TD
A["需求分析<br>确定MCP服务器功能边界"] --> B["选择技术栈<br>TypeScript or Python"]
B --> C["项目初始化<br>安装SDK和依赖"]
C --> D["实现三原语<br>Tools / Resources / Prompts"]
D --> E["本地调试<br>MCP Inspector测试"]
E --> F{功能正确?}
F -->|否| G["修复问题<br>检查日志和错误"]
G --> E
F -->|是| H["安全审查<br>输入验证/敏感信息过滤"]
H --> I["注册到Claude Code<br>.mcp.json 或 claude mcp add"]
I --> J["集成测试<br>在Claude Code中实际使用"]
J --> K{满意?}
K -->|否| L["优化改进<br>调整描述/增加功能"]
L --> E
K -->|是| M["生产部署<br>Docker/远程HTTP"]
style A fill:#3498DB,color:#fff
style D fill:#E67E22,color:#fff
style E fill:#9B59B6,color:#fff
style H fill:#E74C3C,color:#fff
style I fill:#27AE60,color:#fff
style M fill:#2C3E50,color:#fff
| 维度 | TypeScript (@modelcontextprotocol/sdk) |
Python (fastmcp) |
|---|---|---|
| 安装 | npm install @modelcontextprotocol/sdk zod |
pip install fastmcp |
| Schema定义 | zod对象（手动构造） | 自动从类型注解推断 |
| 工具描述 | 字符串参数 | 函数docstring |
| 异步支持 | 原生Promise/async | 原生async/await |
| 类型安全 | 编译时检查 | 运行时Pydantic校验 |
| 生态集成 | Node.js生态、前端工具链 | ML/数据科学生态 |
| 适合场景 | 前端/DevOps工具、已有TS项目 | 数据分析、AI/ML工具、快速原型 |
| 调试 | MCP Inspector + VS Code | MCP Inspector + IDE断点 |
问题：注册了工具，但Claude就是不调用。
根因：工具描述太模糊或有歧义，模型无法判断是否匹配用户意图。
解决：
// ❌ 差的描述
server.tool("search", "搜索", ...)
// ✅ 好的描述：说明做什么、什么时候该用、输入输出
server.tool(
"search_code",
"在项目源代码中搜索关键词。当用户要求查找代码、搜索函数定义、定位某个功能的实现位置时使用此工具。",
...
)
问题：Python服务器运行时出现 JSON parse error。
根因：print() 写入stdout，与MCP的JSON-RPC消息混在一起。
解决：
import sys
# ❌ 永远不要用print()
print("调试信息")
# ✅ 写入stderr
print("调试信息", file=sys.stderr)
# ✅ 或使用logging
import logging
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger("mcp-server")
logger.info("这条日志安全地输出到stderr")
问题：在Windows上注册的服务器在macOS上报错。
根因：路径分隔符和可执行文件名不同。
解决：
// .mcp.json - 使用平台无关的写法
{
"mcpServers": {
"my-server": {
"command": "node",
"args": ["./dist/index.js"],
"env": {}
}
}
}
# Windows上用python命令可能指向Python 2，建议明确路径
claude mcp add my-py-server -- C:\Python311\python.exe C:\path\to\server.py
# macOS/Linux上使用虚拟环境的绝对路径
claude mcp add my-py-server -- /abs/path/.venv/bin/python /abs/path/server.py
MCP工具的响应会直接进入模型的上下文窗口。过大的响应会浪费token并降低响应质量。
// ❌ 返回完整的大型JSON
return { content: [{ type: "text", text: JSON.stringify(hugeDataset) }] };
// ✅ 返回摘要 + 提示用户按需获取详情
return {
content: [{
type: "text",
text: `找到 ${results.length} 条记录（已截断显示前10条）：\n${formatResults(results.slice(0, 10))}\n\n如需查看更多，请指定范围或过滤条件。`,
}],
};
import time
from functools import wraps
def cache_result(ttl_seconds: int = 300):
"""简单的TTL缓存装饰器"""
cache: dict = {}
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
key = f"{func.__name__}:{args}:{kwargs}"
if key in cache:
result, timestamp = cache[key]
if time.time() - timestamp < ttl_seconds:
return result
result = await func(*args, **kwargs)
cache[key] = (result, time.time())
return result
return wrapper
return decorator
@mcp.tool()
@cache_result(ttl_seconds=60)
async def query_project_stats(project_path: str) -> str:
"""查询项目统计信息（结果缓存60秒）"""
# 耗时操作...
return "统计数据..."
| 训练营类型 | 适用性 | 建议时长 | 侧重点 |
|---|---|---|---|
| AI工具开发入门营 | ⭐⭐⭐⭐⭐ | 4小时 | Python FastMCP路线，从最小服务器到三原语 |
| 全栈工程师进阶营 | ⭐⭐⭐⭐⭐ | 6小时 | TypeScript完整路线 + 实战案例 |
| 企业AI集成实战营 | ⭐⭐⭐⭐ | 8小时 | 安全加固 + 远程部署 + 多客户端方案 |
| Claude Code深度训练营 | ⭐⭐⭐⭐⭐ | 3小时 | 快速集成 + 调试技巧 + 常见问题 |
| DevOps自动化营 | ⭐⭐⭐ | 4小时 | CI/CD集成 + 监控 + 日志 |
推荐教学路径：Python FastMCP入门（1h）→ TypeScript深度开发（2h）→ 调试与安全（1h）→ 完整实战项目（2h）
[!tip] 51学通信知识星球 星主：爱卫生（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 星球定位：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 当前规模：5200 篇主题 / 990 个回答，198 元/年。 加入福利：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 AI 加持：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。