<!-- title: MCP全原语实战：Tool到Elicitation -->

# MCP全原语实战：从Tool到Elicitation五大能力深度开发

## 概述

MCP（Model Context Protocol）远不止"注册几个工具函数"这么简单。协议定义了**五大服务端原语**——Tools、Resources、Prompts、Sampling、Elicitation——每一种都解决不同的人机交互场景。本教程从零构建一个完整的 MCP Server，逐步解锁全部五种原语能力，适合已有 MCP 基础、想深入掌握协议全貌的开发者。

**学习目标**：掌握 MCP 五大原语的设计意图、实现方式、最佳实践，构建一个生产级多原语 MCP Server。

**适用人群**：已了解 MCP 基本概念、有 TypeScript 或 Python 开发经验的开发者。

## 前置条件

- Node.js 20+ 或 Python 3.11+
- Claude Code CLI 已安装（`npm install -g @anthropic-ai/claude-code`）
- 了解 JSON-RPC 基本概念
- 了解 JSON Schema 基本语法

---

## 第一部分：理解MCP原语全景

### 为什么需要五种原语？

大多数 MCP 教程只讲 Tools——让 AI 调用函数执行操作。但真实的 AI 应用场景远不止"调用函数"：

```mermaid
graph TB
    subgraph "MCP 五大原语"
        T[Tools<br/>执行操作/计算]
        R[Resources<br/>暴露可读数据]
        P[Prompts<br/>提供提示模板]
        S[Sampling<br/>请求LLM推理]
        E[Elicitation<br/>请求用户输入]
    end

    subgraph "使用场景"
        S1[查询数据库 → Tool]
        S2[读取文件内容 → Resource]
        S3[代码审查模板 → Prompt]
        S4[智能分类/摘要 → Sampling]
        S5[缺失参数确认 → Elicitation]
    end

    T --> S1
    R --> S2
    P --> S3
    S --> S4
    E --> S5
```

| 原语 | 方向 | 核心用途 | 类比 |
|------|------|----------|------|
| **Tools** | Client → Server | 让 AI 执行操作 | HTTP POST |
| **Resources** | Client → Server | 让 AI 读取数据 | HTTP GET |
| **Prompts** | Client → Server | 提供预定义提示模板 | HTTP GET (模板) |
| **Sampling** | Server → Client | Server 请求 LLM 推理 | 回调/RPC |
| **Elicitation** | Server → Client → User | Server 请求用户确认 | 对话框弹窗 |

**关键区别**：Tools 和 Resources 是"AI 主动调"，Sampling 和 Elicitation 是"Server 反向请求"。

---

## 第二部分：项目初始化

### 步骤一：创建项目骨架

```bash
mkdir mcp-full-primitives && cd mcp-full-primitives
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
```

创建 `tsconfig.json`：

```json
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "strict": true,
    "esModuleInterop": true
  },
  "include": ["src/**/*"]
}
```

项目结构：

```
mcp-full-primitives/
├── src/
│   ├── index.ts          # 主入口：注册所有原语
│   ├── tools.ts          # Tools 注册
│   ├── resources.ts      # Resources 注册
│   ├── prompts.ts        # Prompts 注册
│   ├── sampling.ts       # Sampling 处理器
│   ├── elicitation.ts    # Elicitation 处理器
│   └── data/             # 示例数据
│       └── knowledge.json
├── package.json
└── tsconfig.json
```

### 步骤二：创建示例数据

创建 `src/data/knowledge.json`：

```json
{
  "articles": [
    {
      "id": "mcp-001",
      "title": "MCP 架构设计指南",
      "content": "MCP 采用客户端-服务器架构...",
      "tags": ["architecture", "guide"],
      "updated": "2026-05-20"
    },
    {
      "id": "mcp-002",
      "title": "Streamable HTTP 传输层实战",
      "content": "Streamable HTTP 是 MCP 2025 规范新增的传输方式...",
      "tags": ["transport", "http"],
      "updated": "2026-05-22"
    },
    {
      "id": "mcp-003",
      "title": "MCP 安全最佳实践",
      "content": "生产环境 MCP Server 需要考虑...",
      "tags": ["security", "production"],
      "updated": "2026-05-25"
    }
  ],
  "status": "active",
  "version": "2.1.0"
}
```

---

## 第三部分：Tools——执行操作的核心原语

Tools 是最常用的原语。AI 根据用户请求自动选择并调用合适的 Tool。

### 步骤三：实现 Tools

创建 `src/tools.ts`：

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";

// 模拟数据库
interface Article {
  id: string;
  title: string;
  content: string;
  tags: string[];
  updated: string;
}

let articles: Article[] = [
  {
    id: "mcp-001",
    title: "MCP 架构设计指南",
    content: "MCP 采用客户端-服务器架构，通过 JSON-RPC 2.0 通信...",
    tags: ["architecture", "guide"],
    updated: "2026-05-20",
  },
  {
    id: "mcp-002",
    title: "Streamable HTTP 传输层实战",
    content: "Streamable HTTP 是 MCP 2025 规范新增的传输方式...",
    tags: ["transport", "http"],
    updated: "2026-05-22",
  },
];

