MCP 协议深度解析与 Server 开发实战

Model Context Protocol 架构原理、TypeScript Server 实现、工具/资源/提示模式与安全设计

引言

Model Context Protocol（MCP）是 Anthropic 于 2024 年底发布的开放协议，旨在为大语言模型提供标准化的上下文接入方式。它解决了一个根本问题：LLM 应用需要连接各种外部数据源和工具，但每个集成都需要定制化开发，导致 M x N 的组合爆炸。

MCP 的核心思想是将"模型如何获取上下文"这一问题抽象为统一协议，就像 USB 协议统一了外设接入标准一样。

协议架构

整体拓扑

┌──────────────────────────────────────────────────────────┐
│                      Host Application                     │
│  (Claude Desktop / VS Code / Cursor / Custom Client)     │
│                                                          │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐   │
│  │ Client A│  │ Client B│  │ Client C│  │ Client D│   │
│  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘   │
└───────┼────────────┼────────────┼────────────┼──────────┘
        │            │            │            │
   ┌────▼────┐  ┌────▼────┐  ┌────▼────┐  ┌────▼────┐
   │Server A │  │Server B │  │Server C │  │Server D │
   │(Files)  │  │(DB)     │  │(API)    │  │(Tools)  │
   └─────────┘  └─────────┘  └─────────┘  └─────────┘

Host 应用内部为每个 MCP Server 维护一个独立的 Client 实例。每个 Client-Server 对通过 JSON-RPC 2.0 通信，互相隔离，互不干扰。

三大核心原语

MCP 定义了三种核心原语（Primitives），分别面向不同的交互场景：

原语	控制方	描述	典型用途
Tools	模型控制	可执行的函数，模型自主决定何时调用	查询数据库、调用 API、执行计算
Resources	应用控制	结构化数据暴露，类似 REST 端点	文件内容、数据库记录、配置
Prompts	用户控制	预定义的提示模板，用户选择触发	代码审查模板、报告生成模板

这种三角设计的精妙之处在于：控制权分散到了模型、应用和用户三方，避免了单点滥用。

传输层

MCP 支持两种传输方式：

stdio 传输（本地进程）：
  Host ──stdin/stdout──▶ Server Process

HTTP + SSE 传输（远程服务）：
  Client ──HTTP POST──▶ Server /message
  Client ◀──SSE Stream── Server /sse

stdio 适用于本地工具（如文件系统、Git），延迟极低。HTTP+SSE 适用于远程服务，支持认证和网络穿透。

TypeScript Server 开发实战

项目初始化

// package.json
{
  "name": "mcp-server-example",
  "version": "1.0.0",
  "type": "module",
  "bin": { "mcp-example": "./dist/index.js" },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.0.0",
    "zod": "^3.23.0"
  }
}

最小可用 Server

// 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: "demo-server",
  version: "1.0.0",
  capabilities: {
    tools: {},
    resources: {},
    prompts: {},
  },
});

// Register a tool
server.tool(
  "calculate",
  "Perform arithmetic calculations",
  {
    expression: z.string().describe("Math expression to evaluate"),
  },
  async ({ expression }) => {
    try {
      // Use Function constructor for safe evaluation
      const result = new Function(`return (${expression})`)();
      return {
        content: [{ type: "text", text: `Result: ${result}` }],
      };
    } catch (error) {
      return {
        content: [{ type: "text", text: `Error: ${error.message}` }],
        isError: true,
      };
    }
  }
);

// Register a resource
server.resource(
  "config",
  "config://app/settings",
  { mimeType: "application/json" },
  async () => ({
    contents: [{
      uri: "config://app/settings",
      text: JSON.stringify({ theme: "dark", language: "zh-CN" }),
    }],
  })
);

// Register a prompt
server.prompt(
  "code-review",
  "Code review template with focus area",
  { focus: z.string().optional().describe("Review focus: security|performance|readability") },
  async ({ focus }) => ({
    messages: [{
      role: "user",
      content: {
        type: "text",
        text: `Please review the following code with focus on ${focus ?? "general quality"}.
Provide specific, actionable feedback with line references.`,
      },
    }],
  })
);

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP server running on stdio");
}

