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

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

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

{
  "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

{
  "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"
}

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;
}

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
            ),
          },
        ],
      };
    }
  );
}

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"),
            },
          },
        ],
      };
    }
  );
}

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,
        };
      }
    }
  );
}

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

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: 展示分析报告

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,
      };
    }
  );
}

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

import { z } from "zod";

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

#!/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);
});

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

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

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

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

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

claude
> /mcp

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

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

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

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

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

pip install fastmcp

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 mcp add knowledge-base-py -- python /path/to/server.py

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

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 审查
    });
  });
});

npx @modelcontextprotocol/inspector node dist/index.js

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");
});

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

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

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

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