export function registerTools(server: McpServer) {
  // Tool 1: 搜索文章
  server.tool(
    "search_articles",
    "根据关键词搜索知识库中的文章，返回匹配的文章列表",
    {
      keyword: z.string().describe("搜索关键词"),
      limit: z.number().optional().default(5).describe("返回结果数量上限"),
    },
    async ({ keyword, limit }) => {
      const results = articles.filter(
        (a) =>
          a.title.includes(keyword) ||
          a.content.includes(keyword) ||
          a.tags.includes(keyword)
      );
      const limited = results.slice(0, limit);
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify(
              {
                total: results.length,
                returned: limited.length,
                articles: limited.map((a) => ({
                  id: a.id,
                  title: a.title,
                  tags: a.tags,
                })),
              },
              null,
              2
            ),
          },
        ],
      };
    }
  );

  // Tool 2: 创建文章（带进度通知）
  server.tool(
    "create_article",
    "创建一篇新文章到知识库",
    {
      title: z.string().describe("文章标题"),
      content: z.string().describe("文章内容"),
      tags: z.array(z.string()).optional().default([]).describe("标签列表"),
    },
    async ({ title, content, tags }, { sendNotification }) => {
      const id = `mcp-${String(articles.length + 1).padStart(3, "0")}`;
      const article: Article = {
        id,
        title,
        content,
        tags,
        updated: new Date().toISOString().split("T")[0],
      };

      // 发送进度通知（Notification 原语的典型用法）
      await sendNotification({
        method: "notifications/progress",
        params: {
          progress: 50,
          total: 100,
          progressToken: "create-article",
        },
      });

      articles.push(article);

      await sendNotification({
        method: "notifications/progress",
        params: {
          progress: 100,
          total: 100,
          progressToken: "create-article",
        },
      });

      return {
        content: [
          {
            type: "text",
            text: `文章创建成功！ID: ${id}, 标题: ${title}`,
          },
        ],
      };
    }
  );

  // Tool 3: 删除文章
  server.tool(
    "delete_article",
    "根据ID删除知识库中的文章",
    {
      article_id: z.string().describe("要删除的文章ID"),
    },
    async ({ article_id }) => {
      const index = articles.findIndex((a) => a.id === article_id);
      if (index === -1) {
        return {
          content: [{ type: "text", text: `错误：未找到ID为 ${article_id} 的文章` }],
          isError: true,
        };
      }
      const removed = articles.splice(index, 1)[0];
      return {
        content: [{ type: "text", text: `已删除文章：${removed.title}` }],
      };
    }
  );
}

// 导出供 Resources 使用
export { articles };
export function getArticles() {
  return articles;
}
```

**关键要点**：

- Tool 的参数使用 Zod schema 定义，SDK 自动转换为 JSON Schema
- `sendNotification` 回调函数可用于发送进度通知
- `isError: true` 标记错误响应，AI 会据此调整后续行为
- Tool 描述（第二个参数）决定了 AI 何时以及如何调用此 Tool

### 常见问题

**Q: Tool 的描述应该写多长？**
A: 一句话说明用途即可。AI 根据描述判断是否调用，描述不清晰会导致误调用或漏调用。避免写实现细节，聚焦于"什么场景用"和"返回什么"。

**Q: 参数用 `.optional()` 和 `.default()` 有什么区别？**
A: `.optional()` 允许 AI 不传该参数（值为 `undefined`），`.default()` 在 AI 不传时自动使用默认值。推荐对可选项使用 `.default()` 以减少 AI 决策负担。

---

## 第四部分：Resources——暴露可读数据

Resources 是 MCP 中最被低估的原语。它让 Server 以 URI 形式暴露数据，AI 可以主动"浏览"而不必通过 Tool 查询。

### Resources vs Tools 的核心区别

| 维度 | Resources | Tools |
|------|-----------|-------|
| 语义 | 读取数据（幂等） | 执行操作（可能有副作用） |
| 寻址 | URI（如 `knowledge://articles/mcp-001`） | 名称 + 参数 |
| AI 行为 | 主动浏览、订阅更新 | 按需调用 |
| 缓存 | 客户端可缓存 | 通常不缓存 |

### 步骤四：实现 Resources

创建 `src/resources.ts`：

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { getArticles } from "./tools.js";