main().catch(console.error);

工具模式进阶

实际开发中，工具设计需要考虑参数校验、错误处理和进度反馈：

// Advanced tool with progress reporting and structured output
server.tool(
  "query-database",
  "Execute read-only SQL queries against the analytics database",
  {
    sql: z.string().describe("SQL SELECT query"),
    limit: z.number().default(100).describe("Maximum rows to return"),
    format: z.enum(["table", "json", "csv"]).default("table"),
  },
  async ({ sql, limit, format }, { reportProgress }) => {
    // Input validation: only allow SELECT
    const normalized = sql.trim().toUpperCase();
    if (!normalized.startsWith("SELECT")) {
      return {
        content: [{ type: "text", text: "Error: Only SELECT queries are allowed" }],
        isError: true,
      };
    }

    // Report progress for long-running queries
    await reportProgress({ progress: 0, total: 100 });

    const db = await getConnection();
    const safeSql = `${sql} LIMIT ${limit}`;
    const rows = await db.query(safeSql);

    await reportProgress({ progress: 100, total: 100 });

    // Format output based on requested format
    let output: string;
    switch (format) {
      case "json":
        output = JSON.stringify(rows, null, 2);
        break;
      case "csv":
        output = rowsToCsv(rows);
        break;
      default:
        output = rowsToMarkdownTable(rows);
    }

    return {
      content: [{
        type: "text",
        text: `Query returned ${rows.length} rows:\n\n${output}`,
      }],
    };
  }
);

资源模式进阶

资源支持 URI 模板，允许动态参数化访问：

// Dynamic resource with URI template
server.resource(
  "user-profile",
  new ResourceTemplate("users://{userId}/profile", { list: undefined }),
  { mimeType: "application/json" },
  async (uri, { userId }) => {
    const user = await userService.findById(userId);
    if (!user) {
      throw new Error(`User ${userId} not found`);
    }
    return {
      contents: [{
        uri: uri.href,
        text: JSON.stringify({
          id: user.id,
          name: user.name,
          role: user.role,
          lastActive: user.lastActive,
        }),
      }],
    };
  }
);

// Resource with subscription support for live updates
server.resource(
  "system-metrics",
  "metrics://system/current",
  { mimeType: "application/json" },
  async () => {
    const metrics = await collectMetrics();
    return {
      contents: [{
        uri: "metrics://system/current",
        text: JSON.stringify(metrics),
      }],
    };
  }
);

// Notify clients when metrics change
setInterval(async () => {
  server.notification({
    method: "notifications/resources/updated",
    params: { uri: "metrics://system/current" },
  });
}, 30000);

安全设计

威胁模型

┌─────────────────────────────────────────────────┐
│               MCP 威胁面分析                      │
├──────────────────┬──────────────────────────────┤
│ 威胁             │ 缓解措施                      │
├──────────────────┼──────────────────────────────┤
│ 提示注入         │ Server 端输入消毒 + 输出标注   │
│ 权限提升         │ 最小权限原则 + 能力声明        │
│ 数据泄露         │ 资源访问控制 + 审计日志        │
│ 拒绝服务         │ 速率限制 + 超时 + 资源配额     │
│ 中间人攻击       │ TLS + 认证令牌 + 签名校验      │
│ 工具滥用         │ 操作确认 + 副作用声明          │
└──────────────────┴──────────────────────────────┘

权限与认证实现

// Middleware-style permission checking
function withPermission(requiredScope: string) {
  return (handler: ToolHandler): ToolHandler => {
    return async (params, context) => {
      const clientId = context.meta?.clientId;
      const allowed = await checkPermission(clientId, requiredScope);

      if (!allowed) {
        return {
          content: [{
            type: "text",
            text: `Permission denied: requires scope '${requiredScope}'`,
          }],
          isError: true,
        };
      }

      // Audit log every tool invocation
      await auditLog({
        clientId,
        tool: context.toolName,
        scope: requiredScope,
        params: sanitize(params),
        timestamp: Date.now(),
      });

      return handler(params, context);
    };
  };
}

