2026 年 5 月，OpenRouter 完成 1.13 亿美元 B 轮融资，其平台月均 API 调用量突破 50 亿次——这组数据背后折射出一个关键趋势：AI Agent 正在从"能聊天"进化到"能干活"，而 Model Context Protocol (MCP) 正是连接 LLM 与外部世界的事实标准。截至目前，GitHub 上已有超过 15,000 个 MCP Server 项目，从数据库查询到 CRM 操作，从代码部署到日志分析，MCP 生态每天都在爆发式增长。
然而，大多数开发者对 MCP 的理解还停留在"JSON-RPC 调个工具"的层面。本文将带你从协议架构到生产部署，用完整的 TypeScript 代码构建一个真正的 MCP Server——不是 Hello World，而是能扛住生产流量的工程实现。
📌 记住： MCP 不仅仅是"给 LLM 加个工具调用"。它定义了一套完整的资源访问、工具执行和提示模板协议，理解这三个原语（Primitives）的区别和协作方式，是构建高质量 MCP Server 的前提。
🔧 一、MCP 核心架构：三原语与传输层
1.1 Tools、Resources、Prompts 的本质区别
MCP 协议定义了三个核心原语，很多开发者把它们混为一谈，这会导致 API 设计混乱。它们各自的职责边界非常清晰：
| 原语 | 控制方 | 用途 | 类比 |
|---|---|---|---|
| Tools | 模型控制（Model-controlled） | LLM 自主决定何时调用，执行操作并返回结果 | REST API 的 POST/PUT/DELETE |
| Resources | 应用控制（Application-controlled） | 客户端决定何时读取，获取上下文数据 | REST API 的 GET |
| Prompts | 用户控制（User-controlled） | 用户主动选择的交互模板，预定义的对话模式 | Slash Commands |
⚠️ 警告： 最常见的设计错误是把所有功能都塞进 Tools。如果你的功能是"读取数据"而非"执行操作"，应该用 Resources；如果是"预定义工作流"，应该用 Prompts。错误选择原语会导致 LLM 上下文膨胀和不必要的 token 消耗。
1.2 传输层：stdio vs Streamable HTTP
MCP 支持两种传输方式，选择直接影响部署架构：
stdio 模式：MCP Server 作为子进程运行，通过标准输入输出通信。适合本地 IDE 集成（如 VS Code、Cursor），延迟低但无法远程部署。
Streamable HTTP 模式（MCP 2025-03-26 规范引入）：基于 HTTP + SSE 的远程传输，支持有状态和无状态两种模式。适合生产部署，支持多客户端并发访问。
// 两种传输层的初始化对比
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
// ✅ stdio 模式：本地 IDE 集成，零配置
const stdioTransport = new StdioServerTransport();
// ✅ Streamable HTTP 模式：远程部署，支持多客户端
const httpTransport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => crypto.randomUUID(), // 有状态模式
enableJsonResponse: true, // 允许 JSON 响应（非 SSE）
});
1.3 协议握手流程
MCP 客户端与服务器之间的通信遵循严格的握手流程。理解这个流程对调试连接问题至关重要：
Client Server
| |
|--- initialize (capabilities)->|
|<-- initialize (capabilities)--|
|--- initialized (notify) ----->|
| |
|--- tools/list --------------->|
|<-- tools (Tool[]) ------------|
| |
|--- tools/call (name, args) -->|
|<-- result (content[]) --------|
🚀 二、从零构建生产级 MCP Server
2.1 项目初始化与依赖
我们构建一个「数据库查询助手」MCP Server，它能让 AI Agent 安全地查询 PostgreSQL 数据库。这个案例覆盖了 Tools（执行查询）、Resources（读取 schema）、Prompts（预定义查询模板）三个原语。
# 初始化项目
mkdir mcp-db-assistant && cd mcp-db-assistant
npm init -y
npm install @modelcontextprotocol/sdk pg zod
npm install -D typescript @types/node @types/pg
npx tsc --init
// tsconfig.json — 关键配置
{
"compilerOptions": {
"target": "ES2022",
"module": "Node16",
"moduleResolution": "Node16",
"outDir": "./dist",
"rootDir": "./src",
"strict": true,
"esModuleInterop": true
}
}
2.2 完整 Server 实现
下面是一个完整的、可直接运行的 MCP Server 实现。注意代码中的安全设计——这是生产环境和 demo 的核心区别：
// src/server.ts — 生产级 MCP 数据库查询服务器
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { z } from "zod";
import pg from "pg";
const { Pool } = pg;
// ========== 数据库连接池（带连接限制） ==========
const pool = new Pool({
host: process.env.DB_HOST || "localhost",
port: parseInt(process.env.DB_PORT || "5432"),
database: process.env.DB_NAME || "mydb",
user: process.env.DB_USER || "readonly_user",
password: process.env.DB_PASSWORD,
max: 5, // 最大连接数限制
idleTimeoutMillis: 30000, // 空闲连接超时
connectionTimeoutMillis: 5000,
});
// ========== 安全层：SQL 白名单校验 ==========
const ALLOWED_TABLES = new Set(["users", "orders", "products"]);
const FORBIDDEN_KEYWORDS = [
"DROP", "DELETE", "TRUNCATE", "ALTER", "CREATE",
"INSERT", "UPDATE", "GRANT", "REVOKE", "EXEC",
];
function validateQuery(sql: string): { valid: boolean; error?: string } {
const upperSql = sql.toUpperCase().trim();
// 必须以 SELECT 开头
if (!upperSql.startsWith("SELECT")) {
return { valid: false, error: "仅允许 SELECT 查询" };
}
// 检查禁用关键字
for (const keyword of FORBIDDEN_KEYWORDS) {
if (upperSql.includes(keyword)) {
return { valid: false, error: `包含禁止的关键字: ${keyword}` };
}
}
// 检查表名白名单
const tableMatch = upperSql.match(/FROM\s+(\w+)/);
if (tableMatch && !ALLOWED_TABLES.has(tableMatch[1].toLowerCase())) {
return { valid: false, error: `不允许访问表: ${tableMatch[1]}` };
}
return { valid: true };
}
// ========== 创建 MCP Server ==========
const server = new McpServer({
name: "db-assistant",
version: "1.0.0",
});
// ---- Tool: 执行 SQL 查询 ----
server.tool(
"query",
"执行只读 SQL 查询，仅支持 SELECT 语句",
{
sql: z.string().describe("要执行的 SQL 查询语句"),
limit: z.number().optional().default(100).describe("最大返回行数"),
},
async ({ sql, limit }) => {
// 安全校验
const validation = validateQuery(sql);
if (!validation.valid) {
return {
content: [{ type: "text", text: `❌ 查询被拒绝: ${validation.error}` }],
isError: true,
};
}
// 添加 LIMIT 保护
const safeSql = sql.includes("LIMIT")
? sql
: `${sql.replace(/;$/, "")} LIMIT ${Math.min(limit, 1000)}`;
try {
const result = await pool.query(safeSql);
return {
content: [{
type: "text",
text: `查询成功，返回 ${result.rows.length} 行:\n${JSON.stringify(result.rows, null, 2)}`,
}],
};
} catch (err: any) {
return {
content: [{ type: "text", text: `❌ 查询执行失败: ${err.message}` }],
isError: true,
};
}
}
);
// ---- Tool: 获取表结构 ----
server.tool(
"describe_table",
"获取指定表的列定义和数据类型",
{
table: z.string().describe("表名"),
},
async ({ table }) => {
if (!ALLOWED_TABLES.has(table.toLowerCase())) {
return {
content: [{ type: "text", text: `❌ 不允许访问表: ${table}` }],
isError: true,
};
}
const result = await pool.query(
`SELECT column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_name = $1
ORDER BY ordinal_position`,
[table]
);
return {
content: [{
type: "text",
text: `表 ${table} 的结构:\n${JSON.stringify(result.rows, null, 2)}`,
}],
};
}
);
// ---- Resource: 数据库 Schema 概览 ----
server.resource(
"db://schema",
"数据库 Schema 概览，包含所有可访问表的信息",
async (uri) => {
const tables = await pool.query(
`SELECT table_name FROM information_schema.tables
WHERE table_schema = 'public' AND table_name = ANY($1)`,
[Array.from(ALLOWED_TABLES)]
);
const schema: Record<string, any> = {};
for (const { table_name } of tables.rows) {
const cols = await pool.query(
`SELECT column_name, data_type FROM information_schema.columns
WHERE table_name = $1 ORDER BY ordinal_position`,
[table_name]
);
schema[table_name] = cols.rows;
}
return {
contents: [{
uri: uri.href,
text: JSON.stringify(schema, null, 2),
mimeType: "application/json",
}],
};
}
);
// ---- Prompt: 预定义查询模板 ----
server.prompt(
"sales_report",
"生成销售数据报告",
{
period: z.enum(["daily", "weekly", "monthly"]).describe("报告周期"),
},
({ period }) => ({
messages: [{
role: "user",
content: {
type: "text",
text: `请帮我生成${period === "daily" ? "日度" : period === "weekly" ? "周度" : "月度"}销售报告。使用 query 工具查询 orders 表，按时间维度汇总销售额、订单数和平均客单价，并给出趋势分析。`,
},
}],
})
);
// ========== 启动服务器 ==========
async function main() {
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => crypto.randomUUID(),
});
await server.connect(transport);
// Express/Connect 兼容的请求处理
const http = await import("http");
const server_http = http.createServer(async (req, res) => {
// CORS 头
res.setHeader("Access-Control-Allow-Origin", "*");
res.setHeader("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS");
res.setHeader("Access-Control-Allow-Headers", "Content-Type, Mcp-Session-Id");
if (req.method === "OPTIONS") {
res.writeHead(204);
res.end();
return;
}
await transport.handleRequest(req, res);
});
const port = parseInt(process.env.PORT || "3001");
server_http.listen(port, () => {
console.log(`MCP Server 启动在 http://localhost:${port}`);
});
}
main().catch(console.error);
2.3 客户端集成：连接 MCP Server
构建完 Server 后，需要在客户端中连接它。以下是三种主流客户端的集成方式：
// src/client.ts — MCP 客户端连接示例
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
async function main() {
// 创建客户端
const client = new Client({
name: "my-app",
version: "1.0.0",
});
// 连接到远程 MCP Server
const transport = new StreamableHTTPClientTransport(
new URL("http://localhost:3001")
);
await client.connect(transport);
console.log("✅ 已连接到 MCP Server");
// 1. 列出所有可用工具
const tools = await client.listTools();
console.log("可用工具:", tools.tools.map(t => t.name));
// 2. 读取数据库 Schema 资源
const schema = await client.readResource({ uri: "db://schema" });
console.log("数据库 Schema:", schema.contents[0].text);
// 3. 调用查询工具
const result = await client.callTool({
name: "query",
arguments: {
sql: "SELECT COUNT(*) as total, SUM(amount) as revenue FROM orders WHERE created_at > NOW() - INTERVAL '7 days'",
limit: 10,
},
});
console.log("查询结果:", result.content[0].text);
// 4. 使用预定义 Prompt
const prompt = await client.getPrompt({
name: "sales_report",
arguments: { period: "weekly" },
});
console.log("报告模板:", prompt.messages[0].content);
}
main().catch(console.error);
💡 三、生产部署与进阶模式
3.1 Session 管理与并发控制
生产环境中，多个客户端可能同时连接同一个 MCP Server。Streamable HTTP 传输层通过 Session ID 管理会话状态：
// 生产级 Session 管理
import { randomUUID } from "node:crypto";
const sessions = new Map<string, StreamableHTTPServerTransport>();
function createSessionTransport(): StreamableHTTPServerTransport {
return new StreamableHTTPServerTransport({
sessionIdGenerator: () => {
const id = randomUUID();
// 设置 Session 过期清理（30 分钟）
setTimeout(() => {
sessions.delete(id);
console.log(`Session ${id} 已过期清理`);
}, 30 * 60 * 1000);
return id;
},
onsessioninitialized: (sessionId) => {
console.log(`新 Session 创建: ${sessionId}`);
},
enableJsonResponse: true, // 简单请求用 JSON，复杂请求用 SSE
});
}
3.2 错误处理与重试策略
MCP 协议定义了标准的错误码体系，合理的错误处理能显著提升 Agent 的容错能力：
| 错误码 | 含义 | 处理策略 |
|---|---|---|
| -32700 | Parse Error（JSON 解析失败） | 返回详细语法错误位置 |
| -32600 | Invalid Request | 校验请求 Schema 后返回 |
| -32601 | Method Not Found | 返回可用方法列表 |
| -32602 | Invalid Params | 返回参数 Schema 提示 |
| -32603 | Internal Error | 记录日志，返回通用错误 |
| -32002 | Resource Not Found | 返回可用资源列表 |
💡 提示： 在 Tool 执行中返回
isError: true而不是抛出异常。异常会导致整个 MCP 连接中断，而isError只会把错误信息返回给 LLM，让它决定下一步操作——这对 Agent 的自主恢复能力至关重要。
3.3 性能优化：连接池与缓存
MCP Server 的性能瓶颈通常不在协议层，而在后端资源访问。以下是经过生产验证的优化策略：
// LRU 缓存封装 —— 避免重复查询相同 Schema
class LRUCache<K, V> {
private cache = new Map<K, { value: V; expires: number }>();
private maxSize: number;
private ttl: number;
constructor(maxSize = 100, ttlMs = 5 * 60 * 1000) {
this.maxSize = maxSize;
this.ttl = ttlMs;
}
get(key: K): V | undefined {
const entry = this.cache.get(key);
if (!entry) return undefined;
if (Date.now() > entry.expires) {
this.cache.delete(key);
return undefined;
}
// LRU: 重新插入到末尾
this.cache.delete(key);
this.cache.set(key, entry);
return entry.value;
}
set(key: K, value: V): void {
if (this.cache.size >= this.maxSize) {
// 删除最旧的条目
const firstKey = this.cache.keys().next().value;
this.cache.delete(firstKey);
}
this.cache.set(key, { value, expires: Date.now() + this.ttl });
}
}
// 使用示例
const schemaCache = new LRUCache<string, any>(50, 10 * 60 * 1000); // 50条目，10分钟TTL
3.4 可观测性：结构化日志与指标
生产环境必须有完善的可观测性。MCP Server 的关键指标包括：
// 结构化日志中间件
interface McpMetrics {
toolCalls: Map<string, { count: number; totalTime: number; errors: number }>;
resourceReads: Map<string, { count: number; cacheHits: number }>;
}
const metrics: McpMetrics = {
toolCalls: new Map(),
resourceReads: new Map(),
};
function logToolCall(toolName: string, durationMs: number, isError: boolean) {
const entry = metrics.toolCalls.get(toolName) || { count: 0, totalTime: 0, errors: 0 };
entry.count++;
entry.totalTime += durationMs;
if (isError) entry.errors++;
metrics.toolCalls.set(toolName, entry);
// 结构化日志
console.log(JSON.stringify({
level: isError ? "error" : "info",
event: "tool_call",
tool: toolName,
duration_ms: durationMs,
is_error: isError,
timestamp: new Date().toISOString(),
}));
}
⚠️ 警告： 永远不要在 MCP Server 日志中记录完整的 SQL 查询或返回结果——这会导致敏感数据泄露。只记录工具名、耗时、行数等元数据。
3.5 与 LLM 框架集成
MCP Server 最终要与 LLM 框架集成才有价值。以下是与主流框架的对接方式：
# Python 客户端：与 Claude/Anthropic SDK 集成
import anthropic
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def chat_with_tools(user_message: str):
async with streamablehttp_client("http://localhost:3001") as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 获取工具列表并转换为 Anthropic 格式
tools_result = await session.list_tools()
anthropic_tools = [{
"name": t.name,
"description": t.description,
"input_schema": t.inputSchema,
} for t in tools_result.tools]
# 与 Claude 对话
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=anthropic_tools,
messages=[{"role": "user", "content": user_message}],
)
# 处理工具调用
for block in response.content:
if block.type == "tool_use":
result = await session.call_tool(block.name, block.input)
print(f"工具 {block.name} 返回: {result}")
📊 四、MCP Server 设计决策矩阵
在实际项目中，你需要根据场景做出关键设计决策。以下矩阵基于我在三个生产项目中的经验总结：
| 设计维度 | 推荐方案 | 不推荐方案 | 适用场景 |
|---|---|---|---|
| 传输层 | Streamable HTTP（远程） | stdio（仅限本地） | 生产部署必须用 HTTP |
| 状态管理 | 无状态 + 外部缓存 | Server 内存状态 | 多实例水平扩展 |
| 错误处理 | isError + 重试提示 | 抛异常中断连接 | Agent 自主恢复 |
| 工具粒度 | 单一职责（< 5 个参数） | 万能工具（10+ 参数） | LLM 调用准确率 |
| 认证方式 | OAuth 2.1 / API Key Header | 无认证 | 生产环境必须 |
| Schema 定义 | Zod（TypeScript）/ Pydantic | 手写 JSON Schema | 类型安全 + 自动文档 |
✅ 总结与最佳实践
构建生产级 MCP Server 需要关注三个核心维度：安全性（输入校验、权限最小化、日志脱敏）、可靠性（错误处理、重试策略、连接管理）、可观测性（结构化日志、指标采集、链路追踪）。
关键建议：
- ✅ 工具设计遵循单一职责原则，每个工具只做一件事
- ✅ 使用 Zod/Pydantic 等 Schema 库做输入校验，不要信任 LLM 的输出
- ✅ 实施 SQL 白名单或 ORM 层，永远不要直接执行用户输入的原始查询
- ❌ 不要在 Tool description 中泄露内部实现细节（防止 Tool Poisoning）
- ❌ 不要把敏感数据放在 MCP 返回结果的明文中
- ⚠️ 定期审计你的 MCP Server 开放的工具列表，移除不必要的能力
相关工具与资源：
- 🔧 MCP TypeScript SDK — 官方 TypeScript 实现
- 🔧 MCP Inspector — 交互式调试工具，开发必备
- 🔧 Smithery — MCP Server 注册与发现平台
- 📖 MCP 规范 — 协议官方文档