export function registerResources(server: McpServer) {
  // Resource 1: 静态资源 - 知识库概览
  server.resource(
    "knowledge-overview",
    "knowledge://overview",
    async (uri) => {
      const articles = getArticles();
      return {
        contents: [
          {
            uri: uri.href,
            mimeType: "application/json",
            text: JSON.stringify({
              totalArticles: articles.length,
              allTags: [...new Set(articles.flatMap((a) => a.tags))],
              lastUpdated: articles.reduce(
                (latest, a) => (a.updated > latest ? a.updated : latest),
                ""
              ),
            }),
          },
        ],
      };
    }
  );

  // Resource 2: 动态资源模板 - 按ID获取文章详情
  server.resource(
    "article-detail",
    new URL("knowledge://articles/{id}"),
    async (uri, { id }) => {
      const articles = getArticles();
      const article = articles.find((a) => a.id === id);

      if (!article) {
        return {
          contents: [
            {
              uri: uri.href,
              mimeType: "text/plain",
              text: `未找到ID为 ${id} 的文章`,
            },
          ],
        };
      }

      return {
        contents: [
          {
            uri: uri.href,
            mimeType: "application/json",
            text: JSON.stringify(article, null, 2),
          },
        ],
      };
    }
  );

  // Resource 3: 按标签筛选的资源模板
  server.resource(
    "articles-by-tag",
    new URL("knowledge://tags/{tag}"),
    async (uri, { tag }) => {
      const articles = getArticles();
      const filtered = articles.filter((a) => a.tags.includes(tag));

      return {
        contents: [
          {
            uri: uri.href,
            mimeType: "application/json",
            text: JSON.stringify(
              {
                tag,
                count: filtered.length,
                articles: filtered.map((a) => ({ id: a.id, title: a.title })),
              },
              null,
              2
            ),
          },
        ],
      };
    }
  );
}
```

**关键要点**：

- 静态资源用字符串 URI，动态资源模板用 `new URL("scheme://path/{param}")`
- Resource 模板中的 `{id}` 会被客户端自动解析并传入回调
- 每个 Resource 可以指定 `mimeType`，帮助 AI 正确解析内容
- Resources 是只读的，AI 通过 `resources/read` 请求访问

### 资源列表与发现

MCP Server 会自动处理 `resources/list` 和 `resources/templates/list` 请求，客户端可以浏览所有可用资源。这意味着在 Claude Code 中，AI 能自动"看到"你的 Resources 并决定何时读取。

---

## 第五部分：Prompts——预定义提示模板

Prompts 让你为 AI 提供预定义的、可参数化的提示模板。用户可以通过 Claude Code 的 `/` 命令选择使用。

### 步骤五：实现 Prompts

创建 `src/prompts.ts`：

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import { getArticles } from "./tools.js";

export function registerPrompts(server: McpServer) {
  // Prompt 1: 文章审查提示
  server.prompt(
    "review-article",
    "对指定文章进行专业审查，给出改进建议",
    {
      article_id: z.string().describe("要审查的文章ID"),
      focus_area: z
        .enum(["accuracy", "completeness", "readability", "all"])
        .optional()
        .default("all")
        .describe("审查重点领域"),
    },
    async ({ article_id, focus_area }) => {
      const articles = getArticles();
      const article = articles.find((a) => a.id === article_id);

      if (!article) {
        return {
          messages: [
            {
              role: "user",
              content: {
                type: "text",
                text: `未找到ID为 ${article_id} 的文章，请检查ID是否正确。`,
              },
            },
          ],
        };
      }

      const focusInstructions: Record<string, string> = {
        accuracy: "重点检查技术准确性，验证所有技术声明",
        completeness: "重点检查内容完整性，是否有遗漏的关键知识点",
        readability: "重点检查可读性，建议改善结构和表达方式",
        all: "全面审查：准确性、完整性、可读性、结构、示例质量",
      };

      return {
        messages: [
          {
            role: "user",
            content: {
              type: "text",
              text: [
                `请对以下文章进行专业审查。`,
                ``,
                `## 审查重点`,
                `${focusInstructions[focus_area]}`,
                ``,
                `## 文章内容`,
                `标题：${article.title}`,
                `标签：${article.tags.join(", ")}`,
                `最后更新：${article.updated}`,
                ``,
                `${article.content}`,
                ``,
                `## 审查要求`,
                `1. 给出总体评分（1-10）`,
                `2. 列出具体问题及改进建议`,
                `3. 提供修改后的优化版本`,
              ].join("\n"),
            },
          },
        ],
      };
    }
  );

  // Prompt 2: 生成文章大纲
  server.prompt(
    "generate-outline",
    "根据主题生成知识库文章的结构大纲",
    {
      topic: z.string().describe("文章主题"),
      depth: z
        .enum(["beginner", "intermediate", "advanced"])
        .optional()
        .default("intermediate")
        .describe("目标读者水平"),
      include_code: z
        .boolean()
        .optional()
        .default(true)
        .describe("是否包含代码示例"),
    },
    async ({ topic, depth, include_code }) => {
      const depthLabels = {
        beginner: "入门级",
        intermediate: "进阶级",
        advanced: "专家级",
      };

      return {
        messages: [
          {
            role: "user",
            content: {
              type: "text",
              text: [
                `请为以下主题生成一篇${depthLabels[depth]}技术文章的详细大纲。`,
                ``,
                `主题：${topic}`,
                `目标读者：${depthLabels[depth]}开发者`,
                `${include_code ? "需要包含代码示例" : "纯概念讲解，不需要代码"}`,
                ``,
                `大纲格式要求：`,
                `- 使用三级标题结构（H1 > H2 > H3）`,
                `- 每个章节附简要说明（1-2句话）`,
                `- 标注预计字数`,
                `- ${include_code ? "标注哪些章节需要代码示例" : ""}`,
              ].join("\n"),
            },
          },
        ],
      };
    }
  );
}
```

**关键要点**：

- Prompt 返回的是 `messages` 数组，而非 `content`，这是与 Tool 的核心区别
- Prompt 的参数同样使用 Zod 定义
- Prompt 的 `description` 参数（第二个）会显示在客户端的提示列表中
- Prompt 适合"模板化"的交互场景，用户选择模板 + 填参数，比自由输入更结构化

---

## 第六部分：Sampling——Server反向请求LLM

Sampling 是 MCP 中最独特的原语：**Server 主动请求 Client 端的 LLM 进行推理**。这让 Server 可以在执行过程中"借助 AI 大脑"做决策。

### 典型使用场景

- Tool 执行中需要对结果进行智能分类
- Server 需要根据上下文生成自然语言摘要
- 需要判断数据中的异常模式

### 步骤六：实现 Sampling

创建 `src/sampling.ts`：

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import { getArticles } from "./tools.js";
import type { CreateMessageRequest } from "@modelcontextprotocol/sdk/types.js";

export function registerSamplingTools(server: McpServer) {
  // 使用 Sampling 的 Tool：智能分析知识库
  server.tool(
    "analyze_knowledge_base",
    "使用AI分析知识库的整体状况，生成智能报告",
    {
      analysis_type: z
        .enum(["summary", "gaps", "trends"])
        .describe("分析类型：summary=概要, gaps=知识盲区, trends=趋势分析"),
    },
    async ({ analysis_type }, { session }) => {
      const articles = getArticles();

      // 检查客户端是否支持 Sampling
      if (!session?.capabilities?.sampling) {
        return {
          content: [
            {
              type: "text",
              text: "当前客户端不支持 Sampling 能力。请使用支持 Sampling 的客户端。",
            },
          ],
          isError: true,
        };
      }

      // 构造 Sampling 请求
      const promptMap = {
        summary: `请用中文对以下知识库内容进行总结概述：\n\n${JSON.stringify(articles, null, 2)}`,
        gaps: `分析以下知识库中的文章，找出知识盲区和缺失的主题领域：\n\n${JSON.stringify(
          articles.map((a) => ({ id: a.id, title: a.title, tags: a.tags })),
          null,
          2
        )}`,
        trends: `分析以下文章的标签和更新时间，识别知识库的发展趋势：\n\n${JSON.stringify(
          articles.map((a) => ({ title: a.title, tags: a.tags, updated: a.updated })),
          null,
          2
        )}`,
      };

      try {
        // 向客户端的 LLM 发送 Sampling 请求
        const result = await session.request(
          {
            method: "sampling/createMessage",
            params: {
              messages: [
                {
                  role: "user",
                  content: {
                    type: "text",
                    text: promptMap[analysis_type],
                  },
                },
              ],
              maxTokens: 1000,
              systemPrompt: "你是一位知识管理专家，擅长分析技术文档库。",
            },
          },
          CreateMessageResultSchema
        );

        return {
          content: [
            {
              type: "text",
              text: `## AI 分析报告\n\n${
                result.content.type === "text" ? result.content.text : "分析完成"
              }`,
            },
          ],
        };
      } catch (error) {
        return {
          content: [
            {
              type: "text",
              text: `Sampling 请求失败：${error instanceof Error ? error.message : String(error)}`,
            },
          ],
          isError: true,
        };
      }
    }
  );
}
```

