7.6 MCP Builder Skill - 构建 MCP 服务器
概述
MCP Builder Skill 是一个专门用于指导开发高质量 MCP (Model Context Protocol) 服务器的技能。MCP 是 Anthropic 推出的模型上下文协议,它允许 LLM 通过精心设计的工具与外部服务进行交互。这个 Skill 帮助开发者创建能够让 AI 真正完成实际任务的 MCP 服务器。
核心理念
MCP 服务器的质量不在于它实现了多少 API 端点,而在于它能多好地帮助 LLM 完成真实世界的任务。一个优秀的 MCP 服务器应该:
- 提供清晰、描述性的工具名称
- 返回简洁、相关的数据
- 给出可操作的错误消息
- 支持分页和过滤
适用场景
MCP Builder Skill 适用于以下场景:
- 集成外部 API:将第三方服务(如 GitHub、Slack、Jira)封装为 MCP 工具
- 构建本地工具服务器:创建文件操作、数据库查询等本地工具
- 开发远程服务:部署可多客户端访问的 MCP 服务
- 扩展 AI 能力:为 Claude 等 AI 模型添加新的工具能力
开发流程四阶段
MCP Builder Skill 定义了一个标准化的四阶段开发流程:
+------------------+ +------------------+ +------------------+ +------------------+
| Phase 1 | | Phase 2 | | Phase 3 | | Phase 4 |
| 深度研究与规划 | --> | 实现 | --> | 审查与测试 | --> | 创建评估 |
+------------------+ +------------------+ +------------------+ +------------------+
| - 理解MCP设计 | | - 项目结构 | | - 代码质量审查 | | - 创建10个问题 |
| - 学习协议文档 | | - 核心基础设施 | | - 构建测试 | | - 验证答案 |
| - 研究框架文档 | | - 实现工具 | | - MCP Inspector | | - 输出XML格式 |
| - 规划实现方案 | | - 错误处理 | | - 质量检查表 | | - 运行评估 |
+------------------+ +------------------+ +------------------+ +------------------+Phase 1: 深度研究与规划
在开始编码之前,需要深入理解以下概念:
API 覆盖 vs 工作流工具
- 全面的 API 端点覆盖给予代理灵活性
- 专门的工作流工具对特定任务更方便
- 当不确定时,优先选择全面的 API 覆盖
工具命名与可发现性
- 使用清晰、描述性的工具名称
- 使用一致的前缀(如
github_create_issue,slack_send_message) - 使用动作导向的命名
上下文管理
- 设计返回聚焦、相关数据的工具
- 支持过滤和分页结果
- 保持工具描述简洁
Phase 2: 实现
支持两种主要的编程语言:
| 特性 | TypeScript | Python |
|---|---|---|
| 框架 | MCP TypeScript SDK | FastMCP |
| 验证库 | Zod | Pydantic |
| 命名约定 | service-mcp-server | service_mcp |
| 示例 | github-mcp-server | github_mcp |
Phase 3: 审查与测试
使用 MCP Inspector 测试服务器:
bash
# TypeScript
npx @modelcontextprotocol/inspector
# Python
python -m py_compile your_server.pyPhase 4: 创建评估
评估是验证 MCP 服务器质量的关键步骤,通过创建 10 个复杂问题来测试 LLM 是否能有效使用你的工具。
TypeScript 实现指南
项目结构
service-mcp-server/
|-- package.json
|-- tsconfig.json
|-- README.md
|-- src/
| |-- index.ts # 主入口,McpServer 初始化
| |-- types.ts # TypeScript 类型定义
| |-- tools/ # 工具实现(每个领域一个文件)
| |-- services/ # API 客户端和共享工具
| |-- schemas/ # Zod 验证模式
| |-- constants.ts # 共享常量
|-- dist/ # 构建输出目录服务器初始化
typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
const server = new McpServer({
name: "service-mcp-server",
version: "1.0.0"
});工具注册模式
typescript
// Zod schema 用于输入验证
const UserSearchInputSchema = z.object({
query: z.string()
.min(2, "查询至少2个字符")
.max(200, "查询不超过200字符")
.describe("搜索字符串,匹配名称/邮箱"),
limit: z.number()
.int()
.min(1)
.max(100)
.default(20)
.describe("最大返回结果数"),
offset: z.number()
.int()
.min(0)
.default(0)
.describe("分页偏移量"),
response_format: z.nativeEnum(ResponseFormat)
.default(ResponseFormat.MARKDOWN)
.describe("输出格式: 'markdown' 或 'json'")
}).strict();
// 注册工具
server.registerTool(
"example_search_users",
{
title: "搜索示例用户",
description: `在 Example 系统中按名称、邮箱或团队搜索用户。
此工具搜索所有用户资料,支持部分匹配和各种搜索过滤器。
它不会创建或修改用户,只搜索现有用户。
Args:
- query (string): 搜索字符串
- limit (number): 最大结果数 1-100 (默认: 20)
- offset (number): 分页偏移量 (默认: 0)
- response_format: 输出格式 (默认: 'markdown')
Returns:
JSON 格式包含 total, count, offset, users, has_more 等字段`,
inputSchema: UserSearchInputSchema,
annotations: {
readOnlyHint: true,
destructiveHint: false,
idempotentHint: true,
openWorldHint: true
}
},
async (params) => {
// 实现逻辑
return {
content: [{ type: "text", text: result }],
structuredContent: output
};
}
);传输选项
typescript
// stdio 传输(本地工具)
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const transport = new StdioServerTransport();
await server.connect(transport);
// Streamable HTTP 传输(远程服务器)
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import express from "express";
const app = express();
app.post('/mcp', async (req, res) => {
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: undefined,
enableJsonResponse: true
});
res.on('close', () => transport.close());
await server.connect(transport);
await transport.handleRequest(req, res, req.body);
});Python 实现指南
服务器初始化
python
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional, List
from enum import Enum
# 初始化 MCP 服务器
mcp = FastMCP("example_mcp")Pydantic 模型定义
python
class ResponseFormat(str, Enum):
"""输出格式选项"""
MARKDOWN = "markdown"
JSON = "json"
class UserSearchInput(BaseModel):
"""用户搜索操作的输入模型"""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
query: str = Field(
...,
description="搜索字符串",
min_length=2,
max_length=200
)
limit: Optional[int] = Field(
default=20,
description="最大返回结果数",
ge=1,
le=100
)
offset: Optional[int] = Field(
default=0,
description="分页偏移量",
ge=0
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="输出格式"
)
@field_validator('query')
@classmethod
def validate_query(cls, v: str) -> str:
if not v.strip():
raise ValueError("查询不能为空")
return v.strip()工具注册模式
python
@mcp.tool(
name="example_search_users",
annotations={
"title": "搜索示例用户",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def example_search_users(params: UserSearchInput) -> str:
"""在 Example 系统中按名称、邮箱或团队搜索用户。
此工具搜索所有用户资料,支持部分匹配和各种搜索过滤器。
Args:
params: 验证后的输入参数
Returns:
str: JSON 格式的搜索结果
"""
try:
data = await _make_api_request("users/search", params={
"q": params.query,
"limit": params.limit,
"offset": params.offset
})
if params.response_format == ResponseFormat.MARKDOWN:
return format_markdown(data)
else:
return json.dumps(data, indent=2)
except Exception as e:
return _handle_api_error(e)高级特性:上下文注入
python
from mcp.server.fastmcp import FastMCP, Context
@mcp.tool()
async def advanced_search(query: str, ctx: Context) -> str:
"""带上下文访问的高级工具"""
# 报告进度
await ctx.report_progress(0.25, "开始搜索...")
# 记录日志
await ctx.log_info("处理查询", {"query": query})
# 执行搜索
results = await search_api(query)
await ctx.report_progress(0.75, "格式化结果...")
return format_results(results)最佳实践
工具命名约定
| 约定 | 示例 |
|---|---|
| 使用 snake_case | search_users, create_project |
| 包含服务前缀 | slack_send_message, github_create_issue |
| 动作导向 | get, list, search, create, update, delete |
| 避免冲突 | 不要用 send_message,用 slack_send_message |
响应格式
支持两种格式以适应不同场景:
JSON 格式 (response_format="json")
- 机器可读的结构化数据
- 包含所有可用字段和元数据
- 用于程序化处理
Markdown 格式 (response_format="markdown")
- 人类可读的格式化文本
- 使用标题、列表等格式
- 将时间戳转换为可读格式
- 显示带ID的显示名称
分页实现
python
# 分页响应结构
{
"total": 150, # 总数
"count": 20, # 当前页数量
"offset": 0, # 当前偏移
"items": [...], # 数据项
"has_more": True, # 是否有更多
"next_offset": 20 # 下一页偏移
}错误处理
提供清晰、可操作的错误消息:
python
def _handle_api_error(e: Exception) -> str:
"""一致的错误格式化"""
if isinstance(e, httpx.HTTPStatusError):
status = e.response.status_code
if status == 404:
return "错误: 资源未找到。请检查 ID 是否正确。"
elif status == 403:
return "错误: 权限被拒绝。您没有访问此资源的权限。"
elif status == 429:
return "错误: 速率限制。请稍后再试。"
return f"错误: API 请求失败,状态码 {status}"
return f"错误: 发生意外错误: {type(e).__name__}"工具注解
| 注解 | 类型 | 默认值 | 描述 |
|---|---|---|---|
readOnlyHint | boolean | false | 工具不修改环境 |
destructiveHint | boolean | true | 工具可能执行破坏性更新 |
idempotentHint | boolean | false | 重复调用无额外效果 |
openWorldHint | boolean | true | 工具与外部实体交互 |
传输选择指南
+------------------+------------------+
| stdio | Streamable HTTP |
+------------------+------------------+
| 本地集成 | 远程服务器 |
| 命令行工具 | Web 服务 |
| 单用户 | 多客户端 |
| 简单设置 | 中等复杂度 |
| 无网络配置 | 需要网络配置 |
+------------------+------------------+选择建议:
- stdio:本地开发环境、桌面应用集成、单用户场景
- Streamable HTTP:云服务部署、多客户端访问、Web 应用集成
评估系统
评估目的
评估的目的是测试 LLM 是否能有效使用你的 MCP 服务器来回答真实、复杂的问题。
创建评估问题的要求
- 独立性:每个问题不依赖其他问题
- 只读:只需要非破坏性操作
- 复杂性:需要多个工具调用
- 真实性:基于真实用例
- 可验证:单一、明确的答案
- 稳定性:答案不会随时间变化
评估文件格式
xml
<evaluation>
<qa_pair>
<question>找到2024年Q2创建的任务完成数最多的项目。
项目名称是什么?</question>
<answer>Website Redesign</answer>
</qa_pair>
<qa_pair>
<question>搜索2024年3月关闭的标记为"bug"的问题。
哪个用户关闭的问题最多?提供用户名。</question>
<answer>sarah_dev</answer>
</qa_pair>
</evaluation>运行评估
bash
# 安装依赖
pip install anthropic mcp
# 设置 API Key
export ANTHROPIC_API_KEY=your_api_key
# 运行 stdio 评估
python scripts/evaluation.py \
-t stdio \
-c python \
-a my_mcp_server.py \
-e API_KEY=abc123 \
evaluation.xml
# 运行 HTTP 评估
python scripts/evaluation.py \
-t http \
-u https://example.com/mcp \
-H "Authorization: Bearer token123" \
evaluation.xml评估报告内容
- 准确率(正确/总数)
- 平均任务时长
- 平均工具调用次数
- 每个任务的详细结果
质量检查表
策略设计
- [ ] 工具支持完整工作流,不仅是 API 端点包装
- [ ] 工具名称反映自然的任务细分
- [ ] 响应格式优化代理上下文效率
- [ ] 使用人类可读的标识符
- [ ] 错误消息引导代理正确使用
实现质量
- [ ] 实现最重要和有价值的工具
- [ ] 所有工具使用
registerTool/@mcp.tool注册 - [ ] 所有工具包含 title、description、inputSchema、annotations
- [ ] 使用 Zod/Pydantic 进行运行时输入验证
- [ ] 描述包含返回值示例和完整模式文档
- [ ] 错误消息清晰、可操作
代码质量
- [ ] 分页正确实现
- [ ] 大响应检查字符限制并截断
- [ ] 提供过滤选项
- [ ] 网络操作正确处理超时和连接错误
- [ ] 公共功能提取为可复用函数
测试
- [ ] 构建成功完成
- [ ] 服务器可以运行
- [ ] 示例工具调用按预期工作
实际应用示例
示例 1: GitHub MCP 服务器
typescript
// github-mcp-server
server.registerTool(
"github_search_issues",
{
title: "搜索 GitHub Issues",
description: "在仓库中搜索 issues...",
inputSchema: z.object({
repo: z.string().describe("仓库名 owner/repo"),
query: z.string().describe("搜索查询"),
state: z.enum(["open", "closed", "all"]).default("open"),
limit: z.number().int().min(1).max(100).default(20)
}),
annotations: {
readOnlyHint: true,
destructiveHint: false,
idempotentHint: true,
openWorldHint: true
}
},
async (params) => {
// 实现搜索逻辑
}
);示例 2: Slack MCP 服务器
python
# slack_mcp
@mcp.tool(
name="slack_send_message",
annotations={
"title": "发送 Slack 消息",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def slack_send_message(params: SendMessageInput) -> str:
"""向指定频道发送消息。
Args:
params: 包含 channel_id 和 text 的输入参数
Returns:
发送结果,包含消息 ID 和时间戳
"""
# 实现发送逻辑总结
MCP Builder Skill 提供了一套完整的方法论来构建高质量的 MCP 服务器:
- 四阶段流程:研究规划 -> 实现 -> 审查测试 -> 评估
- 双语言支持:TypeScript(推荐)和 Python
- 最佳实践:命名约定、响应格式、分页、错误处理
- 评估系统:通过复杂问题验证服务器质量
关键要记住的是:MCP 服务器的质量不在于功能的数量,而在于它能多好地帮助 AI 完成真实世界的任务。设计时要始终从用户(AI 和最终用户)的角度出发,创建直观、高效、可靠的工具。