MCP(Model Context Protocol)正在成为 AI 工具生态的事实标准——截至 2026 年 6 月,GitHub 上已有超过 8000 个 MCP Server 仓库,Claude、Cursor、Windsurf 等主流 AI 客户端全部原生支持 MCP 协议。如果你的工具或服务还没有暴露 MCP 接口,等于主动放弃了被 AI Agent 调用的机会。本文将从协议原理到生产级实现,手把手教你构建一个功能完整的 MCP Server。
🏗️ 一、MCP 协议核心原理
1.1 协议架构:JSON-RPC 2.0 over 双传输层
MCP 的设计哲学是「一个协议,多种传输」。底层通信基于 JSON-RPC 2.0,但传输层支持两种模式:
| 特性 | stdio 模式 | SSE 模式 |
|---|---|---|
| 通信方式 | 标准输入/输出流 | HTTP Server-Sent Events |
| 适用场景 | 本地 CLI 工具、IDE 插件 | 远程服务、Web 应用 |
| 部署复杂度 | ⭐ 低(直接 spawn 进程) | ⭐⭐⭐ 中(需要 HTTP 服务器) |
| 多客户端支持 | ❌ 单客户端独占 | ✅ 天然支持多客户端 |
| 网络穿透 | ❌ 仅限本地 | ✅ 可跨网络 |
| 典型代表 | 文件系统工具、Git 操作 | 数据库查询、API 网关 |
💡 提示: 如果你的工具是「本地优先」的(如操作本地文件、访问本地数据库),用 stdio;如果需要被远程调用或服务多个客户端,用 SSE。
1.2 MCP Server 三大能力支柱
MCP Server 通过三个核心原语向客户端暴露能力:
工具(Tools)——可被 AI 调用的函数,有明确的输入参数和返回结果。这是最常用的能力,类似于 REST API 的 endpoint。
资源(Resources)——可被读取的数据源,用 URI 标识。类似于文件系统中的文件,客户端可以列出和读取。
提示(Prompts)——预定义的提示模板,客户端可以选择性地注入到对话中。
一个完整的 MCP Server 启动握手流程如下:
Client → Server: initialize (协议版本、客户端能力)
Server → Client: initialize result (服务器能力声明)
Client → Server: initialized (确认通知)
Client → Server: tools/list (发现可用工具)
Server → Client: tools/list result (工具定义列表)
Client → Server: tools/call (调用某个工具)
Server → Client: tools/call result (工具执行结果)
🔧 二、从零实现 MCP Server(TypeScript)
2.1 项目初始化与依赖
我们用官方的 @modelcontextprotocol/sdk 来实现,这是目前最成熟的 MCP SDK:
# 初始化项目
mkdir mcp-demo-server && cd mcp-demo-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
npx tsc --init
⚠️ 警告:
@modelcontextprotocol/sdk要求 Node.js >= 18。如果你的环境是 Node 16,需要先升级。推荐使用 Node 20 LTS。
2.2 stdio 模式:本地工具服务器
下面实现一个实用的「JSON 处理工具集」——这正是 jsjson.com 这类工具站可以暴露给 AI Agent 的能力:
// src/stdio-server.ts
// stdio 模式 MCP Server — JSON 处理工具集
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: "json-tools",
version: "1.0.0",
});
// 工具 1:JSON 格式化
server.tool(
"json_format",
"将 JSON 字符串格式化为可读的缩进格式",
{
input: z.string().describe("待格式化的 JSON 字符串"),
indent: z.number().default(2).describe("缩进空格数,默认 2"),
},
async ({ input, indent }) => {
try {
const parsed = JSON.parse(input);
const formatted = JSON.stringify(parsed, null, indent);
return {
content: [{ type: "text", text: formatted }],
};
} catch (err) {
return {
content: [{ type: "text", text: `JSON 解析错误: ${(err as Error).message}` }],
isError: true,
};
}
}
);
// 工具 2:JSON 压缩
server.tool(
"json_minify",
"将 JSON 字符串压缩为最小体积(去除所有空白)",
{
input: z.string().describe("待压缩的 JSON 字符串"),
},
async ({ input }) => {
try {
const parsed = JSON.parse(input);
const minified = JSON.stringify(parsed);
const ratio = ((1 - minified.length / input.length) * 100).toFixed(1);
return {
content: [{
type: "text",
text: `压缩结果 (${ratio}% 体积减少):\n${minified}`,
}],
};
} catch (err) {
return {
content: [{ type: "text", text: `JSON 解析错误: ${(err as Error).message}` }],
isError: true,
};
}
}
);
// 工具 3:JSON Schema 生成
server.tool(
"json_to_schema",
"从 JSON 样本数据自动生成 JSON Schema",
{
input: z.string().describe("JSON 样本数据"),
title: z.string().optional().describe("Schema 标题"),
},
async ({ input, title }) => {
try {
const data = JSON.parse(input);
const schema = generateSchema(data, title || "GeneratedSchema");
return {
content: [{ type: "text", text: JSON.stringify(schema, null, 2) }],
};
} catch (err) {
return {
content: [{ type: "text", text: `错误: ${(err as Error).message}` }],
isError: true,
};
}
}
);
// JSON Schema 生成辅助函数
function generateSchema(data: any, title: string): object {
if (data === null) return { type: "null", title };
if (Array.isArray(data)) {
return {
type: "array",
title,
items: data.length > 0 ? generateSchema(data[0], "Item") : {},
};
}
if (typeof data === "object") {
const properties: Record<string, object> = {};
const required: string[] = [];
for (const [key, value] of Object.entries(data)) {
properties[key] = generateSchema(value, key);
required.push(key);
}
return { type: "object", title, properties, required };
}
const typeMap: Record<string, string> = {
string: "string",
number: "number",
boolean: "boolean",
};
return { type: typeMap[typeof data] || "string", title };
}
// 启动 stdio 传输
const transport = new StdioServerTransport();
await server.connect(transport);
将这个文件配置到 MCP 客户端(如 Claude Desktop)的 mcpServers 配置中:
{
"mcpServers": {
"json-tools": {
"command": "npx",
"args": ["tsx", "src/stdio-server.ts"]
}
}
}
📌 记住: stdio 模式下,MCP Server 的日志绝对不能输出到 stdout,否则会污染 JSON-RPC 通信信道。日志应输出到 stderr:
console.error("log message")。
2.3 SSE 模式:远程可访问的工具服务
当你的工具需要被远程调用时,SSE 模式是正确选择。下面实现一个基于 Express 的 SSE 服务器:
// src/sse-server.ts
// SSE 模式 MCP Server — 支持远程多客户端访问
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from "express";
const app = express();
const server = new McpServer({
name: "remote-tools",
version: "1.0.0",
});
// 注册一个需要网络请求的工具
server.tool(
"http_request",
"发送 HTTP 请求并返回响应状态和头部信息(不返回 body,避免注入风险)",
{
url: z.string().url().describe("目标 URL"),
method: z.enum(["GET", "HEAD"]).default("GET").describe("请求方法"),
},
async ({ url, method }) => {
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 10000);
try {
const resp = await fetch(url, { method, signal: controller.signal });
const headers = Object.fromEntries(resp.headers.entries());
return {
content: [{
type: "text",
text: JSON.stringify({
status: resp.status,
statusText: resp.statusText,
headers,
}, null, 2),
}],
};
} catch (err) {
return {
content: [{ type: "text", text: `请求失败: ${(err as Error).message}` }],
isError: true,
};
} finally {
clearTimeout(timeout);
}
}
);
// SSE 连接管理
let transport: SSEServerTransport | null = null;
app.get("/sse", async (req, res) => {
transport = new SSEServerTransport("/messages", res);
await server.connect(transport);
});
app.post("/messages", async (req, res) => {
if (transport) {
await transport.handlePostMessage(req, res);
} else {
res.status(503).json({ error: "No active SSE connection" });
}
});
app.listen(3100, () => {
console.error("MCP SSE Server running on http://localhost:3100");
});
客户端配置 SSE 模式:
{
"mcpServers": {
"remote-tools": {
"url": "http://localhost:3100/sse"
}
}
}
⚡ 三、生产级实践与避坑指南
3.1 资源暴露:让 AI 能「读取」你的数据
除了工具调用,MCP 还允许 Server 暴露「资源」——类似于一个只读的数据接口。这在数据查询场景中非常有用:
// 资源示例:暴露配置信息
server.resource(
"config",
"config://app/settings",
{ description: "应用当前配置信息", mimeType: "application/json" },
async (uri) => {
const config = {
version: "1.0.0",
features: { darkMode: true, locale: "zh-CN" },
limits: { maxFileSize: "10MB", rateLimit: "100/min" },
};
return {
contents: [{
uri: uri.href,
text: JSON.stringify(config, null, 2),
mimeType: "application/json",
}],
};
}
);
3.2 错误处理与超时控制
MCP 工具调用最常见的生产问题就是超时。AI 客户端通常有 30-60 秒的超时限制,超时后会认为工具失败。以下是关键的防御策略:
// 带超时和重试的工具包装器
function withTimeout<T>(
fn: () => Promise<T>,
timeoutMs: number = 25000
): Promise<T> {
return Promise.race([
fn(),
new Promise<never>((_, reject) =>
setTimeout(() => reject(new Error(`工具执行超时 (${timeoutMs}ms)`)), timeoutMs)
),
]);
}
// 使用示例
server.tool("slow_operation", "可能较慢的操作", {
query: z.string(),
}, async ({ query }) => {
try {
const result = await withTimeout(() => doSlowWork(query), 25000);
return { content: [{ type: "text", text: result }] };
} catch (err) {
return {
content: [{ type: "text", text: `执行失败: ${(err as Error).message}` }],
isError: true,
};
}
});
⚠️ 警告: MCP 工具的
isError字段非常关键。如果你的工具抛出异常但没设置isError: true,客户端会把错误信息当成正常结果展示给 AI,导致 AI 基于错误信息做出错误推理。
3.3 安全红线:绝不暴露的能力
构建 MCP Server 时,安全是第一优先级。以下是几条不可妥协的红线:
❌ 禁止暴露任意代码执行能力 —— 不要提供 run_code、eval、execute_sql 这类工具,除非你有完善的沙箱和权限控制。
❌ 禁止暴露敏感凭据 —— 数据库密码、API Key、私钥等绝对不能作为资源暴露。
❌ 禁止暴露未过滤的文件系统访问 —— 如果提供文件读取工具,必须限制在特定目录内,并校验路径遍历攻击。
❌ 禁止暴露破坏性操作的无确认执行 —— 删除、修改等操作必须有确认机制或审计日志。
✅ 推荐做法 —— 工具设计遵循最小权限原则,只暴露完成特定任务所需的最小能力集。
3.4 工具设计的黄金法则
经过大量实践,我总结出几条工具设计的关键原则:
单一职责 —— 一个工具只做一件事。json_format 和 json_validate 应该是两个工具,而不是一个 json_process 加 mode 参数。AI 对工具名和描述的理解依赖于清晰的语义边界。
参数描述要写成提示词 —— z.string().describe("待格式化的 JSON 字符串") 这个 describe 不只是文档,它直接影响 AI 决策时是否选择你的工具、如何构造参数。把它当成给 AI 的提示词来写。
返回结果要有上下文 —— 不要只返回裸数据,加上元信息。比如 JSON 压缩工具返回「压缩结果 (45.2% 体积减少): {…}」而不是只有 {...},这样 AI 能更好地理解结果含义。
幂等性优先 —— 尽量让工具是幂等的(相同输入多次调用结果一致),这样 AI 可以安全重试。
| 设计原则 | ❌ 错误示例 | ✅ 正确示例 |
|---|---|---|
| 单一职责 | process_json(mode, input) |
json_format(input) + json_validate(input) |
| 参数描述 | z.string().describe("data") |
z.string().describe("待格式化的 JSON 字符串,必须是合法 JSON") |
| 结果上下文 | return { text: result } |
return { text: "格式化完成,共 42 行:\n" + result } |
| 错误信息 | throw new Error("fail") |
return { text: "JSON 解析失败:第3行缺少逗号", isError: true } |
3.5 调试与测试
MCP Server 的调试比普通 HTTP 服务要麻烦,因为 stdio 模式不走网络。以下是实用的调试技巧:
# 用 MCP Inspector 交互式调试(官方工具)
npx @modelcontextprotocol/inspector npx tsx src/stdio-server.ts
# 手动模拟 JSON-RPC 请求(测试单个调用)
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | npx tsx src/stdio-server.ts
# 查看 stderr 日志(stdio 模式)
npx tsx src/stdio-server.ts 2>debug.log
编写自动化测试时,可以用内存传输层避免启动真实进程:
// test/server.test.ts
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
import { createServer } from "../src/server.js";
test("json_format returns formatted output", async () => {
const server = createServer();
const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
await server.connect(serverTransport);
const result = await clientTransport.send({
jsonrpc: "2.0",
id: 1,
method: "tools/call",
params: {
name: "json_format",
arguments: { input: '{"a":1,"b":2}' },
},
});
expect(result.content[0].text).toContain('"a": 1');
});
🎯 总结与工具推荐
MCP 协议的核心价值在于标准化——你的工具只需要实现一次 MCP Server,就能被所有支持 MCP 的 AI 客户端调用。这比为每个 AI 平台单独开发插件要高效得多。
我的建议是: 如果你在开发面向开发者的工具或服务,现在就应该考虑暴露 MCP 接口。这就像 2015 年每个服务都要有 REST API 一样,MCP 正在成为 AI 时代的服务接口标准。
推荐工具和资源:
- 🔧 @modelcontextprotocol/sdk — 官方 TypeScript SDK,API 设计优雅,文档完善
- 🔍 MCP Inspector — 官方调试工具,支持交互式测试工具调用
- 📦 mcp.run — MCP Server 托管平台,一键部署你的工具服务
- 📚 spec.modelcontextprotocol.io — 协议规范文档,解决边界问题时必读
- 🛠️ Smithery — MCP Server 注册与发现平台,让你的工具被更多人使用
⚡ 关键结论: MCP Server 的开发门槛比大多数人想象的低——一个完整的工具服务器只需要 50 行代码。真正的挑战在于工具设计的质量:清晰的职责划分、友好的参数描述、安全的权限控制。投入时间在设计上,而不是框架搭建上。