**注意**：Sampling 的实际代码需要从 SDK 导入 `CreateMessageResultSchema`。在 TypeScript SDK v2 中，完整的 Sampling 集成如下：

```typescript
import { CreateMessageResultSchema } from "@modelcontextprotocol/sdk/types.js";
```

### Sampling 工作流程

```mermaid
sequenceDiagram
    participant User as 用户
    participant CC as Claude Code<br/>(MCP Client)
    participant Server as MCP Server

    User->>CC: "分析知识库盲区"
    CC->>Server: tools/call (analyze_knowledge_base)
    
    Note over Server: 检测到需要AI分析
    Server->>CC: sampling/createMessage<br/>(请求LLM推理)
    CC->>CC: 调用内部LLM生成分析
    CC->>Server: 返回LLM分析结果
    
    Note over Server: 整合分析结果
    Server->>CC: 返回最终Tool结果
    CC->>User: 展示分析报告
```

**安全要点**：

- Sampling 请求由 Client 端控制，Client 可以拒绝任何 Sampling 请求
- Server 不能假设 Sampling 一定可用，必须先检查 `session.capabilities.sampling`
- `maxTokens` 应设置合理上限，避免消耗过多 Token
- `systemPrompt` 可以引导 LLM 的输出格式

---

## 第七部分：Elicitation——Server请求用户输入

Elicitation 是 MCP 2026 规范中最新的原语。它让 Server 能在 Tool 执行过程中**暂停并请求用户输入**，实现真正的 Human-in-the-Loop。

### 为什么需要 Elicitation？

想象你的 Tool 要删除一个重要资源——你可能想在执行前确认。或者 Tool 检测到参数不完整——你可能想追问用户。Elicitation 正是为此而生。

### 步骤七：实现 Elicitation

创建 `src/elicitation.ts`：

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import { getArticles } from "./tools.js";