// Usage: wrap sensitive tools with permission checks
server.tool(
  "delete-record",
  "Delete a database record (requires write scope)",
  { id: z.string(), table: z.string() },
  withPermission("db:write")(async ({ id, table }) => {
    await db.delete(table, id);
    return { content: [{ type: "text", text: `Deleted ${table}/${id}` }] };
  })
);

输入消毒与输出安全

// Sanitize tool inputs to prevent injection
function sanitizeInput(input: string): string {
  // Remove potential prompt injection markers
  return input
    .replace(/\b(ignore|forget|disregard)\s+(previous|above|all)\b/gi, "[FILTERED]")
    .replace(/<\/?system>/gi, "[FILTERED]")
    .trim();
}

// Mark tool outputs to help LLM distinguish data from instructions
function wrapOutput(data: string, source: string): string {
  return `[BEGIN DATA from ${source}]\n${data}\n[END DATA from ${source}]`;
}

生产部署模式

HTTP+SSE 远程部署

import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from "express";

const app = express();

// Health check endpoint
app.get("/health", (req, res) => {
  res.json({ status: "ok", version: "1.0.0" });
});

// SSE endpoint for server-to-client messages
app.get("/sse", async (req, res) => {
  const authHeader = req.headers.authorization;
  if (!validateToken(authHeader)) {
    res.status(401).json({ error: "Unauthorized" });
    return;
  }

  const transport = new SSEServerTransport("/message", res);
  await server.connect(transport);
});

// Message endpoint for client-to-server messages
app.post("/message", async (req, res) => {
  // Route message to the correct transport based on session
  await transport.handlePostMessage(req, res);
});

app.listen(3001, () => {
  console.log("MCP HTTP server listening on :3001");
});

客户端配置

{
  "mcpServers": {
    "remote-analytics": {
      "command": "npx",
      "args": ["-y", "mcp-remote", "https://mcp.example.com/sse"],
      "env": {
        "MCP_AUTH_TOKEN": "${MCP_ANALYTICS_TOKEN}"
      }
    },
    "local-filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/data"],
      "env": {}
    }
  }
}

调试与测试

使用 MCP Inspector

# Start the inspector for interactive testing
npx @modelcontextprotocol/inspector node dist/index.js

# The inspector provides a web UI at http://localhost:5173
# where you can test tools, resources, and prompts interactively

单元测试

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";

describe("MCP Server", () => {
  let client: Client;

  beforeEach(async () => {
    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
    await server.connect(serverTransport);
    client = new Client({ name: "test-client", version: "1.0.0" });
    await client.connect(clientTransport);
  });

  it("should list tools", async () => {
    const result = await client.listTools();
    expect(result.tools).toHaveLength(2);
    expect(result.tools[0].name).toBe("calculate");
  });

  it("should execute calculate tool", async () => {
    const result = await client.callTool({
      name: "calculate",
      arguments: { expression: "2 + 3 * 4" },
    });
    expect(result.content[0].text).toContain("14");
  });
});

设计原则总结

1. 单一职责：每个 MCP Server 聚焦一个领域（文件系统、数据库、特定 API），不要构建全能 Server。
2. 最小权限：Server 只暴露模型真正需要的能力，工具粒度越细越安全。
3. 幂等优先：读操作无条件允许，写操作必须有确认机制或幂等保证。
4. 错误透明：工具失败时返回 isError: true 和人类可读的错误信息，不要隐藏错误。
5. 可观测性：所有工具调用记录审计日志，包含调用方、参数、结果和延迟。

MCP 协议的真正价值不在于"又一个新协议"，而在于它把 LLM 与外部世界的交互标准化了。随着生态成熟，MCP 有望成为 AI 应用的事实标准接口层。

---

Maurice | [email protected]