关于 mcp
用于构建、审查和调试 MCP 服务的开发技能,涵盖实现流程,以及工具、传输、身份验证和注入风险的安全检查。
主要功能
- 01服务构建与审查流程
- 02工具、资源和提示词设计
- 03传输与 OAuth 2.1 指引
- 04注入与工具投毒检查
- 05基于 Inspector 的测试指引
适用场景
- 01创建包含工具和资源的 MCP 服务
- 02加固 Streamable HTTP 传输
- 03检查处理器中的注入风险
- 04排查客户端与服务端的 MCP 连接
MCP 相关
指导开发向 AI 编码 Agent 暴露工具、资源和提示词的 MCP 服务,并涵盖传输配置与连接调试。
安装
复制下面的命令,在终端中运行。
npx skills add https:/ / github.com/ iuliandita/ skills --skill mcp安装后在 Claude Code 中输入/mcp确认已加载。
包含的文件
除 SKILL.md 外,安装时一起下载的参考资料和脚本。
references/agent-hygiene.mdoutput-contract.mdsecurity.md详细介绍
MCP:Model Context Protocol 服务开发
构建、审查和调试 MCP 服务,向 AI 编码助手提供工具、资源和提示词。目标是打造安全、结构良好的服务,遵循协议规范,避免成为又一个存在可预防注入漏洞的服务。
目标版本(2026 年 10 月):
- MCP 规范:2026-07-28(当前稳定版;无状态核心、扩展框架,且无需 initialize/session 握手)
- TypeScript SDK:
@modelcontextprotocol/server、@modelcontextprotocol/client和@modelcontextprotocol/core2.3.0(单体版@modelcontextprotocol/sdk1.32.0 属于旧版系列;2026-10-02 发布的两个版本均将客户端重定向限制为同源,且 2.3.0 使每个Server实例只服务一个连接)。针对高严重性 OAuth 凭据重定向公告(CVE-2026-104850),客户端最低版本为 client 2.2.0 和 sdk 1.31.0 - Python SDK:mcp 2.3.0(2.x 稳定版;请查看 v1 到 v2 迁移指南)。针对高严重性 OAuth 凭据重定向和未回收会话公告(GHSA-qx49-fqc8-xw99、CVE-2026-59951),最低版本为 2.2.0(或 1.x 系列中的 1.30.0)
- 协议传输方式:stdio 和 Streamable HTTP。独立的 HTTP+SSE 传输方式已弃用,仅作为临时旧版桥接方式提供
适用场景
- 构建新的 MCP 服务(工具、资源、提示词)
- 为现有 MCP 服务添加工具处理程序
- 配置 MCP 传输方式(本地使用 stdio,远程使用 streamable HTTP)
- 实现 MCP 身份验证(OAuth 2.1)
- 实现 MCP 引导(交互式对话框)
- 审查 MCP 服务代码,排查注入或工具投毒漏洞
- 调试客户端与服务之间的 MCP 连接问题
- 将自定义工具集成迁移到 MCP
不适用场景
- 不使用 MCP 的通用 REST API 开发——请使用 backend-api
- 在应用中使用 Claude API / Anthropic SDK——请使用 llm-app-development
- 跨整个代码库审计现有服务的安全性——请使用 security-audit(其中包含 MCP 章节)
- 使用 MCP 浏览工具浏览或抓取网页——请使用宿主提供的浏览工具
- 为大模型编写提示词(不是 MCP 提示词资源)——请使用 prompt-generator
AI 自检
生成或审查 MCP 服务代码时,请在提交结果前逐项核验:
- 在服务端验证所有工具处理程序的输入(不得将原始字符串插值到 shell 命令、SQL、文件路径或 URL 中)
- 每个工具在
tools/list中都有非空、准确、简洁的描述(FastMCP 从文档字符串中获取描述,因此未编写文档的函数会注册为空描述;有些客户端会截断过长的描述) - 资源 URI 使用已定义的 scheme,并在使用前进行验证
- 错误响应使用正确的 MCP 错误代码,而非原始堆栈跟踪
- 对处理用户数据的远程传输实现身份验证(使用 PKCE 的 OAuth 2.1)
- 工具处理程序或服务配置中没有硬编码的机密信息
-
inputSchema使用具体的 JSON Schema 类型,并设置required、maxLength和约束 - 服务支持平稳关闭(在 SIGINT/SIGTERM 时清理)
- Streamable HTTP:本地运行时绑定到
127.0.0.1(而非0.0.0.0) - Streamable HTTP:验证
Origin标头(防止 DNS 重绑定) - 对工具调用进行速率限制
- 客户端代码将工具注解视为不可信内容
- 引导请求不会索要密码、令牌或机密信息
- 检查规范版本:传输、身份验证、资源、工具和提示词与当前 MCP 文档及 SDK 行为一致
- 考虑工具投毒:工具描述、动态元数据和服务更新不会在用户不知情的情况下扩大权限
- 核实 SDK 方法:参见常见错误 #8——对照实际 SDK 文档核实每个 API 调用,不要臆造方法名称
- 落实跨领域的 Agent 卫生规范——参见
references/agent-hygiene.md
性能
- 精简工具 schema 和响应;庞大且无结构的工具输出会浪费模型上下文。
- 使用资源提供可重复使用的上下文,而不是在每次工具调用中返回相同的大型负载。
- 对延迟敏感时,可批量处理只读查询,但应将有副作用的工具分开并确保可审计。
最佳实践
- 将 MCP 服务视为安全边界:明确进行身份验证、授权,并记录副作用。
- 保持工具名称和 schema 稳定;对不兼容的更改进行版本管理,而不是直接改变现有语义。
- 对会花费资金、修改基础设施、删除数据或暴露机密信息的工具,要求用户确认。
工作流
**构建与审查:**步骤 1-6 用于构建新服务。审查现有 MCP 服务代码时:(1) 使用步骤 1 的问题确定范围——服务使用哪些工具、传输方式和身份验证;(2) 根据步骤 3 的注入途径和 AI 自检清单审查每个工具处理程序;(3) 对照“常见错误”部分,检查 AI 模型经常引入的模式。
构建服务时复制此清单并跟踪进度:
- 步骤 1:确定用途、工具、传输方式和身份验证
- 步骤 2:搭建服务框架,并实现平稳关闭
- 步骤 3:每个工具处理程序都验证输入,且不使用插值
- 步骤 4:配置并加固传输方式
- 步骤 5:引导处理接受、拒绝和取消
- 步骤 6:Inspector 和恶意输入测试通过(若失败,则修复并返回步骤 3)
步骤 1:确定服务的用途
编写代码前,先明确:
- 将公开哪些工具? 每个工具对应 AI 可调用的一项操作。
- 将提供哪些资源? 资源是 AI 可访问的只读数据。
- 使用哪种传输方式? 本地 CLI 集成使用 stdio,远程或生产环境使用 Streamable HTTP。
- 使用哪种身份验证? stdio 无需身份验证。处理用户数据的远程服务建议使用 OAuth 2.1。
- 使用哪种语言? TypeScript(SDK 最成熟)或 Python(更简单,可使用 FastMCP)。
步骤 2:搭建服务端框架
TypeScript(推荐用于生产环境):
import { McpServer } from "@modelcontextprotocol/server";
import { serveStdio } from "@modelcontextprotocol/server/stdio";
import { readdir, readFile, stat } from "node:fs/promises";
import path from "node:path";
import * as z from "zod/v4";
const DOCS_ROOT = path.resolve(process.env.DOCS_ROOT ?? "./docs");
const MAX_DOCS = 1_000;
const MAX_DOC_BYTES = 256_000;
const config = { mode: "read-only" };
async function loadMarkdownDocs(): Promise<Array<{ title: string; body: string }>> {
const entries = await readdir(DOCS_ROOT, { withFileTypes: true });
const names = entries
.filter((entry) => entry.isFile() && entry.name.endsWith(".md"))
.map((entry) => entry.name)
.sort()
.slice(0, MAX_DOCS);
const docs: Array<{ title: string; body: string }> = [];
for (const name of names) {
const filePath = path.join(DOCS_ROOT, name);
if ((await stat(filePath)).size > MAX_DOC_BYTES) continue;
docs.push({ title: name, body: await readFile(filePath, "utf8") });
}
return docs;
}
let docsPromise: ReturnType<typeof loadMarkdownDocs> | undefined;
function docsIndex(): ReturnType<typeof loadMarkdownDocs> {
docsPromise ??= loadMarkdownDocs();
return docsPromise;
}
function createServer(): McpServer {
const server = new McpServer({ name: "my-server", version: "1.0.0" });
server.registerTool(
"search_docs",
{
title: "Search docs",
description: "Search documentation by keyword",
inputSchema: z.object({
query: z.string().max(200).describe("Search query"),
limit: z.number().int().min(1).max(100).default(10),
}),
outputSchema: z.object({
results: z.array(z.object({ title: z.string(), snippet: z.string() })),
}),
},
async ({ query, limit }) => {
try {
const needle = query.normalize("NFKC").trim().toLocaleLowerCase();
if (!/[\p{L}\p{N}]/u.test(needle)) {
return { isError: true, content: [{ type: "text", text: "Query needs a letter or number." }] };
}
const results = (await docsIndex())
.flatMap(({ title, body }) => {
const text = `${title}\n${body}`;
const index = text.toLocaleLowerCase().indexOf(needle);
if (index < 0) return [];
const start = Math.max(0, index - 160);
const end = Math.min(text.length, index + needle.length + 320);
return [{ title, snippet: text.slice(start, end) }];
})
.slice(0, limit);
const output = { results };
return {
content: [{ type: "text", text: JSON.stringify(output) }],
structuredContent: output,
};
} catch (error: unknown) {
console.error("search_docs failed", error);
return { isError: true, content: [{ type: "text", text: "Search failed while loading documentation." }] };
}
}
);
server.registerResource(
"config",
"config://app/settings",
{ title: "Application config", mimeType: "application/json" },
async (uri) => ({
contents: [{ uri: uri.href, mimeType: "application/json", text: JSON.stringify(config) }],
})
);
return server;
}
async function main(): Promise<void> {
const handle = await serveStdio(createServer);
let closing = false;
const shutdown = async (signal: NodeJS.Signals): Promise<void> => {
if (closing) return;
closing = true;
try {
await handle.close();
} catch (error: unknown) {
console.error(`Failed to shut down after ${signal}:`, error);
process.exitCode = 1;
}
};
process.once("SIGINT", () => void shutdown("SIGINT"));
process.once("SIGTERM", () => void shutdown("SIGTERM"));
}
void main().catch((error: unknown) => {
console.error(error);
process.exitCode = 1;
});
serveStdio 负责 stdio 传输,并返回一个句柄,其 close() 会同时关闭固定的服务端实例和传输。所有 stdio 诊断信息都应输出到 stderr;stdout 是协议通道。
Python(适合快速原型开发的 FastMCP):
import json, re
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-server")
docs = [{"title": "Getting started", "body": "Install the server and connect over stdio."}]
config = {"mode": "read-only"}
@mcp.tool()
def search_docs(query: str, limit: int = 10) -> dict[str, list[dict[str, str]]]:
"""Search documentation; return bounded title/snippet matches."""
if not 1 <= len(query) <= 200 or not 1 <= limit <= 20:
raise ValueError("Require a 1-200 character query and limit 1-20")
needle = re.sub(r"[^\w\s-]", "", query).strip().casefold()
if not any(char.isalnum() for char in needle):
raise ValueError("Query must contain a letter or number")
matches = [doc for doc in docs if needle in f"{doc['title']}\n{doc['body']}".casefold()]
return {"matches": [{"title": doc["title"][:120], "snippet": doc["body"][:240]}
for doc in matches[:limit]]}
@mcp.resource("config://app/settings")
def get_config() -> str:
"""Application configuration."""
return json.dumps(config)
if __name__ == "__main__":
mcp.run()
步骤 3:安全地实现工具
注入是 MCP 中最常见的漏洞类型。每个工具处理程序都是攻击面。
首要规则:绝不要将用户输入插入命令、查询或路径中。
MCP 工具中常见的注入向量:
| 向量 | 不安全的模式 | 安全的模式 |
|---|---|---|
| Shell | 插入用户输入的命令字符串 | 使用参数数组调用 execFile,并验证路径 |
| SQL | 在查询中拼接字符串 | 使用带 $1 占位符的参数化查询 |
| 文件路径 | 直接调用 readFile(userPath) | 解析路径,并根据允许列表验证前缀 |
| URL | 直接调用 fetch(userUrl) | 解析 URL,并根据允许列表验证协议和主机 |
| 模板 | 动态执行代码 | 使用不执行代码的沙箱模板引擎 |
防止路径遍历:
import path from "node:path";
import { realpath } from "node:fs/promises";
async function safeExistingPath(base: string, userInput: string): Promise<string> {
const baseReal = await realpath(base);
const targetReal = await realpath(path.resolve(baseReal, userInput));
const relative = path.relative(baseReal, targetReal);
if (relative === ".." || relative.startsWith(`..${path.sep}`) || path.isAbsolute(relative)) {
throw new Error("Path traversal detected");
}
return targetReal;
}
阅读 references/security.md 中的“路径遍历”,了解使用此辅助函数的工具处理程序改写示例,以及创建新文件的模式。
防止 SSRF(当工具获取用户输入的 URL 时):
- 阻止私有 IP 范围:
10.0.0.0/8、172.16.0.0/12、192.168.0.0/16 - 阻止链路本地地址:
169.254.0.0/16(包含169.254.169.254上的云元数据) - 阻止回环地址:
127.0.0.0/8 - 在生产环境中要求使用 HTTPS
- 在检查和使用之间固定 DNS 解析结果(防范 TOCTOU)
步骤 4:配置传输
| 传输方式 | 用途 | 是否需要身份验证 | 说明 |
|---|---|---|---|
| stdio | 本地工具、CLI 集成 | 否 | 以用户进程运行,安全性最高。 |
| Streamable HTTP | 远程/多客户端服务 | 建议启用 | 单一端点,通过 POST 发送消息,可选 SSE 流式传输。 |
独立的 HTTP+SSE 传输方式(规范版本 2024-11-05)已弃用;远程服务应使用 Streamable HTTP。MCP 2026-07-28 从核心协议中移除了初始化/会话握手。当同一端点还必须为旧版 2025 客户端提供服务时,请使用 SDK 迁移辅助工具。 规范允许不进行身份验证,但处理用户数据的服务强烈建议启用身份验证。实现身份验证时,请使用带 PKCE 的 OAuth 2.1。优先使用 Client ID Metadata Documents,而非 Dynamic Client Registration(DCR 是备用方案,不是必需项)。
Streamable HTTP 安全性:
- 本地服务应绑定到
127.0.0.1(绝不要绑定到0.0.0.0) - 验证所有请求中的
Origin标头(防止 DNS 重绑定) - 将每个 2026-07-28 请求视为独立请求;不要要求提供
MCP-Session-Id - 客户端发送
MCP-Protocol-Version: 2026-07-28,并在_meta中提供身份信息和功能 - 考虑使用
createMcpExpressApp()/createMcpHonoApp()来获得内置的 DNS 重绑定防护——这些功能来自独立的@modelcontextprotocol/express和@modelcontextprotocol/hono包,而非与运行时无关的核心包
步骤 5:安全处理 elicitation
MCP elicitation 允许服务端在任务进行期间向用户请求结构化输入。
架构限制——elicitation 架构仅限于包含基本类型字段的扁平对象:
string(可选使用format:email、uri、date、date-time)number/integer(可设置minimum、maximum)booleanenum(使用带enum的字符串;对于带标签的选项,使用带title的anyOf)enum字符串数组(用于多选)
不支持嵌套对象。请保持架构简单,以便广泛支持客户端。
安全性:绝不要通过 elicitation 请求凭据。客户端应显示请求输入的服务端,并始终允许用户拒绝/取消。处理全部三种响应:accept(带数据)、decline 和 cancel。
第 6 步:测试服务
command -v npx >/dev/null || echo "install Node.js to get npx for the Inspector"
command -v uv >/dev/null || echo "install uv for the Python dev inspector"
# Test with MCP Inspector (official debugging tool)
npx @modelcontextprotocol/inspector your-server-command
# Python alternative
uv run mcp dev server.py
使用以下情况测试每个工具处理器:有效输入(正常路径)、缺少必填字段、恶意输入(注入、路径遍历、超大载荷)以及并发请求。如果某个用例失败,请修复处理器(第 3 步),并重新运行所有用例,直到全部通过。
阅读 references/security.md,了解具体的注入测试载荷。
工具投毒和 Rug Pull 防御
这些攻击针对工具元数据,而非工具执行:工具描述中的隐藏指令,或在批准后发生变化的定义。请编写诚实、简洁且不包含嵌入式指令的描述,并阅读 references/security.md(工具投毒和 Rug Pull 攻击),了解攻击示例以及服务端和客户端的防御措施。
常见错误
AI 模型在生成 MCP 服务代码时经常会犯以下错误:
- 通过字符串插值执行 Shell 命令 - 这是首要漏洞。系统命令始终使用参数数组。
- 缺少服务端验证 - 生成了
inputSchema,却从不在处理器中根据它进行验证。客户端可能会跳过验证。 - Schema 中仅使用
"type": "string"- 没有maxLength、pattern或其他限制。会接受任意长度的任何字符串。 - 将 HTTP 绑定到
0.0.0.0- 会让本地服务暴露到网络中。请使用127.0.0.1。 - 不验证
Origin标头 - 会导致本地服务遭受 DNS 重绑定攻击。 - 泄露错误详情 - 在工具响应中暴露堆栈跟踪、文件路径或数据库错误。
- 令牌透传 - 接受原本用于其他服务的 OAuth 令牌,却不验证受众。
- 臆造 SDK 方法 - 编造不存在的 API 调用。请参阅 AI 自检中的验证清单项。
- 忽略征询操作 - 能处理
accept,但遇到decline/cancel就崩溃。 - 未实现平稳关闭 - stdio 服务缺少 SIGINT/SIGTERM 处理器。
参考文件
references/security.md- OAuth 2.1 详情、已知 CVE、注入测试载荷、路径遍历和 SSRF 防护、会话管理以及工具投毒防御。在实现身份验证、加固处理器或审查服务时阅读此文件。
输出约定
完整约定请参阅 references/output-contract.md。
- 技能名称: MCP
- 交付物类别:
audits - 模式: 有条件。当调用以分析、审查、审计或改进现有仓库内容时,遵循
references/output-contract.md中有关报告篇幅和证据的规则,并将交付物写入docs/local/audits/mcp/<YYYY-MM-DD>-<slug>.md。当调用以回答问题、讲解概念、构建新产物或生成内容时,可自由作答,无需遵循该约定。 - 严重级别:
P0 | P1 | P2 | P3 | info(见共享约定;仅用于审计/审查模式)。
相关技能
- security-audit - 在更广泛的安全审查中审计 MCP 服务。security-audit 技能中的 ASI 和 MCP 部分涵盖漏洞模式;本技能涵盖从一开始就正确构建服务。
- code-review - 审查 MCP 服务代码的正确性,不局限于安全性。
- docker - 使用最小权限容器化 MCP 服务。
- llm-app-development - 在调用 MCP 工具的应用中使用 Claude API / Anthropic SDK。Anthropic SDK 集成代码请使用 llm-app-development,而非本技能。
- backend-api - 开发不使用 MCP 协议的一般 REST/GraphQL API。
- 使用宿主环境的浏览工具读取或访问网页;本技能用于构建 MCP 服务。
- prompt-generator - 编写大模型提示词(而非 MCP 提示词资源);如果请求涉及提示词工程,而非构建 MCP 服务,请转由该技能处理。
规则
- 在服务端验证所有工具输入。 切勿信任客户端或模型。使用架构验证(Zod、Pydantic),明确指定类型、范围和约束。
- 不要使用字符串插值执行 shell 命令。 使用参数数组执行系统命令。
- 保持描述简洁。 部分客户端会截断过长的描述。用几句话说明工具的功能和参数,而非实现细节。
- 处理用户数据时进行身份验证。 对于会访问用户数据的远程服务,使用 OAuth 2.1 和 PKCE。规范允许身份验证为可选,但强烈建议启用。