export function registerElicitationTools(server: McpServer) {
  // 使用 Elicitation 的 Tool：安全删除文章（需用户确认）
  server.tool(
    "safe_delete_article",
    "安全删除文章（需要用户二次确认）",
    {
      article_id: z.string().describe("要删除的文章ID"),
    },
    async ({ article_id }, { session }) => {
      const articles = getArticles();
      const article = articles.find((a) => a.id === article_id);

      if (!article) {
        return {
          content: [
            { type: "text", text: `未找到ID为 ${article_id} 的文章` },
          ],
          isError: true,
        };
      }

      // 检查 Elicitation 能力
      if (!session?.capabilities?.elicitation) {
        // 降级方案：不支持 Elicitation 时直接拒绝
        return {
          content: [
            {
              type: "text",
              text: `安全删除需要客户端支持 Elicitation。文章 "${article.title}" 未被删除。`,
            },
          ],
        };
      }

      // 向用户发送确认请求
      try {
        const elicitResult = await session.request(
          {
            method: "elicitation/create",
            params: {
              message: `确定要删除文章「${article.title}」吗？此操作不可撤销。`,
              requestedSchema: {
                type: "object",
                properties: {
                  confirmed: {
                    type: "boolean",
                    title: "确认删除",
                    description: "勾选以确认删除此文章",
                  },
                  reason: {
                    type: "string",
                    title: "删除原因（可选）",
                    description: "请简要说明删除原因",
                  },
                },
                required: ["confirmed"],
              },
            },
          },
          ElicitResultSchema
        );

        if (!elicitResult.content?.confirmed) {
          return {
            content: [
              { type: "text", text: `用户取消了删除操作。文章 "${article.title}" 保持不变。` },
            ],
          };
        }

        // 用户确认后执行删除
        const index = articles.findIndex((a) => a.id === article_id);
        if (index !== -1) {
          articles.splice(index, 1);
        }

        const reason = elicitResult.content?.reason
          ? `\n删除原因：${elicitResult.content.reason}`
          : "";

        return {
          content: [
            {
              type: "text",
              text: `已成功删除文章「${article.title}」。${reason}`,
            },
          ],
        };
      } catch (error) {
        return {
          content: [
            {
              type: "text",
              text: `确认请求被拒绝或出错，文章未被删除。`,
            },
          ],
        };
      }
    }
  );

  // 使用 Elicitation 的 Tool：交互式创建文章
  server.tool(
    "interactive_create",
    "交互式创建文章（自动补充缺失信息）",
    {
      title: z.string().describe("文章标题"),
      content: z.string().optional().describe("文章内容（可选）"),
    },
    async ({ title, content }, { session }) => {
      // 如果内容缺失，通过 Elicitation 请求用户补充
      if (!content && session?.capabilities?.elicitation) {
        try {
          const result = await session.request(
            {
              method: "elicitation/create",
              params: {
                message: `正在创建文章「${title}」，但缺少文章内容。请补充：`,
                requestedSchema: {
                  type: "object",
                  properties: {
                    content: {
                      type: "string",
                      title: "文章内容",
                      description: "请输入文章正文内容",
                    },
                    tags: {
                      type: "string",
                      title: "标签（逗号分隔）",
                      description: "如: architecture, guide",
                    },
                  },
                  required: ["content"],
                },
              },
            },
            ElicitResultSchema
          );

          const finalContent = result.content?.content || "（无内容）";
          const tags = result.content?.tags
            ? result.content.tags.split(",").map((t: string) => t.trim())
            : [];

          // 现在有完整数据了，创建文章
          const articles = getArticles();
          const id = `mcp-${String(articles.length + 1).padStart(3, "0")}`;
          articles.push({
            id,
            title,
            content: finalContent,
            tags,
            updated: new Date().toISOString().split("T")[0],
          });

          return {
            content: [
              { type: "text", text: `交互式创建成功！文章ID: ${id}` },
            ],
          };
        } catch {
          return {
            content: [
              { type: "text", text: "用户取消了内容补充，文章未创建。" },
            ],
          };
        }
      }

      // 有内容时直接创建
      if (content) {
        const articles = getArticles();
        const id = `mcp-${String(articles.length + 1).padStart(3, "0")}`;
        articles.push({
          id,
          title,
          content,
          tags: [],
          updated: new Date().toISOString().split("T")[0],
        });

        return {
          content: [
            { type: "text", text: `文章创建成功！文章ID: ${id}` },
          ],
        };
      }

      return {
        content: [
          {
            type: "text",
            text: "无法创建文章：缺少内容且客户端不支持交互式输入。",
          },
        ],
        isError: true,
      };
    }
  );
}
```

**注意**：`ElicitResultSchema` 需要从 SDK 类型中导入。在 TypeScript SDK 的最新版本中：

```typescript
import { ElicitResultSchema } from "@modelcontextprotocol/sdk/types.js";
```

如果 SDK 版本尚未包含此 Schema，可以使用自定义 Zod schema：

```typescript
import { z } from "zod";

const ElicitResultSchema = z.object({
  action: z.enum(["accept", "decline", "cancel"]),
  content: z.record(z.unknown()).optional(),
});
```

---

## 第八部分：组装主入口

### 步骤八：整合所有原语

创建 `src/index.ts`：

```typescript
#!/usr/bin/env node

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { registerTools } from "./tools.js";
import { registerResources } from "./resources.js";
import { registerPrompts } from "./prompts.js";
import { registerSamplingTools } from "./sampling.js";
import { registerElicitationTools } from "./elicitation.js";

async function main() {
  // 创建 MCP Server 实例
  const server = new McpServer({
    name: "knowledge-base-full",
    version: "1.0.0",
  });

  // 注册五大原语
  registerTools(server);           // Tools：搜索、创建、删除
  registerResources(server);       // Resources：知识库数据暴露
  registerPrompts(server);         // Prompts：审查、大纲模板
  registerSamplingTools(server);   // Sampling：智能分析
  registerElicitationTools(server); // Elicitation：安全确认

  // 启动 stdio 传输
  const transport = new StdioServerTransport();
  await server.connect(transport);

  console.error("Knowledge Base MCP Server 已启动（全原语模式）");
}

main().catch((error) => {
  console.error("致命错误:", error);
  process.exit(1);
});
```

在 `package.json` 中添加构建和启动脚本：

```json
{
  "name": "mcp-full-primitives",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node dist/index.js",
    "dev": "tsc && node dist/index.js"
  }
}
```

构建并测试：

```bash
npm run build
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}}}' | npm start
```

预期输出包含 `capabilities` 字段，其中 `tools`、`resources`、`prompts` 均为可用状态。

---

## 第九部分：注册到 Claude Code

### 步骤九：配置 MCP Server

```bash
# 方式一：使用 claude mcp add 命令
claude mcp add knowledge-base -- node /path/to/mcp-full-primitives/dist/index.js

# 方式二：直接编辑配置文件
# 全局配置：~/.claude/settings.json
# 项目配置：.claude/settings.json
```

在 `.claude/settings.json` 中添加：

```json
{
  "mcpServers": {
    "knowledge-base": {
      "command": "node",
      "args": ["/absolute/path/to/mcp-full-primitives/dist/index.js"]
    }
  }
}
```

### 验证注册成功

启动 Claude Code 后，输入 `/mcp` 命令查看已注册的 MCP Server：

```bash
claude
> /mcp
```

你应该能看到 `knowledge-base` Server，以及它暴露的所有 Tools、Resources、Prompts。

### 测试各原语

```
# 测试 Tool
> 搜索知识库中关于 HTTP 的文章

# 测试 Resource
> 读取 knowledge://overview 资源

# 测试 Prompt
> 使用 review-article 提示模板审查文章 mcp-001

# 测试 Sampling
> 分析知识库的知识盲区

# 测试 Elicitation
> 安全删除文章 mcp-001
```

---

## 第十部分：Python等效实现（FastMCP）

如果你更偏好 Python，以下是使用 FastMCP 的等效实现。

### 步骤十：Python 全原语 Server

```bash
pip install fastmcp
```

创建 `server.py`：

```python
from fastmcp import FastMCP, Context
from pydantic import BaseModel, Field
from typing import Optional

mcp = FastMCP("knowledge-base-full")

# ========== 数据层 ==========
articles = [
    {
        "id": "mcp-001",
        "title": "MCP 架构设计指南",
        "content": "MCP 采用客户端-服务器架构...",
        "tags": ["architecture", "guide"],
        "updated": "2026-05-20",
    },
    {
        "id": "mcp-002",
        "title": "Streamable HTTP 传输层实战",
        "content": "Streamable HTTP 是 MCP 2025 规范新增的传输方式...",
        "tags": ["transport", "http"],
        "updated": "2026-05-22",
    },
]


# ========== Tools ==========
@mcp.tool()
def search_articles(keyword: str, limit: int = 5) -> str:
    """根据关键词搜索知识库中的文章"""
    results = [
        a for a in articles
        if keyword in a["title"] or keyword in a["content"] or keyword in a["tags"]
    ]
    return str(results[:limit])


@mcp.tool()
async def create_article(
    title: str,
    content: str,
    tags: list[str] = [],
    ctx: Context = None,
) -> str:
    """创建一篇新文章到知识库"""
    new_id = f"mcp-{len(articles) + 1:03d}"

    # 发送进度通知
    await ctx.report_progress(50, 100)

    articles.append({
        "id": new_id,
        "title": title,
        "content": content,
        "tags": tags,
        "updated": "2026-05-27",
    })

    await ctx.report_progress(100, 100)
    return f"文章创建成功！ID: {new_id}"


# ========== Resources ==========
@mcp.resource("knowledge://overview")
def get_overview() -> str:
    """知识库概览"""
    return str({
        "total": len(articles),
        "tags": list(set(t for a in articles for t in a["tags"])),
    })


@mcp.resource("knowledge://articles/{article_id}")
def get_article(article_id: str) -> str:
    """获取指定文章详情"""
    article = next((a for a in articles if a["id"] == article_id), None)
    return str(article) if article else f"未找到 {article_id}"


@mcp.resource("knowledge://tags/{tag}")
def get_articles_by_tag(tag: str) -> str:
    """按标签筛选文章"""
    filtered = [a for a in articles if tag in a["tags"]]
    return str(filtered)


# ========== Prompts ==========
@mcp.prompt()
def review_article(article_id: str, focus_area: str = "all") -> str:
    """对指定文章进行专业审查"""
    article = next((a for a in articles if a["id"] == article_id), None)
    if not article:
        return f"未找到文章 {article_id}"

    return f"""请对以下文章进行专业审查，重点：{focus_area}

标题：{article['title']}
标签：{', '.join(article['tags'])}
内容：{article['content']}

请给出评分（1-10）和改进建议。"""


@mcp.prompt()
def generate_outline(topic: str, depth: str = "intermediate") -> str:
    """根据主题生成文章大纲"""
    return f"""请为以下主题生成一篇{depth}级别的技术文章大纲：

主题：{topic}
要求：三级标题结构，每节附简要说明和预计字数。"""


# ========== Sampling ==========
@mcp.tool()
async def analyze_with_ai(analysis_type: str, ctx: Context) -> str:
    """使用AI分析知识库（需要Sampling支持）"""
    summary = str([{"id": a["id"], "title": a["title"], "tags": a["tags"]} for a in articles])

    # 通过 Sampling 请求客户端 LLM
    result = await ctx.sample(
        f"分析以下知识库文章，找出{analysis_type}：\n\n{summary}",
        system_prompt="你是一位知识管理专家。",
        max_tokens=1000,
    )

    return f"## AI分析报告\n\n{result.text}"


# ========== Elicitation ==========
@mcp.tool()
async def safe_delete(article_id: str, ctx: Context) -> str:
    """安全删除文章（需要用户确认）"""
    article = next((a for a in articles if a["id"] == article_id), None)
    if not article:
        return f"未找到文章 {article_id}"

    # 通过 Elicitation 请求用户确认
    result = await ctx.elicit(
        message=f"确定要删除「{article['title']}」吗？",
        schema={
            "type": "object",
            "properties": {
                "confirmed": {"type": "boolean", "title": "确认删除"},
                "reason": {"type": "string", "title": "删除原因"},
            },
            "required": ["confirmed"],
        },
    )

    if result.get("confirmed"):
        articles.remove(article)
        return f"已删除「{article['title']}」"
    return "用户取消了删除"


if __name__ == "__main__":
    mcp.run()
```

注册到 Claude Code：

```bash
claude mcp add knowledge-base-py -- python /path/to/server.py
```

### TypeScript 与 Python 实现对比

```mermaid
graph LR
    subgraph "TypeScript SDK"
        TS_T[server.tool<br/>+ zod schema]
        TS_R[server.resource<br/>+ URL模板]
        TS_P[server.prompt<br/>+ zod schema]
        TS_S[session.request<br/>sampling/createMessage]
        TS_E[session.request<br/>elicitation/create]
    end

    subgraph "Python FastMCP"
        PY_T["@mcp.tool 装饰器<br/>+ 类型注解"]
        PY_R["@mcp.resource 装饰器<br/>+ URI 模板"]
        PY_P["@mcp.prompt 装饰器<br/>+ 类型注解"]
        PY_S["ctx.sample()<br/>内置方法"]
        PY_E["ctx.elicit()<br/>内置方法"]
    end

    TS_T -.->|"等价"| PY_T
    TS_R -.->|"等价"| PY_R
    TS_P -.->|"等价"| PY_P
    TS_S -.->|"等价"| PY_S
    TS_E -.->|"等价"| PY_E
```

| 特性 | TypeScript SDK | Python FastMCP |
|------|---------------|----------------|
| Schema 定义 | Zod | 类型注解/Pydantic |
| Sampling 调用 | `session.request()` 手动构造 | `ctx.sample()` 内置封装 |
| Elicitation 调用 | `session.request()` 手动构造 | `ctx.elicit()` 内置封装 |
| 进度通知 | `sendNotification()` | `ctx.report_progress()` |
| 装饰器风格 | 函数式 `server.tool()` | `@mcp.tool()` 装饰器 |
| 异步支持 | 原生 async/await | 原生 async/await |

---

## 第十一部分：测试策略

### 步骤十一：单元测试与集成测试

MCP Server 的测试分三层：工具逻辑测试、协议合规测试、端到端集成测试。

创建 `src/__tests__/tools.test.ts`：

```typescript
import { describe, it, expect, beforeEach } from "vitest";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

describe("MCP Server 全原语测试", () => {
  let server: McpServer;
  let client: Client;

  // 单元测试：直接测试工具逻辑
  describe("Tools 逻辑", () => {
    it("search_articles 应返回匹配结果", async () => {
      // 直接调用工具内部逻辑，无需启动 Server
      const keyword = "HTTP";
      const articles = [
        { id: "001", title: "HTTP Guide", content: "...", tags: ["http"] },
        { id: "002", title: "TCP Guide", content: "...", tags: ["tcp"] },
      ];
      const results = articles.filter(
        (a) => a.title.includes(keyword) || a.tags.includes(keyword.toLowerCase())
      );
      expect(results).toHaveLength(1);
      expect(results[0].id).toBe("001");
    });
  });

  // 协议合规测试：通过 Client 连接 Server
  describe("协议合规", () => {
    it("应暴露正确的 capabilities", async () => {
      // 启动 Server 进程
      // 通过 Client 连接
      // 验证 capabilities 包含 tools, resources, prompts
    });

    it("tools/list 应返回所有注册的工具", async () => {
      // 验证工具数量和名称
    });

    it("resources/list 应返回所有资源", async () => {
      // 验证资源 URI 格式
    });

    it("resources/templates/list 应返回模板", async () => {
      // 验证模板参数定义
    });

    it("prompts/list 应返回所有提示模板", async () => {
      // 验证提示模板名称和参数
    });
  });

  // 端到端集成测试
  describe("端到端", () => {
    it("搜索 -> 读取 -> 审查 完整流程", async () => {
      // 1. 调用 search_articles Tool
      // 2. 使用返回的 ID 读取 Resource
      // 3. 使用 review-article Prompt 审查
    });
  });
});
```

使用 MCP Inspector 进行可视化测试：

```bash
npx @modelcontextprotocol/inspector node dist/index.js
```

Inspector 提供了 Web UI，可以：

- 可视化浏览所有 Tools、Resources、Prompts
- 手动调用任意 Tool 并查看返回结果
- 测试 Resource 的读取和模板解析
- 验证 Prompt 的参数渲染

---

## 第十二部分：生产部署要点

### 步骤十二：从 stdio 到 Streamable HTTP

开发阶段使用 stdio 传输，生产环境需要切换到 Streamable HTTP 以支持远程访问。

```typescript
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import express from "express";

const app = express();
app.use(express.json());

const server = new McpServer({
  name: "knowledge-base-full",
  version: "1.0.0",
});

// 注册所有原语（同上）
registerTools(server);
registerResources(server);
registerPrompts(server);
registerSamplingTools(server);
registerElicitationTools(server);

// Streamable HTTP 端点
app.post("/mcp", async (req, res) => {
  const transport = new StreamableHTTPServerTransport({
    sessionIdGenerator: undefined,
  });
  await server.connect(transport);
  await transport.handleRequest(req, res);
});

app.listen(3000, () => {
  console.error("MCP Server 运行在 http://localhost:3000/mcp");
});
```

Claude Code 中配置远程 MCP：

```json
{
  "mcpServers": {
    "knowledge-base-remote": {
      "type": "streamable-http",
      "url": "http://localhost:3000/mcp"
    }
  }
}
```

### 生产检查清单

```mermaid
graph TD
    A[部署前检查] --> B{是否需要认证?}
    B -->|是| C[配置 OAuth 2.1<br/>或 API Key]
    B -->|否| D{是否暴露公网?}
    C --> D
    D -->|是| E[必须启用认证!]
    D -->|否| F[本地/内网安全]
    E --> G[Streamable HTTP<br/>+ TLS]
    F --> G
    G --> H{Sampling/Elicitation<br/>是否必须?}
    H -->|是| I[确认客户端支持<br/>添加降级方案]
    H -->|否| J[纯 Tool 模式<br/>降低复杂度]
    I --> K[性能监控<br/>+ 错误追踪]
    J --> K
```

| 检查项 | 要点 |
|--------|------|
| 错误处理 | 所有 Tool 必须用 `isError: true` 标记错误，不能抛出未捕获异常 |
| 超时控制 | 长 Tool 执行需发送进度通知，避免 Client 超时 |
| 能力降级 | Sampling/Elicitation 不可用时提供替代方案 |
| 输入验证 | Zod/Pydantic schema 是第一道防线，但业务逻辑也要二次验证 |
| 资源限制 | `maxTokens`、文件大小、请求频率都应设上限 |
| 日志记录 | 所有 Tool 调用记录到 `stderr`（stdio 模式下 stdout 被协议占用） |

---

## 完整工作流程回顾

```mermaid
graph TB
    subgraph "开发阶段"
        A[初始化项目] --> B[注册 Tools]
        B --> C[注册 Resources]
        C --> D[注册 Prompts]
        D --> E[实现 Sampling]
        E --> F[实现 Elicitation]
        F --> G[整合主入口]
    end

    subgraph "测试阶段"
        G --> H[单元测试<br/>工具逻辑]
        H --> I[协议合规测试<br/>capabilities]
        I --> J[MCP Inspector<br/>可视化验证]
    end

    subgraph "部署阶段"
        J --> K[stdio → Streamable HTTP]
        K --> L[认证与安全加固]
        L --> M[监控与日志]
    end

    subgraph "使用阶段"
        M --> N[claude mcp add<br/>注册到 Claude Code]
        N --> O[AI 自动调用<br/>Tool/Resource/Prompt]
        O --> P[Sampling/Elicitation<br/>增强交互]
    end
```

### 原语选择决策树

当你设计 MCP Server 时，根据需求选择合适的原语：

```mermaid
graph TD
    Q1{需要让AI<br/>做什么?}
    Q1 -->|执行操作| Q2{有副作用?}
    Q1 -->|读取数据| R[使用 Resource]
    Q1 -->|提供模板| P[使用 Prompt]

    Q2 -->|是| T[使用 Tool]
    Q2 -->|否，纯读取| R2{数据量大?}

    R2 -->|是| R3{需要URI寻址?}
    R2 -->|否| T2[使用 Tool<br/>也行]

    R3 -->|是| R
    R3 -->|否| T2

    T --> Q3{执行中需要<br/>AI辅助?}
    Q3 -->|是| S[使用 Sampling]
    Q3 -->|否| Q4{需要用户确认?}

    Q4 -->|是| E[使用 Elicitation]
    Q4 -->|否| DONE[完成]

    S --> DONE
    E --> DONE
```

---

## 延伸阅读

- [Model Context Protocol 官方规范](https://modelcontextprotocol.io/specification/2025-06-18) — MCP 协议完整规范，包含所有原语定义
- [TypeScript SDK GitHub 仓库](https://github.com/modelcontextprotocol/typescript-sdk) — 官方 TypeScript SDK 源码和文档
- [FastMCP Python 文档](https://fastmcp.wiki/) — Python MCP 开发框架完整文档
- [MCP Primitives 概念详解](https://docs.mcp-agent.com/concepts/mcp-primitives) — 五大原语的设计理念和适用场景
- [MCP Inspector 工具](https://github.com/modelcontextprotocol/inspector) — 官方可视化测试工具

## 参考来源

1. [MCP Architecture Overview](https://modelcontextprotocol.io/docs/learn/architecture) — 官方架构文档，详细说明了协议层次和原语设计
2. [Complete Guide to MCP in 2026](https://dev.to/x4nent/complete-guide-to-mcp-model-context-protocol-in-2026-architecture-implementation-and-4a11) — 2026年MCP完整指南，涵盖架构和实现
3. [MCP Resources, Prompts, and Sampling](https://kindatechnical.com/agentic-ai/mcp-resources-prompts-and-sampling.html) — Tool之外三种能力的深度教程
4. [MCP Elicitation: Human-in-the-Loop](https://dev.to/kachurun/mcp-elicitation-human-in-the-loop-for-mcp-servers-m6a) — Elicitation原语的实战指南
5. [FastMCP Streaming Tutorial](https://www.maximofn.com/en/streamable-mcp/) — FastMCP框架的流式传输和进度通知教程

---

**训练营适用性评估**：

| 维度 | 说明 |
|------|------|
| 适合训练营 | 中高级 AI 工具开发训练营、MCP 生态开发者工作坊 |
| 教学时长 | 约 6-8 小时（含动手实践） |
| 学员基础 | 需掌握 TypeScript 或 Python、了解 MCP 基本概念 |
| 实验环境 | 本地 Node.js/Python 环境 + Claude Code CLI |
| 分组建议 | 每 2-3 人一组，分别实现不同原语后合并 |

> [!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 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)